diff --git a/qrexec/policy/parser.py b/qrexec/policy/parser.py index 114ef0c0..7709cd12 100644 --- a/qrexec/policy/parser.py +++ b/qrexec/policy/parser.py @@ -236,6 +236,14 @@ def match_strings(info: SystemInfo, self: str, other: str) -> bool: return self == other +def is_dom0(token) -> bool: + return token in ( + "@adminvm", + "dom0", + "uuid:00000000-0000-0000-0000-000000000000", + ) + + class VMToken(str, metaclass=VMTokenMeta): """A domain specification @@ -273,11 +281,6 @@ def __new__( "invalid empty {} token".format(cls.__name__.lower()), ) - # first, adjust some aliases - if token in {"dom0", "uuid:00000000-0000-0000-0000-000000000000"}: - # TODO: log a warning in Qubes 4.3 - token = "@adminvm" - # if user specified just qube name or UUID, use it directly if not (token.startswith("@") or token == "*"): return super().__new__(cls, token) @@ -351,9 +354,10 @@ def match( source: Optional["VMToken"] = None, ) -> bool: """Check if this token matches opposite token""" - # pylint: disable=unused-argument,too-many-return-statements - if self == "@adminvm": - return other == "@adminvm" + # pylint: disable=unused-argument + if is_dom0(self): + # see note in AdminVM.match + return is_dom0(other) return match_strings(system_info["domains"], self, other) def is_special_value(self) -> bool: @@ -486,6 +490,20 @@ def expand(self, *, system_info: FullSystemInfo) -> Iterable["AdminVM"]: def verify(self, *, system_info: FullSystemInfo) -> "AdminVM": return self + # Currently there's only one valid AdminVM nameley "@adminvm". It refers to + # (the local) dom0. We match dom0 and @adminvm commutatively for backward + # compatibility. If you are going to change that, you need to check all + # code that tests for dom0, uuid:0, @adminvm or AdminVM and adapt it, if + # necessary. + def match( + self, + other: Optional[str], + *, + system_info: FullSystemInfo, + source: Optional[VMToken] = None, + ) -> bool: + return is_dom0(other) + class AnyVM(Source, Target): # pylint: disable=missing-docstring,unused-argument @@ -498,13 +516,13 @@ def match( system_info: FullSystemInfo, source: Optional[VMToken] = None, ) -> bool: - return other != "@adminvm" + return not is_dom0(other) def expand(self, *, system_info: FullSystemInfo) -> Iterable[VMToken]: for name, domain in system_info["domains"].items(): if name.startswith("uuid:"): continue - if domain["type"] != "AdminVM": + if not is_dom0(name): yield IntendedTarget(name) if domain["template_for_dispvms"]: yield DispVMTemplate("@dispvm:" + name) @@ -767,16 +785,12 @@ async def execute(self) -> str: lines = [] # Adminvm/dom0 special case - if target in ( - "@adminvm", - "dom0", - "uuid:00000000-0000-0000-0000-000000000000", - ): + if is_dom0(target): lines.extend( [ f"user={self.user or 'DEFAULT'}", "result=allow", - "target=@adminvm", + "target=dom0", f"autostart={self.autostart}", f"requested_target={request.target}", ] @@ -1079,7 +1093,7 @@ def allow_no_autostart(target: str, system_info: FullSystemInfo) -> bool: """ Should we allow this target when autostart is disabled """ - if target == "@adminvm": + if is_dom0(target): return True if target.startswith("@dispvm"): return False @@ -1189,7 +1203,7 @@ def evaluate(self, request: Request) -> AllowResolution: ) target = target_ del target_ - # expand default AdminVM + # expand @adminvm but keep uuid:0 elif target == "@adminvm": target = "dom0" @@ -1262,7 +1276,10 @@ def evaluate(self, request: Request) -> AskResolution: targets_for_ask: Iterable[str] if self.target is not None: - targets_for_ask = [self.target] + if is_dom0(self.target): + targets_for_ask = ["dom0"] + else: + targets_for_ask = [self.target] else: targets_for_ask = list( self.rule.policy.collect_targets_for_ask(request) @@ -1293,8 +1310,8 @@ def evaluate(self, request: Request) -> AskResolution: default_target = default_target.get_dispvm_template( request.source, system_info=request.system_info ) - # expand default AdminVM - elif isinstance(default_target, AdminVM): + # expand @adminvm and uuid:0 to dom0 + elif is_dom0(default_target): default_target = "dom0" return request.ask_resolution_type( diff --git a/qrexec/tests/policy_graph.py b/qrexec/tests/policy_graph.py index 457db3a7..b23c5985 100644 --- a/qrexec/tests/policy_graph.py +++ b/qrexec/tests/policy_graph.py @@ -277,7 +277,7 @@ def test_simple_redirect(): ) content = output.read().decode() expected = """digraph g { - "work" -> "@adminvm" [label="test.Service" color=red]; + "work" -> "dom0" [label="test.Service" color=red]; } """ assert content == expected diff --git a/qrexec/tests/policy_parser.py b/qrexec/tests/policy_parser.py index 0d26277a..e8e4df57 100644 --- a/qrexec/tests/policy_parser.py +++ b/qrexec/tests/policy_parser.py @@ -301,7 +301,7 @@ def test_021_Target_expand(self): ) self.assertEqual( list(parser.Target("dom0").expand(system_info=self.system_info)), - ["@adminvm"], + ["dom0"], ) self.assertEqual( list( @@ -309,7 +309,7 @@ def test_021_Target_expand(self): "uuid:00000000-0000-0000-0000-000000000000" ).expand(system_info=self.system_info) ), - ["@adminvm"], + ["dom0"], ) self.assertEqual( list( @@ -354,12 +354,12 @@ def test_021_Target_expand(self): set(parser.Target("*").expand(system_info=self.system_info)) ), [ - "@adminvm", "@dispvm", "@dispvm:default-dvm", "@dispvm:test-vm3", "@dispvm:test-vm4", "default-dvm", + "dom0", "test-invalid-dvm", "test-no-dvm", "test-relayvm1", @@ -564,7 +564,14 @@ def test_100_match_single(self): ("@adminvm", "@adminvm", True), ("@adminvm", "dom0", True), ("dom0", "@adminvm", True), + ("@adminvm", "uuid:00000000-0000-0000-0000-000000000000", True), ("dom0", "dom0", True), + ("test-vm3", "dom0", False), + ("dom0", "test-vm3", False), + ("test-vm3", "@adminvm", False), + ("@adminvm", "test-vm3", False), + ("test-vm3", "uuid:00000000-0000-0000-0000-000000000000", False), + ("uuid:00000000-0000-0000-0000-000000000000", "test-vm3", False), ("@dispvm:default-dvm", "@dispvm:default-dvm", True), ("@anyvm", "@dispvm", True), ("*", "test-vm1", True), @@ -583,8 +590,8 @@ def test_100_match_single(self): ("@anyvm", "@adminvm", False), ("@tag:dom0-tag", "@adminvm", False), ("@type:AdminVM", "@adminvm", False), - ("@tag:dom0-tag", "dom0", False), - ("@type:AdminVM", "dom0", False), + ("@tag:dom0-tag", "dom0", True), + ("@type:AdminVM", "dom0", True), ("@tag:tag1", "dom0", False), ("@dispvm", "test-vm1", False), ("@dispvm", "default-dvm", False), @@ -1869,7 +1876,7 @@ def test_060_eval_to_dom0(self): self.assertIsInstance(resolution, parser.AllowResolution) self.assertEqual(resolution.rule, policy.rules[0]) self.assertEqual(resolution.target, "dom0") - self.assertEqual(resolution.request.target, "@adminvm") + self.assertEqual(resolution.request.target, "dom0") def test_061_eval_to_dom0_keyword(self): policy = parser.StringPolicy( @@ -1883,6 +1890,45 @@ def test_061_eval_to_dom0_keyword(self): self.assertEqual(resolution.target, "dom0") self.assertEqual(resolution.request.target, "@adminvm") + def test_062_eval_to_dom0_literal(self): + policy = parser.StringPolicy( + policy="""\ + * * test-vm3 dom0 allow""" + ) + resolution = policy.evaluate(self.gen_req("test-vm3", "dom0")) + + self.assertIsInstance(resolution, parser.AllowResolution) + self.assertEqual(resolution.rule, policy.rules[0]) + self.assertEqual(resolution.target, "dom0") + self.assertEqual(resolution.request.target, "dom0") + + def test_063_eval_to_dom0_literal_policy(self): + policy = parser.StringPolicy( + policy="""\ + * * test-vm3 dom0 allow""" + ) + resolution = policy.evaluate(self.gen_req("test-vm3", "@adminvm")) + + self.assertIsInstance(resolution, parser.AllowResolution) + self.assertEqual(resolution.rule, policy.rules[0]) + self.assertEqual(resolution.target, "dom0") + self.assertEqual(resolution.request.target, "@adminvm") + + def test_064_eval_to_dom0_deny(self): + names = ( + "dom0", + "@adminvm", + "uuid:00000000-0000-0000-0000-000000000000", + ) + for target in names: + policy = parser.StringPolicy(policy=f"* * test-vm3 test-vm2 allow") + with self.assertRaises(exc.AccessDenied): + policy.evaluate(self.gen_req("test-vm3", target)) + + policy = parser.StringPolicy(policy=f"* * test-vm3 {target} allow") + with self.assertRaises(exc.AccessDenied): + policy.evaluate(self.gen_req("test-vm3", "test-vm2")) + def test_070_eval_to_dom0_ask_default_target(self): policy = parser.StringPolicy( policy="""\ @@ -1893,7 +1939,7 @@ def test_070_eval_to_dom0_ask_default_target(self): self.assertIsInstance(resolution, parser.AskResolution) self.assertEqual(resolution.rule, policy.rules[0]) self.assertEqual(resolution.default_target, "dom0") - self.assertEqual(resolution.request.target, "@adminvm") + self.assertEqual(resolution.request.target, "dom0") self.assertEqual(resolution.targets_for_ask, ["dom0"]) def test_071_eval_to_dom0_ask_default_target(self): @@ -1906,7 +1952,7 @@ def test_071_eval_to_dom0_ask_default_target(self): self.assertIsInstance(resolution, parser.AskResolution) self.assertEqual(resolution.rule, policy.rules[0]) self.assertEqual(resolution.default_target, "dom0") - self.assertEqual(resolution.request.target, "@adminvm") + self.assertEqual(resolution.request.target, "dom0") self.assertEqual(resolution.targets_for_ask, ["dom0"]) def test_072_eval_to_dom0_ask_default_target(self): @@ -1919,7 +1965,7 @@ def test_072_eval_to_dom0_ask_default_target(self): self.assertIsInstance(resolution, parser.AskResolution) self.assertEqual(resolution.rule, policy.rules[0]) self.assertEqual(resolution.default_target, "dom0") - self.assertEqual(resolution.request.target, "@adminvm") + self.assertEqual(resolution.request.target, "dom0") self.assertEqual(resolution.targets_for_ask, ["dom0"]) def test_073_eval_to_dom0_ask_default_target(self): @@ -1932,9 +1978,30 @@ def test_073_eval_to_dom0_ask_default_target(self): self.assertIsInstance(resolution, parser.AskResolution) self.assertEqual(resolution.rule, policy.rules[0]) self.assertEqual(resolution.default_target, "dom0") - self.assertEqual(resolution.request.target, "@adminvm") + self.assertEqual(resolution.request.target, "dom0") self.assertEqual(resolution.targets_for_ask, ["dom0"]) + def test_074_eval_to_default_dom0(self): + names = ( + "dom0", + "@adminvm", + "uuid:00000000-0000-0000-0000-000000000000", + ) + for target in names: + for default_target in names: + policy = parser.StringPolicy( + policy=f"* * test-vm3 @default ask target={target} default_target={default_target}" + ) + resolution = policy.evaluate( + self.gen_req("test-vm3", "@default") + ) + + self.assertIsInstance(resolution, parser.AskResolution) + self.assertEqual(resolution.rule, policy.rules[0]) + self.assertEqual(resolution.default_target, "dom0") + self.assertEqual(resolution.request.target, "@default") + self.assertEqual(resolution.targets_for_ask, ["dom0"]) + def test_080_eval_override_target(self): policy = parser.StringPolicy( policy="""\ @@ -2034,7 +2101,9 @@ def test_088_eval_override_target_uuid_dom0(self): self.assertIsInstance(resolution, parser.AllowResolution) self.assertEqual(resolution.rule, policy.rules[0]) - self.assertEqual(resolution.target, "dom0") + self.assertEqual( + resolution.target, "uuid:00000000-0000-0000-0000-000000000000" + ) self.assertEqual(resolution.request.target, "test-vm1") def test_089_eval_override_target_dispvm_uuid(self): @@ -2231,9 +2300,9 @@ async def _test_121_execute_dom0(self): """\ user=DEFAULT result=allow -target=@adminvm +target=dom0 autostart=True -requested_target=@adminvm\ +requested_target=dom0\ """, ) @@ -2258,7 +2327,7 @@ async def _test_121_execute_dom0_keyword(self): """\ user=DEFAULT result=allow -target=@adminvm +target=dom0 autostart=True requested_target=@adminvm\ """, diff --git a/qrexec/tests/qrexec_legacy_convert.py b/qrexec/tests/qrexec_legacy_convert.py index 667b828e..f4838de9 100644 --- a/qrexec/tests/qrexec_legacy_convert.py +++ b/qrexec/tests/qrexec_legacy_convert.py @@ -147,7 +147,7 @@ def test_simplest_convert(mock_policy_dirs: Tuple[pathlib.Path, pathlib.Path]): assert ( result.read_text() == qrexec_legacy_convert.TOOL_DISCLAIMER - + """qubes.Filecopy\t*\t@adminvm\t@anyvm\task + + """qubes.Filecopy\t*\tdom0\t@anyvm\task qubes.Filecopy\t*\twork\t@anyvm\task qubes.Filecopy\t*\tpersonal\t@anyvm\tdeny """ @@ -273,9 +273,9 @@ def test_input_rules_simple( assert set(input_result.read_text().split("\n")) == set( ( qrexec_legacy_convert.TOOL_DISCLAIMER - + """qubes.InputKeyboard\t*\tsys-usb\t@adminvm\task -qubes.InputMouse\t*\tsys-usb\t@adminvm\tallow -qubes.InputTablet\t*\tsys-usb\t@adminvm\tdeny + + """qubes.InputKeyboard\t*\tsys-usb\tdom0\task +qubes.InputMouse\t*\tsys-usb\tdom0\tallow +qubes.InputTablet\t*\tsys-usb\tdom0\tdeny """ ).split("\n") ) @@ -293,7 +293,7 @@ def test_input_multiple_rules( # the deny @anyvm should be move to 30-user file (old_policy_dir / "qubes.InputKeyboard").write_text( """ -sys-usb dom0 ask default_target=@adminvm +sys-usb dom0 ask default_target=dom0 sys-usb @anyvm deny """ ) @@ -321,9 +321,9 @@ def test_input_multiple_rules( assert set(input_result.read_text().split("\n")) == set( ( qrexec_legacy_convert.TOOL_DISCLAIMER - + """qubes.InputKeyboard\t*\tsys-usb\t@adminvm\task default_target=@adminvm -qubes.InputMouse\t*\tsys-usb\t@adminvm\tallow -qubes.InputTablet\t*\tsys-usb\t@adminvm\tdeny + + """qubes.InputKeyboard\t*\tsys-usb\tdom0\task default_target=dom0 +qubes.InputMouse\t*\tsys-usb\tdom0\tallow +qubes.InputTablet\t*\tsys-usb\tdom0\tdeny """ ).split("\n") ) @@ -355,9 +355,9 @@ def test_input_multiple_sys_usbs( # the sys-usb-2 @anyvm rule should go to 30-user (old_policy_dir / "qubes.InputMouse").write_text( """ -sys-usb dom0 allow +sys-usb @adminvm allow sys-usb-2 @anyvm deny -sys-usb-2 dom0 allow +sys-usb-2 @adminvm allow """ ) # first rule goes to 50-config @@ -379,11 +379,11 @@ def test_input_multiple_sys_usbs( assert set(input_result.read_text().split("\n")) == set( ( qrexec_legacy_convert.TOOL_DISCLAIMER - + """qubes.InputKeyboard\t*\tsys-usb\t@adminvm\task -qubes.InputKeyboard\t*\tsys-usb-2\t@adminvm\tdeny + + """qubes.InputKeyboard\t*\tsys-usb\tdom0\task +qubes.InputKeyboard\t*\tsys-usb-2\tdom0\tdeny qubes.InputMouse\t*\tsys-usb\t@adminvm\tallow qubes.InputMouse\t*\tsys-usb-2\t@adminvm\tallow -qubes.InputTablet\t*\tsys-usb\t@adminvm\task +qubes.InputTablet\t*\tsys-usb\tdom0\task """ ).split("\n") )