Automatic ruff fixes

dom0/remove-tags.py

@@ -15,17 +15,17 @@ def main():
     for vm in q.domains:
         for tag in TAGS_TO_REMOVE:
             if tag in q.domains[vm].tags:
-                print("Removing tag '{}' from VM '{}'.".format(tag, vm))
+                print(f"Removing tag '{tag}' from VM '{vm}'.")
                 try:
                     q.domains[vm].tags.remove(tag)
                 except Exception as error:
-                    print("Error removing tag: '{}'".format(error))
+                    print(f"Error removing tag: '{error}'")
                     print("Aborting.")
                     exit(1)
                 tags_removed = True
 
     if tags_removed is False:
-        print("Tags {} not set on any VMs, nothing removed.".format(TAGS_TO_REMOVE))
+        print(f"Tags {TAGS_TO_REMOVE} not set on any VMs, nothing removed.")
 
 
 if __name__ == "__main__":

files/destroy-vm.py

@@ -41,7 +41,7 @@ def destroy_vm(vm):
     assert SDW_DEFAULT_TAG in vm.tags
     if vm.is_running():
         vm.kill()
-    print("Destroying VM '{}'... ".format(vm.name), end="")
+    print(f"Destroying VM '{vm.name}'... ", end="")
     subprocess.check_call(["qvm-remove", "-f", vm.name])
     print("OK")
 

files/sdw-admin.py

@@ -70,7 +70,7 @@ def install_pvh_support():
     TODO: install this via package requirements instead if possible
     """
     try:
-        subprocess.run(["sudo", "qubes-dom0-update", "-y", "-q", "grub2-xen-pvh"])
+        subprocess.run(["sudo", "qubes-dom0-update", "-y", "-q", "grub2-xen-pvh"], check=False)
     except subprocess.CalledProcessError:
         raise SDWAdminException("Error installing grub2-xen-pvah: local PVH not available.")
 

files/sdw-login.py

@@ -3,6 +3,7 @@
 Utility script for SecureDrop Workstation. Launches the SecureDrop Workstation
 updater on boot. It will prompt users to apply template and dom0 updates
 """
+
 import logging
 import os
 import subprocess

files/sdw-notify.py

@@ -4,6 +4,7 @@
 for too long without checking for security updates. Writes output to a logfile,
 not stdout. All settings are in Notify utility module.
 """
+
 import sys
 
 from PyQt5.QtWidgets import QApplication
@@ -55,7 +56,7 @@ def show_update_warning():
     notify script runs.
     """
 
-    app = QApplication([])  # noqa: F841
+    app = QApplication([])
     dialog = NotifyApp.NotifyDialog(Util.is_sdapp_halted())
     result = dialog.run()
 

files/validate_config.py

