Add ruff

Signed-off-by: Stephen Finucane <stephenfin@redhat.com>
Change-Id: Ia1d7e0af838401517c6ed2073b6e4ce2bed3ad27

.pre-commit-config.yaml

@@ -3,7 +3,6 @@ repos:
     rev: v6.0.0
     hooks:
       - id: trailing-whitespace
-      # Replaces or checks mixed line ending
       - id: mixed-line-ending
         args: ['--fix', 'lf']
         exclude: '.*\.(svg)$'
@@ -13,18 +12,15 @@ repos:
       - id: debug-statements
       - id: check-yaml
         files: .*\.(yaml|yml)$
+  - repo: https://github.com/astral-sh/ruff-pre-commit
+    rev: v0.14.8
+    hooks:
+      - id: ruff-check
+        args: ['--fix', '--unsafe-fixes']
+      - id: ruff-format
   - repo: https://opendev.org/openstack/hacking
-    rev: 7.0.0
+    rev: 8.0.0
     hooks:
       - id: hacking
         additional_dependencies: []
-  - repo: https://github.com/PyCQA/bandit
-    rev: 1.8.6
-    hooks:
-      - id: bandit
-        args: ['-c', 'pyproject.toml']
-  - repo: https://github.com/asottile/pyupgrade
-    rev: v3.20.0
-    hooks:
-      - id: pyupgrade
-        args: [--py310-plus]
+        exclude: '^(doc|releasenotes)/.*$'

benchmark/benchmark.py

@@ -27,10 +27,12 @@ num_iterations = 100
 
 
 def run_plain(cmd):
-    obj = subprocess.Popen(cmd,
-                           stdin=subprocess.PIPE,
-                           stdout=subprocess.PIPE,
-                           stderr=subprocess.PIPE)
+    obj = subprocess.Popen(
+        cmd,
+        stdin=subprocess.PIPE,
+        stdout=subprocess.PIPE,
+        stderr=subprocess.PIPE,
+    )
     out, err = obj.communicate()
     out = os.fsdecode(out)
     err = os.fsdecode(err)
@@ -42,14 +44,27 @@ def run_sudo(cmd):
 
 
 def run_rootwrap(cmd):
-    return run_plain([
-        "sudo", sys.executable, "-c",
-        "from oslo_rootwrap import cmd; cmd.main()", config_path] + cmd)
+    return run_plain(
+        [
+            "sudo",
+            sys.executable,
+            "-c",
+            "from oslo_rootwrap import cmd; cmd.main()",
+            config_path,
+        ]
+        + cmd
+    )
 
 
-run_daemon = client.Client([
-    "sudo", sys.executable, "-c",
-    "from oslo_rootwrap import cmd; cmd.daemon()", config_path]).execute
+run_daemon = client.Client(
+    [
+        "sudo",
+        sys.executable,
+        "-c",
+        "from oslo_rootwrap import cmd; cmd.daemon()",
+        config_path,
+    ]
+).execute
 
 
 def run_one(runner, cmd):
@@ -57,6 +72,7 @@ def run_one(runner, cmd):
         code, out, err = runner(cmd)
         assert err == "", "Stderr not empty:\n" + err
         assert code == 0, "Command failed"
+
     return __inner
 
 
@@ -81,17 +97,22 @@ def run_bench(cmd, runners):
     strcmd = ' '.join(cmd)
     max_name_len = max(len(name) for name, _ in runners) + len(strcmd) - 3
     print(f"Running '{strcmd}':")
-    print("{0:^{1}} :".format("method", max_name_len),
-          "".join(map("{:^10}".format, ["min", "avg", "max", "dev"])))
+    print(
+        "{0:^{1}} :".format("method", max_name_len),
+        "".join(map("{:^10}".format, ["min", "avg", "max", "dev"])),
+    )
     for name, runner in runners:
-        results = timeit.repeat(run_one(runner, cmd), repeat=num_iterations,
-                                number=1)
+        results = timeit.repeat(
+            run_one(runner, cmd), repeat=num_iterations, number=1
+        )
         avg = sum(results) / num_iterations
         min_ = min(results)
         max_ = max(results)
         dev = math.sqrt(sum((r - avg) ** 2 for r in results) / num_iterations)
-        print("{0:>{1}} :".format(name.format(strcmd), max_name_len),
-              " ".join(map(get_time_string, [min_, avg, max_, dev])))
+        print(
+            "{0:>{1}} :".format(name.format(strcmd), max_name_len),
+            " ".join(map(get_time_string, [min_, avg, max_, dev])),
+        )
 
 
 def main():

doc/source/conf.py

@@ -64,15 +64,15 @@ pygments_style = 'sphinx'
 # html_static_path = ['static']
 html_theme = 'openstackdocs'
 
-# Output file base name for HTML help builder.
-htmlhelp_basename = '%sdoc' % project
-
 # Grouping the document tree into LaTeX files. List of tuples
 # (source start file, target name, title, author, documentclass
 # [howto/manual]).
 latex_documents = [
-    ('index',
-     '%s.tex' % project,
-     '%s Documentation' % project,
-     'OpenStack Foundation', 'manual'),
+    (
+        'index',
+        f'{project}.tex',
+        f'{project} Documentation',
+        'OpenStack Foundation',
+        'manual',
+    ),
 ]

oslo_rootwrap/__init__.py

@@ -23,12 +23,14 @@ except ImportError:
     _patched_socket = False
 else:
     # In tests patching happens later, so we'll rely on environment variable
-    _patched_socket = (eventlet.patcher.is_monkey_patched('socket') or
-                       os.environ.get('TEST_EVENTLET', False))
+    _patched_socket = eventlet.patcher.is_monkey_patched(
+        'socket'
+    ) or os.environ.get('TEST_EVENTLET', False)
 
 if not _patched_socket:
     import subprocess
 else:
     debtcollector.deprecate(
-        "Eventlet support is deprecated and will be soon removed.")
-    from eventlet.green import subprocess   # noqa
+        "Eventlet support is deprecated and will be soon removed."
+    )
+    from eventlet.green import subprocess  # noqa

oslo_rootwrap/client.py

@@ -36,9 +36,12 @@ if oslo_rootwrap._patched_socket:
 try:
     finalize = weakref.finalize
 except AttributeError:
+
     def finalize(obj, func, *args, **kwargs):
-        return mp_util.Finalize(obj, func, args=args, kwargs=kwargs,
-                                exitpriority=0)
+        return mp_util.Finalize(
+            obj, func, args=args, kwargs=kwargs, exitpriority=0
+        )
+
 
 ClientManager = daemon.get_manager_class()
 LOG = logging.getLogger(__name__)
@@ -62,17 +65,22 @@ class Client:
 
     def _initialize(self):
         if self._process is not None and self._process.poll() is not None:
-            LOG.warning("Leaving behind already spawned process with pid %d, "
-                        "root should kill it if it's still there (I can't)",
-                        self._process.pid)
+            LOG.warning(
+                "Leaving behind already spawned process with pid %d, "
+                "root should kill it if it's still there (I can't)",
+                self._process.pid,
+            )
 
-        process_obj = subprocess.Popen(self._start_command,
-                                       stdin=subprocess.PIPE,
-                                       stdout=subprocess.PIPE,
-                                       stderr=subprocess.PIPE,
-                                       close_fds=True)
-        LOG.debug("Popen for %s command has been instantiated",
-                  self._start_command)
+        process_obj = subprocess.Popen(
+            self._start_command,
+            stdin=subprocess.PIPE,
+            stdout=subprocess.PIPE,
+            stderr=subprocess.PIPE,
+            close_fds=True,
+        )
+        LOG.debug(
+            "Popen for %s command has been instantiated", self._start_command
+        )
 
         self._process = process_obj
         socket_path = process_obj.stdout.readline()[:-1]
@@ -83,15 +91,18 @@ class Client:
         if process_obj.poll() is not None:
             stderr = process_obj.stderr.read()
             # NOTE(yorik-sar): don't expose stdout here
-            raise Exception("Failed to spawn rootwrap process.\nstderr:\n%s" %
-                            (stderr,))
-        LOG.info("Spawned new rootwrap daemon process with pid=%d",
-                 process_obj.pid)
+            raise Exception(
+                f"Failed to spawn rootwrap process.\nstderr:\n{stderr}"
+            )
+        LOG.info(
+            "Spawned new rootwrap daemon process with pid=%d", process_obj.pid
+        )
 
         def wait_process():
             return_code = process_obj.wait()
-            LOG.info("Rootwrap daemon process exit with status: %d",
-                     return_code)
+            LOG.info(
+                "Rootwrap daemon process exit with status: %d", return_code
+            )
 
         reap_process = threading.Thread(target=wait_process)
         reap_process.daemon = True
@@ -99,8 +110,9 @@ class Client:
         self._manager = ClientManager(socket_path, authkey)
         self._manager.connect()
         self._proxy = self._manager.rootwrap()
