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

@@ -2038,3 +2038,43 @@ 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()
+        try:
+            notes = self.dest.get_notes()
+        except Exception as e:
+            raise qubes.exc.QubesException(
+                "Could not read qube notes: " + str(e)
+            )
+        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)
+                )
+            )
+        allowed_chars = string.printable
+        notes = "".join(
+            [
+                c if c in allowed_chars else "_"
+                for c in untrusted_payload.decode("ascii")
+            ]
+        )
+        try:
+            self.dest.set_notes(notes)
+        except Exception as e:
+            raise qubes.exc.QubesException(
+                "Could not write qube notes: " + str(e)
+            )

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,50 @@ 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": Exception()}
+        )
+        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()
+        with self.assertRaises(qubes.exc.ProtocolError):
+            payload = ("x" * 256001).encode()
+            self.call_mgmt_func(
+                b"admin.vm.notes.Set",
+                b"test-vm1",
+                payload=payload,
+            )
+        with self.assertRaises(qubes.exc.QubesException):
+            payload = "For Your Eyes Only".encode()
+            self.call_mgmt_func(
+                b"admin.vm.notes.Set",
+                b"test-vm1",
+                payload=payload,
+            )
+            self.app.domains["test-vm1"].set_notes.configure_mock(
+                **{"side_effect": Exception()}
+            )
+            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

@@ -2947,3 +2947,18 @@ def test_801_ordering(self):
         assert qubes.vm.qubesvm.QubesVM(
             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(Exception):
+                vm.set_notes(notes)
+            self.assertEqual(vm.get_notes(), notes)
+            mock_open.side_effect = FileNotFoundError()
+            self.assertEqual(vm.get_notes(), "")
+            with self.assertRaises(Exception):
+                mock_open.side_effect = PermissionError()
+                vm.get_notes()

qubes/vm/qubesvm.py

@@ -2585,6 +2585,41 @@ def kernelopts_common(self):
         else:
             return base_kernelopts + qubes.config.defaults["kernelopts_common"]
 
+    #
+    # 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="ascii"
+            ) as fd:
+                return fd.read()
+        except FileNotFoundError:
+            return ""
+        except Exception as exc:
+            self.log.error("Failed to read notes file: %s", str(exc))
+            raise
+
+    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="ascii",
+            ) as fd:
+                fd.write(notes)
+        except Exception as exc:
+            self.log.error("Failed to write notes file: %s", str(exc))
+            raise
+
     #
     # helper methods
     #

rpm_spec/core-dom0.spec.in

@@ -286,6 +286,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