Don't translate dom0 to @adminvm

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(

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

qrexec/tests/policy_parser.py

@@ -301,15 +301,15 @@ def test_021_Target_expand(self):
         )
         self.assertEqual(
             list(parser.Target("dom0").expand(system_info=self.system_info)),
-            ["@adminvm"],
+            ["dom0"],
         )
         self.assertEqual(
             list(
                 parser.Target(
                     "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\
 """,