From c11aa04f5b1bc637e24cd5b8f8626e5bddab80d7 Mon Sep 17 00:00:00 2001 From: Piotr Bartman-Szwarc Date: Thu, 17 Oct 2024 13:51:04 +0200 Subject: [PATCH] q-dev: cleanup mic.py --- qubesguidaemon/mic.py | 110 ++++++++++++++++++++---------------------- 1 file changed, 53 insertions(+), 57 deletions(-) diff --git a/qubesguidaemon/mic.py b/qubesguidaemon/mic.py index 7a62822e..e35338be 100644 --- a/qubesguidaemon/mic.py +++ b/qubesguidaemon/mic.py @@ -31,6 +31,8 @@ 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" @@ -55,9 +57,6 @@ class MicDeviceExtension(qubes.ext.Extension): Extension to control microphone access """ - def __init__(self): - super(MicDeviceExtension, self).__init__() - @staticmethod def get_device(app): return MicDevice( @@ -80,7 +79,7 @@ def on_device_get_mic(self, vm, event, port_id): 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 @@ -93,6 +92,7 @@ def on_device_get_mic(self, vm, event, port_id): @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 @@ -103,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) @@ -112,24 +112,23 @@ 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( - 'Microphone assignment does not support user 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( @@ -137,18 +136,17 @@ async def on_device_pre_attach_mic(self, vm, event, device, options): ): 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") @@ -161,13 +159,11 @@ async def on_device_pre_detach_mic(self, vm, event, 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( @@ -175,97 +171,96 @@ async def on_device_pre_detach_mic(self, vm, event, port): ): 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') + @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') + "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("/audio-input/{}".format(vm.name)) - vm.audiovm.untrusted_qdb.rm( - "/audio-input-request/{}".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}) + device, {vm: assignment} + ) allowed = allowed.strip() if vm.name != allowed: return await self.on_device_pre_attach_mic( - vm, 'device-pre-attach:mic', device, assignment.options) + vm, "device-pre-attach:mic", device, assignment.options + ) await vm.fire_event_async( - 'device-attach:mic', device=device, options=assignment.options) + "device-attach:mic", device=device, options=assignment.options + ) - @qubes.ext.handler('domain-start') + @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() + assignments = vm.devices["mic"].get_assigned_devices() # the most specific assignments first for assignment in reversed(sorted(assignments)): for device in assignment.devices: @@ -276,7 +271,9 @@ async def on_domain_start(self, vm, _event, **_kwargs): if not assignment.matches(device): print( "Unrecognized identity, skipping attachment of device " - f"from the port {assignment}", file=sys.stderr) + f"from the port {assignment}", + file=sys.stderr, + ) continue # chose first assignment (the most specific) and ignore rest if device not in to_attach: @@ -285,12 +282,11 @@ async def on_domain_start(self, vm, _event, **_kwargs): for assignment in to_attach.values(): asyncio.ensure_future(self.attach_and_notify(vm, assignment)) - @qubes.ext.handler('domain-shutdown') + @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( - f'device-detach:mic', port=mic.port - )) - + if mic in vm.devices["mic"].get_attached_devices(): + asyncio.ensure_future( + vm.fire_event_async("device-detach:mic", port=mic.port) + )