Add free-form text to qube for notes, comments, ...

Core and API part of adding free-form text to each qube for comments,
notes, descriptions, remarks, reminders, etc.

fixes: QubesOS/qubes-issues#899

Author: alimirjamali

Makefile

@@ -109,6 +109,8 @@ ADMIN_API_METHODS_SIMPLE = \
 	admin.vm.firewall.GetPolicy \
 	admin.vm.firewall.SetPolicy \
 	admin.vm.firewall.Reload \
+	admin.vm.notes.Get \
+	admin.vm.notes.Set \
 	admin.vm.property.Get \
 	admin.vm.property.GetAll \
 	admin.vm.property.GetDefault \

qubes/api/admin.py

@@ -28,6 +28,8 @@
 import subprocess
 import pathlib
 
+from ctypes import CDLL
+
 import libvirt
 import lxml.etree
 import importlib.metadata
@@ -53,6 +55,9 @@
     DeviceInterface,
 )
 
+# To validate & sanitise UTF8 strings
+LIBQUBES_PURE = "libqubes-pure.so.0"
+
 
 class QubesMgmtEventsDispatcher:
     def __init__(self, filters, send_event):
@@ -2036,3 +2041,48 @@ async def vm_current_state(self):
             "power_state": self.dest.get_power_state(),
         }
         return " ".join("{}={}".format(k, v) for k, v in state.items())
+
+    @qubes.api.method(
+        "admin.vm.notes.Get", no_payload=True, scope="local", read=True
+    )
+    async def vm_notes_get(self):
+        """Get qube notes"""
+        self.enforce(self.dest.name != "dom0")
+        self.fire_event_for_permission()
+        notes = self.dest.get_notes()
+        return notes
+
+    @qubes.api.method("admin.vm.notes.Set", scope="local", write=True)
+    async def vm_notes_set(self, untrusted_payload):
+        """Set qube notes"""
+        self.enforce(self.dest.name != "dom0")
+        self.fire_event_for_permission()
+        if len(untrusted_payload) > 256000:
+            raise qubes.exc.ProtocolError(
+                "Maximum note size is 256000 bytes ({} bytes received)".format(
+                    len(untrusted_payload)
+                )
+            )
+
+        # Sanitise the incoming utf8 notes with libqubes-pure
+        # use encode() to catch invalid UTF8 chars even earlier
+        try:
+            libqubespure = CDLL(LIBQUBES_PURE)
+            notes = "".join(
+                [
+                    (
+                        c
+                        if libqubespure.qubes_pure_string_safe_for_display(
+                            f"{c}\0".encode(), 0
+                        )
+                        else "_"
+                    )
+                    for c in untrusted_payload.decode("utf8")
+                ]
+            )
+        except Exception as e:
+            raise qubes.exc.ProtocolError(
+                "Unable to sanitise qube notes: " + str(e)
+            )
+
+        self.dest.set_notes(notes)

qubes/backup.py

@@ -433,6 +433,10 @@ def get_files_to_backup(self):
             if os.path.exists(firewall_conf):
                 vm_files.append(self.FileToBackup(firewall_conf, subdir))
 
+            notes_file_path = os.path.join(vm.dir_path, vm.notes_file)
+            if os.path.exists(notes_file_path):
+                vm_files.append(self.FileToBackup(notes_file_path, subdir))
+
             if not vm_files:
                 # subdir/ is needed in the tar file, otherwise restore
                 # of a (Disp)VM without any backed up files is going

qubes/tests/api_admin.py

@@ -2130,6 +2130,44 @@ def test_450_property_reset(self):
         self.assertIsNone(value)
         self.app.save.assert_called_once_with()
 
