admin.py 5.4 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144
  1. # -*- encoding: utf8 -*-
  2. #
  3. # The Qubes OS Project, http://www.qubes-os.org
  4. #
  5. # Copyright (C) 2017 Wojtek Porczyk <woju@invisiblethingslab.com>
  6. #
  7. # This library is free software; you can redistribute it and/or
  8. # modify it under the terms of the GNU Lesser General Public
  9. # License as published by the Free Software Foundation; either
  10. # version 2.1 of the License, or (at your option) any later version.
  11. #
  12. # This library is distributed in the hope that it will be useful,
  13. # but WITHOUT ANY WARRANTY; without even the implied warranty of
  14. # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
  15. # Lesser General Public License for more details.
  16. #
  17. # You should have received a copy of the GNU Lesser General Public
  18. # License along with this library; if not, see <https://www.gnu.org/licenses/>.
  19. import qubes.api
  20. import qubes.api.internal
  21. import qubes.ext
  22. import qubes.vm.adminvm
  23. from qrexec.policy import utils, parser
  24. class JustEvaluateAskResolution(parser.AskResolution):
  25. async def execute(self, caller_ident):
  26. pass
  27. class JustEvaluateAllowResolution(parser.AllowResolution):
  28. async def execute(self, caller_ident):
  29. pass
  30. class AdminExtension(qubes.ext.Extension):
  31. def __init__(self):
  32. super(AdminExtension, self).__init__()
  33. # during tests, __init__() of the extension can be called multiple
  34. # times, because there are multiple Qubes() object instances
  35. if not hasattr(self, 'policy_cache'):
  36. self.policy_cache = utils.PolicyCache(lazy_load=True)
  37. self.policy_cache.initialize_watcher()
  38. # pylint: disable=too-few-public-methods
  39. @qubes.ext.handler(
  40. 'admin-permission:admin.vm.tag.Set',
  41. 'admin-permission:admin.vm.tag.Remove')
  42. def on_tag_set_or_remove(self, vm, event, arg, **kwargs):
  43. '''Forbid changing specific tags'''
  44. # pylint: disable=no-self-use,unused-argument
  45. if arg.startswith('created-by-') and \
  46. not isinstance(vm, qubes.vm.adminvm.AdminVM):
  47. raise qubes.api.PermissionDenied(
  48. 'changing this tag is prohibited by {}.{}'.format(
  49. __name__, type(self).__name__))
  50. # TODO create that tag here (need to figure out how to pass mgmtvm name)
  51. @qubes.ext.handler('admin-permission:admin.vm.List')
  52. def admin_vm_list(self, vm, event, arg, **kwargs):
  53. '''When called with target 'dom0' (aka "get full list"), exclude domains
  54. that the caller don't have permission to list
  55. '''
  56. # pylint: disable=unused-argument
  57. if vm.klass == 'AdminVM':
  58. # dom0 can always list everything
  59. return None
  60. policy = self.policy_cache.get_policy()
  61. system_info = qubes.api.internal.get_system_info(vm.app)
  62. def filter_vms(dest_vm):
  63. request = parser.Request(
  64. 'admin.vm.List',
  65. '+' + arg,
  66. vm.name,
  67. dest_vm.name,
  68. system_info=system_info,
  69. ask_resolution_type=JustEvaluateAskResolution,
  70. allow_resolution_type=JustEvaluateAllowResolution)
  71. try:
  72. resolution = policy.evaluate(request)
  73. # do not consider 'ask' as allow here,
  74. # this needs to be not interactive
  75. return isinstance(resolution, parser.AllowResolution)
  76. except parser.AccessDenied:
  77. return False
  78. return (filter_vms,)
  79. @qubes.ext.handler('admin-permission:admin.Events')
  80. def admin_events(self, vm, event, arg, **kwargs):
  81. '''When called with target 'dom0' (aka "get all events"),
  82. exclude domains that the caller don't have permission to receive
  83. events about
  84. '''
  85. # pylint: disable=unused-argument
  86. if vm.klass == 'AdminVM':
  87. # dom0 can always list everything
  88. return None
  89. def filter_events(event):
  90. subject, event, kwargs = event
  91. try:
  92. dest = subject.name
  93. except AttributeError:
  94. # domain-add and similar events fired on the Qubes() object
  95. if 'vm' in kwargs:
  96. dest = kwargs['vm'].name
  97. else:
  98. dest = '@adminvm'
  99. policy = self.policy_cache.get_policy()
  100. # TODO: cache system_info (based on last qubes.xml write time?)
  101. system_info = qubes.api.internal.get_system_info(vm.app)
  102. request = parser.Request(
  103. 'admin.Events',
  104. '+' + event.replace(':', '_'),
  105. vm.name,
  106. dest,
  107. system_info=system_info,
  108. ask_resolution_type=JustEvaluateAskResolution,
  109. allow_resolution_type=JustEvaluateAllowResolution)
  110. try:
  111. resolution = policy.evaluate(request)
  112. # do not consider 'ask' as allow here,
  113. # this needs to be not interactive
  114. return isinstance(resolution, parser.AllowResolution)
  115. except parser.AccessDenied:
  116. return False
  117. return (filter_events,)
  118. @qubes.ext.handler('qubes-close', system=True)
  119. def on_qubes_close(self, app, event, **kwargs):
  120. """Unregister policy file watches on app.close()."""
  121. # pylint: disable=unused-argument
  122. if hasattr(self, 'policy_cache'):
  123. self.policy_cache.cleanup()
  124. del self.policy_cache