-        self._finalize = finalize(self, self._shutdown, self._process,
-                                  self._manager)
+        self._finalize = finalize(
+            self, self._shutdown, self._process, self._manager
+        )
         self._initialized = True
 
     @staticmethod
@@ -108,8 +120,9 @@ class Client:
         # Storing JsonClient in arguments because globals are set to None
         # before executing atexit routines in Python 2.x
         if process.poll() is None:
-            LOG.info('Stopping rootwrap daemon process with pid=%s',
-                     process.pid)
+            LOG.info(
+                'Stopping rootwrap daemon process with pid=%s', process.pid
+            )
             for _ in range(SHUTDOWN_RETRIES):
                 try:
                     manager.rootwrap().shutdown()

oslo_rootwrap/cmd.py

@@ -15,20 +15,21 @@
 
 """Root wrapper for OpenStack services
 
-   Filters which commands a service is allowed to run as another user.
+Filters which commands a service is allowed to run as another user.
 
-   To use this with oslo, you should set the following in
-   oslo.conf:
-   rootwrap_config=/etc/oslo/rootwrap.conf
+To use this with oslo, you should set the following in
+oslo.conf:
+rootwrap_config=/etc/oslo/rootwrap.conf
 
-   You also need to let the oslo user run oslo-rootwrap
-   as root in sudoers:
-   oslo ALL = (root) NOPASSWD: /usr/bin/oslo-rootwrap
-                                   /etc/oslo/rootwrap.conf *
+You also need to let the oslo user run oslo-rootwrap
+as root in sudoers:
+oslo ALL = (root) NOPASSWD: /usr/bin/oslo-rootwrap
+                                /etc/oslo/rootwrap.conf *
 
-   Service packaging should deploy .filters files only on nodes where
-   they are needed, to avoid allowing more than is necessary.
+Service packaging should deploy .filters files only on nodes where
+they are needed, to avoid allowing more than is necessary.
 """
+
 import configparser
 import logging
 import os
@@ -43,6 +44,8 @@ try:
 except ImportError:
     resource = None
 
+LOG = logging.getLogger(__name__)
+
 RC_UNAUTHORIZED = 99
 RC_NOCOMMAND = 98
 RC_BADCONFIG = 97
@@ -53,7 +56,7 @@ SIGNAL_BASE = 128
 def _exit_error(execname, message, errorcode, log=True):
     print(f"{execname}: {message}", file=sys.stderr)
     if log:
-        logging.error(message)
+        LOG.error(message)
     sys.exit(errorcode)
 
 
@@ -66,12 +69,14 @@ def main(run_daemon=False):
     execname = sys.argv.pop(0)
     if run_daemon:
         if len(sys.argv) != 1:
-            _exit_error(execname, "Extra arguments to daemon", RC_NOCOMMAND,
-                        log=False)
+            _exit_error(
+                execname, "Extra arguments to daemon", RC_NOCOMMAND, log=False
+            )
     else:
         if len(sys.argv) < 2:
-            _exit_error(execname, "No command specified", RC_NOCOMMAND,
-                        log=False)
+            _exit_error(
+                execname, "No command specified", RC_NOCOMMAND, log=False
+            )
 
     configfile = sys.argv.pop(0)
 
@@ -84,8 +89,12 @@ def main(run_daemon=False):
         msg = f"Incorrect value in {configfile}: {exc.args[0]}"
         _exit_error(execname, msg, RC_BADCONFIG, log=False)
     except configparser.Error:
-        _exit_error(execname, "Incorrect configuration file: %s" % configfile,
-                    RC_BADCONFIG, log=False)
+        _exit_error(
+            execname,
+            f"Incorrect configuration file: {configfile}",
+            RC_BADCONFIG,
+            log=False,
+        )
 
     if resource:
         # When use close_fds=True on Python 2.x, calling subprocess with
@@ -96,7 +105,7 @@ def main(run_daemon=False):
         # Lower our ulimit to a reasonable value to regain performance.
         fd_limits = resource.getrlimit(resource.RLIMIT_NOFILE)
         sensible_fd_limit = min(config.rlimit_nofile, fd_limits[0])
-        if (fd_limits[0] > sensible_fd_limit):
+        if fd_limits[0] > sensible_fd_limit:
             # Close any fd beyond sensible_fd_limit prior adjusting our
             # rlimit to ensure all fds are closed
             for fd_entry in os.listdir('/proc/self/fd'):
@@ -109,18 +118,21 @@ def main(run_daemon=False):
             # Unfortunately this inherits to our children, so allow them to
             # re-raise by passing through the hard limit unmodified
             resource.setrlimit(
-                resource.RLIMIT_NOFILE, (sensible_fd_limit, fd_limits[1]))
+                resource.RLIMIT_NOFILE, (sensible_fd_limit, fd_limits[1])
+            )
             # This is set on import to the hard ulimit. if its defined we
             # already have imported it, so we need to update it to the new
             # limit.
-            if (hasattr(subprocess, 'MAXFD') and
-                    subprocess.MAXFD > sensible_fd_limit):
+            if (
+                hasattr(subprocess, 'MAXFD')
+                and subprocess.MAXFD > sensible_fd_limit
+            ):
                 subprocess.MAXFD = sensible_fd_limit
 
     if config.use_syslog:
-        wrapper.setup_syslog(execname,
-                             config.syslog_log_facility,
-                             config.syslog_log_level)
+        wrapper.setup_syslog(
+            execname, config.syslog_log_facility, config.syslog_log_level
+        )
 
     filters = wrapper.load_filters(config.filters_path)
 
@@ -129,6 +141,7 @@ def main(run_daemon=False):
         # slows us down just a bit. So moving it here so we have
         # it only when we need it.
         from oslo_rootwrap import daemon as daemon_mod
+
         daemon_mod.daemon_start(config, filters)
     else:
         run_one_command(execname, config, filters, sys.argv)
@@ -138,24 +151,27 @@ def run_one_command(execname, config, filters, userargs):
     # Execute command if it matches any of the loaded filters
     try:
         obj = wrapper.start_subprocess(
-            filters, userargs,
+            filters,
+            userargs,
             exec_dirs=config.exec_dirs,
             log=config.use_syslog,
             stdin=sys.stdin,
             stdout=sys.stdout,
-            stderr=sys.stderr)
+            stderr=sys.stderr,
+        )
         returncode = obj.wait()
         # Fix returncode of Popen
         if returncode < 0:
             returncode = SIGNAL_BASE - returncode
         sys.exit(returncode)
-
     except wrapper.FilterMatchNotExecutable as exc:
-        msg = ("Executable not found: %s (filter match = %s)"
-               % (exc.match.exec_path, exc.match.name))
+        msg = (
+            f"Executable not found: {exc.match.exec_path} "
+            f"(filter match = {exc.match.name})"
+        )
         _exit_error(execname, msg, RC_NOEXECFOUND, log=config.use_syslog)
-
     except wrapper.NoFilterMatched:
-        msg = ("Unauthorized command: %s (no filter matched)"
-               % ' '.join(userargs))
+        msg = "Unauthorized command: {} (no filter matched)".format(
+            ' '.join(userargs)
+        )
         _exit_error(execname, msg, RC_UNAUTHORIZED, log=config.use_syslog)

oslo_rootwrap/daemon.py

@@ -49,21 +49,24 @@ class RootwrapClass:
         self.reset_timer()
         try:
             obj = wrapper.start_subprocess(
-                self.filters, userargs,
+                self.filters,
+                userargs,
                 exec_dirs=self.config.exec_dirs,
                 log=self.config.use_syslog,
                 close_fds=True,
                 stdin=subprocess.PIPE,
                 stdout=subprocess.PIPE,
-                stderr=subprocess.PIPE)
+                stderr=subprocess.PIPE,
+            )
         except wrapper.FilterMatchNotExecutable:
-            LOG.warning("Executable not found for: %s",
-                        ' '.join(userargs))
+            LOG.warning("Executable not found for: %s", ' '.join(userargs))
             return cmd.RC_NOEXECFOUND, "", ""
 
         except wrapper.NoFilterMatched:
-            LOG.warning("Unauthorized command: %s (no filter matched)",
-                        ' '.join(userargs))
+            LOG.warning(
+                "Unauthorized command: %s (no filter matched)",
+                ' '.join(userargs),
+            )
             return cmd.RC_UNAUTHORIZED, "", ""
 
         if stdin is not None:
@@ -89,9 +92,9 @@ class RootwrapClass:
         if config is not None:
             cls.daemon_timeout = config.daemon_timeout
         # Wait a bit longer to avoid rounding errors
-        timeout = max(
-            cls.last_called + cls.daemon_timeout - time.time(),
-            0) + 1
+        timeout = (
+            max(cls.last_called + cls.daemon_timeout - time.time(), 0) + 1
+        )
         if getattr(cls, 'timeout', None):
             # Another timer is already initialized
             return
@@ -115,8 +118,7 @@ def get_manager_class(config=None, filters=None):
     class RootwrapManager(managers.BaseManager):
         def __init__(self, address=None, authkey=None):
             # Force jsonrpc because neither pickle nor xmlrpclib is secure
