Support for AudioVM

qubes/app.py

@@ -743,6 +743,12 @@ class Qubes(qubes.PropertyHolder):
         default=lambda app: app.domains['dom0'], allow_none=True,
         doc='Default GuiVM for VMs.')
 
+    default_audiovm = qubes.VMProperty(
+        'default_audiovm',
+        load_stage=3,
+        default=lambda app: app.domains['dom0'], allow_none=True,
+        doc='Default AudioVM for VMs.')
+
     default_netvm = qubes.VMProperty(
         'default_netvm',
         load_stage=3,

qubes/ext/audio.py

@@ -0,0 +1,64 @@
+#
+# The Qubes OS Project, https://www.qubes-os.org/
+#
+# Copyright (C) 2019 Frédéric Pierret <frederic.pierret@qubes-os.org>
+#
+# This library is free software; you can redistribute it and/or
+# modify it under the terms of the GNU Lesser General Public
+# License as published by the Free Software Foundation; either
+# version 2.1 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 Lesser General Public
+# License along with this library; if not, see <https://www.gnu.org/licenses/>.
+#
+
+import qubes.config
+import qubes.ext
+
+
+class AUDIO(qubes.ext.Extension):
+    # pylint: disable=too-few-public-methods
+    # property-del <=> property-reset-to-default
+    @qubes.ext.handler('property-del:audiovm')
+    def on_property_del(self, subject, event, name, oldvalue=None):
+        newvalue = getattr(subject, 'audiovm', None)
+        self.on_property_set(subject, event, name, newvalue, oldvalue)
+
+    @qubes.ext.handler('property-set:audiovm')
+    def on_property_set(self, subject, event, name, newvalue, oldvalue=None):
+        # pylint: disable=unused-argument,no-self-use
+
+        # Clean other 'audiovm-XXX' tags.
+        # pulseaudio agent (module-vchan-sink) can connect to only one domain
+        tags_list = list(subject.tags)
+        for tag in tags_list:
+            if 'audiovm-' in tag:
+                subject.tags.remove(tag)
+
+        if newvalue:
+            audiovm = 'audiovm-' + newvalue.name
+            subject.tags.add(audiovm)
+
+    @qubes.ext.handler('domain-qdb-create')
+    def on_domain_qdb_create(self, vm, event):
+        # pylint: disable=unused-argument,no-self-use
+        # Add AudioVM Xen ID for gui-agent
+        if getattr(vm, 'audiovm', None):
+            if vm != vm.audiovm:
+                vm.untrusted_qdb.write('/qubes-audio-domain-xid',
+                                       str(vm.audiovm.xid))
+
+    @qubes.ext.handler('property-set:default_audiovm', system=True)
+    def on_property_set_default_audiovm(self, app, event, name, newvalue,
+                                      oldvalue=None):
+        # pylint: disable=unused-argument,no-self-use
+        for vm in app.domains:
+            if hasattr(vm, 'audiovm') and vm.property_is_default('audiovm'):
+                vm.fire_event('property-set:audiovm',
+                              name='audiovm', newvalue=newvalue,
+                              oldvalue=oldvalue)

qubes/tests/app.py

@@ -608,6 +608,47 @@ class TC_90_Qubes(qubes.tests.QubesTestCase):
 
         self.assertIn('guivm-sys-gui', appvm.tags)
 
+    def test_114_default_audiovm(self):
+        class MyTestHolder(qubes.tests.TestEmitter, qubes.PropertyHolder):
+            default_audiovm = qubes.property('default_audiovm',
+                                           default=(lambda self: 'dom0'))
+
+        holder = MyTestHolder(None)
+        audiovm = self.app.add_new_vm('AppVM', name='sys-audio', audiovm='dom0',
+                                    template=self.template, label='red')
+        appvm = self.app.add_new_vm('AppVM', name='test-vm',
+                                    template=self.template, label='red')
+        holder.default_audiovm = 'sys-audio'
+        self.assertEqual(holder.default_audiovm, 'sys-audio')
+        self.assertIsNotNone(self.app.default_audiovm)
+        self.assertTrue(appvm.property_is_default('audiovm'))
+        self.app.default_audiovm = audiovm
+        self.assertEventFired(holder, 'property-set:default_audiovm',
+                              kwargs={'name': 'default_audiovm',
+                                      'newvalue': 'sys-audio'})
+
+        self.assertIn('audiovm-sys-audio', appvm.tags)
+
+    def test_115_audiovm(self):
+        class MyTestHolder(qubes.tests.TestEmitter, qubes.PropertyHolder):
+            audiovm = qubes.property('audiovm',
+                                   default=(lambda self: 'dom0'))
+
+        holder = MyTestHolder(None)
+        audiovm = self.app.add_new_vm('AppVM', name='sys-audio', audiovm='dom0',
+                                    template=self.template, label='red')
+        appvm = self.app.add_new_vm('AppVM', name='test-vm', audiovm='dom0',
+                                    template=self.template, label='red')
+        holder.audiovm = 'sys-audio'
+        self.assertEqual(holder.audiovm, 'sys-audio')
+        self.assertFalse(appvm.property_is_default('audiovm'))
+        appvm.audiovm = audiovm
+        self.assertEventFired(holder, 'property-set:audiovm',
+                              kwargs={'name': 'audiovm',
+                                      'newvalue': 'sys-audio'})
+
+        self.assertIn('audiovm-sys-audio', appvm.tags)
+
     def test_200_remove_template(self):
         appvm = self.app.add_new_vm('AppVM', name='test-vm',
                                     template=self.template,

qubes/tests/vm/qubesvm.py

@@ -1819,7 +1819,7 @@ class TC_90_QubesVM(QubesVMTestsMixin, qubes.tests.QubesTestCase):
     @unittest.mock.patch('qubes.utils.get_timezone')
     @unittest.mock.patch('qubes.utils.urandom')
     @unittest.mock.patch('qubes.vm.qubesvm.QubesVM.untrusted_qdb')
-    def test_622_qdb_keyboard_layout(self, mock_qubesdb, mock_urandom,
+    def test_622_qdb_guivm_keyboard_layout(self, mock_qubesdb, mock_urandom,
             mock_timezone):
         mock_urandom.return_value = b'A' * 64
         mock_timezone.return_value = 'UTC'
@@ -1871,6 +1871,48 @@ class TC_90_QubesVM(QubesVMTestsMixin, qubes.tests.QubesTestCase):
             '/connected-ips6': '',
         })
 
+    @unittest.mock.patch('qubes.utils.get_timezone')
+    @unittest.mock.patch('qubes.utils.urandom')
+    @unittest.mock.patch('qubes.vm.qubesvm.QubesVM.untrusted_qdb')
+    def test_623_qdb_audiovm(self, mock_qubesdb, mock_urandom,
+            mock_timezone):
+        mock_urandom.return_value = b'A' * 64
+        mock_timezone.return_value = 'UTC'
+        template = self.get_vm(
+            cls=qubes.vm.templatevm.TemplateVM, name='template')
+        template.netvm = None
+        audiovm = self.get_vm(cls=qubes.vm.appvm.AppVM, template=template,
+            name='sys-audio', qid=2, provides_network=False)
+        vm = self.get_vm(cls=qubes.vm.appvm.AppVM, template=template,
+            name='appvm', qid=3)
+        vm.netvm = None
+        vm.audiovm = audiovm
+        vm.events_enabled = True
+        test_qubesdb = TestQubesDB()
+        mock_qubesdb.write.side_effect = test_qubesdb.write
+        mock_qubesdb.rm.side_effect = test_qubesdb.rm
+        vm.create_qdb_entries()
+        self.maxDiff = None
+        self.assertEqual(test_qubesdb.data, {
+            '/name': 'test-inst-appvm',
+            '/type': 'AppVM',
+            '/default-user': 'user',
+            '/qubes-vm-type': 'AppVM',
+            '/qubes-audio-domain-xid': '{}'.format(audiovm.xid),
+            '/qubes-debug-mode': '0',
+            '/qubes-base-template': 'test-inst-template',
+            '/qubes-timezone': 'UTC',
+            '/qubes-random-seed': base64.b64encode(b'A' * 64),
+            '/qubes-vm-persistence': 'rw-only',
+            '/qubes-vm-updateable': 'False',
+            '/qubes-block-devices': '',
+            '/qubes-usb-devices': '',
+            '/qubes-iptables': 'reload',
+            '/qubes-iptables-error': '',
+            '/qubes-iptables-header': unittest.mock.ANY,
+            '/qubes-service/qubes-update-check': '0',
+            '/qubes-service/meminfo-writer': '1',
+        })
 
     @asyncio.coroutine
     def coroutine_mock(self, mock, *args, **kwargs):

qubes/vm/qubesvm.py

@@ -519,6 +519,10 @@ class QubesVM(qubes.vm.mix.net.NetVMMixin, qubes.vm.BaseVM):
                              default=(lambda self: self.app.default_guivm),
                              doc='VM used for Gui')
 
