device identity

qubesguidaemon/mic.py

@@ -19,38 +19,44 @@
 # with this program; if not, see <http://www.gnu.org/licenses/>.
 
 """Microphone control extension"""
-
+import asyncio
 import subprocess
+import sys
 
-import qubes.device_protocol
 import qubes.ext
 import qubes.vm.adminvm
+from qubes.device_protocol import Port, DeviceInterface, DeviceInfo
 
 
-class MicDevice(qubes.device_protocol.DeviceInfo):
+class MicDevice(DeviceInfo):
     """Microphone device info class"""
 
+    # pylint: disable=too-few-public-methods)
+
     def __init__(self, backend_domain, product, manufacturer):
+        port = Port(
+            backend_domain=backend_domain, port_id="mic", devclass="mic"
+        )
         super().__init__(
-            backend_domain=backend_domain,
-            ident="mic",
-            devclass="mic",
+            port,
             product=product,
             manufacturer=manufacturer,
         )
-        self._interfaces = [
-            qubes.device_protocol.DeviceInterface("******", devclass="mic")
-        ]
+        self._interfaces = [DeviceInterface("000000", devclass="mic")]
+
+    @property
+    def device_id(self) -> str:
+        """
+        Get identification of a device not related to port.
+        """
+        return "dom0:mic::m000000"
 
 
 class MicDeviceExtension(qubes.ext.Extension):
     """
     Extension to control microphone access
     """
 