-            super().__init__(address, authkey,
-                             serializer='jsonrpc')
+            super().__init__(address, authkey, serializer='jsonrpc')
 
     if config is not None:
         partial_class = functools.partial(RootwrapClass, config, filters)
@@ -132,9 +134,13 @@ def daemon_start(config, filters):
     LOG.debug("Created temporary directory %s", temp_dir)
     try:
         # allow everybody to find the socket
-        rwxr_xr_x = (stat.S_IRWXU |
-                     stat.S_IRGRP | stat.S_IXGRP |
-                     stat.S_IROTH | stat.S_IXOTH)
+        rwxr_xr_x = (
+            stat.S_IRWXU
+            | stat.S_IRGRP
+            | stat.S_IXGRP
+            | stat.S_IROTH
+            | stat.S_IXOTH
+        )
         os.chmod(temp_dir, rwxr_xr_x)
         socket_path = os.path.join(temp_dir, "rootwrap.sock")
         LOG.debug("Will listen on socket %s", socket_path)
@@ -143,9 +149,14 @@ def daemon_start(config, filters):
         server = manager.get_server()
         try:
             # allow everybody to connect to the socket
-            rw_rw_rw_ = (stat.S_IRUSR | stat.S_IWUSR |
-                         stat.S_IRGRP | stat.S_IWGRP |
-                         stat.S_IROTH | stat.S_IWOTH)
+            rw_rw_rw_ = (
+                stat.S_IRUSR
+                | stat.S_IWUSR
+                | stat.S_IRGRP
+                | stat.S_IWGRP
+                | stat.S_IROTH
+                | stat.S_IWOTH
+            )
             os.chmod(socket_path, rw_rw_rw_)
             try:
                 # In Python 3 we have to use buffer to push in bytes directly

oslo_rootwrap/filters.py

@@ -101,11 +101,11 @@ class RegExpFilter(CommandFilter):
 
     def match(self, userargs):
         # Early skip if command or number of args don't match
-        if (not userargs or len(self.args) != len(userargs)):
+        if not userargs or len(self.args) != len(userargs):
             # DENY: argument numbers don't match
             return False
         # Compare each arg (anchoring pattern explicitly at end of string)
-        for (pattern, arg) in zip(self.args, userargs):
+        for pattern, arg in zip(self.args, userargs):
             try:
                 if not re.match(pattern + '$', arg):
                     # DENY: Some arguments did not match
@@ -120,14 +120,14 @@ class RegExpFilter(CommandFilter):
 class PathFilter(CommandFilter):
     """Command filter checking that path arguments are within given dirs
 
-        One can specify the following constraints for command arguments:
-            1) pass     - pass an argument as is to the resulting command
-            2) some_str - check if an argument is equal to the given string
-            3) abs path - check if a path argument is within the given base dir
+    One can specify the following constraints for command arguments:
+        1) pass     - pass an argument as is to the resulting command
+        2) some_str - check if an argument is equal to the given string
+        3) abs path - check if a path argument is within the given base dir
 
-        A typical rootwrapper filter entry looks like this:
-            # cmdname: filter name, raw command, user, arg_i_constraint [, ...]
-            chown: PathFilter, /bin/chown, root, nova, /var/lib/images
+    A typical rootwrapper filter entry looks like this:
+        # cmdname: filter name, raw command, user, arg_i_constraint [, ...]
+        chown: PathFilter, /bin/chown, root, nova, /var/lib/images
 
     """
 
@@ -150,33 +150,36 @@ class PathFilter(CommandFilter):
             if os.path.isabs(arg)  # arguments specifying abs paths
         )
 
-        return (equal_args_num and
-                exec_is_valid and
-                args_equal_or_pass and
-                paths_are_within_base_dirs)
+        return (
+            equal_args_num
+            and exec_is_valid
+            and args_equal_or_pass
+            and paths_are_within_base_dirs
+        )
 
     def get_command(self, userargs, exec_dirs=None):
         exec_dirs = exec_dirs or []
         command, arguments = userargs[0], userargs[1:]
 
         # convert path values to canonical ones; copy other args as is
-        args = [realpath(value) if os.path.isabs(arg) else value
-                for arg, value in zip(self.args, arguments)]
+        args = [
+            realpath(value) if os.path.isabs(arg) else value
+            for arg, value in zip(self.args, arguments)
+        ]
 
-        return super().get_command([command] + args,
-                                   exec_dirs)
+        return super().get_command([command] + args, exec_dirs)
 
 
 class KillFilter(CommandFilter):
     """Specific filter for the kill calls.
 
-       1st argument is the user to run /bin/kill under
-       2nd argument is the location of the affected executable
-           if the argument is not absolute, it is checked against $PATH
-       Subsequent arguments list the accepted signals (if any)
+    1st argument is the user to run /bin/kill under
+    2nd argument is the location of the affected executable
+        if the argument is not absolute, it is checked against $PATH
+    Subsequent arguments list the accepted signals (if any)
 
-       This filter relies on /proc to accurately determine affected
-       executable, so it will only work on procfs-capable systems (not OSX).
+    This filter relies on /proc to accurately determine affected
+    executable, so it will only work on procfs-capable systems (not OSX).
     """
 
     def __init__(self, *args):
@@ -208,7 +211,7 @@ class KillFilter(CommandFilter):
         """Determine the program associated with pid"""
 
         try:
-            command = os.readlink("/proc/%d/exe" % int(pid))
+            command = os.readlink(f"/proc/{int(pid)}/exe")
         except (ValueError, OSError):
             # Incorrect PID
             return None
@@ -221,7 +224,7 @@ class KillFilter(CommandFilter):
         # NOTE(dprince): /proc/PID/exe may have ' (deleted)' on
         # the end if an executable is updated or deleted
         if command.endswith(" (deleted)"):
-            command = command[:-len(" (deleted)")]
+            command = command[: -len(" (deleted)")]
 
         if os.path.isfile(command):
             return command
@@ -230,7 +233,7 @@ class KillFilter(CommandFilter):
         # a ';......' or '.#prelink#......' suffix etc.
         # So defer to /proc/PID/cmdline in that case.
         try:
-            with open("/proc/%d/cmdline" % int(pid)) as pfile:
+            with open(f"/proc/{int(pid)}/cmdline") as pfile:
                 cmdline = pfile.read().partition('\0')[0]
 
             cmdline = self._program_path(cmdline)
@@ -271,10 +274,12 @@ class KillFilter(CommandFilter):
         if os.path.isabs(kill_command):
             return kill_command == command
 
-        return (os.path.isabs(command) and
-                kill_command == os.path.basename(command) and
-                os.path.dirname(command) in os.environ.get('PATH', ''
-                                                           ).split(':'))
+        return (
+            os.path.isabs(command)
+            and kill_command == os.path.basename(command)
+            and os.path.dirname(command)
+            in os.environ.get('PATH', '').split(':')
+        )
 
 
 class ReadFileFilter(CommandFilter):
@@ -285,7 +290,7 @@ class ReadFileFilter(CommandFilter):
         super().__init__("/bin/cat", "root", *args)
 
     def match(self, userargs):
-        return (userargs == ['cat', self.file_path])
+        return userargs == ['cat', self.file_path]
 
 
 class IpFilter(CommandFilter):
@@ -339,11 +344,14 @@ class EnvFilter(CommandFilter):
         # extract all env args
         user_envs = self._extract_env(userargs)
         filter_envs = self._extract_env(self.args)
-        user_command = userargs[len(user_envs):len(user_envs) + 1]
+        user_command = userargs[len(user_envs) : len(user_envs) + 1]
 
         # match first non-env argument with CommandFilter
-        return (super().match(user_command) and
-                len(filter_envs) and user_envs == filter_envs)
+        return (
+            super().match(user_command)
+            and len(filter_envs)
+            and user_envs == filter_envs
+        )
 
     def exec_args(self, userargs):
         args = userargs[:]
@@ -394,8 +402,11 @@ class IpNetnsExecFilter(ChainingFilter):
         if self.run_as != "root" or len(userargs) < 4:
             return False
 
-        return (userargs[0] == 'ip' and userargs[1] in NETNS_VARS and
-                userargs[2] in EXEC_VARS)
+        return (
+            userargs[0] == 'ip'
+            and userargs[1] in NETNS_VARS
+            and userargs[2] in EXEC_VARS
+        )
 
     def exec_args(self, userargs):
         args = userargs[4:]
@@ -413,10 +424,10 @@ class ChainingRegExpFilter(ChainingFilter):
 
     def match(self, userargs):
         # Early skip if number of args is smaller than the filter
-        if (not userargs or len(self.args) > len(userargs)):
+        if not userargs or len(self.args) > len(userargs):
             return False
         # Compare each arg (anchoring pattern explicitly at end of string)
-        for (pattern, arg) in zip(self.args, userargs):
+        for pattern, arg in zip(self.args, userargs):
             try:
                 if not re.match(pattern + '$', arg):
                     # DENY: Some arguments did not match
