From 714d224010034b23edb6804655ad22b2f8d4b1e9 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marek=20Marczykowski-G=C3=B3recki?= Date: Fri, 21 Jun 2024 18:07:52 +0200 Subject: [PATCH 1/5] Move audio tests to separate module There are quite a lot of them now, move the a separate module to allow scheduling them separately. --- qubes/tests/__init__.py | 1 + qubes/tests/integ/audio.py | 478 +++++++++++++++++++++++++++++ qubes/tests/integ/vm_qrexec_gui.py | 438 -------------------------- rpm_spec/core-dom0.spec.in | 1 + 4 files changed, 480 insertions(+), 438 deletions(-) create mode 100644 qubes/tests/integ/audio.py diff --git a/qubes/tests/__init__.py b/qubes/tests/__init__.py index cd583bbe8..61d571762 100644 --- a/qubes/tests/__init__.py +++ b/qubes/tests/__init__.py @@ -1536,6 +1536,7 @@ def load_tests(loader, tests, pattern): # pylint: disable=unused-argument 'qubes.tests.integ.network_ipv6', 'qubes.tests.integ.dispvm', 'qubes.tests.integ.vm_qrexec_gui', + 'qubes.tests.integ.audio', 'qubes.tests.integ.mime', 'qubes.tests.integ.salt', 'qubes.tests.integ.backup', diff --git a/qubes/tests/integ/audio.py b/qubes/tests/integ/audio.py new file mode 100644 index 000000000..42c0a1c65 --- /dev/null +++ b/qubes/tests/integ/audio.py @@ -0,0 +1,478 @@ +# +# The Qubes OS Project, https://www.qubes-os.org/ +# +# Copyright (C) 2024 Marek Marczykowski-Górecki +# +# +# This library is free software; you can redistribute it and/or +# modify it under the terms of the GNU General Public +# License as published by the Free Software Foundation; either +# version 2 of the License, or (at your option) any later version. +# +# This library is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU +# Lesser General Public License for more details. +# +# You should have received a copy of the GNU General Public +# License along with this library; if not, see . +import asyncio +import grp +import json +import os +import signal +import subprocess +import sys +import tempfile +import unittest +from distutils import spawn + +import numpy as np + +import qubes.vm +from qubes.tests.integ.vm_qrexec_gui import TC_00_AppVMMixin, in_qemu + + +class TC_00_AudioMixin(TC_00_AppVMMixin): + def wait_for_pulseaudio_startup(self, vm): + self.loop.run_until_complete( + self.wait_for_session(self.testvm1)) + try: + self.loop.run_until_complete(vm.run_for_stdio( + "timeout 30s sh -c 'while ! pactl info; do sleep 1; done'" + )) + except subprocess.CalledProcessError as e: + self.fail('Timeout waiting for pulseaudio start in {}: {}{}'.format( + vm.name, e.stdout, e.stderr)) + # then wait for the stream to appear in dom0 + local_user = grp.getgrnam('qubes').gr_mem[0] + p = self.loop.run_until_complete(asyncio.create_subprocess_shell( + "sudo -E -u {} timeout 60s sh -c '" + "while ! pactl list sink-inputs | grep -q :{}; do sleep 1; done'" + .format(local_user, vm.name))) + self.loop.run_until_complete(p.wait()) + # and some more... + self.loop.run_until_complete(asyncio.sleep(1)) + + def prepare_audio_test(self, backend): + if 'whonix-g' in self.template: + self.skipTest('whonix gateway have no audio') + self.loop.run_until_complete(self.testvm1.start()) + pulseaudio_units = 'pulseaudio.socket pulseaudio.service' + pipewire_units = 'pipewire.socket wireplumber.service pipewire.service' + if backend == 'pipewire': + if not self.testvm1.features.check_with_template('supported-service.pipewire', False): + self.skipTest('PipeWire not supported in VM') + if 'debian-11' in self.template or ( + 'whonix' in self.template and '16' in self.template): + self.skipTest('PipeWire audio not supported in Debian 11') + self.testvm1.features['service.pipewire'] = True + elif backend == 'pulseaudio': + # Use PulseAudio if it is installed. If it is not installed, + # PipeWire will still run, and its PulseAudio emulation will + # be tested. + self.testvm1.features['service.pipewire'] = False + else: + self.fail('bad audio backend') + self.wait_for_pulseaudio_startup(self.testvm1) + + def create_audio_vm(self, backend, start=True): + self.audiovm = self.app.add_new_vm( + qubes.vm.appvm.AppVM, + label='red', + name=self.make_vm_name('audiovm'), + template=self.app.domains[self.template]) + self.loop.run_until_complete(self.audiovm.create_on_disk()) + with open("/etc/qubes/policy.d/10-test-audiovm.policy", "w") as f: + f.write(""" +admin.Events * {vm} {vm} allow target=dom0 +admin.Events * {vm} @adminvm allow target=dom0 +admin.Events * {vm} @tag:audiovm-{vm} allow target=dom0 +admin.vm.CurrentState * {vm} {vm} allow target=dom0 +admin.vm.CurrentState * {vm} @adminvm allow target=dom0 +admin.vm.CurrentState * {vm} @tag:audiovm-{vm} allow target=dom0 +admin.vm.List * {vm} {vm} allow target=dom0 +admin.vm.List * {vm} @adminvm allow target=dom0 +admin.vm.List * {vm} @tag:audiovm-{vm} allow target=dom0 +admin.vm.property.Get +audiovm {vm} @tag:audiovm-{vm} allow target=dom0 +admin.vm.property.Get +xid {vm} @tag:audiovm-{vm} allow target=dom0 +admin.vm.property.Get +stubdom_xid {vm} @tag:audiovm-{vm} allow target=dom0 +admin.vm.property.Get +virt_mode {vm} @tag:audiovm-{vm} allow target=dom0 +admin.vm.feature.CheckWithTemplate +audio {vm} @tag:audiovm-{vm} allow target=dom0 +admin.vm.feature.CheckWithTemplate +audio-model {vm} @tag:audiovm-{vm} allow target=dom0 +""".format(vm=self.audiovm.name)) + self.addCleanup(os.unlink, "/etc/qubes/policy.d/10-test-audiovm.policy") + self.audiovm.features['service.audiovm'] = True + if start: + self.loop.run_until_complete(self.audiovm.start()) + + def check_pacat_running(self, audiovm, xid): + pidfile = f"/run/qubes/pacat.{xid}" + if audiovm.qid == 0: + try: + with open(pidfile) as f: + pid = int(f.readline()) + os.kill(pid, 0) + running = True + except (FileNotFoundError, ProcessLookupError, ValueError): + running = False + else: + try: + self.loop.run_until_complete(audiovm.run_for_stdio(f"kill -0 $(cat {pidfile})")) + running = True + except subprocess.CalledProcessError: + running = False + return running + + def assert_pacat_running(self, audiovm, testvm, expected=True): + if testvm.features.get('audio-model', None): + xid = testvm.stubdom_xid + else: + xid = testvm.xid + running = None + for attempt in range(10): + running = self.check_pacat_running(audiovm, xid) + if running == expected: + break + self.loop.run_until_complete(asyncio.sleep(1)) + if expected != running: + self.fail(f"pacat for {testvm.name} (xid {xid}) running({running}) " + f"in {audiovm.name} while expected running({expected})") + + def check_audio_sample(self, sample, sfreq): + rec = np.fromstring(sample, dtype=np.float32) + # determine sample size using silence threshold + threshold = 10**-3 + rec_size = np.count_nonzero((rec > threshold) | (rec < -threshold)) + if not rec_size: + self.fail('only silence detected, no useful audio data') + margin = 0.95 + if in_qemu and self.testvm1.features.get('audio-model'): + # be less strict on HVM tests in nested virt, the test environment + # has huge overhead already + margin = 0.80 + if rec_size < margin*441000: + fname = f"/tmp/audio-sample-{self.id()}.raw" + with open(fname, "wb") as f: + f.write(sample) + self.fail(f'too short audio, expected 10s, got {rec_size/44100}, saved to {fname}') + # find zero crossings + crossings = np.nonzero((rec[1:] > threshold) & + (rec[:-1] < -threshold))[0] + np.seterr('raise') + # compare against sine wave frequency + rec_freq = 44100/np.mean(np.diff(crossings)) + if not sfreq*0.8 < rec_freq < sfreq*1.2: + fname = f"/tmp/audio-sample-{self.id()}.raw" + with open(fname, "wb") as f: + f.write(sample) + self.fail('frequency {} not in specified range, saved to {}' + .format(rec_freq, fname)) + + def common_audio_playback(self): + # sine frequency + sfreq = 4400 + # generate signal + audio_in = np.sin(2*np.pi*np.arange(441000)*sfreq/44100) + # Need to use .snd extension so that pw-play (really libsndfile) + # recognizes the file as raw audio. + self.loop.run_until_complete( + self.testvm1.run_for_stdio('cat > audio_in.snd', + input=audio_in.astype(np.float32).tobytes())) + local_user = grp.getgrnam('qubes').gr_mem[0] + if self.testvm1.features['service.pipewire']: + cmd = 'timeout 20s pw-play --format=f32 --rate=44100 --channels=1 - < audio_in.snd' + else: + cmd = ('timeout 20s paplay --format=float32le --rate=44100 --channels=1 ' + '--raw audio_in.snd') + with tempfile.NamedTemporaryFile() as recorded_audio: + os.chmod(recorded_audio.name, 0o666) + p = subprocess.Popen(['sudo', '-E', '-u', local_user, + 'parecord', '-d', '@DEFAULT_MONITOR@', '--raw', + '--format=float32le', '--rate=44100', '--channels=1', + recorded_audio.name], stdout=subprocess.PIPE, stderr=subprocess.PIPE) + try: + self.loop.run_until_complete(self.testvm1.run_for_stdio(cmd)) + except subprocess.CalledProcessError as err: + self.fail('{} stderr: {}'.format(str(err), err.stderr)) + # wait for possible parecord buffering + self.loop.run_until_complete(asyncio.sleep(2)) + if p.returncode is not None: + self.fail("Recording process ended prematurely: exit code {}, stderr: {}".format( + p.returncode, p.stderr.read())) + p.send_signal(signal.SIGINT) + p.wait() + self.check_audio_sample(recorded_audio.file.read(), sfreq) + + def _configure_audio_recording(self, vm): + """Connect VM's output-source to sink monitor instead of mic""" + local_user = grp.getgrnam("qubes").gr_mem[0] + sudo = ["sudo", "-E", "-u", local_user] + source_outputs = json.loads(subprocess.check_output( + sudo + ["pactl", "-f", "json", "list", "source-outputs"])) + + try: + output_index = [s["index"] for s in source_outputs + if s["properties"].get("application.name") + == vm.name][0] + except IndexError: + self.fail("source-output for VM {} not found".format(vm.name)) + # self.fail never returns + assert False + + sources = json.loads(subprocess.check_output( + sudo + ["pactl", "-f", "json", "list", "sources"])) + try: + source_index = [s["index"] for s in sources + if s["name"].endswith(".monitor")][0] + except IndexError: + self.fail("monitor source not found") + # self.fail never returns + assert False + + subprocess.check_call(sudo + + ["pactl", "move-source-output", str(output_index), str(source_index)]) + + def common_audio_record_muted(self): + # connect VM's recording source output monitor (instead of mic) + self._configure_audio_recording(self.testvm1) + + # generate some "audio" data + audio_in = b'\x20' * 4 * 44100 + local_user = grp.getgrnam('qubes').gr_mem[0] + # Need to use .snd extension so that pw-play (really libsndfile) + # recognizes the file as raw audio. + if self.testvm1.features['service.pipewire']: + cmd = 'pw-record --format=f32 --rate=44100 --channels=1 audio_rec.snd' + kill_cmd = 'pkill --signal SIGINT pw-record' + else: + cmd = 'parecord --raw audio_rec.snd' + kill_cmd = 'pkill --signal SIGINT parecord' + record = self.loop.run_until_complete(self.testvm1.run(cmd, + stdout=subprocess.PIPE, stderr=subprocess.PIPE)) + # give it time to start recording + self.loop.run_until_complete(asyncio.sleep(0.5)) + p = subprocess.Popen(['sudo', '-E', '-u', local_user, + 'paplay', '--raw'], + stdin=subprocess.PIPE) + p.communicate(audio_in) + # wait for possible parecord buffering + self.loop.run_until_complete(asyncio.sleep(2)) + if record.returncode is not None: + self.fail("Recording process ended prematurely: exit code {}, stderr: {}".format( + record.returncode, self.loop.run_until_complete(record.stderr.read()))) + try: + self.loop.run_until_complete( + self.testvm1.run_for_stdio(kill_cmd)) + except subprocess.CalledProcessError: + pass + self.loop.run_until_complete(record.wait()) + recorded_audio, _ = self.loop.run_until_complete( + self.testvm1.run_for_stdio('cat audio_rec.snd')) + # should be empty or silence, so check just a little fragment + if audio_in[:32] in recorded_audio: + self.fail('VM recorded something, even though mic disabled') + + def common_audio_record_unmuted(self): + deva = qubes.device_protocol.DeviceAssignment(self.app.domains[0], 'mic') + self.loop.run_until_complete( + self.testvm1.devices['mic'].attach(deva)) + # connect VM's recording source output monitor (instead of mic) + self._configure_audio_recording(self.testvm1) + sfreq = 4400 + audio_in = np.sin(2*np.pi*np.arange(441000)*sfreq/44100) + local_user = grp.getgrnam('qubes').gr_mem[0] + # Need to use .snd extension so that pw-play (really libsndfile) + # recognizes the file as raw audio. + if self.testvm1.features['service.pipewire']: + record_cmd = ('pw-record --format=f32 --rate=44100 --channels=1 ' + 'audio_rec.snd') + kill_cmd = 'pkill --signal SIGINT pw-record' + else: + record_cmd = ('parecord --raw --format=float32le --rate=44100 ' + '--channels=1 audio_rec.snd') + kill_cmd = 'pkill --signal SIGINT parecord' + record = self.loop.run_until_complete(self.testvm1.run(record_cmd)) + # give it time to start recording + self.loop.run_until_complete(asyncio.sleep(0.5)) + p = subprocess.Popen(['sudo', '-E', '-u', local_user, + 'paplay', '--raw', '--format=float32le', + '--rate=44100', '--channels=1'], + stdin=subprocess.PIPE) + p.communicate(audio_in.astype(np.float32).tobytes()) + # wait for possible parecord buffering + self.loop.run_until_complete(asyncio.sleep(2)) + if record.returncode is not None: + self.fail("Recording process ended prematurely: exit code {}, stderr: {}".format( + record.returncode, self.loop.run_until_complete(record.stderr.read()))) + try: + self.loop.run_until_complete(self.testvm1.run_for_stdio(kill_cmd)) + except subprocess.CalledProcessError: + pass + _, record_stderr = self.loop.run_until_complete(record.communicate()) + if record_stderr: + self.fail('parecord printed something on stderr: {}'.format( + record_stderr)) + + recorded_audio, _ = self.loop.run_until_complete( + self.testvm1.run_for_stdio('cat audio_rec.snd')) + self.check_audio_sample(recorded_audio, sfreq) + + +class TC_20_AudioVM_Pulse(TC_00_AudioMixin): + @unittest.skipUnless(spawn.find_executable('parecord'), + "pulseaudio-utils not installed in dom0") + def test_220_audio_play_pulseaudio(self): + self.prepare_audio_test('pulseaudio') + self.common_audio_playback() + + @unittest.skipUnless(spawn.find_executable('parecord'), + "pulseaudio-utils not installed in dom0") + def test_221_audio_rec_muted_pulseaudio(self): + self.prepare_audio_test('pulseaudio') + self.common_audio_record_muted() + + @unittest.skipUnless(spawn.find_executable('parecord'), + "pulseaudio-utils not installed in dom0") + def test_222_audio_rec_unmuted_pulseaudio(self): + self.prepare_audio_test('pulseaudio') + self.common_audio_record_unmuted() + + @unittest.skipUnless(spawn.find_executable('parecord'), + "pulseaudio-utils not installed in dom0") + def test_223_audio_play_hvm(self): + self.testvm1.virt_mode = 'hvm' + self.testvm1.features['audio-model'] = 'ich6' + self.prepare_audio_test('pulseaudio') + try: + self.loop.run_until_complete( + self.testvm1.run_for_stdio( + 'systemctl --user is-active pipewire-pulse.socket || ' + 'pacmd unload-module module-vchan-sink')) + except subprocess.CalledProcessError: + self.skipTest('PipeWire modules cannot be unloaded') + self.common_audio_playback() + + @unittest.skipUnless(spawn.find_executable('parecord'), + "pulseaudio-utils not installed in dom0") + def test_224_audio_rec_muted_hvm(self): + self.testvm1.virt_mode = 'hvm' + self.testvm1.features['audio-model'] = 'ich6' + self.prepare_audio_test('pulseaudio') + try: + # if pulseaudio is really emulated by pipewire, nothing needs to be + # done - pipewire-qubes won't register output before connecting to + # dom0, and with emulated sound active, it won't connect + self.loop.run_until_complete( + self.testvm1.run_for_stdio( + 'systemctl --user is-active pipewire-pulse.socket || ' + 'pacmd unload-module module-vchan-sink')) + except subprocess.CalledProcessError: + self.skipTest('PipeWire modules cannot be unloaded') + self.common_audio_record_muted() + + @unittest.skipUnless(spawn.find_executable('parecord'), + "pulseaudio-utils not installed in dom0") + def test_225_audio_rec_unmuted_hvm(self): + self.testvm1.virt_mode = 'hvm' + self.testvm1.features['audio-model'] = 'ich6' + self.prepare_audio_test('pulseaudio') + pa_info = self.loop.run_until_complete( + self.testvm1.run_for_stdio("pactl info"))[0] + # Server Name: PulseAudio (on PipeWire 0.3.65) + if b"on PipeWire 0.3." in pa_info: + self.skipTest("Known-buggy pipewire runs inside VM") + try: + sinks = self.loop.run_until_complete( + self.testvm1.run_for_stdio("pactl -f json list sinks"))[0] + sink_index = json.loads(sinks)[0]["index"] + self.loop.run_until_complete( + self.testvm1.run_for_stdio( + f"pactl set-sink-volume {sink_index!s} 0x10000")) + self.loop.run_until_complete( + self.testvm1.run_for_stdio( + 'systemctl --user is-active pipewire-pulse.socket || ' + 'pacmd unload-module module-vchan-sink')) + except subprocess.CalledProcessError: + self.skipTest('PipeWire modules cannot be unloaded') + self.common_audio_record_unmuted() + + @unittest.skipUnless(spawn.find_executable('parecord'), + "pulseaudio-utils not installed in dom0") + def test_252_audio_playback_audiovm_switch_hvm(self): + self.create_audio_vm('pulseaudio') + self.testvm1.audiovm = self.audiovm + self.testvm1.virt_mode = 'hvm' + self.testvm1.features['audio-model'] = 'ich6' + self.testvm1.features['stubdom-qrexec'] = '1' + self.prepare_audio_test('pulseaudio') + self.assert_pacat_running(self.audiovm, self.testvm1, True) + self.assert_pacat_running(self.app.domains[0], self.testvm1, False) + self.common_audio_playback() + self.testvm1.audiovm = 'dom0' + self.assert_pacat_running(self.audiovm, self.testvm1, False) + self.assert_pacat_running(self.app.domains[0], self.testvm1, True) + self.common_audio_playback() + self.testvm1.audiovm = None + self.assert_pacat_running(self.audiovm, self.testvm1, False) + + +class TC_20_AudioVM_PipeWire(TC_00_AudioMixin): + @unittest.skipUnless(spawn.find_executable('parecord'), + "pulseaudio-utils not installed in dom0") + def test_226_audio_playback_pipewire(self): + self.prepare_audio_test('pipewire') + self.common_audio_playback() + + @unittest.skipUnless(spawn.find_executable('parecord'), + "pulseaudio-utils not installed in dom0") + def test_227_audio_rec_muted_pipewire(self): + self.prepare_audio_test('pipewire') + self.common_audio_record_muted() + + @unittest.skipUnless(spawn.find_executable('parecord'), + "pulseaudio-utils not installed in dom0") + def test_228_audio_rec_unmuted_pipewire(self): + self.prepare_audio_test('pipewire') + self.common_audio_record_unmuted() + + @unittest.skipUnless(spawn.find_executable('parecord'), + "pulseaudio-utils not installed in dom0") + def test_250_audio_playback_audiovm_pipewire(self): + self.create_audio_vm('pipewire') + self.testvm1.audiovm = self.audiovm + self.prepare_audio_test('pipewire') + self.assert_pacat_running(self.audiovm, self.testvm1, True) + self.assert_pacat_running(self.app.domains[0], self.testvm1, False) + self.common_audio_playback() + self.testvm1.audiovm = None + self.assert_pacat_running(self.audiovm, self.testvm1, False) + + @unittest.skipUnless(spawn.find_executable('parecord'), + "pulseaudio-utils not installed in dom0") + def test_251_audio_playback_audiovm_pipewire_late_start(self): + self.create_audio_vm('pipewire', start=False) + self.testvm1.audiovm = self.audiovm + self.prepare_audio_test('pipewire') + self.loop.run_until_complete(self.audiovm.start()) + self.assert_pacat_running(self.audiovm, self.testvm1, True) + self.assert_pacat_running(self.app.domains[0], self.testvm1, False) + self.common_audio_playback() + + +def create_testcases_for_templates(): + yield from qubes.tests.create_testcases_for_templates( + 'TC_20_AudioVM_Pulse', TC_20_AudioVM_Pulse, + qubes.tests.SystemTestCase, module=sys.modules[__name__]) + yield from qubes.tests.create_testcases_for_templates( + 'TC_20_AudioVM_PipeWire', TC_20_AudioVM_PipeWire, + qubes.tests.SystemTestCase, module=sys.modules[__name__]) + + +def load_tests(loader, tests, pattern): + tests.addTests(loader.loadTestsFromNames( + create_testcases_for_templates())) + return tests + + +qubes.tests.maybe_create_testcases_on_import(create_testcases_for_templates) diff --git a/qubes/tests/integ/vm_qrexec_gui.py b/qubes/tests/integ/vm_qrexec_gui.py index 9ab23b2a4..8385be8f0 100644 --- a/qubes/tests/integ/vm_qrexec_gui.py +++ b/qubes/tests/integ/vm_qrexec_gui.py @@ -22,24 +22,17 @@ import asyncio import os import subprocess -import signal import sys -import tempfile -import json import unittest from distutils import spawn -import grp - import qubes.config import qubes.devices import qubes.tests import qubes.vm.appvm import qubes.vm.templatevm -import numpy as np - in_qemu = os.path.exists("/sys/firmware/qemu_fw_cfg") class TC_00_AppVMMixin(object): @@ -68,431 +61,6 @@ def setUp(self): self.testvm2.provides_network = True self.app.save() -class TC_00_AudioMixin(TC_00_AppVMMixin): - def wait_for_pulseaudio_startup(self, vm): - self.loop.run_until_complete( - self.wait_for_session(self.testvm1)) - try: - self.loop.run_until_complete(vm.run_for_stdio( - "timeout 30s sh -c 'while ! pactl info; do sleep 1; done'" - )) - except subprocess.CalledProcessError as e: - self.fail('Timeout waiting for pulseaudio start in {}: {}{}'.format( - vm.name, e.stdout, e.stderr)) - # then wait for the stream to appear in dom0 - local_user = grp.getgrnam('qubes').gr_mem[0] - p = self.loop.run_until_complete(asyncio.create_subprocess_shell( - "sudo -E -u {} timeout 60s sh -c '" - "while ! pactl list sink-inputs | grep -q :{}; do sleep 1; done'" - .format(local_user, vm.name))) - self.loop.run_until_complete(p.wait()) - # and some more... - self.loop.run_until_complete(asyncio.sleep(1)) - - def prepare_audio_test(self, backend): - if 'whonix-g' in self.template: - self.skipTest('whonix gateway have no audio') - self.loop.run_until_complete(self.testvm1.start()) - pulseaudio_units = 'pulseaudio.socket pulseaudio.service' - pipewire_units = 'pipewire.socket wireplumber.service pipewire.service' - if backend == 'pipewire': - if not self.testvm1.features.check_with_template('supported-service.pipewire', False): - self.skipTest('PipeWire not supported in VM') - if 'debian-11' in self.template or ( - 'whonix' in self.template and '16' in self.template): - self.skipTest('PipeWire audio not supported in Debian 11') - self.testvm1.features['service.pipewire'] = True - elif backend == 'pulseaudio': - # Use PulseAudio if it is installed. If it is not installed, - # PipeWire will still run, and its PulseAudio emulation will - # be tested. - self.testvm1.features['service.pipewire'] = False - else: - self.fail('bad audio backend') - self.wait_for_pulseaudio_startup(self.testvm1) - - def create_audio_vm(self, backend, start=True): - self.audiovm = self.app.add_new_vm( - qubes.vm.appvm.AppVM, - label='red', - name=self.make_vm_name('audiovm'), - template=self.app.domains[self.template]) - self.loop.run_until_complete(self.audiovm.create_on_disk()) - with open("/etc/qubes/policy.d/10-test-audiovm.policy", "w") as f: - f.write(""" -admin.Events * {vm} {vm} allow target=dom0 -admin.Events * {vm} @adminvm allow target=dom0 -admin.Events * {vm} @tag:audiovm-{vm} allow target=dom0 -admin.vm.CurrentState * {vm} {vm} allow target=dom0 -admin.vm.CurrentState * {vm} @adminvm allow target=dom0 -admin.vm.CurrentState * {vm} @tag:audiovm-{vm} allow target=dom0 -admin.vm.List * {vm} {vm} allow target=dom0 -admin.vm.List * {vm} @adminvm allow target=dom0 -admin.vm.List * {vm} @tag:audiovm-{vm} allow target=dom0 -admin.vm.property.Get +audiovm {vm} @tag:audiovm-{vm} allow target=dom0 -admin.vm.property.Get +xid {vm} @tag:audiovm-{vm} allow target=dom0 -admin.vm.property.Get +stubdom_xid {vm} @tag:audiovm-{vm} allow target=dom0 -admin.vm.property.Get +virt_mode {vm} @tag:audiovm-{vm} allow target=dom0 -admin.vm.feature.CheckWithTemplate +audio {vm} @tag:audiovm-{vm} allow target=dom0 -admin.vm.feature.CheckWithTemplate +audio-model {vm} @tag:audiovm-{vm} allow target=dom0 -""".format(vm=self.audiovm.name)) - self.addCleanup(os.unlink, "/etc/qubes/policy.d/10-test-audiovm.policy") - self.audiovm.features['service.audiovm'] = True - if start: - self.loop.run_until_complete(self.audiovm.start()) - - def check_pacat_running(self, audiovm, xid): - pidfile = f"/run/qubes/pacat.{xid}" - if audiovm.qid == 0: - try: - with open(pidfile) as f: - pid = int(f.readline()) - os.kill(pid, 0) - running = True - except (FileNotFoundError, ProcessLookupError, ValueError): - running = False - else: - try: - self.loop.run_until_complete(audiovm.run_for_stdio(f"kill -0 $(cat {pidfile})")) - running = True - except subprocess.CalledProcessError: - running = False - return running - - def assert_pacat_running(self, audiovm, testvm, expected=True): - if testvm.features.get('audio-model', None): - xid = testvm.stubdom_xid - else: - xid = testvm.xid - running = None - for attempt in range(10): - running = self.check_pacat_running(audiovm, xid) - if running == expected: - break - self.loop.run_until_complete(asyncio.sleep(1)) - if expected != running: - self.fail(f"pacat for {testvm.name} (xid {xid}) running({running}) " - f"in {audiovm.name} while expected running({expected})") - - def check_audio_sample(self, sample, sfreq): - rec = np.fromstring(sample, dtype=np.float32) - # determine sample size using silence threshold - threshold = 10**-3 - rec_size = np.count_nonzero((rec > threshold) | (rec < -threshold)) - if not rec_size: - self.fail('only silence detected, no useful audio data') - margin = 0.95 - if in_qemu and self.testvm1.features.get('audio-model'): - # be less strict on HVM tests in nested virt, the test environment - # has huge overhead already - margin = 0.80 - if rec_size < margin*441000: - fname = f"/tmp/audio-sample-{self.id()}.raw" - with open(fname, "wb") as f: - f.write(sample) - self.fail(f'too short audio, expected 10s, got {rec_size/44100}, saved to {fname}') - # find zero crossings - crossings = np.nonzero((rec[1:] > threshold) & - (rec[:-1] < -threshold))[0] - np.seterr('raise') - # compare against sine wave frequency - rec_freq = 44100/np.mean(np.diff(crossings)) - if not sfreq*0.8 < rec_freq < sfreq*1.2: - fname = f"/tmp/audio-sample-{self.id()}.raw" - with open(fname, "wb") as f: - f.write(sample) - self.fail('frequency {} not in specified range, saved to {}' - .format(rec_freq, fname)) - - def common_audio_playback(self): - # sine frequency - sfreq = 4400 - # generate signal - audio_in = np.sin(2*np.pi*np.arange(441000)*sfreq/44100) - # Need to use .snd extension so that pw-play (really libsndfile) - # recognizes the file as raw audio. - self.loop.run_until_complete( - self.testvm1.run_for_stdio('cat > audio_in.snd', - input=audio_in.astype(np.float32).tobytes())) - local_user = grp.getgrnam('qubes').gr_mem[0] - if self.testvm1.features['service.pipewire']: - cmd = 'timeout 20s pw-play --format=f32 --rate=44100 --channels=1 - < audio_in.snd' - else: - cmd = ('timeout 20s paplay --format=float32le --rate=44100 --channels=1 ' - '--raw audio_in.snd') - with tempfile.NamedTemporaryFile() as recorded_audio: - os.chmod(recorded_audio.name, 0o666) - p = subprocess.Popen(['sudo', '-E', '-u', local_user, - 'parecord', '-d', '@DEFAULT_MONITOR@', '--raw', - '--format=float32le', '--rate=44100', '--channels=1', - recorded_audio.name], stdout=subprocess.PIPE, stderr=subprocess.PIPE) - try: - self.loop.run_until_complete(self.testvm1.run_for_stdio(cmd)) - except subprocess.CalledProcessError as err: - self.fail('{} stderr: {}'.format(str(err), err.stderr)) - # wait for possible parecord buffering - self.loop.run_until_complete(asyncio.sleep(2)) - if p.returncode is not None: - self.fail("Recording process ended prematurely: exit code {}, stderr: {}".format( - p.returncode, p.stderr.read())) - p.send_signal(signal.SIGINT) - p.wait() - self.check_audio_sample(recorded_audio.file.read(), sfreq) - - def _configure_audio_recording(self, vm): - """Connect VM's output-source to sink monitor instead of mic""" - local_user = grp.getgrnam("qubes").gr_mem[0] - sudo = ["sudo", "-E", "-u", local_user] - source_outputs = json.loads(subprocess.check_output( - sudo + ["pactl", "-f", "json", "list", "source-outputs"])) - - try: - output_index = [s["index"] for s in source_outputs - if s["properties"].get("application.name") - == vm.name][0] - except IndexError: - self.fail("source-output for VM {} not found".format(vm.name)) - # self.fail never returns - assert False - - sources = json.loads(subprocess.check_output( - sudo + ["pactl", "-f", "json", "list", "sources"])) - try: - source_index = [s["index"] for s in sources - if s["name"].endswith(".monitor")][0] - except IndexError: - self.fail("monitor source not found") - # self.fail never returns - assert False - - subprocess.check_call(sudo + - ["pactl", "move-source-output", str(output_index), str(source_index)]) - - def common_audio_record_muted(self): - # connect VM's recording source output monitor (instead of mic) - self._configure_audio_recording(self.testvm1) - - # generate some "audio" data - audio_in = b'\x20' * 4 * 44100 - local_user = grp.getgrnam('qubes').gr_mem[0] - # Need to use .snd extension so that pw-play (really libsndfile) - # recognizes the file as raw audio. - if self.testvm1.features['service.pipewire']: - cmd = 'pw-record --format=f32 --rate=44100 --channels=1 audio_rec.snd' - kill_cmd = 'pkill --signal SIGINT pw-record' - else: - cmd = 'parecord --raw audio_rec.snd' - kill_cmd = 'pkill --signal SIGINT parecord' - record = self.loop.run_until_complete(self.testvm1.run(cmd, - stdout=subprocess.PIPE, stderr=subprocess.PIPE)) - # give it time to start recording - self.loop.run_until_complete(asyncio.sleep(0.5)) - p = subprocess.Popen(['sudo', '-E', '-u', local_user, - 'paplay', '--raw'], - stdin=subprocess.PIPE) - p.communicate(audio_in) - # wait for possible parecord buffering - self.loop.run_until_complete(asyncio.sleep(2)) - if record.returncode is not None: - self.fail("Recording process ended prematurely: exit code {}, stderr: {}".format( - record.returncode, self.loop.run_until_complete(record.stderr.read()))) - try: - self.loop.run_until_complete( - self.testvm1.run_for_stdio(kill_cmd)) - except subprocess.CalledProcessError: - pass - self.loop.run_until_complete(record.wait()) - recorded_audio, _ = self.loop.run_until_complete( - self.testvm1.run_for_stdio('cat audio_rec.snd')) - # should be empty or silence, so check just a little fragment - if audio_in[:32] in recorded_audio: - self.fail('VM recorded something, even though mic disabled') - - def common_audio_record_unmuted(self): - deva = qubes.device_protocol.DeviceAssignment(self.app.domains[0], 'mic') - self.loop.run_until_complete( - self.testvm1.devices['mic'].attach(deva)) - # connect VM's recording source output monitor (instead of mic) - self._configure_audio_recording(self.testvm1) - sfreq = 4400 - audio_in = np.sin(2*np.pi*np.arange(441000)*sfreq/44100) - local_user = grp.getgrnam('qubes').gr_mem[0] - # Need to use .snd extension so that pw-play (really libsndfile) - # recognizes the file as raw audio. - if self.testvm1.features['service.pipewire']: - record_cmd = ('pw-record --format=f32 --rate=44100 --channels=1 ' - 'audio_rec.snd') - kill_cmd = 'pkill --signal SIGINT pw-record' - else: - record_cmd = ('parecord --raw --format=float32le --rate=44100 ' - '--channels=1 audio_rec.snd') - kill_cmd = 'pkill --signal SIGINT parecord' - record = self.loop.run_until_complete(self.testvm1.run(record_cmd)) - # give it time to start recording - self.loop.run_until_complete(asyncio.sleep(0.5)) - p = subprocess.Popen(['sudo', '-E', '-u', local_user, - 'paplay', '--raw', '--format=float32le', - '--rate=44100', '--channels=1'], - stdin=subprocess.PIPE) - p.communicate(audio_in.astype(np.float32).tobytes()) - # wait for possible parecord buffering - self.loop.run_until_complete(asyncio.sleep(2)) - if record.returncode is not None: - self.fail("Recording process ended prematurely: exit code {}, stderr: {}".format( - record.returncode, self.loop.run_until_complete(record.stderr.read()))) - try: - self.loop.run_until_complete(self.testvm1.run_for_stdio(kill_cmd)) - except subprocess.CalledProcessError: - pass - _, record_stderr = self.loop.run_until_complete(record.communicate()) - if record_stderr: - self.fail('parecord printed something on stderr: {}'.format( - record_stderr)) - - recorded_audio, _ = self.loop.run_until_complete( - self.testvm1.run_for_stdio('cat audio_rec.snd')) - self.check_audio_sample(recorded_audio, sfreq) - -class TC_20_AudioVM_Pulse(TC_00_AudioMixin): - @unittest.skipUnless(spawn.find_executable('parecord'), - "pulseaudio-utils not installed in dom0") - def test_220_audio_play_pulseaudio(self): - self.prepare_audio_test('pulseaudio') - self.common_audio_playback() - - @unittest.skipUnless(spawn.find_executable('parecord'), - "pulseaudio-utils not installed in dom0") - def test_221_audio_rec_muted_pulseaudio(self): - self.prepare_audio_test('pulseaudio') - self.common_audio_record_muted() - - @unittest.skipUnless(spawn.find_executable('parecord'), - "pulseaudio-utils not installed in dom0") - def test_222_audio_rec_unmuted_pulseaudio(self): - self.prepare_audio_test('pulseaudio') - self.common_audio_record_unmuted() - - @unittest.skipUnless(spawn.find_executable('parecord'), - "pulseaudio-utils not installed in dom0") - def test_223_audio_play_hvm(self): - self.testvm1.virt_mode = 'hvm' - self.testvm1.features['audio-model'] = 'ich6' - self.prepare_audio_test('pulseaudio') - try: - self.loop.run_until_complete( - self.testvm1.run_for_stdio( - 'systemctl --user is-active pipewire-pulse.socket || ' - 'pacmd unload-module module-vchan-sink')) - except subprocess.CalledProcessError: - self.skipTest('PipeWire modules cannot be unloaded') - self.common_audio_playback() - - @unittest.skipUnless(spawn.find_executable('parecord'), - "pulseaudio-utils not installed in dom0") - def test_224_audio_rec_muted_hvm(self): - self.testvm1.virt_mode = 'hvm' - self.testvm1.features['audio-model'] = 'ich6' - self.prepare_audio_test('pulseaudio') - try: - # if pulseaudio is really emulated by pipewire, nothing needs to be - # done - pipewire-qubes won't register output before connecting to - # dom0, and with emulated sound active, it won't connect - self.loop.run_until_complete( - self.testvm1.run_for_stdio( - 'systemctl --user is-active pipewire-pulse.socket || ' - 'pacmd unload-module module-vchan-sink')) - except subprocess.CalledProcessError: - self.skipTest('PipeWire modules cannot be unloaded') - self.common_audio_record_muted() - - @unittest.skipUnless(spawn.find_executable('parecord'), - "pulseaudio-utils not installed in dom0") - def test_225_audio_rec_unmuted_hvm(self): - self.testvm1.virt_mode = 'hvm' - self.testvm1.features['audio-model'] = 'ich6' - self.prepare_audio_test('pulseaudio') - pa_info = self.loop.run_until_complete( - self.testvm1.run_for_stdio("pactl info"))[0] - # Server Name: PulseAudio (on PipeWire 0.3.65) - if b"on PipeWire 0.3." in pa_info: - self.skipTest("Known-buggy pipewire runs inside VM") - try: - sinks = self.loop.run_until_complete( - self.testvm1.run_for_stdio("pactl -f json list sinks"))[0] - sink_index = json.loads(sinks)[0]["index"] - self.loop.run_until_complete( - self.testvm1.run_for_stdio( - f"pactl set-sink-volume {sink_index!s} 0x10000")) - self.loop.run_until_complete( - self.testvm1.run_for_stdio( - 'systemctl --user is-active pipewire-pulse.socket || ' - 'pacmd unload-module module-vchan-sink')) - except subprocess.CalledProcessError: - self.skipTest('PipeWire modules cannot be unloaded') - self.common_audio_record_unmuted() - - @unittest.skipUnless(spawn.find_executable('parecord'), - "pulseaudio-utils not installed in dom0") - def test_252_audio_playback_audiovm_switch_hvm(self): - self.create_audio_vm('pulseaudio') - self.testvm1.audiovm = self.audiovm - self.testvm1.virt_mode = 'hvm' - self.testvm1.features['audio-model'] = 'ich6' - self.testvm1.features['stubdom-qrexec'] = '1' - self.prepare_audio_test('pulseaudio') - self.assert_pacat_running(self.audiovm, self.testvm1, True) - self.assert_pacat_running(self.app.domains[0], self.testvm1, False) - self.common_audio_playback() - self.testvm1.audiovm = 'dom0' - self.assert_pacat_running(self.audiovm, self.testvm1, False) - self.assert_pacat_running(self.app.domains[0], self.testvm1, True) - self.common_audio_playback() - self.testvm1.audiovm = None - self.assert_pacat_running(self.audiovm, self.testvm1, False) - - -class TC_20_AudioVM_PipeWire(TC_00_AudioMixin): - @unittest.skipUnless(spawn.find_executable('parecord'), - "pulseaudio-utils not installed in dom0") - def test_226_audio_playback_pipewire(self): - self.prepare_audio_test('pipewire') - self.common_audio_playback() - - @unittest.skipUnless(spawn.find_executable('parecord'), - "pulseaudio-utils not installed in dom0") - def test_227_audio_rec_muted_pipewire(self): - self.prepare_audio_test('pipewire') - self.common_audio_record_muted() - - @unittest.skipUnless(spawn.find_executable('parecord'), - "pulseaudio-utils not installed in dom0") - def test_228_audio_rec_unmuted_pipewire(self): - self.prepare_audio_test('pipewire') - self.common_audio_record_unmuted() - - @unittest.skipUnless(spawn.find_executable('parecord'), - "pulseaudio-utils not installed in dom0") - def test_250_audio_playback_audiovm_pipewire(self): - self.create_audio_vm('pipewire') - self.testvm1.audiovm = self.audiovm - self.prepare_audio_test('pipewire') - self.assert_pacat_running(self.audiovm, self.testvm1, True) - self.assert_pacat_running(self.app.domains[0], self.testvm1, False) - self.common_audio_playback() - self.testvm1.audiovm = None - self.assert_pacat_running(self.audiovm, self.testvm1, False) - - @unittest.skipUnless(spawn.find_executable('parecord'), - "pulseaudio-utils not installed in dom0") - def test_251_audio_playback_audiovm_pipewire_late_start(self): - self.create_audio_vm('pipewire', start=False) - self.testvm1.audiovm = self.audiovm - self.prepare_audio_test('pipewire') - self.loop.run_until_complete(self.audiovm.start()) - self.assert_pacat_running(self.audiovm, self.testvm1, True) - self.assert_pacat_running(self.app.domains[0], self.testvm1, False) - self.common_audio_playback() - class TC_20_NonAudio(TC_00_AppVMMixin): def test_000_start_shutdown(self): @@ -1160,12 +728,6 @@ def test_000_anyvm_deny_dom0(self): ' qrexec-client-vm output: {} {}'.format(stdout, stderr)) def create_testcases_for_templates(): - yield from qubes.tests.create_testcases_for_templates( - 'TC_20_AudioVM_Pulse', TC_20_AudioVM_Pulse, - qubes.tests.SystemTestCase, module=sys.modules[__name__]) - yield from qubes.tests.create_testcases_for_templates( - 'TC_20_AudioVM_PipeWire', TC_20_AudioVM_PipeWire, - qubes.tests.SystemTestCase, module=sys.modules[__name__]) yield from qubes.tests.create_testcases_for_templates( 'TC_20_NonAudio', TC_20_NonAudio, qubes.tests.SystemTestCase, module=sys.modules[__name__]) diff --git a/rpm_spec/core-dom0.spec.in b/rpm_spec/core-dom0.spec.in index 3e092681b..b7f907dc8 100644 --- a/rpm_spec/core-dom0.spec.in +++ b/rpm_spec/core-dom0.spec.in @@ -506,6 +506,7 @@ done %dir %{python3_sitelib}/qubes/tests/integ/__pycache__ %{python3_sitelib}/qubes/tests/integ/__pycache__/* %{python3_sitelib}/qubes/tests/integ/__init__.py +%{python3_sitelib}/qubes/tests/integ/audio.py %{python3_sitelib}/qubes/tests/integ/backup.py %{python3_sitelib}/qubes/tests/integ/backupcompatibility.py %{python3_sitelib}/qubes/tests/integ/backupdispvm.py From 14b5612c1d40356b61605ecb1a44f39a9dd02851 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marek=20Marczykowski-G=C3=B3recki?= Date: Sat, 22 Jun 2024 01:48:29 +0200 Subject: [PATCH 2/5] tests: extend simple_proxyvm test Add netvm reconnect to the test. This is mostly to test if DNS forwarding gets properly updated on dynamic network attach too. Suggested at https://github.com/QubesOS/qubes-core-agent-linux/pull/505#issuecomment-2182108738 --- qubes/tests/integ/network.py | 18 +++++++++++++++++- 1 file changed, 17 insertions(+), 1 deletion(-) diff --git a/qubes/tests/integ/network.py b/qubes/tests/integ/network.py index 7d4e42ffe..484b9dae0 100644 --- a/qubes/tests/integ/network.py +++ b/qubes/tests/integ/network.py @@ -163,7 +163,6 @@ def test_000_simple_networking(self): self.assertEqual(self.run_cmd(self.testvm1, self.ping_ip), 0) self.assertEqual(self.run_cmd(self.testvm1, self.ping_name), 0) - def test_010_simple_proxyvm(self): ''' :type self: qubes.tests.SystemTestCase | VMNetworkingMixin @@ -188,6 +187,23 @@ def test_010_simple_proxyvm(self): self.assertEqual(self.run_cmd(self.testvm1, self.ping_name), 0, "Ping by IP from AppVM failed") + self.proxy.netvm = None + self.loop.run_until_complete(self.testnetvm.shutdown(wait=True)) + # change IP to test if all info is updated, especially DNS redirect + self.test_ip = '192.168.45.123' + self.ping_ip = self.ping_cmd.format(target=self.test_ip) + self.configure_netvm() + self.proxy.netvm = self.testnetvm + self.loop.run_until_complete(asyncio.sleep(1)) + + self.assertEqual(self.run_cmd(self.proxy, self.ping_ip), 0, + "Ping by IP from ProxyVM failed (after switch)") + self.assertEqual(self.run_cmd(self.proxy, self.ping_name), 0, + "Ping by name from ProxyVM failed (after switch)") + self.assertEqual(self.run_cmd(self.testvm1, self.ping_ip), 0, + "Ping by IP from AppVM failed (after switch)") + self.assertEqual(self.run_cmd(self.testvm1, self.ping_name), 0, + "Ping by IP from AppVM failed (after switch)") @qubes.tests.expectedFailureIfTemplate('debian-7') @unittest.skipUnless(spawn.find_executable('xdotool'), From 566d1178ab7a6289152a42e7b8dd1a54a793148b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marek=20Marczykowski-G=C3=B3recki?= Date: Sat, 22 Jun 2024 03:28:37 +0200 Subject: [PATCH 3/5] vm: add domain-pre-spawn event Add an event after domain-pre-start but still before creating libvirt object (creating paused domain). This allows plugging in some handler after storage is prepared and netvm is started. QubesOS/qubes-issues#9318 --- qubes/vm/qubesvm.py | 17 +++++++++++++++++ 1 file changed, 17 insertions(+) diff --git a/qubes/vm/qubesvm.py b/qubes/vm/qubesvm.py index a38d46432..04889dcc5 100644 --- a/qubes/vm/qubesvm.py +++ b/qubes/vm/qubesvm.py @@ -280,6 +280,19 @@ class QubesVM(qubes.vm.mix.net.NetVMMixin, qubes.vm.BaseVM): *other arguments are as in :py:meth:`start`* + .. event:: domain-pre-spawn (subject, event, start_guid) + + Fired just before creating libvirt domain. + But after preparation steps - verifying storage, requesting memory, + starting netvm etc. + + :param subject: Event emitter (the qube object) + :param event: Event name (``'domain-pre-spawn'``) + + Handler for this event may be asynchronous. + + *other arguments are as in :py:meth:`start`* + .. event:: domain-spawn (subject, event, start_guid) Fired after creating libvirt domain. @@ -1208,6 +1221,10 @@ async def start(self, start_guid=True, notify_function=None, raise try: + await self.fire_event_async('domain-pre-spawn', + pre_event=True, + start_guid=start_guid) + self._update_libvirt_domain() self.libvirt_domain.createWithFlags( From 9bbac9005ffb31eb1a903b33041d7f80e44cbb51 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marek=20Marczykowski-G=C3=B3recki?= Date: Sat, 22 Jun 2024 03:36:50 +0200 Subject: [PATCH 4/5] Create qubesdb entries in netvm earlier in qube startup sequence vif hotplug script is called in netvm when libvirt creates connected the domain, even if it's created paused. Set qubesdb entries used by the vif script earlier, using new domain-pre-spawn event. Do not check is_running() here, because domain isn't running yet. Fixes QubesOS/qubes-issues#9318 --- qubes/vm/mix/net.py | 10 +++++++++- 1 file changed, 9 insertions(+), 1 deletion(-) diff --git a/qubes/vm/mix/net.py b/qubes/vm/mix/net.py index f3e2474d9..af50359d7 100644 --- a/qubes/vm/mix/net.py +++ b/qubes/vm/mix/net.py @@ -583,11 +583,19 @@ def on_domain_qdb_create(self, event): self.set_mapped_ip_info_for_vm(vm) self.reload_firewall_for_vm(vm) - @qubes.events.handler('firewall-changed', 'domain-spawn') + @qubes.events.handler('firewall-changed') def on_firewall_changed(self, event, **kwargs): ''' Reloads the firewall if vm is running and has a NetVM assigned ''' # pylint: disable=unused-argument if self.is_running() and self.netvm: + self.netvm.reload_firewall_for_vm(self) # pylint: disable=no-member + + @qubes.events.handler('domain-pre-spawn') + def on_pre_spawn(self, event, **kwargs): + """ Prepare qubesdb in netvm entries before relevant interface + is created """ + # pylint: disable=unused-argument + if self.netvm: self.netvm.reload_connected_ips() self.netvm.set_mapped_ip_info_for_vm(self) self.netvm.reload_firewall_for_vm(self) # pylint: disable=no-member From 872119fc134384df90cec3316ab38c5cafd799f0 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marek=20Marczykowski-G=C3=B3recki?= Date: Tue, 25 Jun 2024 01:53:42 +0200 Subject: [PATCH 5/5] tests: adjust qubes.Filecopy argument It was renamed from allow-unsafe-characters to allow-all-names See https://github.com/QubesOS/qubes-core-agent-linux/pull/508 https://github.com/QubesOS/qubes-issues/issues/8332 --- qubes/tests/integ/vm_qrexec_gui.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/qubes/tests/integ/vm_qrexec_gui.py b/qubes/tests/integ/vm_qrexec_gui.py index 8385be8f0..1b0bda315 100644 --- a/qubes/tests/integ/vm_qrexec_gui.py +++ b/qubes/tests/integ/vm_qrexec_gui.py @@ -362,7 +362,7 @@ def test_140_qrexec_filecopy_unsafe_name(self): self.loop.run_until_complete(self.testvm1.run_for_stdio( f"cp /etc/passwd /tmp/{name}")) with self.qrexec_policy('qubes.Filecopy+', self.testvm1, self.testvm2), \ - self.qrexec_policy('qubes.Filecopy+allow-unsafe-characters', self.testvm1, self.testvm2, allow=False): + self.qrexec_policy('qubes.Filecopy+allow-all-names', self.testvm1, self.testvm2, allow=False): with self.assertRaises(subprocess.CalledProcessError): self.loop.run_until_complete( self.testvm1.run_for_stdio(