-    def __init__(self):
-        super(MicDeviceExtension, self).__init__()
-
     @staticmethod
     def get_device(app):
         return MicDevice(
@@ -67,25 +73,26 @@ def on_device_list_mic(self, vm, event):
         return self.on_device_get_mic(vm, event, "mic")
 
     @qubes.ext.handler("device-get:mic")
-    def on_device_get_mic(self, vm, event, ident):
+    def on_device_get_mic(self, vm, event, port_id):
         """Get microphone device
 
         Currently, this assumes audio being handled in dom0. When adding support
         for GUI domain, this needs to be changed
         """
-        # pylint: disable=unused-argument,no-self-use
+        # pylint: disable=unused-argument
 
         if not isinstance(vm, qubes.vm.adminvm.AdminVM):
             return
 
-        if ident != "mic":
+        if port_id != "mic":
             return
 
         yield self.get_device(vm.app)
 
     @qubes.ext.handler("device-list-attached:mic")
     def on_device_list_attached_mic(self, vm, event, persistent=None):
         """List attached microphone to the VM"""
+        # pylint: disable=unused-argument
 
         if persistent is True:
             return
@@ -96,7 +103,7 @@ def on_device_list_attached_mic(self, vm, event, persistent=None):
             return
 
         untrusted_audio_input = audiovm.untrusted_qdb.read(
-            "/audio-input-config/{}".format(vm.name)
+            f"/audio-input-config/{vm.name}"
         )
         if untrusted_audio_input == b"1":
             # (device, options)
@@ -105,129 +112,181 @@ def on_device_list_attached_mic(self, vm, event, persistent=None):
     @qubes.ext.handler("device-pre-attach:mic")
     async def on_device_pre_attach_mic(self, vm, event, device, options):
         """Attach microphone to the VM"""
+        # pylint: disable=unused-argument
 
         # there is only one microphone
         assert device == self.get_device(vm.app)
         if options:
             raise qubes.exc.QubesException(
-                "mic device does not support options"
+                "Microphone assignment does not support user options"
             )
 
         audiovm = getattr(vm, "audiovm", None)
 
         if audiovm is None:
-            raise qubes.exc.QubesException(
-                "VM {} has no AudioVM set".format(vm)
-            )
+            raise qubes.exc.QubesException(f"VM {vm} has no AudioVM set")
 
         if not audiovm.is_running():
             raise qubes.exc.QubesVMNotRunningError(
-                audiovm, "Audio VM {} isn't running".format(audiovm)
+                audiovm, f"Audio VM {audiovm} isn't running"
             )
 
         if audiovm.features.check_with_netvm(
             "supported-rpc.qubes.AudioInputEnable", False
         ):
             try:
                 await audiovm.run_service_for_stdio(
-                    "qubes.AudioInputEnable+{}".format(vm.name)
+                    f"qubes.AudioInputEnable+{vm.name}"
                 )
             except subprocess.CalledProcessError:
+                # pylint: disable=raise-missing-from
                 raise qubes.exc.QubesVMError(
                     vm,
-                    "Failed to attach audio input from {!s} to {!s}: "
-                    "pulseaudio agent not running".format(audiovm, vm),
+                    f"Failed to attach audio input from {audiovm} to {vm}: "
+                    "pulseaudio agent not running",
                 )
         else:
-            audiovm.untrusted_qdb.write(
-                "/audio-input-config/{}".format(vm.name), "1"
-            )
+            audiovm.untrusted_qdb.write(f"/audio-input-config/{vm.name}", "1")
 
     # pylint: disable=unused-argument
     @qubes.ext.handler("device-pre-detach:mic")
-    async def on_device_pre_detach_mic(self, vm, event, device):
+    async def on_device_pre_detach_mic(self, vm, event, port):
         """Detach microphone from the VM"""
 
         # there is only one microphone
-        assert device == self.get_device(vm.app)
+        assert port == self.get_device(vm.app).port
 
         audiovm = getattr(vm, "audiovm", None)
 
         if audiovm is None:
-            raise qubes.exc.QubesException(
-                "VM {} has no AudioVM set".format(vm)
-            )
+            raise qubes.exc.QubesException(f"VM {vm} has no AudioVM set")
 
         if not audiovm.is_running():
             raise qubes.exc.QubesVMNotRunningError(
-                audiovm, "Audio VM {} isn't running".format(audiovm)
+                audiovm, f"Audio VM {audiovm} isn't running"
             )
 
         if audiovm.features.check_with_netvm(
             "supported-rpc.qubes.AudioInputDisable", False
         ):
             try:
                 await audiovm.run_service_for_stdio(
-                    "qubes.AudioInputDisable+{}".format(vm.name)
+                    f"qubes.AudioInputDisable+{vm.name}"
                 )
             except subprocess.CalledProcessError:
+                # pylint: disable=raise-missing-from
                 raise qubes.exc.QubesVMError(
                     vm,
-                    "Failed to detach audio input from {!s} to {!s}: "
-                    "pulseaudio agent not running".format(audiovm, vm),
+                    f"Failed to detach audio input from {audiovm} to {vm}: "
+                    "pulseaudio agent not running",
                 )
         else:
-            audiovm.untrusted_qdb.write(
-                "/audio-input-config/{}".format(vm.name), "0"
+            audiovm.untrusted_qdb.write(f"/audio-input-config/{vm.name}", "0")
+
+    @qubes.ext.handler("device-pre-assign:mic")
+    async def on_device_assign_mic(self, vm, event, device, options):
+        # pylint: disable=unused-argument
+
+        if options:
+            raise qubes.exc.QubesException(
+                "Microphone assignment does not support user options"
             )
 
     @qubes.ext.handler("property-set:audiovm")
     def on_property_set(self, subject, event, name, newvalue, oldvalue=None):
+        # pylint: disable=too-many-arguments
         if not subject.is_running() or not newvalue:
             return
         if not newvalue.is_running():
             subject.log.warning(
-                "Cannot attach mic to {!s}: "
-                "AudioVM '{!s}' is powered off.".format(subject, newvalue)
+                f"Cannot attach mic to {subject}: "
+                f"AudioVM '{newvalue}' is powered off."
             )
         if newvalue == oldvalue:
             return
         if oldvalue and oldvalue.is_running():
             mic_allowed = oldvalue.untrusted_qdb.read(
-                "/audio-input-config/{}".format(subject.name)
+                f"/audio-input-config/{subject.name}"
             )
             if mic_allowed is None:
                 return
             try:
                 mic_allowed_value = mic_allowed.decode("ascii")
             except UnicodeError:
+                # pylint: disable=raise-missing-from
                 raise qubes.exc.QubesVMError(
                     subject,
-                    "Cannot decode ASCII value for '/audio-input-config/{!s}'".format(
-                        subject.name
-                    ),
+                    f"Cannot decode ASCII value for "
+                    f"'/audio-input-config/{subject.name}'",
                 )
             if mic_allowed_value in ("0", "1"):
                 newvalue.untrusted_qdb.write(
-                    "/audio-input-config/{}".format(subject.name),
+                    f"/audio-input-config/{subject.name}",
                     mic_allowed_value,
                 )
             else:
                 raise qubes.exc.QubesVMError(
                     subject,
-                    "Invalid value '{!s}' for '/audio-input-config/{!s}' from {!s}".format(
-                        mic_allowed_value, subject.name, oldvalue
-                    ),
+                    f"Invalid value '{mic_allowed_value}' for "
+                    f"'/audio-input-config/{subject.name}' from {oldvalue}",
                 )
 
     @qubes.ext.handler("domain-qdb-create")
     def on_domain_qdb_create(self, vm, event):
         if vm.audiovm and vm.audiovm.is_running():
             # Remove previous config, status and request entries on audiovm start
-            vm.audiovm.untrusted_qdb.rm(
-                "/audio-input-config/{}".format(vm.name)
+            vm.audiovm.untrusted_qdb.rm(f"/audio-input-config/{vm.name}")
+            vm.audiovm.untrusted_qdb.rm(f"/audio-input/{vm.name}")
+            vm.audiovm.untrusted_qdb.rm(f"/audio-input-request/{vm.name}")
+
+    async def attach_and_notify(self, vm, assignment):
+        # bypass DeviceCollection logic preventing double attach
+        device = assignment.device
+        if assignment.mode.value == "ask-to-attach":
+            allowed = await qubes.ext.utils.confirm_device_attachment(
+                device, {vm: assignment}
             )
-            vm.audiovm.untrusted_qdb.rm("/audio-input/{}".format(vm.name))
-            vm.audiovm.untrusted_qdb.rm(
-                "/audio-input-request/{}".format(vm.name)
+            allowed = allowed.strip()
+            if vm.name != allowed:
+                return
+        await self.on_device_pre_attach_mic(
+            vm, "device-pre-attach:mic", device, assignment.options
+        )
+        await vm.fire_event_async(
+            "device-attach:mic", device=device, options=assignment.options
+        )
+
+    @qubes.ext.handler("domain-start")
+    async def on_domain_start(self, vm, _event, **_kwargs):
+        # pylint: disable=unused-argument
+        to_attach = {}
+        assignments = vm.devices["mic"].get_assigned_devices()
+        # the most specific assignments first
+        for assignment in reversed(sorted(assignments)):
+            for device in assignment.devices:
+                if isinstance(device, qubes.device_protocol.UnknownDevice):
+                    continue
+                if device.attachment:
+                    continue
+                if not assignment.matches(device):
+                    print(
+                        "Unrecognized identity, skipping attachment of device "
+                        f"from the port {assignment}",
+                        file=sys.stderr,
+                    )
+                    continue
+                # chose first assignment (the most specific) and ignore rest
+                if device not in to_attach:
+                    # make it unique
+                    to_attach[device] = assignment.clone(device=device)
+        for assignment in to_attach.values():
+            asyncio.ensure_future(self.attach_and_notify(vm, assignment))
+
+    @qubes.ext.handler("domain-shutdown")
+    async def on_domain_shutdown(self, vm, _event, **_kwargs):
+        # pylint: disable=unused-argument
+        mic = self.get_device(vm.app)
+        if mic in vm.devices["mic"].get_attached_devices():
+            asyncio.ensure_future(
+                vm.fire_event_async("device-detach:mic", port=mic.port)
             )