@@ -3,6 +3,7 @@
 Utility to verify that SecureDrop Workstation config is properly structured.
 Checks for
 """
+
 import json
 import os
 import re
@@ -24,7 +25,7 @@ class ValidationError(Exception):
     pass
 
 
-class SDWConfigValidator(object):
+class SDWConfigValidator:
     def __init__(self, config_base_dir=None):
         if config_base_dir:
             self.config_filepath = os.path.join(config_base_dir, CONFIG_FILEPATH)
@@ -42,7 +43,7 @@ def __init__(self, config_base_dir=None):
 
     def confirm_config_file_exists(self):
         if not os.path.exists(self.config_filepath):
-            msg = "Config file does not exist: {}".format(self.config_filepath)
+            msg = f"Config file does not exist: {self.config_filepath}"
             msg += "Create from config.json.example"
             raise ValidationError(msg)
 
@@ -107,9 +108,7 @@ def confirm_submission_privkey_file(self):
                 pass
 
         if not result:
-            raise ValidationError(
-                "GPG private key is not valid: {}".format(self.secret_key_filepath)
-            )
+            raise ValidationError(f"GPG private key is not valid: {self.secret_key_filepath}")
 
     def confirm_submission_privkey_fingerprint(self):
         if "submission_key_fpr" not in self.config:
@@ -126,12 +125,10 @@ def confirm_submission_privkey_fingerprint(self):
                 raise ValidationError("Configured fingerprint does not match key!")
 
         except subprocess.CalledProcessError as e:
-            raise ValidationError(
-                "Key validation failed: {}".format(e.output.decode(sys.stdout.encoding))
-            )
+            raise ValidationError(f"Key validation failed: {e.output.decode(sys.stdout.encoding)}")
 
     def read_config_file(self):
-        with open(self.config_filepath, "r") as f:
+        with open(self.config_filepath) as f:
             j = json.load(f)
         return j
 

launcher/tests/conftest.py

@@ -4,7 +4,7 @@
 import pytest
 
 
-@pytest.fixture
+@pytest.fixture()
 def tmpdir():
     """Run the test in a temporary directory"""
     cwd = os.getcwd()

launcher/tests/test_updater.py

@@ -153,11 +153,11 @@ def test_write_updates_status_flag_to_disk(
         Updater._write_updates_status_flag_to_disk(status)
 
     mocked_call.assert_called_once_with(
-        ["qvm-run", "sd-app", "echo '{}' > {}".format(status.value, flag_file_sd_app)]
+        ["qvm-run", "sd-app", f"echo '{status.value}' > {flag_file_sd_app}"]
     )
 
     assert os.path.exists(flag_file_dom0)
-    with open(flag_file_dom0, "r") as f:
+    with open(flag_file_dom0) as f:
         contents = json.load(f)
         assert contents["status"] == status.value
     assert "tmp" in flag_file_dom0
@@ -207,12 +207,12 @@ def test_write_last_updated_flags_to_disk(mocked_info, mocked_error, mocked_call
     subprocess_command = [
         "qvm-run",
         "sd-app",
-        "echo '{}' > {}".format(current_time, flag_file_sd_app),
+        f"echo '{current_time}' > {flag_file_sd_app}",
     ]
     mocked_call.assert_called_once_with(subprocess_command)
     assert not mocked_error.called
     assert os.path.exists(flag_file_dom0)
-    contents = open(flag_file_dom0, "r").read()
+    contents = open(flag_file_dom0).read()
     assert contents == current_time
 
 
@@ -311,7 +311,7 @@ def test_apply_updates_vms(mocked_info, mocked_error, mocked_call, vm):
 @mock.patch("sdw_updater.Updater.sdlog.info")
 def test_apply_updates_vms_fails(mocked_info, mocked_error, mocked_call, vm):
     error_calls = [
-        call("An error has occurred updating {}. Please contact your administrator.".format(vm)),
+        call(f"An error has occurred updating {vm}. Please contact your administrator."),
         call("Command 'check_call' returned non-zero exit status 1."),
     ]
     result = Updater._apply_updates_vm(vm)
@@ -403,7 +403,7 @@ def test_overall_update_status_reboot_not_done_previously(
 @mock.patch("sdw_updater.Updater.sdlog.error")
 @mock.patch("sdw_updater.Updater.sdlog.info")
 def test_safely_shutdown(mocked_info, mocked_error, mocked_output, vm):
-    call_list = [call(["qvm-shutdown", "--wait", "{}".format(vm)], stderr=-1)]
+    call_list = [call(["qvm-shutdown", "--wait", f"{vm}"], stderr=-1)]
 
     Updater._safely_shutdown_vm(vm)
     mocked_output.assert_has_calls(call_list)
@@ -431,7 +431,7 @@ def test_safely_start(mocked_info, mocked_error, mocked_output, vm):
 @mock.patch("sdw_updater.Updater.sdlog.info")
 def test_safely_start_fails(mocked_info, mocked_error, mocked_output, vm):
     call_list = [
-        call("Error while starting {}".format(vm)),
+        call(f"Error while starting {vm}"),
         call("Command 'check_output' returned non-zero exit status 1."),
     ]
 
@@ -445,7 +445,7 @@ def test_safely_start_fails(mocked_info, mocked_error, mocked_output, vm):
 @mock.patch("sdw_updater.Updater.sdlog.info")
 def test_safely_shutdown_fails(mocked_info, mocked_error, mocked_call, vm):
     call_list = [
-        call("Failed to shut down {}".format(vm)),
+        call(f"Failed to shut down {vm}"),
         call("Command 'check_output' returned non-zero exit status 1."),
     ]
 
@@ -474,8 +474,8 @@ def test_shutdown_and_start_vms(
     ]
     template_vm_calls = [
         call("fedora-39-xfce"),
-        call("sd-large-{}-template".format(DEBIAN_VERSION)),
-        call("sd-small-{}-template".format(DEBIAN_VERSION)),
+        call(f"sd-large-{DEBIAN_VERSION}-template"),
+        call(f"sd-small-{DEBIAN_VERSION}-template"),
         call("whonix-gateway-17"),
     ]
     app_vm_calls = [
@@ -520,8 +520,8 @@ def test_shutdown_and_start_vms_sysvm_fail(
     ]
     template_vm_calls = [
         call("fedora-39-xfce"),
-        call("sd-large-{}-template".format(DEBIAN_VERSION)),
-        call("sd-small-{}-template".format(DEBIAN_VERSION)),
+        call(f"sd-large-{DEBIAN_VERSION}-template"),
+        call(f"sd-small-{DEBIAN_VERSION}-template"),
         call("whonix-gateway-17"),
     ]
     error_calls = [
@@ -550,7 +550,7 @@ def test_read_dom0_update_flag_from_disk(mocked_error, mocked_subprocess, status
         flag_file_dom0 = Updater.get_dom0_path(Updater.FLAG_FILE_STATUS_DOM0)
 
         assert os.path.exists(flag_file_dom0)
-        with open(flag_file_dom0, "r") as f:
+        with open(flag_file_dom0) as f:
             contents = json.load(f)
             assert contents["status"] == status.value
         assert "tmp" in flag_file_dom0
@@ -807,7 +807,7 @@ def test_run_full_install(mocked_call, mocked_output, mocked_info):
     """
     # subprocess.check_call is mocked, so this directory should never be accessed
     # by the test.
