add custom-persist extension

doc/qubes-ext.rst

@@ -14,6 +14,7 @@ Extensions defined here
 .. autoclass:: qubes.ext.admin.AdminExtension
 .. autoclass:: qubes.ext.block.BlockDeviceExtension
 .. autoclass:: qubes.ext.core_features.CoreFeatures
+.. autoclass:: qubes.ext.custom_persist.CustomPersist
 .. autoclass:: qubes.ext.gui.GUI
 .. autoclass:: qubes.ext.pci.PCIDeviceExtension
 .. autoclass:: qubes.ext.r3compatibility.R3Compatibility

qubes/ext/custom_persist.py

@@ -0,0 +1,134 @@
+# -*- encoding: utf-8 -*-
+#
+# The Qubes OS Project, http://www.qubes-os.org
+#
+# Copyright (C) 2024 Guillaume Chinal <[email protected]>
+#
+# 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 os
+import qubes.ext
+import qubes.config
+
+FEATURE_PREFIX = "custom-persist."
+QDB_PREFIX = "/persist/"
+QDB_KEY_LIMIT = 63
+
+
+class CustomPersist(qubes.ext.Extension):
+    """This extension allows to create minimal-state APP with by configuring an
+    exhaustive list of bind dirs(and files)
+    """
+
+    @staticmethod
+    def _extract_key_from_feature(feature) -> str:
+        return feature[len(FEATURE_PREFIX) :]
+
+    @staticmethod
+    def _is_expected_feature(feature) -> bool:
+        return feature.startswith(FEATURE_PREFIX)
+
+    @staticmethod
+    def _check_key(key):
+        if not key:
+            raise qubes.exc.QubesValueError(
+                "custom-persist key cannot be empty"
+            )
+
+        # QubesDB key length limit
+        key_maxlen = QDB_KEY_LIMIT - len(QDB_PREFIX)
+        if len(key) > key_maxlen:
+            raise qubes.exc.QubesValueError(
+                "custom-persist key is too long (max {}), ignoring: "
+                "{}".format(key_maxlen, key)
+            )
+
+    @staticmethod
+    def _check_value_path(value):
+        if not os.path.isabs(value):
+            raise qubes.exc.QubesValueError(f"invalid path '{value}'")
+
+    def _check_value(self, value):
+        if value.startswith("/"):
+            self._check_value_path(value)
+        else:
+            options = value.split(":")
+            if len(options) < 5 or not options[4].startswith("/"):
+                raise qubes.exc.QubesValueError(
+                    f"invalid value format: '{value}'"
+                )
+
+            resource_type = options[0]
+            mode = options[3]
+            if resource_type not in ("file", "dir"):
+                raise qubes.exc.QubesValueError(
+                    f"invalid resource type option '{resource_type}' "
+                    f"in value '{value}'"
+                )
+            try:
+                if not 0 <= int(mode, 8) <= 0o7777:
+                    raise qubes.exc.QubesValueError(
+                        f"invalid mode option '{mode}' in value '{value}'"
+                    )
+            except ValueError:
+                raise qubes.exc.QubesValueError(
+                    f"invalid mode option '{mode}' in value '{value}'"
+                )
+
+            self._check_value_path(":".join(options[4:]))
+
+    def _write_db_value(self, feature, value, vm):
+        vm.untrusted_qdb.write(
+            "{}{}".format(QDB_PREFIX, self._extract_key_from_feature(feature)),
+            str(value),
+        )
+
+    @qubes.ext.handler("domain-qdb-create")
+    def on_domain_qdb_create(self, vm, event):
+        """Actually export features"""
+        # pylint: disable=unused-argument
+        for feature, value in vm.features.items():
+            if self._is_expected_feature(feature):
+                self._check_key(self._extract_key_from_feature(feature))
+                self._check_value(value)
+                self._write_db_value(feature, value, vm)
+
+    @qubes.ext.handler("domain-feature-set:*")
+    def on_domain_feature_set(self, vm, event, feature, value, oldvalue=None):
+        """Inject persist keys in QubesDB in runtime"""
+        # pylint: disable=unused-argument
+
+        if not self._is_expected_feature(feature):
+            return
+
+        self._check_key(self._extract_key_from_feature(feature))
+        self._check_value(value)
+
+        if not vm.is_running():
+            return
+
+        self._write_db_value(feature, value, vm)
+
+    @qubes.ext.handler("domain-feature-delete:*")
+    def on_domain_feature_delete(self, vm, event, feature):
+        """Update /persist/ QubesDB tree in runtime"""
+        # pylint: disable=unused-argument
+        if not vm.is_running():
+            return
+        if not feature.startswith(FEATURE_PREFIX):
+            return
+
+        vm.untrusted_qdb.rm(
+            "{}{}".format(QDB_PREFIX, self._extract_key_from_feature(feature))
+        )

