security: quote untrusted values in command construction across connectors, operations, and util (#1664)

Author: wowi42

src/pyinfra/connectors/chroot.py

@@ -1,4 +1,5 @@
 import os
+import shlex
 from tempfile import mkstemp
 from typing import TYPE_CHECKING, Optional
 
@@ -64,7 +65,7 @@ def connect(self) -> None:
         try:
             with progress_spinner({"chroot run"}):
                 local.shell(
-                    "chroot {0} ls".format(chroot_directory),
+                    "chroot {0} ls".format(shlex.quote(chroot_directory)),
                     splitlines=True,
                 )
         except PyinfraError as e:
@@ -91,7 +92,7 @@ def run_shell_command(
 
         chroot_command = StringCommand(
             "chroot",
-            chroot_directory,
+            QuoteString(chroot_directory),
             "sh",
             "-c",
             command,
@@ -130,8 +131,8 @@ def put_file(
             chroot_directory = self.host.connector_data["chroot_directory"]
             chroot_command = StringCommand(
                 "cp",
-                temp_filename,
-                f"{chroot_directory}/{remote_filename}",
+                QuoteString(temp_filename),
+                QuoteString(f"{chroot_directory}/{remote_filename}"),
             )
 
             status, output = self.local.run_shell_command(
@@ -172,8 +173,8 @@ def get_file(
             chroot_directory = self.host.connector_data["chroot_directory"]
             chroot_command = StringCommand(
                 "cp",
-                f"{chroot_directory}/{remote_filename}",
-                temp_filename,
+                QuoteString(f"{chroot_directory}/{remote_filename}"),
+                QuoteString(temp_filename),
             )
 
             status, output = self.local.run_shell_command(

src/pyinfra/connectors/docker.py

@@ -2,6 +2,7 @@
 
 import json
 import os
+import shlex
 from tempfile import mkstemp
 from typing import TYPE_CHECKING
 
@@ -106,12 +107,13 @@ def make_names_data(name=None):
 
     # 2 helper functions
     def _find_start_docker_container(self, container_id) -> tuple[str, bool]:
-        docker_info = local.shell(f"{self.docker_cmd} container inspect {container_id}")
+        quoted_container_id = shlex.quote(container_id)
+        docker_info = local.shell(f"{self.docker_cmd} container inspect {quoted_container_id}")
         assert isinstance(docker_info, str)
         docker_info = json.loads(docker_info)[0]
         if docker_info["State"]["Running"] is False:
             logger.info(f"Starting stopped container: {container_id}")
-            local.shell(f"{self.docker_cmd} container start {container_id}")
+            local.shell(f"{self.docker_cmd} container start {quoted_container_id}")
             return container_id, False
         return container_id, True
 
@@ -123,13 +125,13 @@ def _start_docker_image(self, image_name):
         ]
 
         if self.data.get("docker_platform"):
-            docker_cmd_parts.extend(["--platform", self.data["docker_platform"]])
+            docker_cmd_parts.extend(["--platform", shlex.quote(self.data["docker_platform"])])
         if self.data.get("docker_architecture"):
-            docker_cmd_parts.extend(["--arch", self.data["docker_architecture"]])
+            docker_cmd_parts.extend(["--arch", shlex.quote(self.data["docker_architecture"])])
 
         docker_cmd_parts.extend(
             [
-                image_name,
+                shlex.quote(image_name),
                 "tail",
                 "-f",
                 "/dev/null",
@@ -173,14 +175,16 @@ def disconnect(self) -> None:
             )
             return
 
+        quoted_container_id = shlex.quote(container_id)
+
         with progress_spinner({f"{self.docker_cmd} commit"}):
-            image_id = local.shell(f"{self.docker_cmd} commit {container_id}", splitlines=True)[-1][
-                7:19
-            ]  # last line is the image ID, get sha256:[XXXXXXXXXX]...
+            image_id = local.shell(
+                f"{self.docker_cmd} commit {quoted_container_id}", splitlines=True
+            )[-1][7:19]  # last line is the image ID, get sha256:[XXXXXXXXXX]...
 
         with progress_spinner({f"{self.docker_cmd} rm"}):
             local.shell(
-                f"{self.docker_cmd} rm -f {container_id}",
+                f"{self.docker_cmd} rm -f {quoted_container_id}",
             )
 
         logger.info(
@@ -255,8 +259,8 @@ def put_file(
             docker_command = StringCommand(
                 self.docker_cmd,
                 "cp",
-                temp_filename,
-                f"{self.container_id}:{remote_filename}",
+                QuoteString(temp_filename),
+                QuoteString(f"{self.container_id}:{remote_filename}"),
             )
 
             status, output = self.local.run_shell_command(
@@ -303,8 +307,8 @@ def get_file(
             docker_command = StringCommand(
                 self.docker_cmd,
                 "cp",
-                f"{self.container_id}:{remote_filename}",
-                temp_filename,
+                QuoteString(f"{self.container_id}:{remote_filename}"),
+                QuoteString(temp_filename),
             )
 
             status, output = self.local.run_shell_command(

src/pyinfra/connectors/dockerssh.py

@@ -90,7 +90,7 @@ def connect(self) -> None:
                         self.docker_cmd,
                         "run",
                         "-d",
-                        self.host.data.docker_image,
+                        QuoteString(self.host.data.docker_image),
                         "tail",
                         "-f",
                         "/dev/null",
@@ -111,15 +111,15 @@ def disconnect(self) -> None:
 
         with progress_spinner({f"{self.docker_cmd} commit"}):
             _, output = self.ssh.run_shell_command(
-                StringCommand(self.docker_cmd, "commit", container_id)
+                StringCommand(self.docker_cmd, "commit", QuoteString(container_id))
             )
 
             # Last line is the image ID, get sha256:[XXXXXXXXXX]...
             image_id = output.stdout_lines[-1][7:19]
 
         with progress_spinner({f"{self.docker_cmd} rm"}):
             self.ssh.run_shell_command(
-                StringCommand(self.docker_cmd, "rm", "-f", container_id),
+                StringCommand(self.docker_cmd, "rm", "-f", QuoteString(container_id)),
             )
 
         logger.info(
@@ -150,7 +150,7 @@ def run_shell_command(
             self.docker_cmd,
             "exec",
             docker_flags,
-            container_id,
+            QuoteString(container_id),
             "sh",
             "-c",
             command,
@@ -203,8 +203,8 @@ def put_file(
             docker_command = StringCommand(
                 self.docker_cmd,
                 "cp",
-                remote_temp_filename,
-                f"{docker_id}:{remote_filename}",
+                QuoteString(remote_temp_filename),
+                QuoteString(f"{docker_id}:{remote_filename}"),
             )
 
             status, output = self.ssh.run_shell_command(
@@ -257,8 +257,8 @@ def get_file(
             docker_command = StringCommand(
                 self.docker_cmd,
                 "cp",
-                f"{docker_id}:{remote_filename}",
-                remote_temp_filename,
+                QuoteString(f"{docker_id}:{remote_filename}"),
+                QuoteString(remote_temp_filename),
             )
 
             status, output = self.ssh.run_shell_command(

src/pyinfra/connectors/util.py

@@ -235,12 +235,12 @@ def write_stdin(stdin, buffer):
 def remove_any_sudo_askpass_file(host) -> None:
     sudo_askpass_path = host.connector_data.get("sudo_askpass_path")
     if sudo_askpass_path:
-        host.run_shell_command("rm -f {0}".format(sudo_askpass_path))
+        host.run_shell_command(StringCommand("rm", "-f", QuoteString(sudo_askpass_path)))
         host.connector_data["sudo_askpass_path"] = None
 
     su_askpass_path = host.connector_data.get("su_askpass_path")
     if su_askpass_path:
-        host.run_shell_command("rm -f {0}".format(su_askpass_path))
+        host.run_shell_command(StringCommand("rm", "-f", QuoteString(su_askpass_path)))
         host.connector_data["su_askpass_path"] = None
 
 
@@ -293,7 +293,7 @@ def _ensure_askpass_set_for_host(host: "Host", key: str, env_var: str):
             )
         )
 
-    host.connector_data[key] = StringCommand(QuoteString(output.stdout_lines[0])).get_raw_value()
+    host.connector_data[key] = output.stdout_lines[0]
 
 
 def make_unix_command_for_host(
@@ -367,31 +367,38 @@ def make_unix_command(
         _shell_executable = "sh"
 
     if _env:
-        env_string = " ".join(['"{0}={1}"'.format(key, value) for key, value in _env.items()])
-        command = StringCommand("export", env_string, "&&", command)
+        env_bits: list[Union[str, StringCommand, QuoteString]] = ["export"]
+        for key, value in _env.items():
+            # Quote the whole `key=value` pair so arbitrary values cannot break
+            # out into additional shell tokens. Invalid identifiers in `key` will
+            # fail safely when the shell rejects the resulting `export` statement.
+            env_bits.append(QuoteString("{0}={1}".format(key, value)))
+        env_bits.append("&&")
+        env_bits.append(command)
+        command = StringCommand(*env_bits)
 
     if _chdir:
-        command = StringCommand("cd", _chdir, "&&", command)
+        command = StringCommand("cd", QuoteString(_chdir), "&&", command)
 
     command_bits: list[Union[str, StringCommand, QuoteString]] = []
 
     if _doas:
         command_bits.extend(["doas", "-n"])
 
         if _doas_user:
-            command_bits.extend(["-u", _doas_user])
+            command_bits.extend(["-u", QuoteString(_doas_user)])
 
     if _dzdo:
         command_bits.extend(["dzdo", "-H", "-n"])
 
         if _dzdo_user:
-            command_bits.extend(["-u", _dzdo_user])
+            command_bits.extend(["-u", QuoteString(_dzdo_user)])
 
     if _sudo_password and _sudo_askpass_path:
         command_bits.extend(
             [
                 "env",
-                "SUDO_ASKPASS={0}".format(_sudo_askpass_path),
+                StringCommand("SUDO_ASKPASS=", QuoteString(_sudo_askpass_path), _separator=""),
                 MaskString(
                     "{0}={1}".format(
                         SUDO_ASKPASS_ENV_VAR,
@@ -416,7 +423,7 @@ def make_unix_command(
             command_bits.append("-E")
 
         if _sudo_user:
-            command_bits.extend(("-u", _sudo_user))
+            command_bits.extend(("-u", QuoteString(_sudo_user)))
 
     if _su_user:
         if _su_password and _su_askpass_path:
@@ -429,7 +436,7 @@ def make_unix_command(
                             StringCommand(QuoteString(_su_password)).get_raw_value(),
                         )
                     ),
-                    _su_askpass_path,
+                    QuoteString(_su_askpass_path),
                     "|",
                 ],
             )
@@ -444,9 +451,16 @@ def make_unix_command(
             command_bits.append("-m")
 
         if _su_shell:
-            command_bits.extend(["-s", "`which {0}`".format(_su_shell)])
+            # Resolve the shell via `command -v`, with the user-supplied shell
+            # name safely quoted so it cannot inject extra shell syntax.
+            command_bits.extend(
+                [
+                    "-s",
+                    StringCommand("$(command -v ", QuoteString(_su_shell), ")", _separator=""),
+                ]
+            )
 
-        command_bits.extend([_su_user, "-c"])
+        command_bits.extend([QuoteString(_su_user), "-c"])
 
         if _shell_executable is not None:
             # Quote the whole shell -c 'command' as BSD `su` does not have a shell option

src/pyinfra/operations/selinux.py

@@ -60,8 +60,12 @@ def boolean(bool_name: str, value: Boolean, persistent=False):
         raise OperationValueError(f"Invalid value '{value}' for boolean operation")
 
     if host.get_fact(SEBoolean, boolean=bool_name) != value_str:
-        persist = "-P " if persistent else ""
-        yield StringCommand("setsebool", f"{persist}{bool_name}", value_str)
+        command_bits: list = ["setsebool"]
+        if persistent:
+            command_bits.append("-P")
+        command_bits.append(QuoteString(bool_name))
+        command_bits.append(value_str)
+        yield StringCommand(*command_bits)
     else:
         host.noop(f"boolean '{bool_name}' already had the value '{value_str}'")
 

src/pyinfra/operations/util/files.py

@@ -6,7 +6,7 @@
 from enum import Enum
 from typing import Callable, Generator
 
-from pyinfra.api import QuoteString, StringCommand
+from pyinfra.api import OperationError, QuoteString, StringCommand
 from pyinfra.api.output import format_text
 
 
@@ -146,7 +146,7 @@ def chown(
     dereference=True,
 ) -> StringCommand:
     command = "chown"
-    user_group = None
+    user_group: str | None = None
 
     if user and group:
         user_group = "{0}:{1}".format(user, group)
@@ -158,14 +158,17 @@ def chown(
         command = "chgrp"
         user_group = group
 
+    if user_group is None:
+        raise OperationError("chown() requires at least one of user or group")
+
     args = [command]
     if recursive:
         args.append("-R")
 
     if not dereference:
         args.append("-h")
 
-    return StringCommand(" ".join(args), user_group, QuoteString(target))
+    return StringCommand(" ".join(args), QuoteString(user_group), QuoteString(target))
 
 
 # like the touch command, but only supports setting one field at a time, and expects any

tests/operations/files.file/add_user_group_with_metachars.json

@@ -0,0 +1,18 @@
+{
+    "args": ["testfile"],
+    "kwargs": {
+        "user": "root; rm -rf /",
+        "group": "wheel$(id)",
+        "mode": "777"
+    },
+    "facts": {
+        "files.File": {
+            "path=testfile": null
+        }
+    },
+    "commands": [
+        "touch testfile",
+        "chmod 777 testfile",
+        "chown 'root; rm -rf /:wheel$(id)' testfile"
+    ]
+}

tests/operations/selinux.boolean/change_bool_name_with_metachars.json

@@ -0,0 +1,14 @@
+{
+    "args": ["httpd`id`", "on"],
+    "kwargs": {
+        "persistent": true
+    },
+    "facts": {
+        "selinux.SEBoolean": {
+            "boolean=httpd`id`": "off"
+        }
+    },
+    "commands": [
+        "setsebool -P 'httpd`id`' on"
+    ]
+}