-    MIGRATION_DIR = "/tmp/potato"  # nosec
+    MIGRATION_DIR = "/tmp/potato"
     with mock.patch("sdw_updater.Updater.MIGRATION_DIR", MIGRATION_DIR):
         result = Updater.run_full_install()
     check_outputs = [call(["sdw-admin", "--apply"])]
@@ -833,7 +833,7 @@ def test_run_full_install_with_error(mocked_call, mocked_output, mocked_error):
       And the error is logged
       And the failure enum is returned
     """
-    MIGRATION_DIR = "/tmp/potato"  # nosec
+    MIGRATION_DIR = "/tmp/potato"
     with mock.patch("sdw_updater.Updater.MIGRATION_DIR", MIGRATION_DIR):
         result = Updater.run_full_install()
     calls = [call(["sdw-admin", "--apply"])]
@@ -857,7 +857,7 @@ def test_run_full_install_with_flag_error(mocked_call, mocked_output, mocked_err
     Then the error is logged
       And the failure enum is returned
     """
-    MIGRATION_DIR = "/tmp/potato"  # nosec
+    MIGRATION_DIR = "/tmp/potato"
     with mock.patch("sdw_updater.Updater.MIGRATION_DIR", MIGRATION_DIR):
         result = Updater.run_full_install()
     check_outputs = [call(["sdw-admin", "--apply"])]

launcher/tests/test_util.py

@@ -33,7 +33,7 @@ def test_obtain_lock(mocked_info, mocked_warning, mocked_error, tmp_path):
 
         basename = "test-obtain-lock.lock"
         pid_str = str(os.getpid())
-        lh = Util.obtain_lock(basename)  # noqa: F841
+        lh = Util.obtain_lock(basename)
         # No handled exception should occur
         assert not mocked_error.called
         # We should be getting a lock handle back
