From ee39d272ac4912653ddbe7d5f364ce8dd5dee08b Mon Sep 17 00:00:00 2001 From: Ali Mirjamali Date: Sat, 15 Feb 2025 20:48:27 +0330 Subject: [PATCH] Add free-form text to qube for notes, comments, ... qubesadmin part of adding free-form text to each qube for comments, notes, descriptions, remarks, reminders, etc. fixes: https://github.com/QubesOS/qubes-issues/issues/899 Additional options to skip legacy backup tests or super slow tests --- doc/conf.py | 5 +- doc/manpages/qvm-notes.rst | 84 ++++++ qubesadmin/app.py | 11 + qubesadmin/backup/__init__.py | 4 + qubesadmin/backup/core2.py | 4 + qubesadmin/backup/core3.py | 7 + qubesadmin/backup/restore.py | 2 + qubesadmin/exc.py | 4 + qubesadmin/tests/app.py | 49 ++++ .../tests/backup/backupcompatibility.py | 34 +++ qubesadmin/tests/tools/qvm_create.py | 3 + qubesadmin/tests/tools/qvm_notes.py | 265 ++++++++++++++++++ qubesadmin/tools/qvm_notes.py | 202 +++++++++++++ qubesadmin/vm/__init__.py | 14 + 14 files changed, 686 insertions(+), 2 deletions(-) create mode 100644 doc/manpages/qvm-notes.rst create mode 100644 qubesadmin/tests/tools/qvm_notes.py create mode 100644 qubesadmin/tools/qvm_notes.py diff --git a/doc/conf.py b/doc/conf.py index 5bdaae1dc..c5d5b7063 100644 --- a/doc/conf.py +++ b/doc/conf.py @@ -84,7 +84,7 @@ # # This is also used if you do content translation via gettext catalogs. # Usually you set "language" from the command line for these cases. -language = None +language = 'en' # There are two options for replacing |today|: either, you set today to some # non-false value, then it is used: @@ -344,6 +344,8 @@ u'Kill the specified qube', _man_pages_author, 1), ('manpages/qvm-ls', 'qvm-ls', u'List VMs and various information about them', _man_pages_author, 1), + ('manpages/qvm-notes', 'qvm-notes', + u'Manipulate qube notes', _man_pages_author, 1), ('manpages/qvm-pause', 'qvm-pause', u'Pause a specified qube(s)', _man_pages_author, 1), ('manpages/qvm-pool', 'qvm-pool', @@ -370,7 +372,6 @@ u'Pause a qube', _man_pages_author, 1), ('manpages/qvm-volume', 'qvm-volume', u'Manage storage volumes of a qube', _man_pages_author, 1), - ('manpages/qubes-prefs', 'qubes-prefs', u'Display system-wide Qubes settings', _man_pages_author, 1), ] diff --git a/doc/manpages/qvm-notes.rst b/doc/manpages/qvm-notes.rst new file mode 100644 index 000000000..35476ced7 --- /dev/null +++ b/doc/manpages/qvm-notes.rst @@ -0,0 +1,84 @@ +.. program:: qvm-notes + +:program:`qvm-notes` -- Manipulate qube notes +============================================= + +Synopsis +-------- + +:command:`qvm-notes` [options] *VMNAME* [--edit | --print | --import *FILENAME* | --set '*NOTES*' | --append '*NOTES*' | --delete] + +Description +----------- + +This command is used to manipulate individual qube notes. Each qube notes is +limited to 256KB of clear text which could contain most UTF-8 characters. +However, some UTF-8 characters will be replaced with underline (`_`) due to +security limitations. Qube notes will be included in backup/restore. + +If this command is run outside dom0, it will require `admin.vm.notes.Get` and/or +`admin.vm.notes.Set` access privileges for the target qube in the RPC policies. + +General options +--------------- + +.. option:: --verbose, -v + + increase verbosity + +.. option:: --quiet, -q + + decrease verbosity + +.. option:: --help, -h + + show this help message and exit + +.. option:: --version + + show program's version number and exit + +.. option:: --force, -f + + Do not prompt for confirmation; assume `yes` + +Action options +-------------- + +.. option:: --edit, -e + + Edit qube notes in $EDITOR (default text editor). This is the default action. + +.. option:: --print, -p + + Print qube notes + +.. option:: --import=FILENAME, -i FILENAME + + Import qube notes from file + +.. option:: --set='NOTES', -s 'NOTES' + + Set qube notes from the provided string + +.. option:: --append='NOTES' + + Append the provided string to qube notes. If the last line of existing note + is not empty, a new line will be automatically inserted. + + Note that by design, qube notes is not suitable for appending automated logs + because of 256KB size limit and infrior performance compared to alternatives. + +.. option:: --delete, -d + + Delete qube notes + +Authors +------- + +| Marek Marczykowski +| Ali Mirjamali + +| For complete author list see: https://github.com/QubesOS/qubes-core-admin-client.git + +.. vim: ts=3 sw=3 et tw=80 diff --git a/qubesadmin/app.py b/qubesadmin/app.py index d2c980287..896d4880f 100644 --- a/qubesadmin/app.py +++ b/qubesadmin/app.py @@ -365,6 +365,7 @@ def clone_vm(self, src_vm, new_name, new_cls=None, *, pool=None, pools=None, ignore_errors=False, ignore_volumes=None, ignore_devices=False): # pylint: disable=too-many-statements + # pylint: disable=too-many-branches """Clone Virtual Machine Example usage with custom storage pools: @@ -474,6 +475,16 @@ def clone_vm(self, src_vm, new_name, new_cls=None, *, pool=None, pools=None, if not ignore_errors: raise + try: + vm_notes = src_vm.get_notes() + if vm_notes: + dst_vm.set_notes(vm_notes) + except qubesadmin.exc.QubesException as e: + dst_vm.log.error( + 'Failed to clone qube notes: {!s}'.format(e)) + if not ignore_errors: + raise + try: dst_vm.firewall.save_rules(src_vm.firewall.rules) except qubesadmin.exc.QubesException as e: diff --git a/qubesadmin/backup/__init__.py b/qubesadmin/backup/__init__.py index ced44d78c..f307646dc 100644 --- a/qubesadmin/backup/__init__.py +++ b/qubesadmin/backup/__init__.py @@ -71,3 +71,7 @@ def included_in_backup(self): def handle_firewall_xml(self, vm, stream): '''Import appropriate format of firewall.xml''' raise NotImplementedError + + def handle_notes_txt(self, vm, stream): + '''Import qubes notes.txt''' + raise NotImplementedError # pragma: no cover diff --git a/qubesadmin/backup/core2.py b/qubesadmin/backup/core2.py index d46f4495d..5b8958739 100644 --- a/qubesadmin/backup/core2.py +++ b/qubesadmin/backup/core2.py @@ -140,6 +140,10 @@ def _translate_action(key): except: # pylint: disable=bare-except vm.log.exception('Failed to set firewall') + def handle_notes_txt(self, vm, stream): + '''Qube notes did not exist at this time''' + raise NotImplementedError # pragma: no cover + class Core2Qubes(qubesadmin.backup.BackupApp): '''Parsed qubes.xml''' diff --git a/qubesadmin/backup/core3.py b/qubesadmin/backup/core3.py index 80efb3357..e1d9d7d93 100644 --- a/qubesadmin/backup/core3.py +++ b/qubesadmin/backup/core3.py @@ -52,6 +52,13 @@ def handle_firewall_xml(self, vm, stream): except: # pylint: disable=bare-except vm.log.exception('Failed to set firewall') + def handle_notes_txt(self, vm, stream): + '''Load new (Qubes >= 4.2) notes''' + try: + vm.set_notes(stream.read().decode()) + except: # pylint: disable=bare-except + vm.log.exception('Failed to set notes') + class Core3Qubes(qubesadmin.backup.BackupApp): '''Parsed qubes.xml''' def __init__(self, store=None): diff --git a/qubesadmin/backup/restore.py b/qubesadmin/backup/restore.py index db8c2250f..f6956f07e 100644 --- a/qubesadmin/backup/restore.py +++ b/qubesadmin/backup/restore.py @@ -1963,6 +1963,8 @@ def restore_do(self, restore_info): handlers[img_path] = data_func handlers[os.path.join(vm_info.subdir, 'firewall.xml')] = \ functools.partial(vm_info.vm.handle_firewall_xml, vm) + handlers[os.path.join(vm_info.subdir, 'notes.txt')] = \ + functools.partial(vm_info.vm.handle_notes_txt, vm) handlers[os.path.join(vm_info.subdir, 'whitelisted-appmenus.list')] = \ functools.partial(self._handle_appmenus_list, vm) diff --git a/qubesadmin/exc.py b/qubesadmin/exc.py index 3efbab7e7..737bd6d9f 100644 --- a/qubesadmin/exc.py +++ b/qubesadmin/exc.py @@ -232,5 +232,9 @@ def __init__(self, prop): super().__init__("Failed to access '%s' property" % prop) + +class QubesNotesError(QubesException): + """Some problem with qube notes.""" + # legacy name QubesDaemonNoResponseError = QubesDaemonAccessError diff --git a/qubesadmin/tests/app.py b/qubesadmin/tests/app.py index 1f920d2be..31dc705da 100644 --- a/qubesadmin/tests/app.py +++ b/qubesadmin/tests/app.py @@ -358,6 +358,11 @@ def clone_setup_common_calls(self, src, dst): (dst, 'admin.vm.firewall.Set', None, rules)] = \ b'0\x00' + # notes + self.app.expected_calls[ + (src, 'admin.vm.notes.Get', None, None)] = \ + b'0\0' + # storage for vm in (src, dst): self.app.expected_calls[ @@ -813,6 +818,50 @@ def test_044_clone_devices_fail(self): self.assertAllCalled() + def test_045_clone_notes_fail(self): + self.app.expected_calls[ + ('test-vm', 'admin.vm.notes.Get', None, None)] = \ + b'0\0Secret Note\0' + self.app.expected_calls[ + ('new-name', 'admin.vm.notes.Set', None, b'Secret Note\0')] = \ + b'2\0QubesNotesException\0\0It was For Your Eyes Only!\0' + self.app.expected_calls[ + ('test-vm', 'admin.vm.property.List', None, None)] = \ + b'0\0qid\nname\ntemplate\nlabel\nmemory\n' + self.app.expected_calls[ + ('test-vm', 'admin.vm.volume.List', None, None)] = \ + b'0\x00' + self.app.expected_calls[ + ('test-vm', 'admin.vm.property.Get', 'label', None)] = \ + b'0\0default=False type=label red' + self.app.expected_calls[ + ('test-vm', 'admin.vm.property.Get', 'template', None)] = \ + b'0\0default=False type=vm test-template' + self.app.expected_calls[ + ('test-vm', 'admin.vm.property.Get', 'memory', None)] = \ + b'0\0default=False type=int 400' + self.app.expected_calls[ + ('new-name', 'admin.vm.property.Set', 'memory', b'400')] = \ + b'0\0' + self.app.expected_calls[ + ('test-vm', 'admin.vm.tag.List', None, None)] = \ + b'0\0' + self.app.expected_calls[ + ('test-vm', 'admin.vm.feature.List', None, None)] = \ + b'0\0' + self.app.expected_calls[('dom0', 'admin.vm.List', None, None)] = \ + b'0\x00new-name class=AppVM state=Halted\n' \ + b'test-vm class=AppVM state=Halted\n' \ + b'test-template class=TemplateVM state=Halted\n' \ + b'test-net class=AppVM state=Halted\n' + self.app.expected_calls[('dom0', 'admin.vm.Create.AppVM', + 'test-template', b'name=new-name label=red')] = b'0\x00' + self.app.expected_calls[('new-name', 'admin.vm.Remove', None, None)] = \ + b'0\x00' + with self.assertRaises(qubesadmin.exc.QubesException): + self.app.clone_vm('test-vm', 'new-name') + self.assertAllCalled() + def test_050_automatic_reset_cache(self): self.app.cache_enabled = True dispatcher = qubesadmin.events.EventsDispatcher(self.app) diff --git a/qubesadmin/tests/backup/backupcompatibility.py b/qubesadmin/tests/backup/backupcompatibility.py index 3c5dab887..1e5ee221e 100644 --- a/qubesadmin/tests/backup/backupcompatibility.py +++ b/qubesadmin/tests/backup/backupcompatibility.py @@ -1050,6 +1050,14 @@ def create_v4_files(self): "tests/backup/v4-firewall.xml" f_firewall.write(xml_path.read_bytes()) + # setup notes only on one VM + with open( + self.fullpath("appvms/test-work/notes.txt"), + "w+", + encoding="utf-8", + ) as notes: + notes.write("For Your Eyes Only") + # StandaloneVMs for vm in ('test-standalonevm', 'test-hvm'): os.mkdir(self.fullpath('appvms/{}'.format(vm))) @@ -1528,6 +1536,8 @@ def create_limited_tmpdir(self, size): self.addCleanup(self.cleanup_tmpdir, tmpdir) return tmpdir.name + @unittest.skipIf(os.environ.get('DISABLE_LEGACY_TESTS', False), + 'Set DISABLE_LEGACY_TESTS=1 environment variable to skip this test') def test_210_r2(self): self.create_v3_backup(False) self.app.expected_calls[('dom0', 'admin.vm.List', None, None)] = ( @@ -1601,6 +1611,8 @@ def test_210_r2(self): self.assertDom0Restored(dummy_timestamp) + @unittest.skipIf(os.environ.get('DISABLE_LEGACY_TESTS', False), + 'Set DISABLE_LEGACY_TESTS=1 environment variable to skip this test') def test_220_r2_encrypted(self): self.create_v3_backup(True) @@ -1676,6 +1688,8 @@ def test_220_r2_encrypted(self): self.assertDom0Restored(dummy_timestamp) + @unittest.skipIf(os.environ.get('DISABLE_LEGACY_TESTS', False), + 'Set DISABLE_LEGACY_TESTS=1 environment variable to skip this test') def test_230_r2_uncompressed(self): self.create_v3_backup(False, False) self.app.expected_calls[('dom0', 'admin.vm.List', None, None)] = ( @@ -1781,6 +1795,9 @@ def test_230_r4(self): self.app.expected_calls[ ('test-work', 'admin.vm.firewall.Set', None, firewall_data.encode())] = b'0\0' + self.app.expected_calls[ + ('test-work', 'admin.vm.notes.Set', None, + b'For Your Eyes Only')] = b'0\0' qubesd_calls_queue = multiprocessing.Queue() @@ -1858,6 +1875,9 @@ def test_230_r4_compressed(self): self.app.expected_calls[ ('test-work', 'admin.vm.firewall.Set', None, firewall_data.encode())] = b'0\0' + self.app.expected_calls[ + ('test-work', 'admin.vm.notes.Set', None, + b'For Your Eyes Only')] = b'0\0' qubesd_calls_queue = multiprocessing.Queue() @@ -1935,6 +1955,9 @@ def test_230_r4_custom_cmpression(self): self.app.expected_calls[ ('test-work', 'admin.vm.firewall.Set', None, firewall_data.encode())] = b'0\0' + self.app.expected_calls[ + ('test-work', 'admin.vm.notes.Set', None, + b'For Your Eyes Only')] = b'0\0' qubesd_calls_queue = multiprocessing.Queue() @@ -2049,6 +2072,9 @@ def test_230_r4_uncommon_compression_forced(self): self.app.expected_calls[ ('test-work', 'admin.vm.firewall.Set', None, firewall_data.encode())] = b'0\0' + self.app.expected_calls[ + ('test-work', 'admin.vm.notes.Set', None, + b'For Your Eyes Only')] = b'0\0' qubesd_calls_queue = multiprocessing.Queue() @@ -2126,6 +2152,9 @@ def test_230_r4_optional_compression(self): self.app.expected_calls[ ('test-work', 'admin.vm.firewall.Set', None, firewall_data.encode())] = b'0\0' + self.app.expected_calls[ + ('test-work', 'admin.vm.notes.Set', None, + b'For Your Eyes Only')] = b'0\0' qubesd_calls_queue = multiprocessing.Queue() @@ -2171,6 +2200,8 @@ def test_230_r4_optional_compression(self): self.assertDom0Restored(dummy_timestamp) + @unittest.skipIf(os.environ.get('DISABLE_SUPER_SLOW_TESTS', False), + 'Set DISABLE_SUPER_SLOW_TESTS=1 environment variable to skip this test') @unittest.skipUnless(shutil.which('scrypt'), "scrypt not installed") def test_300_r4_no_space(self): @@ -2202,6 +2233,9 @@ def test_300_r4_no_space(self): self.app.expected_calls[ ('test-work', 'admin.vm.firewall.Set', None, firewall_data.encode())] = b'0\0' + self.app.expected_calls[ + ('test-work', 'admin.vm.notes.Set', None, + b'For Your Eyes Only')] = b'0\0' qubesd_calls_queue = multiprocessing.Queue() diff --git a/qubesadmin/tests/tools/qvm_create.py b/qubesadmin/tests/tools/qvm_create.py index 78c6ddc74..e1f29c671 100644 --- a/qubesadmin/tests/tools/qvm_create.py +++ b/qubesadmin/tests/tools/qvm_create.py @@ -268,6 +268,9 @@ def test_011_standalonevm(self, check_output_mock): self.app.expected_calls[ ('template', 'admin.vm.volume.CloneFrom', 'root', None)] = \ b'0\0clone-cookie' + self.app.expected_calls[ + ('template', 'admin.vm.notes.Get', None, None)] = \ + b'0\0' self.app.expected_calls[ ('new-vm', 'admin.vm.volume.CloneTo', 'root', b'clone-cookie')] = \ b'0\0' diff --git a/qubesadmin/tests/tools/qvm_notes.py b/qubesadmin/tests/tools/qvm_notes.py new file mode 100644 index 000000000..20956a006 --- /dev/null +++ b/qubesadmin/tests/tools/qvm_notes.py @@ -0,0 +1,265 @@ +# -*- encoding: utf-8 -*- +# +# The Qubes OS Project, http://www.qubes-os.org +# +# Copyright (C) 2025 Marek Marczykowski-Górecki +# +# Copyright (C) 2025 Ali Mirjamali +# +# This program 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 program 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 program; if not, see . + +# pylint: disable=missing-docstring + +import tempfile +from unittest.mock import patch + +import qubesadmin.exc +import qubesadmin.tests +import qubesadmin.tools.qvm_notes + + +class TC_00_qvm_notes(qubesadmin.tests.QubesTestCase): + + @patch("subprocess.run") + @patch("os.path.getmtime", side_effect=[2025, 1984]) + def test_001_edit(self, run, getmtime): + # pylint: disable=w0613 + self.app.expected_calls[("dom0", "admin.vm.List", None, None)] = ( + b"0\x00vm class=AppVM state=Running\n" + ) + self.app.expected_calls[("vm", "admin.vm.notes.Get", None, None)] = ( + b"0\x00For Your Eyes Only" + ) + self.app.expected_calls[ + ("vm", "admin.vm.notes.Set", None, b"For Your Eyes Only") + ] = b"0\x00" + self.assertEqual( + qubesadmin.tools.qvm_notes.main(["vm"], app=self.app), 0 + ) + self.assertAllCalled() + + def test_002_print(self): + self.app.expected_calls[("dom0", "admin.vm.List", None, None)] = ( + b"0\x00vm class=AppVM state=Running\n" + ) + self.app.expected_calls[("vm", "admin.vm.notes.Get", None, None)] = ( + b"0\x00For Your Eyes Only\n" + ) + self.assertEqual( + qubesadmin.tools.qvm_notes.main(["vm", "--print"], app=self.app), 0 + ) + self.assertAllCalled() + + def test_003_set(self): + self.app.expected_calls[("dom0", "admin.vm.List", None, None)] = ( + b"0\x00vm class=AppVM state=Running\n" + ) + self.app.expected_calls[ + ("vm", "admin.vm.notes.Set", None, b"For Your Eyes Only") + ] = b"0\x00" + self.assertEqual( + qubesadmin.tools.qvm_notes.main( + ["vm", "--force", "--set", "For Your Eyes Only"], app=self.app + ), + 0, + ) + self.assertAllCalled() + + def test_004_import(self): + self.app.expected_calls[("dom0", "admin.vm.List", None, None)] = ( + b"0\x00vm class=AppVM state=Running\n" + ) + self.app.expected_calls[ + ("vm", "admin.vm.notes.Set", None, b"For Your Eyes Only") + ] = b"0\x00" + with tempfile.NamedTemporaryFile( + mode="w+", + delete_on_close=False, + ) as temp: + temp.write("For Your Eyes Only") + temp.close() + self.assertEqual( + qubesadmin.tools.qvm_notes.main( + [ + "vm", + "--force", + "--import", + temp.name, + ], + app=self.app, + ), + 0, + ) + self.assertAllCalled() + + def test_005_append(self): + self.app.expected_calls[("dom0", "admin.vm.List", None, None)] = ( + b"0\x00vm class=AppVM state=Running\n" + ) + self.app.expected_calls[("vm", "admin.vm.notes.Get", None, None)] = ( + b"0\0Note 1" + ) + self.app.expected_calls[ + ("vm", "admin.vm.notes.Set", None, b"Note 1\nNote 2") + ] = b"0\x00" + self.assertEqual( + qubesadmin.tools.qvm_notes.main( + ["vm", "--force", "--append", "Note 2"], app=self.app + ), + 0, + ) + self.assertAllCalled() + + def test_006_delete(self): + self.app.expected_calls[("dom0", "admin.vm.List", None, None)] = ( + b"0\x00vm class=AppVM state=Running\n" + ) + self.app.expected_calls[("vm", "admin.vm.notes.Set", None, b"")] = ( + b"0\x00" + ) + self.assertEqual( + qubesadmin.tools.qvm_notes.main( + [ + "vm", + "--force", + "--delete", + ], + app=self.app, + ), + 0, + ) + self.assertAllCalled() + + @patch("builtins.input", return_value="NO") + def test_007_delete_canceled(self, input): + # pylint: disable=w0613,w0622 + self.app.expected_calls[("dom0", "admin.vm.List", None, None)] = ( + b"0\x00vm class=AppVM state=Running\n" + ) + with self.assertRaises(SystemExit): + qubesadmin.tools.qvm_notes.main( + [ + "vm", + "--delete", + ], + app=self.app, + ) + self.assertAllCalled() + + def test_010_no_read_access(self): + self.app.expected_calls[("dom0", "admin.vm.List", None, None)] = ( + b"0\x00vm class=AppVM state=Running\n" + ) + self.app.expected_calls[("vm", "admin.vm.notes.Get", None, None)] = ( + b"2\0QubesNotesException\0\0You do not have read access to notes!\0" + ) + self.assertEqual( + qubesadmin.tools.qvm_notes.main( + [ + "vm", + "--print", + ], + app=self.app, + ), + 1, + ) + self.assertEqual( + qubesadmin.tools.qvm_notes.main( + [ + "vm", + ], + app=self.app, + ), + 1, + ) + self.assertEqual( + qubesadmin.tools.qvm_notes.main( + [ + "vm", + "--append", + "Some note", + ], + app=self.app, + ), + 1, + ) + self.assertAllCalled() + + def test_011_no_write_access(self): + self.app.expected_calls[("dom0", "admin.vm.List", None, None)] = ( + b"0\x00vm class=AppVM state=Running\nq2 class=AppVM state=Running\n" + ) + self.app.expected_calls[("vm", "admin.vm.notes.Get", None, None)] = ( + b"0\x00" + ) + self.app.expected_calls[ + ("vm", "admin.vm.notes.Set", None, b"New note") + ] = b"2\0QubesNotesException\0\0You do not have read access to notes!\0" + self.app.expected_calls[("q2", "admin.vm.notes.Set", None, b"")] = ( + b"2\0QubesNotesException\0\0You do not have read access to notes!\0" + ) + self.assertEqual( + qubesadmin.tools.qvm_notes.main( + [ + "vm", + "--force", + "--append", + "New note", + ], + app=self.app, + ), + 1, + ) + self.assertEqual( + qubesadmin.tools.qvm_notes.main( + [ + "vm", + "--force", + "--set", + "New note", + ], + app=self.app, + ), + 1, + ) + with tempfile.NamedTemporaryFile( + mode="w+", + delete_on_close=False, + ) as temp: + temp.write("New note") + temp.close() + self.assertEqual( + qubesadmin.tools.qvm_notes.main( + [ + "vm", + "--force", + "--import", + temp.name, + ], + app=self.app, + ), + 1, + ) + self.assertEqual( + qubesadmin.tools.qvm_notes.main( + [ + "q2", + "--force", + "--delete", + ], + app=self.app, + ), + 1, + ) + self.assertAllCalled() diff --git a/qubesadmin/tools/qvm_notes.py b/qubesadmin/tools/qvm_notes.py new file mode 100644 index 000000000..70f7d8a10 --- /dev/null +++ b/qubesadmin/tools/qvm_notes.py @@ -0,0 +1,202 @@ +# -*- encoding: utf-8 -*- +# +# The Qubes OS Project, http://www.qubes-os.org +# +# Copyright (C) 2025 Marek Marczykowski-Górecki +# +# Copyright (C) 2025 Ali Mirjamali +# +# This program 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 program 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 program; if not, see . + +""" Qube notes manipulation tool """ + + +import logging +import os +import subprocess +import sys +import tempfile + +import qubesadmin +import qubesadmin.exc +import qubesadmin.tools + + +class ConfirmAction: + """Confirmation for set, delete and import actions""" + + use_the_force = False + + def __init__(self, message: str) -> None: + if self.use_the_force: + return + print(message) + if input("Are you certain? [y/N]").upper() != "Y": + sys.exit(2) + + +def get_parser(): + """Create :py:class:`argparse.ArgumentParser` suitable for + :program:`qvm-notes`. + """ + parser = qubesadmin.tools.QubesArgumentParser( + description="Manipulate qube notes", + vmname_nargs=1, + epilog=( + "Each qube notes is limited to 256KB of clear-text. " + "See program manpage for more information on other limitations." + ), + ) + parser.add_argument( + "--force", + "-f", + action="store_true", + help="Do not prompt for confirmation; assume `yes`", + ) + action_group = parser.add_argument_group( + title="Notes action options", + description="note: `--edit` is the default action for qube notes.", + ) + group = action_group.add_mutually_exclusive_group() + group.add_argument( + "--edit", + "-e", + action="store_const", + dest="action", + const="edit", + help="Edit qube notes in $EDITOR (default text editor)", + ) + group.add_argument( + "--print", + "-p", + action="store_const", + dest="action", + const="print", + help="Print qube notes", + ) + group.add_argument( + "--import", + "-i", + dest="filename", + metavar="FILENAME", + help="Import qube notes from file", + ) + group.add_argument( + "--set", + "-s", + dest="notes", + metavar="'NOTES'", + help="Set qube notes from provided string", + ) + group.add_argument( + "--append", + dest="append", + metavar="'NOTES'", + help="Append the provided string to qube notes", + ) + group.add_argument( + "--delete", + "-d", + action="store_const", + dest="action", + const="delete", + help="Delete qube notes", + ) + + # Setting notes editing as default preferred action + parser.set_defaults(action="edit") + return parser + + +def main(args=None, app=None): + """Main function of Program:`qvm-notes`.""" + app = app or qubesadmin.Qubes() + parser = get_parser() + args = parser.parse_args(args, app=app) + qube = args.domains.pop() + + ConfirmAction.use_the_force = args.force + + if args.filename: + args.action = "import" + if args.notes: + args.action = "set" + if args.append: + args.action = "append" + + exit_code: int = 0 + + match args.action: + case "edit": + try: + with tempfile.NamedTemporaryFile( + mode="w+", + prefix=qube.name + "_qube_", + suffix="_notes.txt", + delete_on_close=False, + ) as temp: + temp.write(qube.get_notes()) + temp.close() + last_modified = os.path.getmtime(temp.name) + edit_cmd = "${VISUAL:-${EDITOR:-vi}} " + temp.name + subprocess.run(edit_cmd, shell=True, check=True) + if last_modified != os.path.getmtime(temp.name): + with open(temp.name, encoding="utf-8") as notes_file: + qube.set_notes(notes_file.read()) + # os.unlink(temp.name) + except qubesadmin.exc.QubesException as e: + logging.error("Failed to edit qube notes: %s", str(e)) + exit_code = 1 + case "print": + try: + print(qube.get_notes()) + except qubesadmin.exc.QubesException as e: + logging.error("Unable to get qube notes: %s", str(e)) + exit_code = 1 + case "set": + try: + qube.set_notes(args.notes) + except qubesadmin.exc.QubesException as e: + logging.error("Unable to set qube notes: %s", str(e)) + exit_code = 1 + case "import": + try: + with open(args.filename, encoding="utf-8") as notes_file: + notes = notes_file.read() + qube.set_notes(notes) + except qubesadmin.exc.QubesException as e: + logging.error("Unable to import notes file: %s", str(e)) + exit_code = 1 + case "append": + try: + notes = qube.get_notes() + if notes.split("\n")[-1]: + notes += "\n" + qube.set_notes(notes + args.append) + except qubesadmin.exc.QubesException as e: + logging.error("Unable to append qube notes: %s", str(e)) + exit_code = 1 + case "delete": + try: + ConfirmAction("You are about to delete existing qube notes") + qube.set_notes("") + except qubesadmin.exc.QubesException as e: + logging.error("Unable to delete qube notes: %s", str(e)) + exit_code = 1 + + return exit_code + + +if __name__ == "__main__": + sys.exit(main()) # pragma: no cover diff --git a/qubesadmin/vm/__init__.py b/qubesadmin/vm/__init__.py index 8f0534502..0af6af318 100644 --- a/qubesadmin/vm/__init__.py +++ b/qubesadmin/vm/__init__.py @@ -424,6 +424,20 @@ def klass(self): self._klass = super().klass return self._klass + def get_notes(self) -> str: + ''' Get qube notes ''' + response = self.qubesd_call(self._method_dest, 'admin.vm.notes.Get') + return response.decode() + + def set_notes(self, notes: str): + ''' Set qube notes ''' + self.qubesd_call( + self._method_dest, + 'admin.vm.notes.Set', + payload=str(notes).encode(encoding='utf-8') + ) + + class DispVMWrapper(QubesVM): '''Wrapper class for new DispVM, supporting only service call