@@ -428,7 +439,7 @@ class ChainingRegExpFilter(ChainingFilter):
         return True
 
     def exec_args(self, userargs):
-        args = userargs[len(self.args):]
+        args = userargs[len(self.args) :]
         if args:
             args[0] = os.path.basename(args[0])
         return args

oslo_rootwrap/jsonrpc.py

@@ -35,8 +35,10 @@ class RpcJSONEncoder(json.JSONEncoder):
         if isinstance(o, wrapper.NoFilterMatched):
             return {"__exception__": "NoFilterMatched"}
         elif isinstance(o, wrapper.FilterMatchNotExecutable):
-            return {"__exception__": "FilterMatchNotExecutable",
-                    "match": o.match}
+            return {
+                "__exception__": "FilterMatchNotExecutable",
+                "match": o.match,
+            }
         # Other errors will fail to pass JSON encoding and will be visible on
         # client side
         else:
@@ -105,6 +107,7 @@ if hasattr(managers.Server, 'accepter'):
             old_accepter(self)
         except EOFError:
             pass
+
     old_accepter = managers.Server.accepter
     managers.Server.accepter = silent_accepter
 
@@ -179,8 +182,9 @@ class JsonConnection:
             pass
         else:
             # In Python 2 json returns unicode while multiprocessing needs str
-            if (kind in ("#TRACEBACK", "#UNSERIALIZABLE") and
-                    not isinstance(res[1], str)):
+            if kind in ("#TRACEBACK", "#UNSERIALIZABLE") and not isinstance(
+                res[1], str
+            ):
                 res[1] = res[1].encode('utf-8', 'replace')
         return res
 

oslo_rootwrap/tests/run_daemon.py

@@ -36,6 +36,7 @@ def forwarding_popen(f, old_popen=subprocess.Popen):
         t.daemon = True
         t.start()
         return p
+
     return popen
 
 
@@ -50,8 +51,7 @@ class nonclosing:
         pass
 
 
-log_format = ("%(asctime)s | [%(process)5s]+%(levelname)5s | "
-              "%(message)s")
+log_format = "%(asctime)s | [%(process)5s]+%(levelname)5s | %(message)s"
 if __name__ == '__main__':
     logging.basicConfig(level=logging.DEBUG, format=log_format)
     sys.stderr = nonclosing(sys.stderr)

oslo_rootwrap/tests/test_functional.py

@@ -51,19 +51,19 @@ class _FunctionalBase:
         filters_file = os.path.join(tmpdir, 'filters.d', 'test.filters')
         os.mkdir(filters_dir)
         with open(self.config_file, 'w') as f:
-            f.write("""[DEFAULT]
-filters_path={}
+            f.write(f"""[DEFAULT]
+filters_path={filters_dir}
 daemon_timeout=10
-exec_dirs=/bin""".format(filters_dir))
+exec_dirs=/bin""")
         with open(filters_file, 'w') as f:
-            f.write("""[Filters]
+            f.write(f"""[Filters]
 echo: CommandFilter, /bin/echo, root
 cat: CommandFilter, /bin/cat, root
 sh: CommandFilter, /bin/sh, root
 id: CommandFilter, /usr/bin/id, nobody
 unknown_cmd: CommandFilter, /unknown/unknown_cmd, root
-later_install_cmd: CommandFilter, %s, root
-""" % self.later_cmd)
+later_install_cmd: CommandFilter, {self.later_cmd}, root
+""")
 
     def _test_run_once(self, expect_byte=True):
         code, out, err = self.execute(['echo', 'teststr'])
@@ -122,9 +122,11 @@ class RootwrapTest(_FunctionalBase, testtools.TestCase):
     def setUp(self):
         super().setUp()
         self.cmd = [
-            sys.executable, '-c',
+            sys.executable,
+            '-c',
             'from oslo_rootwrap import cmd; cmd.main()',
-            self.config_file]
+            self.config_file,
+        ]
 
     def execute(self, cmd, stdin=None):
         proc = subprocess.Popen(
@@ -134,10 +136,12 @@ class RootwrapTest(_FunctionalBase, testtools.TestCase):
             stderr=subprocess.PIPE,
         )
         out, err = proc.communicate(stdin)
-        self.addDetail('stdout',
-                       content.text_content(out.decode('utf-8', 'replace')))
-        self.addDetail('stderr',
-                       content.text_content(err.decode('utf-8', 'replace')))
+        self.addDetail(
+            'stdout', content.text_content(out.decode('utf-8', 'replace'))
+        )
+        self.addDetail(
+            'stderr', content.text_content(err.decode('utf-8', 'replace'))
+        )
         return proc.returncode, out, err
 
     def test_run_once(self):
@@ -151,8 +155,10 @@ class RootwrapDaemonTest(_FunctionalBase, testtools.TestCase):
     def assert_unpatched(self):
         # We need to verify that these tests are run without eventlet patching
         if eventlet and eventlet.patcher.is_monkey_patched('socket'):
-            self.fail("Standard library should not be patched by eventlet"
-                      " for this test")
+            self.fail(
+                "Standard library should not be patched by eventlet"
+                " for this test"
+            )
 
     def setUp(self):
         self.assert_unpatched()
@@ -161,8 +167,10 @@ class RootwrapDaemonTest(_FunctionalBase, testtools.TestCase):
 
         # Collect daemon logs
         daemon_log = io.BytesIO()
-        p = mock.patch('oslo_rootwrap.subprocess.Popen',
-                       run_daemon.forwarding_popen(daemon_log))
+        p = mock.patch(
+            'oslo_rootwrap.subprocess.Popen',
+            run_daemon.forwarding_popen(daemon_log),
+        )
         p.start()
         self.addCleanup(p.stop)
 
@@ -179,17 +187,24 @@ class RootwrapDaemonTest(_FunctionalBase, testtools.TestCase):
         # Add all logs as details
         @self.addCleanup
         def add_logs():
-            self.addDetail('daemon_log', content.Content(
-                content.UTF8_TEXT,
-                lambda: [daemon_log.getvalue()]))
-            self.addDetail('client_log', content.Content(
-                content.UTF8_TEXT,
-                lambda: [client_log.getvalue().encode('utf-8')]))
+            self.addDetail(
+                'daemon_log',
+                content.Content(
+                    content.UTF8_TEXT, lambda: [daemon_log.getvalue()]
+                ),
+            )
+            self.addDetail(
+                'client_log',
+                content.Content(
+                    content.UTF8_TEXT,
+                    lambda: [client_log.getvalue().encode('utf-8')],
+                ),
+            )
 
         # Create client
-        self.client = client.Client([
-            sys.executable, run_daemon.__file__,
-            self.config_file])
+        self.client = client.Client(
+            [sys.executable, run_daemon.__file__, self.config_file]
+        )
 
         # _finalize is set during Client.execute()
         @self.addCleanup
@@ -235,8 +250,9 @@ class RootwrapDaemonTest(_FunctionalBase, testtools.TestCase):
         try:
             # Run a shell script that signals calling process through FIFO and
             # then hangs around for 1 sec
-            self._thread_res = self.execute([
-                'sh', '-c', 'echo > "%s"; sleep 1; echo OK' % fifo_path])
+            self._thread_res = self.execute(
+                ['sh', '-c', f'echo > "{fifo_path}"; sleep 1; echo OK']
+            )
         except Exception as e:
             self._thread_res = e
 
@@ -280,17 +296,23 @@ class RootwrapDaemonTest(_FunctionalBase, testtools.TestCase):
             stop.wait(3)
             if not stop.is_set():
                 os.kill(process.pid, signal.SIGKILL)
+
         threading.Thread(target=sleep_kill).start()
         # Wait for process to finish one way or another
         self.client._process.wait()
         # Notify background thread that process is dead (no need to kill it)
         stop.set()
         # Fail if the process got killed by the background thread
-        self.assertNotEqual(-signal.SIGKILL, process.returncode,
-                            "Server haven't stopped in one second")
+        self.assertNotEqual(
+            -signal.SIGKILL,
+            process.returncode,
+            "Server haven't stopped in one second",
+        )
         # Verify that socket is deleted
-        self.assertFalse(os.path.exists(socket_path),
-                         "Server didn't remove its temporary directory")
+        self.assertFalse(
+            os.path.exists(socket_path),
+            "Server didn't remove its temporary directory",
+        )
 
     def test_daemon_cleanup_client(self):
         # Run _test_daemon_cleanup stopping daemon as Client instance would

oslo_rootwrap/tests/test_functional_eventlet.py

@@ -17,6 +17,7 @@ import os
 
 if os.environ.get('TEST_EVENTLET', False):
     import eventlet
+
     eventlet.monkey_patch()
 
     from oslo_rootwrap.tests import test_functional
