Support using UUIDs instead of VM names

qubes/api/__init__.py

@@ -24,10 +24,13 @@
 import functools
 import io
 import os
+import re
 import shutil
 import socket
 import struct
 import traceback
+from typing import Union, Any
+import uuid
 
 import qubes.exc
 from qubes.exc import ProtocolError, PermissionDenied
@@ -88,6 +91,32 @@
         iterable = filter(selector, iterable)
     return iterable
 
+# This regex allows incorrect-length UUIDs,
+# but there is an explicit length check to catch that.
+_uuid_regex = re.compile(rb"\Auuid:[0-9a-f]{8}(?:-[0-9a-f]{4}){3}-[0-9a-f]*")
+def decode_vm(
+    untrusted_input: bytes, domains: qubes.app.VMCollection
+) -> qubes.vm.qubesvm.QubesVM:
+    lookup: Union[uuid.UUID, str]
+    vm = untrusted_input.decode("ascii", "strict")
+    if untrusted_input.startswith(b"uuid:"):
+        if (len(untrusted_input) != 41 or
+            not _uuid_regex.match(untrusted_input)):
+            raise qubes.exc.QubesVMInvalidUUIDError(vm[5:])
+        lookup = uuid.UUID(vm[5:])
+    else:
+        # throws if name is invalid
+        qubes.vm.validate_name(None, None, vm)
+        lookup = vm
+    try:
+        return domains[lookup]
+    except KeyError:
+        # normally this should filtered out by qrexec policy, but there are
+        # two cases it might not be:
+        # 1. The call comes from dom0, which bypasses qrexec policy
+        # 2. Domain was removed between checking the policy and here
+        # we inform the client accordingly
+        raise qubes.exc.QubesVMNotFoundError(vm)
 
 class AbstractQubesAPI:
     '''Common code for Qubes Management Protocol handling
@@ -108,25 +137,23 @@
     #: the preferred socket location (to be overridden in child's class)
     SOCKNAME = None
 
-    def __init__(self, app, src, method_name, dest, arg, send_event=None):
+    app: qubes.Qubes
+    src: qubes.vm.qubesvm.QubesVM
+    def __init__(self,
+                 app: qubes.Qubes,
+                 src: bytes,
+                 method_name: bytes,
+                 dest: bytes,
+                 arg: qubes.Qubes,
+                 send_event: Any = None) -> None:
         #: :py:class:`qubes.Qubes` object
         self.app = app
 
-        try:
-            vm = src.decode('ascii')
-            #: source qube
-            self.src = self.app.domains[vm]
-
-            vm = dest.decode('ascii')
-            #: destination qube
-            self.dest = self.app.domains[vm]
-        except KeyError:
-            # normally this should be filtered out by qrexec policy, but there
-            # are two cases it might not be:
-            # 1. The call comes from dom0, which bypasses qrexec policy
-            # 2. Domain was removed between checking the policy and here
-            # we inform the client accordingly
-            raise qubes.exc.QubesVMNotFoundError(vm)
+        #: source qube
+        self.src = decode_vm(src, app.domains)
+
+        #: destination qube
+        self.dest = decode_vm(dest, app.domains)
 
         #: argument
         self.arg = arg.decode('ascii')

qubes/api/admin.py

@@ -121,10 +121,11 @@
         else:
             domains = self.fire_event_for_filter([self.dest])
 
-        return ''.join('{} class={} state={}\n'.format(
+        return ''.join('{} class={} state={} uuid={}\n'.format(
                 vm.name,
                 vm.__class__.__name__,
-                vm.get_power_state())
+                vm.get_power_state(),
+                vm.uuid)
             for vm in sorted(domains))
 
     @qubes.api.method('admin.vm.property.List', no_payload=True,
@@ -1138,10 +1139,14 @@
             raise
         self.app.save()
 
-    @qubes.api.method('admin.vm.CreateDisposable', no_payload=True,
+    @qubes.api.method('admin.vm.CreateDisposable',
         scope='global', write=True)
-    async def create_disposable(self):
+    async def create_disposable(self, untrusted_payload):
         self.enforce(not self.arg)
+        if untrusted_payload not in (b"", b"uuid"):
+            raise qubes.exc.QubesValueError(
+                    "Invalid payload for admin.vm.CreateDisposable: "
+                    "expected the empty string or 'uuid'")
 
         if self.dest.name == 'dom0':
             dispvm_template = self.src.default_dispvm
@@ -1155,7 +1160,9 @@
         dispvm.tags.add('created-by-' + str(self.src))
         dispvm.tags.add('disp-created-by-' + str(self.src))
 
-        return dispvm.name
+        return (("uuid:" + str(dispvm.uuid))
+                if untrusted_payload
+                else dispvm.name)
 
     @qubes.api.method(
         'admin.vm.Remove', no_payload=True, scope='global', write=True)

qubes/api/internal.py

@@ -43,6 +43,7 @@ def get_system_info(app):
             'guivm': (domain.guivm.name if getattr(domain, 'guivm', None)
                       else None),
             'power_state': domain.get_power_state(),
+            'uuid': str(domain.uuid),
         } for domain in app.domains
     }}
     return system_info

qubes/app.py

@@ -516,6 +516,7 @@ def __getitem__(self, key):
 
         if isinstance(key, uuid.UUID):
             for vm in self:
+                assert isinstance(vm.uuid, uuid.UUID)
                 if vm.uuid == key:
                     return vm
             raise KeyError(key)
@@ -540,7 +541,7 @@ def __delitem__(self, key):
         self.app.fire_event('domain-delete', vm=vm)
 
     def __contains__(self, key):
-        return any((key in (vm, vm.qid, vm.name))
+        return any((key in (vm, vm.qid, vm.name, vm.uuid))
                    for vm in self)
 
     def __len__(self):

qubes/exc.py

@@ -37,6 +37,14 @@ def __str__(self):
         # KeyError overrides __str__ method
         return QubesException.__str__(self)
 
+class QubesVMInvalidUUIDError(QubesException):
+    """Domain UUID is invalid"""
+    # pylint: disable = super-init-not-called
+    def __init__(self, uuid: str) -> None:
+        # QubesVMNotFoundError overrides __init__ method
+        # pylint: disable = non-parent-init-called
+        QubesException.__init__(self, f"VM UUID is not valid: {uuid!r}")
+        self.vmname = uuid
 
 class QubesVMError(QubesException):
     '''Some problem with domain state.'''

qubes/tests/api_admin.py

@@ -23,9 +23,11 @@
 import asyncio
 import operator
 import os
+import re
 import shutil
 import tempfile
 import unittest.mock
+import uuid
 
 import libvirt
 import copy
@@ -46,6 +48,8 @@
     'pool', 'vid', 'size', 'usage', 'rw', 'source', 'path',
     'save_on_stop', 'snap_on_start', 'revisions_to_keep', 'ephemeral']
 
+_uuid_regex = re.compile(
+        r"\Auuid:[0-9a-f]{8}(?:-[0-9a-f]{4}){3}-[0-9a-f]{12}\Z")
 
 class AdminAPITestCase(qubes.tests.QubesTestCase):
     def setUp(self):
@@ -78,6 +82,7 @@ def setUp(self):
         app.save = unittest.mock.Mock()
         self.vm = app.add_new_vm('AppVM', label='red', name='test-vm1',
             template='test-template')
+        self.uuid = b"uuid:" + str(self.vm.uuid).encode("ascii", "strict")
         self.app = app
         libvirt_attrs = {
             'libvirt_conn.lookupByUUID.return_value.isActive.return_value':
@@ -130,14 +135,33 @@ class TC_00_VMs(AdminAPITestCase):
     def test_000_vm_list(self):
         value = self.call_mgmt_func(b'admin.vm.List', b'dom0')
         self.assertEqual(value,
-            'dom0 class=AdminVM state=Running\n'
-            'test-template class=TemplateVM state=Halted\n'
-            'test-vm1 class=AppVM state=Halted\n')
+            'dom0 class=AdminVM state=Running '
+            'uuid=00000000-0000-0000-0000-000000000000\n'
+            'test-template class=TemplateVM state=Halted '
+            f'uuid={self.template.uuid}\n'
+            f'test-vm1 class=AppVM state=Halted uuid={self.vm.uuid}\n')
 
     def test_001_vm_list_single(self):
         value = self.call_mgmt_func(b'admin.vm.List', b'test-vm1')
         self.assertEqual(value,
-            'test-vm1 class=AppVM state=Halted\n')
+            f"test-vm1 class=AppVM state=Halted uuid={self.vm.uuid}\n")
+
+    def test_001_vm_list_single_uuid(self):
+        value = self.call_mgmt_func(b'admin.vm.List', self.uuid)
+        self.assertEqual(value,
+            f"test-vm1 class=AppVM state=Halted uuid={self.vm.uuid}\n")
+
+    def test_001_vm_list_single_invalid_name(self):
+        with self.assertRaisesRegex(qubes.exc.QubesValueError,
+                                    r"\AVM name contains illegal characters\Z"):
+            self.call_mgmt_func(b'admin.vm.CreateDisposable', b'.test-vm1')
+        self.assertFalse(self.app.save.called)
+
+    def test_001_vm_list_single_invalid_uuid(self):
+        with self.assertRaisesRegex(qubes.exc.QubesVMInvalidUUIDError,
+                                    r"\AVM UUID is not valid: ''\Z"):
+            self.call_mgmt_func(b'admin.vm.CreateDisposable', b"uuid:")
+        self.assertFalse(self.app.save.called)
 
     def test_002_vm_list_filter(self):
         with tempfile.TemporaryDirectory() as tmpdir:
@@ -154,8 +178,9 @@ def test_002_vm_list_filter(self):
                 value = loop.run_until_complete(
                     mgmt_obj.execute(untrusted_payload=b''))
                 self.assertEqual(value,
-                    'dom0 class=AdminVM state=Running\n'
-                    'test-vm1 class=AppVM state=Halted\n')
+                    'dom0 class=AdminVM state=Running '
+                    'uuid=00000000-0000-0000-0000-000000000000\n'
+                    f"test-vm1 class=AppVM state=Halted uuid={self.vm.uuid}\n")
 
     def test_010_vm_property_list(self):
         # this test is kind of stupid, but at least check if appropriate
@@ -2816,6 +2841,26 @@ def test_640_vm_create_disposable(self, mock_storage):
         mock_storage.assert_called_once_with()
         self.assertTrue(self.app.save.called)
 
+    @unittest.mock.patch('qubes.storage.Storage.create')
+    def test_640_vm_create_disposable_uuid(self, mock_storage):
+        mock_storage.side_effect = self.dummy_coro
+        self.vm.template_for_dispvms = True
+        retval = self.call_mgmt_func(b'admin.vm.CreateDisposable',
+                b'test-vm1', payload=b'uuid')
+        self.assertRegex(retval, _uuid_regex)
+        found = False
+        for i in self.app.domains:
+            print(i.uuid)
+            if i.uuid == uuid.UUID(retval):
+                found = True
+            self.assertIsInstance(i.uuid, uuid.UUID)
+        self.assertTrue(found)
+        self.assertIn(uuid.UUID(retval), self.app.domains)
+        dispvm = self.app.domains[uuid.UUID(retval)]
+        self.assertEqual(dispvm.template, self.vm)
+        mock_storage.assert_called_once_with()
+        self.assertTrue(self.app.save.called)
+
     @unittest.mock.patch('qubes.storage.Storage.create')
     def test_641_vm_create_disposable_default(self, mock_storage):
         mock_storage.side_effect = self.dummy_coro

qubes/tests/api_internal.py

@@ -23,13 +23,16 @@
 import qubes.vm.adminvm
 from unittest import mock
 import json
+import uuid
 
 def mock_coro(f):
     async def coro_f(*args, **kwargs):
         return f(*args, **kwargs)
 
     return coro_f
 
+TEST_UUID = uuid.UUID("50c7dad4-5f1e-4586-9f6a-bf10a86ba6f0")
+
 class TC_00_API_Misc(qubes.tests.QubesTestCase):
     def setUp(self):
         super().setUp()
@@ -150,6 +153,7 @@ def test_010_get_system_info(self):
         self.dom0.template_for_dispvms = False
         self.dom0.label.icon = 'icon-dom0'
         self.dom0.get_power_state.return_value = 'Running'
+        self.dom0.uuid = uuid.UUID("00000000-0000-0000-0000-000000000000")
         del self.dom0.guivm
 
         vm = mock.NonCallableMock(spec=qubes.vm.qubesvm.QubesVM)
@@ -160,6 +164,7 @@ def test_010_get_system_info(self):
         vm.label.icon = 'icon-vm'
         vm.guivm = vm
         vm.get_power_state.return_value = 'Halted'
+        vm.uuid = TEST_UUID
         self.domains['vm'] = vm
 
         ret = json.loads(self.call_mgmt_func(b'internal.GetSystemInfo'))
@@ -173,6 +178,7 @@ def test_010_get_system_info(self):
                     'icon': 'icon-dom0',
                     'guivm': None,
                     'power_state': 'Running',
+                    'uuid': "00000000-0000-0000-0000-000000000000",
                 },
                 'vm': {
                     'tags': ['tag3', 'tag4'],
@@ -182,6 +188,7 @@ def test_010_get_system_info(self):
                     'icon': 'icon-vm',
                     'guivm': 'vm',
                     'power_state': 'Halted',
+                    "uuid": str(TEST_UUID),
                 }
             }
         })

qubes/vm/adminvm.py

@@ -25,6 +25,7 @@
 import grp
 import subprocess
 import libvirt
+import uuid
 
 import qubes
 import qubes.exc
@@ -45,7 +46,7 @@ class AdminVM(BaseVM):
         default=0, type=int, setter=qubes.property.forbidden)
 
     uuid = qubes.property('uuid',
-        default='00000000-0000-0000-0000-000000000000',
+        default=uuid.UUID('00000000-0000-0000-0000-000000000000'),
         setter=qubes.property.forbidden)
 
     default_dispvm = qubes.VMProperty('default_dispvm',

qubes/vm/qubesvm.py

@@ -61,6 +61,7 @@
 MEM_OVERHEAD_BASE = (3 + 1) * 1024 * 1024
 MEM_OVERHEAD_PER_VCPU = 3 * 1024 * 1024 / 2
 
+_vm_uuid_re = re.compile(rb"\A/vm/[0-9a-f]{8}(?:-[0-9a-f]{4}){4}[0-9a-f]{8}\Z")
 
 def _setter_kernel(self, prop, value):
     """ Helper for setting the domain kernel and running sanity checks on it.
@@ -801,6 +802,17 @@
                 e.get_error_code()))
             raise
 
+    @qubes.stateless_property
+    def stubdom_uuid(self):
+        stubdom_xid = self.stubdom_xid
+        if stubdom_xid == -1:
+            return ""
+        stubdom_uuid = self.app.vmm.xs.read(
+            '', '/local/domain/{}/vm'.format(
+                stubdom_xid))
+        assert _vm_uuid_re.match(stubdom_uuid), "Invalid UUID in XenStore"
+        return stubdom_uuid[4:].decode("ascii", "strict")
+
     @qubes.stateless_property
     def stubdom_xid(self):
         if not self.is_running():
@@ -1807,9 +1819,11 @@
 
         self.log.debug('Starting the qrexec daemon')
         if stubdom:
-            qrexec_args = [str(self.stubdom_xid), self.name + '-dm', 'root']
+            qrexec_args = ["-u", self.stubdom_uuid, "--",
+                           str(self.stubdom_xid), self.name + '-dm', 'root']
         else:
-            qrexec_args = [str(self.xid), self.name, self.default_user]
+            qrexec_args = ["-u", str(self.uuid), "--",
+                           str(self.xid), self.name, self.default_user]
 
         if not self.debug:
             qrexec_args.insert(0, "-q")