qubes/tests/ext.py

@@ -22,6 +22,7 @@
 import unittest.mock
 
 import qubes.ext.core_features
+import qubes.ext.custom_persist
 import qubes.ext.services
 import qubes.ext.windows
 import qubes.ext.supported_features
@@ -1045,3 +1046,218 @@ def test_022_supported_rpc_remove(self):
                 "supported-rpc.test2": True,
             },
         )
+
+
+class TC_40_CustomPersist(qubes.tests.QubesTestCase):
+    def setUp(self):
+        super().setUp()
+        self.ext = qubes.ext.custom_persist.CustomPersist()
+        self.features = {}
+        specs = {
+            "features.get.side_effect": self.features.get,
+            "features.items.side_effect": self.features.items,
+            "features.__iter__.side_effect": self.features.__iter__,
+            "features.__contains__.side_effect": self.features.__contains__,
+            "features.__setitem__.side_effect": self.features.__setitem__,
+            "features.__delitem__.side_effect": self.features.__delitem__,
+        }
+
+        vmspecs = {
+            **specs,
+            **{
+                "template": None,
+            },
+        }
+        self.vm = mock.MagicMock()
+        self.vm.configure_mock(**vmspecs)
+
+    def test_000_write_to_qdb(self):
+        self.features["custom-persist.home"] = "/home"
+        self.features["custom-persist.usrlocal"] = "/usr/local"
+        self.features["custom-persist.var_test"] = "/var/test"
+
+        self.ext.on_domain_qdb_create(self.vm, "domain-qdb-create")
+        self.assertEqual(
+            sorted(self.vm.untrusted_qdb.mock_calls),
+            [
+                mock.call.write("/persist/home", "/home"),
+                mock.call.write("/persist/usrlocal", "/usr/local"),
+                mock.call.write("/persist/var_test", "/var/test"),
+            ],
+        )
+
+    def test_001_feature_set(self):
+        self.ext.on_domain_feature_set(
+            self.vm,
+            "feature-set:custom-persist.test_no_oldvalue",
+            "custom-persist.test_no_oldvalue",
+            "/test_no_oldvalue",
+        )
+        self.ext.on_domain_feature_set(
+            self.vm,
+            "feature-set:custom-persist.test_oldvalue",
+            "custom-persist.test_oldvalue",
+            "/newvalue",
+            "",
+        )
+
+        self.assertEqual(
+            sorted(self.vm.untrusted_qdb.mock_calls),
+            [
+                mock.call.write(
+                    "/persist/test_no_oldvalue", "/test_no_oldvalue"
+                ),
+                mock.call.write("/persist/test_oldvalue", "/newvalue"),
+            ],
+        )
+
+    def test_002_feature_delete(self):
+        self.ext.on_domain_feature_delete(
+            self.vm, "feature-delete:custom-persist.test", "custom-persist.test"
+        )
+        self.vm.untrusted_qdb.rm.assert_called_with("/persist/test")
+
+    def test_003_empty_key(self):
+        with self.assertRaises(qubes.exc.QubesValueError) as e:
+            self.ext.on_domain_feature_set(
+                self.vm,
+                "feature-set:custom-persist.",
+                "custom-persist.",
+                "/test",
+                "",
+            )
+        self.assertEqual(str(e.exception), "custom-persist key cannot be empty")
+        self.vm.untrusted_qdb.write.assert_not_called()
+
+    def test_004_key_too_long(self):
+        with self.assertRaises(qubes.exc.QubesValueError) as e:
+            self.ext.on_domain_feature_set(
+                self.vm,
+                "feature-set:custom-persist." + "X" * 55,
+                "custom-persist." + "X" * 55,
+                "/test",
+                "",
+            )
+
+        self.assertEqual(
+            str(e.exception),
+            "custom-persist key is too long (max 54), ignoring: " + "X" * 55,
+        )
+        self.vm.untrusted_qdb.assert_not_called()
+
+    def test_005_other_feature_deletion(self):
+        self.ext.on_domain_feature_delete(
+            self.vm, "feature-delete:otherfeature.test", "otherfeature.test"
+        )
+        self.vm.untrusted_qdb.assert_not_called()
+
+    def test_006_feature_set_while_vm_is_not_running(self):
+        self.vm.is_running.return_value = False
+        self.ext.on_domain_feature_set(
+            self.vm,
+            "feature-set:custom-persist.test",
+            "custom-persist.test",
+            "/test",
+        )
+        self.vm.untrusted_qdb.write.assert_not_called()
+
+    def test_007_feature_set_value_with_option(self):
+        self.ext.on_domain_feature_set(
+            self.vm,
+            "feature-set:custom-persist.test",
+            "custom-persist.test",
+            "dir:root:root:0755:/var/test",
+            "",
+        )
+        self.vm.untrusted_qdb.write.assert_called_with(
+            "/persist/test", "dir:root:root:0755:/var/test"
+        )
+
+    def test_008_feature_set_invalid_path(self):
+        with self.assertRaises(qubes.exc.QubesValueError):
+            self.ext.on_domain_feature_set(
+                self.vm,
+                "feature-set:custom-persist.test",
+                "custom-persist.test",
+                "test",
+                "",
+            )
+        self.vm.untrusted_qdb.write.assert_not_called()
+
+    def test_009_feature_set_invalid_option_type(self):
+        with self.assertRaises(qubes.exc.QubesValueError):
+            self.ext.on_domain_feature_set(
+                self.vm,
+                "feature-set:custom-persist.test",
+                "custom-persist.test",
+                "bad:root:root:0755:/var/test",
+                "",
+            )
+        self.vm.untrusted_qdb.write.assert_not_called()
+
+    def test_010_feature_set_invalid_option_mode_too_high(self):
+        with self.assertRaises(qubes.exc.QubesValueError):
+            self.ext.on_domain_feature_set(
+                self.vm,
+                "feature-set:custom-persist.test",
+                "custom-persist.test",
+                "file:root:root:9750:/var/test",
+                "",
+            )
+        self.vm.untrusted_qdb.write.assert_not_called()
+
+    def test_011_feature_set_invalid_option_mode_negative_high(self):
+        with self.assertRaises(qubes.exc.QubesValueError):
+            self.ext.on_domain_feature_set(
+                self.vm,
+                "feature-set:custom-persist.test",
+                "custom-persist.test",
+                "file:root:root:-755:/var/test",
+                "",
+            )
+        self.vm.untrusted_qdb.write.assert_not_called()
+
+    def test_012_feature_set_option_mode_without_leading_zero(self):
+        self.ext.on_domain_feature_set(
+            self.vm,
+            "feature-set:custom-persist.test",
+            "custom-persist.test",
+            "file:root:root:755:/var/test",
+            "",
+        )
+        self.vm.untrusted_qdb.write.assert_called_with(
+            "/persist/test", "file:root:root:755:/var/test"
+        )
+
+    def test_013_feature_set_invalid_path_with_option(self):
+        with self.assertRaises(qubes.exc.QubesValueError):
+            self.ext.on_domain_feature_set(
+                self.vm,
+                "feature-set:custom-persist.test",
+                "custom-persist.test",
+                "dir:root:root:0755:var/test",
+                "",
+            )
+        self.vm.untrusted_qdb.write.assert_not_called()
+
+    def test_014_feature_set_path_with_colon_with_options(self):
+        self.ext.on_domain_feature_set(
+            self.vm,
+            "feature-set:custom-persist.test",
+            "custom-persist.test",
+            "file:root:root:755:/var/test:dir:with:colon",
+            "",
+        )
+        self.vm.untrusted_qdb.write.assert_called()
+
+    def test_015_feature_set_path_with_colon_without_options(self):
+        self.ext.on_domain_feature_set(
+            self.vm,
+            "feature-set:custom-persist.test",
+            "custom-persist.test",
+            "/var/test:dir:with:colon",
+            "",
+        )
+        self.vm.untrusted_qdb.write.assert_called_with(
+            "/persist/test", "/var/test:dir:with:colon"
+        )

rpm_spec/core-dom0.spec.in

@@ -443,6 +443,7 @@ done
 %{python3_sitelib}/qubes/ext/backup_restore.py
 %{python3_sitelib}/qubes/ext/block.py
 %{python3_sitelib}/qubes/ext/core_features.py
+%{python3_sitelib}/qubes/ext/custom_persist.py
 %{python3_sitelib}/qubes/ext/gui.py
 %{python3_sitelib}/qubes/ext/audio.py
 %{python3_sitelib}/qubes/ext/pci.py

setup.py

@@ -64,6 +64,7 @@ def run(self):
                 'qubes.ext.backup_restore = '
                 'qubes.ext.backup_restore:BackupRestoreExtension',
                 'qubes.ext.core_features = qubes.ext.core_features:CoreFeatures',
+                'qubes.ext.custom_persist = qubes.ext.custom_persist:CustomPersist',
                 'qubes.ext.gui = qubes.ext.gui:GUI',
                 'qubes.ext.audio = qubes.ext.audio:AUDIO',
                 'qubes.ext.r3compatibility = qubes.ext.r3compatibility:R3Compatibility',