@@ -28,7 +29,8 @@ if os.environ.get('TEST_EVENTLET', False):
 
         def _thread_worker(self, seconds, msg):
             code, out, err = self.execute(
-                ['sh', '-c', 'sleep %d; echo %s' % (seconds, msg)])
+                ['sh', '-c', f'sleep {seconds}; echo {msg}']
+            )
             # Ignore trailing newline
             self.assertEqual(msg, out.rstrip())
 
@@ -50,10 +52,13 @@ if os.environ.get('TEST_EVENTLET', False):
             # 10 was not enough for some reason.
             for i in range(15):
                 th.append(
-                    eventlet.spawn(self._thread_worker, i % 3, 'abc%d' % i))
+                    eventlet.spawn(self._thread_worker, i % 3, f'abc{i}')
+                )
             for i in [5, 17, 20, 25]:
                 th.append(
-                    eventlet.spawn(self._thread_worker_timeout, 2,
-                                   'timeout%d' % i, i))
+                    eventlet.spawn(
+                        self._thread_worker_timeout, 2, f'timeout{i}', i
+                    )
+                )
             for thread in th:
                 thread.wait()

oslo_rootwrap/tests/test_rootwrap.py

@@ -31,7 +31,6 @@ from oslo_rootwrap import wrapper
 
 
 class RootwrapLoaderTestCase(testtools.TestCase):
-
     def test_privsep_in_loader(self):
         privsep = ["privsep-helper", "--context", "foo"]
         filterlist = wrapper.load_filters([])
@@ -42,8 +41,10 @@ class RootwrapLoaderTestCase(testtools.TestCase):
             filtermatch = wrapper.match_filter(filterlist, privsep)
 
             self.assertIsNotNone(filtermatch)
-            self.assertEqual(["/fake/privsep-helper", "--context", "foo"],
-                             filtermatch.get_command(privsep))
+            self.assertEqual(
+                ["/fake/privsep-helper", "--context", "foo"],
+                filtermatch.get_command(privsep),
+            )
 
     def test_strict_switched_off_in_configparser(self):
         temp_dir = self.useFixture(fixtures.TempDir()).path
@@ -72,7 +73,7 @@ class RootwrapTestCase(testtools.TestCase):
             filters.CommandFilter("/usr/bin/foo_bar_not_exist", "root"),
             filters.RegExpFilter("/bin/cat", "root", 'cat', '/[a-z]+'),
             filters.CommandFilter("/nonexistent/cat", "root"),
-            filters.CommandFilter("/bin/cat", "root")  # Keep this one last
+            filters.CommandFilter("/bin/cat", "root"),  # Keep this one last
         ]
 
     def test_CommandFilter(self):
@@ -105,28 +106,50 @@ class RootwrapTestCase(testtools.TestCase):
         usercmd = ["ls", "/root"]
         filtermatch = wrapper.match_filter(self.filters, usercmd)
         self.assertFalse(filtermatch is None)
-        self.assertEqual(["/bin/ls", "/root"],
-                         filtermatch.get_command(usercmd))
+        self.assertEqual(
+            ["/bin/ls", "/root"], filtermatch.get_command(usercmd)
+        )
 
     def test_RegExpFilter_reject(self):
         usercmd = ["ls", "root"]
-        self.assertRaises(wrapper.NoFilterMatched,
-                          wrapper.match_filter, self.filters, usercmd)
+        self.assertRaises(
+            wrapper.NoFilterMatched,
+            wrapper.match_filter,
+            self.filters,
+            usercmd,
+        )
 
     def test_missing_command(self):
         valid_but_missing = ["foo_bar_not_exist"]
         invalid = ["foo_bar_not_exist_and_not_matched"]
-        self.assertRaises(wrapper.FilterMatchNotExecutable,
-                          wrapper.match_filter,
-                          self.filters, valid_but_missing)
-        self.assertRaises(wrapper.NoFilterMatched,
-                          wrapper.match_filter, self.filters, invalid)
+        self.assertRaises(
+            wrapper.FilterMatchNotExecutable,
+            wrapper.match_filter,
+            self.filters,
+            valid_but_missing,
+        )
+        self.assertRaises(
+            wrapper.NoFilterMatched,
+            wrapper.match_filter,
+            self.filters,
+            invalid,
+        )
 
     def _test_EnvFilter_as_DnsMasq(self, config_file_arg):
-        usercmd = ['env', config_file_arg + '=A', 'NETWORK_ID=foobar',
-                   'dnsmasq', 'foo']
-        f = filters.EnvFilter("env", "root", config_file_arg + '=A',
-                              'NETWORK_ID=', "/usr/bin/dnsmasq")
+        usercmd = [
+            'env',
+            config_file_arg + '=A',
+            'NETWORK_ID=foobar',
+            'dnsmasq',
+            'foo',
+        ]
+        f = filters.EnvFilter(
+            "env",
+            "root",
+            config_file_arg + '=A',
+            'NETWORK_ID=',
+            "/usr/bin/dnsmasq",
+        )
         self.assertTrue(f.match(usercmd))
         self.assertEqual(['/usr/bin/dnsmasq', 'foo'], f.get_command(usercmd))
         env = f.get_environment(usercmd)
@@ -192,11 +215,14 @@ class RootwrapTestCase(testtools.TestCase):
         self.assertNotIn('sleep', env.keys())
 
     def test_KillFilter(self):
-        if not os.path.exists("/proc/%d" % os.getpid()):
+        if not os.path.exists(f"/proc/{os.getpid()}"):
             self.skipTest("Test requires /proc filesystem (procfs)")
-        p = subprocess.Popen(["cat"], stdin=subprocess.PIPE,
-                             stdout=subprocess.PIPE,
-                             stderr=subprocess.STDOUT)
+        p = subprocess.Popen(
+            ["cat"],
+            stdin=subprocess.PIPE,
+            stdout=subprocess.PIPE,
+            stderr=subprocess.STDOUT,
+        )
         try:
             f = filters.KillFilter("root", "/bin/cat", "-9", "-HUP")
             f2 = filters.KillFilter("root", "/usr/bin/cat", "-9", "-HUP")
@@ -209,9 +235,9 @@ class RootwrapTestCase(testtools.TestCase):
             self.assertFalse(f.match(usercmd) or f2.match(usercmd))
             # Providing matching signal should be allowed
             usercmd = ['kill', '-9', p.pid]
-            self.assertTrue(f.match(usercmd) or
-                            f2.match(usercmd) or
-                            f3.match(usercmd))
+            self.assertTrue(
+                f.match(usercmd) or f2.match(usercmd) or f3.match(usercmd)
+            )
 
             f = filters.KillFilter("root", "/bin/cat")
             f2 = filters.KillFilter("root", "/usr/bin/cat")
@@ -224,9 +250,9 @@ class RootwrapTestCase(testtools.TestCase):
             self.assertFalse(f.match(usercmd) or f2.match(usercmd))
             usercmd = ['kill', p.pid]
             # Providing no signal should work
-            self.assertTrue(f.match(usercmd) or
-                            f2.match(usercmd) or
-                            f3.match(usercmd))
+            self.assertTrue(
+                f.match(usercmd) or f2.match(usercmd) or f3.match(usercmd)
+            )
 
             # verify that relative paths are matched against $PATH
             f = filters.KillFilter("root", "cat")
@@ -270,8 +296,10 @@ class RootwrapTestCase(testtools.TestCase):
         with mock.patch('os.readlink') as readlink:
             readlink.return_value = command + ' (deleted)'
             with mock.patch('os.path.isfile') as exists:
+
                 def fake_exists(path):
                     return path == command
+
                 exists.side_effect = fake_exists
                 self.assertTrue(f.match(usercmd))
 
@@ -294,8 +322,9 @@ class RootwrapTestCase(testtools.TestCase):
     @mock.patch('os.path.isfile')
     @mock.patch('os.path.exists')
     @mock.patch('os.access')
-    def test_KillFilter_renamed_exe(self, mock_access, mock_exists,
-                                    mock_isfile, mock_readlink):
+    def test_KillFilter_renamed_exe(
+        self, mock_access, mock_exists, mock_isfile, mock_readlink
+    ):
         """Makes sure renamed exe's are killed correctly."""
         command = "/bin/commandddddd"
         f = filters.KillFilter("root", command)
@@ -326,8 +355,9 @@ class RootwrapTestCase(testtools.TestCase):
         self.assertTrue(f.match(['ip', 'link', 'list']))
         self.assertTrue(f.match(['ip', '-s', 'link', 'list']))
         self.assertTrue(f.match(['ip', '-s', '-v', 'netns', 'add']))
-        self.assertTrue(f.match(['ip', 'link', 'set', 'interface',
-                                 'netns', 'somens']))
+        self.assertTrue(
+            f.match(['ip', 'link', 'set', 'interface', 'netns', 'somens'])
+        )
 
     def test_IpFilter_netns(self):
         f = filters.IpFilter(self._ip, 'root')