@@ -73,7 +73,7 @@ def test_cannot_obtain_exclusive_lock_when_busy(
 
         # We're running in the same process, so obtaining a lock will succeed.
         # Instead we're mocking the IOError lockf would raise.
-        with mock.patch("fcntl.lockf", side_effect=IOError()) as mocked_lockf:
+        with mock.patch("fcntl.lockf", side_effect=OSError()) as mocked_lockf:
             lh2 = Util.obtain_lock(basename)
             mocked_lockf.assert_called_once()
             assert lh2 is None
@@ -100,7 +100,7 @@ def test_cannot_obtain_shared_lock_when_busy(mocked_info, mocked_warning, mocked
 
         # We're running in the same process, so obtaining a lock will succeed.
         # Instead we're mocking the IOError lockf would raise.
-        with mock.patch("fcntl.lockf", side_effect=IOError()) as mocked_lockf:
+        with mock.patch("fcntl.lockf", side_effect=OSError()) as mocked_lockf:
             can_get_lock = Util.can_obtain_lock(basename)
             mocked_lockf.assert_called_once()
             assert can_get_lock is False
@@ -129,7 +129,7 @@ def test_permission_error_is_handled(mocked_info, mocked_warning, mocked_error):
     """
     Test whether permission errors obtaining a lock are handled correctly
     """
-    with mock.patch("builtins.open", side_effect=PermissionError()) as mocked_open:  # noqa: F821
+    with mock.patch("builtins.open", side_effect=PermissionError()) as mocked_open:
         lock = Util.obtain_lock("test-open-error.lock")
         assert lock is None
         mocked_open.assert_called_once()
@@ -236,7 +236,7 @@ def test_get_logger():
     logger = Util.get_logger(prefix=test_prefix)
     assert logger.name == test_prefix
     logger = Util.get_logger(prefix=test_prefix, module=test_module)
-    assert logger.name == "{}.{}".format(test_prefix, test_module)
+    assert logger.name == f"{test_prefix}.{test_module}"
 
 
 @pytest.mark.parametrize(
@@ -255,7 +255,7 @@ def test_is_sdapp_halted_yes(os_release_fixture, version_contains):
     """
     output = bytes(
         "NAME     STATE     CLASS     LABEL     TEMPLATE\nsd-app"
-        "    Halted    AppVM   yellow     sd-small-{}-template\n".format(DEBIAN_VERSION),
+        f"    Halted    AppVM   yellow     sd-small-{DEBIAN_VERSION}-template\n",
         "utf-8",
     )
 
@@ -280,7 +280,7 @@ def test_is_sdapp_halted_no(os_release_fixture, version_contains):
     """
     output = bytes(
         "NAME     STATE     CLASS     LABEL     TEMPLATE\nsd-app"
-        "    Paused    AppVM   yellow     sd-small-{}-template\n".format(DEBIAN_VERSION),
+        f"    Paused    AppVM   yellow     sd-small-{DEBIAN_VERSION}-template\n",
         "utf-8",
     )
 

scripts/configure-environment.py

@@ -3,6 +3,7 @@
 Updates the config.json in-place in dom0 to set the environment to 'dev' or
 'staging'.
 """
+
 import argparse
 import json
 import os
@@ -27,7 +28,7 @@ def parse_args():
     )
     args = parser.parse_args()
     if not os.path.exists(args.config):
-        msg = "Config file not found: {}\n".format(args.config)
+        msg = f"Config file not found: {args.config}\n"
         sys.stderr.write(msg)
         parser.print_help(sys.stderr)
         sys.exit(1)
@@ -39,14 +40,14 @@ def parse_args():
 
 
 def set_env_in_config(args):
-    with open(args.config, "r") as f:
+    with open(args.config) as f:
         old_config = json.load(f)
 
     new_config = dict(old_config)
     new_config["environment"] = args.environment
 
     if new_config != old_config:
-        msg = "Updated config environment to '{}'...\n".format(args.environment)
+        msg = f"Updated config environment to '{args.environment}'...\n"
         sys.stderr.write(msg)
 
         with open(args.config, "w") as f:

scripts/fake-setarch.py

@@ -6,4 +6,4 @@
 from sys import argv
 
 if "sh" in argv:
-    run(argv[argv.index("sh") :])
+    run(argv[argv.index("sh") :], check=False)