+    audiovm = qubes.VMProperty('audiovm', load_stage=4, allow_none=True,
+                             default=(lambda self: self.app.default_audiovm),
+                             doc='VM used for Audio')
+
     virt_mode = qubes.property(
         'virt_mode',
         type=str, setter=_setter_virt_mode,

rpm_spec/core-dom0.spec.in

@@ -276,6 +276,7 @@ fi
 %{python3_sitelib}/qubes/ext/block.py
 %{python3_sitelib}/qubes/ext/core_features.py
 %{python3_sitelib}/qubes/ext/gui.py
+%{python3_sitelib}/qubes/ext/audio.py
 %{python3_sitelib}/qubes/ext/pci.py
 %{python3_sitelib}/qubes/ext/qubesmanager.py
 %{python3_sitelib}/qubes/ext/r3compatibility.py

setup.py

@@ -64,6 +64,7 @@ if __name__ == '__main__':
                 'qubes.ext.core_features = qubes.ext.core_features:CoreFeatures',
                 'qubes.ext.qubesmanager = qubes.ext.qubesmanager:QubesManager',
                 'qubes.ext.gui = qubes.ext.gui:GUI',
+                'qubes.ext.audio = qubes.ext.audio:AUDIO',
                 'qubes.ext.r3compatibility = qubes.ext.r3compatibility:R3Compatibility',
                 'qubes.ext.pci = qubes.ext.pci:PCIDeviceExtension',
                 'qubes.ext.block = qubes.ext.block:BlockDeviceExtension',