@@ -354,7 +384,8 @@ class RootwrapTestCase(testtools.TestCase):
     def test_IpNetnsExecFilter_match(self):
         f = filters.IpNetnsExecFilter(self._ip, 'root')
         self.assertTrue(
-            f.match(['ip', 'netns', 'exec', 'foo', 'ip', 'link', 'list']))
+            f.match(['ip', 'netns', 'exec', 'foo', 'ip', 'link', 'list'])
+        )
         self.assertTrue(f.match(['ip', 'net', 'exec', 'foo', 'bar']))
         self.assertTrue(f.match(['ip', 'netn', 'e', 'foo', 'bar']))
         self.assertTrue(f.match(['ip', 'net', 'e', 'foo', 'bar']))
@@ -375,64 +406,97 @@ class RootwrapTestCase(testtools.TestCase):
     def test_IpNetnsExecFilter_nomatch_nonroot(self):
         f = filters.IpNetnsExecFilter(self._ip, 'user')
         self.assertFalse(
-            f.match(['ip', 'netns', 'exec', 'foo', 'ip', 'link', 'list']))
+            f.match(['ip', 'netns', 'exec', 'foo', 'ip', 'link', 'list'])
+        )
 
     def test_match_filter_recurses_exec_command_filter_matches(self):
-        filter_list = [filters.IpNetnsExecFilter(self._ip, 'root'),
-                       filters.IpFilter(self._ip, 'root')]
+        filter_list = [
+            filters.IpNetnsExecFilter(self._ip, 'root'),
+            filters.IpFilter(self._ip, 'root'),
+        ]
         args = ['ip', 'netns', 'exec', 'foo', 'ip', 'link', 'list']
 
         self.assertIsNotNone(wrapper.match_filter(filter_list, args))
 
     def test_match_filter_recurses_exec_command_matches_user(self):
-        filter_list = [filters.IpNetnsExecFilter(self._ip, 'root'),
-                       filters.IpFilter(self._ip, 'user')]
+        filter_list = [
+            filters.IpNetnsExecFilter(self._ip, 'root'),
+            filters.IpFilter(self._ip, 'user'),
+        ]
         args = ['ip', 'netns', 'exec', 'foo', 'ip', 'link', 'list']
 
         # Currently ip netns exec requires root, so verify that
         # no non-root filter is matched, as that would escalate privileges
-        self.assertRaises(wrapper.NoFilterMatched,
-                          wrapper.match_filter, filter_list, args)
+        self.assertRaises(
+            wrapper.NoFilterMatched, wrapper.match_filter, filter_list, args
+        )
 
     def test_match_filter_recurses_exec_command_filter_does_not_match(self):
-        filter_list = [filters.IpNetnsExecFilter(self._ip, 'root'),
-                       filters.IpFilter(self._ip, 'root')]
-        args = ['ip', 'netns', 'exec', 'foo', 'ip', 'netns', 'exec', 'bar',
-                'ip', 'link', 'list']
+        filter_list = [
+            filters.IpNetnsExecFilter(self._ip, 'root'),
+            filters.IpFilter(self._ip, 'root'),
+        ]
+        args = [
+            'ip',
+            'netns',
+            'exec',
+            'foo',
+            'ip',
+            'netns',
+            'exec',
+            'bar',
+            'ip',
+            'link',
+            'list',
+        ]
 
-        self.assertRaises(wrapper.NoFilterMatched,
-                          wrapper.match_filter, filter_list, args)
+        self.assertRaises(
+            wrapper.NoFilterMatched, wrapper.match_filter, filter_list, args
+        )
 
     def test_ChainingRegExpFilter_match(self):
-        filter_list = [filters.ChainingRegExpFilter('nice', 'root',
-                                                    'nice', r'-?\d+'),
-                       filters.CommandFilter('cat', 'root')]
+        filter_list = [
+            filters.ChainingRegExpFilter('nice', 'root', 'nice', r'-?\d+'),
+            filters.CommandFilter('cat', 'root'),
+        ]
         args = ['nice', '5', 'cat', '/a']
         dirs = ['/bin', '/usr/bin']
 
         self.assertIsNotNone(wrapper.match_filter(filter_list, args, dirs))
 
     def test_ChainingRegExpFilter_not_match(self):
-        filter_list = [filters.ChainingRegExpFilter('nice', 'root',
-                                                    'nice', r'-?\d+'),
-                       filters.CommandFilter('cat', 'root')]
-        args_invalid = (['nice', '5', 'ls', '/a'],
-                        ['nice', '--5', 'cat', '/a'],
-                        ['nice2', '5', 'cat', '/a'],
-                        ['nice', 'cat', '/a'],
-                        ['nice', '5'])
+        filter_list = [
+            filters.ChainingRegExpFilter('nice', 'root', 'nice', r'-?\d+'),
+            filters.CommandFilter('cat', 'root'),
+        ]
+        args_invalid = (
+            ['nice', '5', 'ls', '/a'],
+            ['nice', '--5', 'cat', '/a'],
+            ['nice2', '5', 'cat', '/a'],
+            ['nice', 'cat', '/a'],
+            ['nice', '5'],
+        )
         dirs = ['/bin', '/usr/bin']
 
         for args in args_invalid:
-            self.assertRaises(wrapper.NoFilterMatched,
-                              wrapper.match_filter, filter_list, args, dirs)
+            self.assertRaises(
+                wrapper.NoFilterMatched,
+                wrapper.match_filter,
+                filter_list,
+                args,
+                dirs,
+            )
 
     def test_ChainingRegExpFilter_multiple(self):
-        filter_list = [filters.ChainingRegExpFilter('ionice', 'root', 'ionice',
-                                                    '-c[0-3]'),
-                       filters.ChainingRegExpFilter('ionice', 'root', 'ionice',
-                                                    '-c[0-3]', '-n[0-7]'),
-                       filters.CommandFilter('cat', 'root')]
+        filter_list = [
+            filters.ChainingRegExpFilter(
+                'ionice', 'root', 'ionice', '-c[0-3]'
+            ),
+            filters.ChainingRegExpFilter(
+                'ionice', 'root', 'ionice', '-c[0-3]', '-n[0-7]'
+            ),
+            filters.CommandFilter('cat', 'root'),
+        ]
         # both filters match to ['ionice', '-c2'], but only the second accepts
         args = ['ionice', '-c2', '-n7', 'cat', '/a']
         dirs = ['/bin', '/usr/bin']
@@ -450,9 +514,10 @@ class RootwrapTestCase(testtools.TestCase):
         f = filters.CommandFilter("cat", "root")
         usercmd = ['cat', '/f']
         self.assertTrue(f.match(usercmd))
-        self.assertTrue(f.get_command(usercmd,
-                                      exec_dirs=['/bin', '/usr/bin'])
-                        in (['/bin/cat', '/f'], ['/usr/bin/cat', '/f']))
+        self.assertTrue(
+            f.get_command(usercmd, exec_dirs=['/bin', '/usr/bin'])
+            in (['/bin/cat', '/f'], ['/usr/bin/cat', '/f'])
+        )
 
     def test_skips(self):
         # Check that all filters are skipped and that the last matches
@@ -464,8 +529,7 @@ class RootwrapTestCase(testtools.TestCase):
         raw = configparser.RawConfigParser()
 
         # Empty config should raise configparser.Error
-        self.assertRaises(configparser.Error,
-                          wrapper.RootwrapConfig, raw)
+        self.assertRaises(configparser.Error, wrapper.RootwrapConfig, raw)
 
         # Check default values
         raw.set('DEFAULT', 'filters_path', '/a,/b')
@@ -478,8 +542,10 @@ class RootwrapTestCase(testtools.TestCase):
             self.assertEqual([], c.exec_dirs)
 
         self.assertFalse(config.use_syslog)
-        self.assertEqual(logging.handlers.SysLogHandler.LOG_SYSLOG,
-                         config.syslog_log_facility)
+        self.assertEqual(
+            logging.handlers.SysLogHandler.LOG_SYSLOG,
+            config.syslog_log_facility,
+        )
         self.assertEqual(logging.ERROR, config.syslog_log_level)
 
         # Check general values
@@ -497,12 +563,15 @@ class RootwrapTestCase(testtools.TestCase):
         self.assertRaises(ValueError, wrapper.RootwrapConfig, raw)
         raw.set('DEFAULT', 'syslog_log_facility', 'local0')
         config = wrapper.RootwrapConfig(raw)
-        self.assertEqual(logging.handlers.SysLogHandler.LOG_LOCAL0,
-                         config.syslog_log_facility)
+        self.assertEqual(
+            logging.handlers.SysLogHandler.LOG_LOCAL0,
+            config.syslog_log_facility,
+        )
         raw.set('DEFAULT', 'syslog_log_facility', 'LOG_AUTH')
         config = wrapper.RootwrapConfig(raw)
-        self.assertEqual(logging.handlers.SysLogHandler.LOG_AUTH,
-                         config.syslog_log_facility)
+        self.assertEqual(
+            logging.handlers.SysLogHandler.LOG_AUTH, config.syslog_log_facility
+        )
 
         raw.set('DEFAULT', 'syslog_log_level', 'bar')
         self.assertRaises(ValueError, wrapper.RootwrapConfig, raw)
