Skip to content

Commit

Permalink
q-dev: cleanup mic.py
Browse files Browse the repository at this point in the history
  • Loading branch information
@piotrbartman
piotrbartman committed Oct 17, 2024
1 parent 43d85da commit c11aa04
Showing 1 changed file with 53 additions and 57 deletions.
110 changes: 53 additions & 57 deletions qubesguidaemon/mic.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand All @@ -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(
Expand All @@ -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
Expand All @@ -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
Expand All @@ -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)
Expand All @@ -112,43 +112,41 @@ 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(
"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")
Expand All @@ -161,111 +159,108 @@ 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(
"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')
@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:
Expand All @@ -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:
Expand All @@ -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)
)

0 comments on commit c11aa04

Please sign in to comment.