+    def test_notes_get(self):
+        notes = "For Your Eyes Only"
+        self.app.domains["test-vm1"].get_notes = unittest.mock.Mock()
+        self.app.domains["test-vm1"].get_notes.configure_mock(
+            **{"return_value": notes}
+        )
+        value = self.call_mgmt_func(b"admin.vm.notes.Get", b"test-vm1")
+        self.assertEqual(value, notes)
+        self.app.domains["test-vm1"].get_notes.configure_mock(
+            **{"side_effect": qubes.exc.QubesException()}
+        )
+        with self.assertRaises(qubes.exc.QubesException):
+            self.call_mgmt_func(b"admin.vm.notes.Get", b"test-vm1")
+        self.assertEqual(
+            self.app.domains["test-vm1"].get_notes.mock_calls,
+            [unittest.mock.call(), unittest.mock.call()],
+        )
+        self.assertFalse(self.app.save.called)
+
+    def test_notes_set(self):
+        self.app.domains["test-vm1"].set_notes = unittest.mock.Mock()
+        # Acceptable note
+        payload = "For Your Eyes Only".encode()
+        self.call_mgmt_func(
+            b"admin.vm.notes.Set",
+            b"test-vm1",
+            payload=payload,
+        )
+
+        # Unacceptable oversized note
+        with self.assertRaises(qubes.exc.ProtocolError):
+            payload = ("x" * 256001).encode()
+            self.call_mgmt_func(
+                b"admin.vm.notes.Set",
+                b"test-vm1",
+                payload=payload,
+            )
+
     def device_list_testclass(self, vm, event):
         if vm is not self.vm:
             return

qubes/tests/vm/qubesvm.py

@@ -3096,6 +3096,21 @@ def test_801_ordering(self):
             self.app, None, qid=1, name="bogus"
         ) > qubes.vm.adminvm.AdminVM(self.app, None)
 
+    def test_802_notes(self):
+        vm = self.get_vm()
+        notes = "For Your Eyes Only"
+        with unittest.mock.patch(
+            "builtins.open", unittest.mock.mock_open(read_data=notes)
+        ) as mock_open:
+            with self.assertNotRaises(qubes.exc.QubesException):
+                vm.set_notes(notes)
+            self.assertEqual(vm.get_notes(), notes)
+            mock_open.side_effect = FileNotFoundError()
+            self.assertEqual(vm.get_notes(), "")
+            with self.assertRaises(qubes.exc.QubesException):
+                mock_open.side_effect = PermissionError()
+                vm.get_notes()
+
     def test_810_bootmode_kernelopts(self):
         vm = self.get_vm(cls=qubes.vm.appvm.AppVM)
         vm.template = self.get_vm(cls=qubes.vm.templatevm.TemplateVM)

qubes/vm/qubesvm.py

@@ -2677,6 +2677,43 @@ def kernelopts_common(self):
 
         return result
 
+    #
+    # free-form text for descriptions, notes, comments, remarks, etc.
+    #
+
+    @property
+    def notes_file(self) -> str:
+        """Notes file name within /var/lib/qubes (per each qube sub-dir)"""
+        return "notes.txt"
+
+    def get_notes(self) -> str:
+        """Read the notes file and return its content"""
+        try:
+            with open(
+                os.path.join(self.dir_path, self.notes_file), encoding="utf8"
+            ) as fd:
+                return fd.read()
+        except FileNotFoundError:
+            return ""
+        except Exception as exc:
+            raise qubes.exc.QubesException(
+                "Failed to read notes file: " + str(exc)
+            )
+
+    def set_notes(self, notes: str):
+        """Write to notes file. Return True on success, False on error"""
+        try:
+            with open(
+                os.path.join(self.dir_path, self.notes_file),
+                "w",
+                encoding="utf8",
+            ) as fd:
+                fd.write(notes)
+        except Exception as exc:
+            raise qubes.exc.QubesException(
+                "Failed to write notes file: " + str(exc)
+            )
+
     #
     # helper methods
     #

rpm_spec/core-dom0.spec.in

@@ -108,6 +108,9 @@ Conflicts:      qubes-audio-dom0 < 4.3.5
 # Required for qvm-console* tools
 Requires:       socat
 
+# Requires libqubes-pure for qube notes utf8 sanitisation
+Requires:       qubes-utils-libs
+
 %{?systemd_requires}
 
 Obsoletes:	qubes-core-dom0-doc <= 4.0
@@ -286,6 +289,8 @@ admin.vm.feature.Set
 admin.vm.firewall.Get
 admin.vm.firewall.Reload
 admin.vm.firewall.Set
+admin.vm.notes.Get
+admin.vm.notes.Set
 admin.vm.property.Get
 admin.vm.property.GetAll
 admin.vm.property.GetDefault