@@ -520,7 +589,8 @@ class RootwrapTestCase(testtools.TestCase):
             with mock.patch('os.getlogin') as os_getlogin:
                 os_getenv.side_effect = [None, None, 'bar']
                 os_getlogin.side_effect = OSError(
-                    '[Errno 22] Invalid argument')
+                    '[Errno 22] Invalid argument'
+                )
                 self.assertEqual('bar', wrapper._getlogin())
                 os_getlogin.assert_called_once_with()
                 self.assertEqual(3, os_getenv.call_count)
@@ -536,30 +606,40 @@ class PathFilterTestCase(testtools.TestCase):
 
         self.f = filters.PathFilter('/bin/chown', 'root', 'nova', tmpdir.path)
 
-        gen_name = lambda: str(uuid.uuid4())
+        def gen_name():
+            return str(uuid.uuid4())
 
         self.SIMPLE_FILE_WITHIN_DIR = os.path.join(tmpdir.path, 'some')
         self.SIMPLE_FILE_OUTSIDE_DIR = os.path.join(self.tmp_root_dir, 'some')
-        self.TRAVERSAL_WITHIN_DIR = os.path.join(tmpdir.path, 'a', '..',
-                                                 'some')
+        self.TRAVERSAL_WITHIN_DIR = os.path.join(
+            tmpdir.path, 'a', '..', 'some'
+        )
         self.TRAVERSAL_OUTSIDE_DIR = os.path.join(tmpdir.path, '..', 'some')
 
-        self.TRAVERSAL_SYMLINK_WITHIN_DIR = os.path.join(tmpdir.path,
-                                                         gen_name())
-        os.symlink(os.path.join(tmpdir.path, 'a', '..', 'a'),
-                   self.TRAVERSAL_SYMLINK_WITHIN_DIR)
+        self.TRAVERSAL_SYMLINK_WITHIN_DIR = os.path.join(
+            tmpdir.path, gen_name()
+        )
+        os.symlink(
+            os.path.join(tmpdir.path, 'a', '..', 'a'),
+            self.TRAVERSAL_SYMLINK_WITHIN_DIR,
+        )
 
-        self.TRAVERSAL_SYMLINK_OUTSIDE_DIR = os.path.join(tmpdir.path,
-                                                          gen_name())
-        os.symlink(os.path.join(tmpdir.path, 'a', '..', '..', '..', 'etc'),
-                   self.TRAVERSAL_SYMLINK_OUTSIDE_DIR)
+        self.TRAVERSAL_SYMLINK_OUTSIDE_DIR = os.path.join(
+            tmpdir.path, gen_name()
+        )
+        os.symlink(
+            os.path.join(tmpdir.path, 'a', '..', '..', '..', 'etc'),
+            self.TRAVERSAL_SYMLINK_OUTSIDE_DIR,
+        )
 
         self.SYMLINK_WITHIN_DIR = os.path.join(tmpdir.path, gen_name())
         os.symlink(os.path.join(tmpdir.path, 'a'), self.SYMLINK_WITHIN_DIR)
 
         self.SYMLINK_OUTSIDE_DIR = os.path.join(tmpdir.path, gen_name())
-        os.symlink(os.path.join(self.tmp_root_dir, 'some_file'),
-                   self.SYMLINK_OUTSIDE_DIR)
+        os.symlink(
+            os.path.join(self.tmp_root_dir, 'some_file'),
+            self.SYMLINK_OUTSIDE_DIR,
+        )
 
     def test_empty_args(self):
         self.assertFalse(self.f.match([]))
@@ -629,22 +709,31 @@ class PathFilterTestCase(testtools.TestCase):
 
     def test_get_command_traversal(self):
         args = ['chown', 'nova', self.TRAVERSAL_WITHIN_DIR]
-        expected = ['/bin/chown', 'nova',
-                    os.path.realpath(self.TRAVERSAL_WITHIN_DIR)]
+        expected = [
+            '/bin/chown',
+            'nova',
+            os.path.realpath(self.TRAVERSAL_WITHIN_DIR),
+        ]
 
         self.assertEqual(expected, self.f.get_command(args))
 
     def test_get_command_symlink(self):
         args = ['chown', 'nova', self.SYMLINK_WITHIN_DIR]
-        expected = ['/bin/chown', 'nova',
-                    os.path.realpath(self.SYMLINK_WITHIN_DIR)]
+        expected = [
+            '/bin/chown',
+            'nova',
+            os.path.realpath(self.SYMLINK_WITHIN_DIR),
+        ]
 
         self.assertEqual(expected, self.f.get_command(args))
 
     def test_get_command_traversal_symlink(self):
         args = ['chown', 'nova', self.TRAVERSAL_SYMLINK_WITHIN_DIR]
-        expected = ['/bin/chown', 'nova',
-                    os.path.realpath(self.TRAVERSAL_SYMLINK_WITHIN_DIR)]
+        expected = [
+            '/bin/chown',
+            'nova',
+            os.path.realpath(self.TRAVERSAL_SYMLINK_WITHIN_DIR),
+        ]
 
         self.assertEqual(expected, self.f.get_command(args))
 
@@ -669,13 +758,18 @@ class DaemonCleanupException(Exception):
 
 
 class DaemonCleanupTestCase(testtools.TestCase):
-
     @mock.patch('os.chmod')
     @mock.patch('shutil.rmtree')
     @mock.patch('tempfile.mkdtemp')
-    @mock.patch('multiprocessing.managers.BaseManager.get_server',
-                side_effect=DaemonCleanupException)
+    @mock.patch(
+        'multiprocessing.managers.BaseManager.get_server',
+        side_effect=DaemonCleanupException,
+    )
     def test_daemon_no_cleanup_for_uninitialized_server(self, gs, mkd, *args):
         mkd.return_value = '/just_dir/123'
-        self.assertRaises(DaemonCleanupException, daemon.daemon_start,
-                          config=None, filters=None)
+        self.assertRaises(
+            DaemonCleanupException,
+            daemon.daemon_start,
+            config=None,
+            filters=None,
+        )

oslo_rootwrap/wrapper.py

@@ -26,20 +26,23 @@ from oslo_rootwrap import subprocess
 if sys.platform != 'win32':
     import pwd
 
+LOG = logging.getLogger(__name__)
+
 
 class NoFilterMatched(Exception):
     """This exception is raised when no filter matched."""
+
     pass
 
 
 class FilterMatchNotExecutable(Exception):
     """Raised when a filter matched but no executable was found."""
+
     def __init__(self, match=None, **kwargs):
         self.match = match
 
 
 class RootwrapConfig:
-
     def __init__(self, config):
         # filters_path
         self.filters_path = config.get("DEFAULT", "filters_path").split(",")
@@ -57,12 +60,13 @@ class RootwrapConfig:
         if config.has_option("DEFAULT", "syslog_log_facility"):
             v = config.get("DEFAULT", "syslog_log_facility")
             facility_names = logging.handlers.SysLogHandler.facility_names
-            self.syslog_log_facility = getattr(logging.handlers.SysLogHandler,
-                                               v, None)
+            self.syslog_log_facility = getattr(
+                logging.handlers.SysLogHandler, v, None
+            )
             if self.syslog_log_facility is None and v in facility_names:
                 self.syslog_log_facility = facility_names.get(v)
             if self.syslog_log_facility is None:
-                raise ValueError('Unexpected syslog_log_facility: %s' % v)
+                raise ValueError(f'Unexpected syslog_log_facility: {v}')
         else:
             default_facility = logging.handlers.SysLogHandler.LOG_SYSLOG
             self.syslog_log_facility = default_facility
@@ -72,8 +76,8 @@ class RootwrapConfig:
             v = config.get("DEFAULT", "syslog_log_level")
             level = v.upper()
             self.syslog_log_level = logging.getLevelName(level)
-            if (self.syslog_log_level == "Level %s" % level):
-                raise ValueError('Unexpected syslog_log_level: %r' % v)
+            if self.syslog_log_level == f"Level {level}":
+                raise ValueError(f'Unexpected syslog_log_level: {v!r}')
         else:
             self.syslog_log_level = logging.ERROR
 
@@ -98,26 +102,33 @@ class RootwrapConfig:
 
 def setup_syslog(execname, facility, level):
     try:
-        handler = logging.handlers.SysLogHandler(address='/dev/log',
-                                                 facility=facility)
+        handler = logging.handlers.SysLogHandler(
+            address='/dev/log', facility=facility
+        )
     except OSError:
-        logging.warning("Unable to setup syslog, maybe /dev/log socket needs "
-                        "to be restarted. Ignoring syslog configuration "
-                        "options.")
+        LOG.warning(
+            "Unable to setup syslog, maybe /dev/log socket needs "
+            "to be restarted. Ignoring syslog configuration "
+            "options."
+        )
         return
 
     rootwrap_logger = logging.getLogger()
     rootwrap_logger.setLevel(level)
-    handler.setFormatter(logging.Formatter(
-                         os.path.basename(execname) + ': %(message)s'))
+    handler.setFormatter(
+        logging.Formatter(os.path.basename(execname) + ': %(message)s')
+    )
     rootwrap_logger.addHandler(handler)
 
 
 def build_filter(class_name, *args):
     """Returns a filter object of class class_name."""
     if not hasattr(filters, class_name):
-        logging.warning("Skipping unknown filter class (%s) specified "
-                        "in filter definitions" % class_name)
+        LOG.warning(
+            "Skipping unknown filter class (%s) specified "
+            "in filter definitions",
+            class_name,
+        )
         return None
     filterclass = getattr(filters, class_name)
     return filterclass(*args)
@@ -129,15 +140,16 @@ def load_filters(filters_path):
     for filterdir in filters_path:
         if not os.path.isdir(filterdir):
             continue
-        for filterfile in filter(lambda f: not f.startswith('.'),
-                                 os.listdir(filterdir)):
+        for filterfile in filter(
+            lambda f: not f.startswith('.'), os.listdir(filterdir)
+        ):
             filterfilepath = os.path.join(filterdir, filterfile)
             if not os.path.isfile(filterfilepath):
                 continue
             kwargs = {"strict": False}
             filterconfig = configparser.RawConfigParser(**kwargs)
             filterconfig.read(filterfilepath)
-            for (name, value) in filterconfig.items("Filters"):
+            for name, value in filterconfig.items("Filters"):
                 filterdefinition = [s.strip() for s in value.split(',')]
                 newfilter = build_filter(*filterdefinition)
                 if newfilter is None:
@@ -169,11 +181,13 @@ def match_filter(filter_list, userargs, exec_dirs=None):
                 # This command calls exec verify that remaining args
                 # matches another filter.
                 def non_chain_filter(fltr):
-                    return (fltr.run_as == f.run_as and
-                            not isinstance(fltr, filters.ChainingFilter))
+                    return fltr.run_as == f.run_as and not isinstance(
+                        fltr, filters.ChainingFilter
+                    )
 
-                leaf_filters = [fltr for fltr in filter_list
-                                if non_chain_filter(fltr)]
+                leaf_filters = [
+                    fltr for fltr in filter_list if non_chain_filter(fltr)
+                ]
                 args = f.exec_args(userargs)
                 if not args:
                     continue
@@ -202,9 +216,9 @@ def _getlogin():
     try:
         return os.getlogin()
     except OSError:
-        return (os.getenv('USER') or
-                os.getenv('USERNAME') or
-                os.getenv('LOGNAME'))
+        return (
+            os.getenv('USER') or os.getenv('USERNAME') or os.getenv('LOGNAME')
+        )
 
 
 def start_subprocess(filter_list, userargs, exec_dirs=[], log=False, **kwargs):
@@ -212,9 +226,13 @@ def start_subprocess(filter_list, userargs, exec_dirs=[], log=False, **kwargs):
 
     command = filtermatch.get_command(userargs, exec_dirs)
     if log:
-        logging.info("({} > {}) Executing {} (filter match = {})".format(
-            _getlogin(), pwd.getpwuid(os.getuid())[0],
-            command, filtermatch.name))
+        LOG.info(
+            "(%s > %s) Executing %s (filter match = %s)",
+            _getlogin(),
+            pwd.getpwuid(os.getuid())[0],
+            command,
+            filtermatch.name,
+        )
 
     def preexec():
         # Python installs a SIGPIPE handler by default. This is
@@ -222,8 +240,10 @@ def start_subprocess(filter_list, userargs, exec_dirs=[], log=False, **kwargs):
         signal.signal(signal.SIGPIPE, signal.SIG_DFL)
         filtermatch.preexec()
 
-    obj = subprocess.Popen(command,
-                           preexec_fn=preexec,
-                           env=filtermatch.get_environment(userargs),
-                           **kwargs)
+    obj = subprocess.Popen(
+        command,
+        preexec_fn=preexec,
+        env=filtermatch.get_environment(userargs),
+        **kwargs,
+    )
     return obj

pyproject.toml

@@ -38,6 +38,15 @@ oslo-rootwrap-daemon = "oslo_rootwrap.cmd:daemon"
 [tool.setuptools]
 packages = ["oslo_rootwrap"]
 
-[tool.bandit]
-exclude_dirs = ["tests", "benchmark"]
-skips = ['B404']
+[tool.ruff]
+line-length = 79
+
+[tool.ruff.format]
+quote-style = "preserve"
+docstring-code-format = true
+
+[tool.ruff.lint]
+select = ["E4", "E5", "E7", "E9", "F", "G", "LOG", "S", "UP"]
+
+[tool.ruff.lint.per-file-ignores]
+"benchmark/benchmark.py" = ["S"]

releasenotes/source/conf.py

@@ -193,9 +193,13 @@ htmlhelp_basename = 'oslo.rootwrapReleaseNotesDoc'
 # (source start file, target name, title,
 #  author, documentclass [howto, manual, or own class]).
 latex_documents = [
-    ('index', 'oslo.rootwrapReleaseNotes.tex',
-     'oslo.rootwrap Release Notes Documentation',
-     'oslo.rootwrap Developers', 'manual'),
+    (
+        'index',
+        'oslo.rootwrapReleaseNotes.tex',
+        'oslo.rootwrap Release Notes Documentation',
+        'oslo.rootwrap Developers',
+        'manual',
+    ),
 ]
 
 # The name of an image file (relative to this directory) to place at the top of
@@ -224,9 +228,13 @@ latex_documents = [
 # One entry per manual page. List of tuples
 # (source start file, name, description, authors, manual section).
 man_pages = [
-    ('index', 'oslo.rootwrapReleaseNotes',
-     'oslo.rootwrap Release Notes Documentation',
-     ['oslo.rootwrap Developers'], 1)
+    (
+        'index',
+        'oslo.rootwrapReleaseNotes',
+        'oslo.rootwrap Release Notes Documentation',
+        ['oslo.rootwrap Developers'],
+        1,
+    )
 ]
 
 # If true, show URL addresses after external links.
@@ -239,12 +247,16 @@ man_pages = [
 # (source start file, target name, title, author,
 #  dir menu entry, description, category)
 texinfo_documents = [
-    ('index', 'oslo.rootwrapReleaseNotes',
-     'oslo.rootwrap Release Notes Documentation',
-     'oslo.rootwrap Developers', 'oslo.rootwrapReleaseNotes',
-     'Allows fine-grained filtering of shell commands to run as root from'
-     ' OpenStack services.',
-     'Miscellaneous'),
+    (
+        'index',
+        'oslo.rootwrapReleaseNotes',
+        'oslo.rootwrap Release Notes Documentation',
+        'oslo.rootwrap Developers',
+        'oslo.rootwrapReleaseNotes',
+        'Allows fine-grained filtering of shell commands to run as root from'
+        ' OpenStack services.',
+        'Miscellaneous',
+    ),
 ]
 
 # Documents to append as an appendix to all manuals.

setup.py

@@ -15,6 +15,4 @@
 
 import setuptools
 
-setuptools.setup(
-    setup_requires=['pbr>=2.0.0'],
-    pbr=True)
+setuptools.setup(setup_requires=['pbr>=2.0.0'], pbr=True)

tox.ini

@@ -14,13 +14,6 @@ commands =
   stestr run --slowest (?!tests.test_functional_eventlet)tests {posargs}
   env TEST_EVENTLET=1 stestr run --slowest tests.test_functional_eventlet
 
-[testenv:pep8]
-skip_install = true
-deps =
-  pre-commit>=2.6.0 # MIT
-commands =
-  pre-commit run -a
-
 [testenv:cover]
 deps = {[testenv]deps}
     coverage
@@ -37,6 +30,10 @@ commands =
 [testenv:venv]
 commands = {posargs}
 
+[testenv:benchmark]
+commands =
+  python3 benchmark/benchmark.py
+
 [testenv:docs]
 allowlist_externals = rm
 deps =
@@ -46,19 +43,6 @@ commands =
   rm -fr doc/build
   sphinx-build -W --keep-going -b html doc/source doc/build/html
 
-[flake8]
-show-source = True
-# H904: Delay string interpolations at logging calls
-enable-extensions = H904
-# E731 skipped as assign a lambda expression
-# W504 line break after binary operator
-ignore = E731,W504
-exclude = .tox,dist,doc,*.egg,build
-
-[testenv:benchmark]
-commands =
-  python3 benchmark/benchmark.py
-
 [testenv:releasenotes]
 allowlist_externals = rm
 deps =
@@ -68,3 +52,13 @@ commands =
   rm -rf releasenotes/build
   sphinx-build -a -E -W -d releasenotes/build/doctrees --keep-going -b html releasenotes/source releasenotes/build/html
 
+[testenv:pep8]
+skip_install = true
+deps =
+  pre-commit>=2.6.0 # MIT
+commands =
+  pre-commit run -a
+
+[flake8]
+# We only enable the hacking (H) checks
+select = H