Modoboa has an OS Command Injection

@@ -274,7 +274,7 @@ def validate_default_mailbox_quota(self, value):
     def validate_dkim_keys_storage_dir(self, value):
         """Check that directory exists."""
         if value:
-            code, output = exec_cmd("which openssl")
+            code, output = exec_cmd("which openssl", shell=True)
             if code:
                 raise serializers.ValidationError(
                     _("openssl not found, please make sure it is installed.")

@@ -298,7 +298,7 @@ def init_structure() -> dict:
     structure = copy.deepcopy(GLOBAL_PARAMETERS_STRUCT)
     hide_fields = False
     dpath = None
-    code, output = exec_cmd("which dovecot")
+    code, output = exec_cmd("which dovecot", shell=True)
     if not code:
         dpath = force_str(output).strip()
     else:
@@ -312,7 +312,7 @@ def init_structure() -> dict:
                 dpath = fpath
     if dpath:
         try:
-            code, version = exec_cmd(f"{dpath} --version")
+            code, version = exec_cmd(f"{dpath} --version", shell=True)
         except OSError:
             hide_fields = True
         else:

@@ -35,7 +35,7 @@ def rename_mailbox(operation):
                 f"renaming of {operation.argument} to {new_mail_home} failed (reason: {reason})"
             )
             return
-    code, output = exec_cmd(f"mv {operation.argument} {new_mail_home}")
+    code, output = exec_cmd(["mv", operation.argument, new_mail_home])
     if code:
         logger.critical(f"Renaming of {new_mail_home} failed (reason: {output})")
         return

@@ -44,7 +44,9 @@ def create_new_dkim_key(self, domain):
             if domain.dkim_key_length
             else self.default_key_length
         )
-        code, output = sysutils.exec_cmd(f"openssl genrsa -out {pkey_path} {key_size}")
+        code, output = sysutils.exec_cmd(
+            ["openssl", "genrsa", "-out", pkey_path, str(key_size)]
+        )
         if code:
             print(
                 f"Failed to generate DKIM private key for domain {domain.name}: {smart_str(output)}"
@@ -54,7 +56,9 @@ def create_new_dkim_key(self, domain):
             )
             return
         domain.dkim_private_key_path = pkey_path
-        code, output = sysutils.exec_cmd(f"openssl rsa -in {pkey_path} -pubout")
+        code, output = sysutils.exec_cmd(
+            ["openssl", "rsa", "-in", pkey_path, "-pubout"]
+        )
         if code:
             print(
                 f"Failed to generate DKIM public key for domain {domain.name}: {smart_str(output)}"

@@ -147,7 +147,7 @@ def mail_home(self):
         if not admin_params.get("handle_mailboxes"):
             return None
         if self.__mail_home is None:
-            code, output = doveadm_cmd(f"user -f home {self.full_address}")
+            code, output = doveadm_cmd(["user", "-f", "home", self.full_address])
             if code:
                 raise lib_exceptions.InternalError(
                     _("Failed to retrieve mailbox location (%s)") % output

@@ -92,37 +92,28 @@ class SpamassassinClient:
 
     def __init__(self, user, recipient_db):
         """Constructor."""
-        conf = dict(param_tools.get_global_parameters("amavis"))
-        self._sa_is_local = conf["sa_is_local"]
-        self._default_username = conf["default_user"]
+        self.conf = dict(param_tools.get_global_parameters("amavis"))
+        self._sa_is_local = self.conf["sa_is_local"]
+        self._default_username = self.conf["default_user"]
         self._recipient_db = recipient_db
         self._setup_cache = {}
         self._username_cache = []
         if user.role == "SimpleUsers":
-            if conf["user_level_learning"]:
+            if self.conf["user_level_learning"]:
                 self._username = user.email
         else:
             self._username = None
         self.error = None
         if self._sa_is_local:
-            self._learn_cmd = self._find_binary("sa-learn")
-            self._learn_cmd += " --{0} --no-sync -u {1}"
             self._learn_cmd_kwargs = {}
             self._expected_exit_codes = [0]
-            self._sync_cmd = self._find_binary("sa-learn")
-            self._sync_cmd += " -u {0} --sync"
         else:
-            self._learn_cmd = self._find_binary("spamc")
-            self._learn_cmd += " -d {} -p {}".format(
-                conf["spamd_address"], conf["spamd_port"]
-            )
-            self._learn_cmd += " -L {0} -u {1}"
             self._learn_cmd_kwargs = {}
             self._expected_exit_codes = [5, 6]
 
     def _find_binary(self, name):
         """Find path to binary."""
-        code, output = exec_cmd(f"which {name}")
+        code, output = exec_cmd(f"which {name}", shell=True)
         if not code:
             return smart_str(output).strip()
         known_paths = getattr(settings, "SA_LOOKUP_PATH", ("/usr/bin",))
@@ -132,6 +123,32 @@ def _find_binary(self, name):
                 return bpath
         raise InternalError(_("Failed to find {} binary").format(name))
 
+    def get_learn_cmd(self, mtype: str, username: str) -> list[str]:
+        result = []
+        if self._sa_is_local:
+            result += [self._find_binary("sa-learn")]
+            result += [f"--{mtype}", "--no-sync", "-u", username]
+        else:
+            result += [self._find_binary("spamc")]
+            result += [
+                "-d",
+                self.conf["spamd_address"],
+                "-p",
+                self.conf["spamd_port"],
+                "-L",
+                mtype,
+                "-u",
+                username,
+            ]
+        return result
+
+    def get_sync_cmd(self, username: str) -> list[str]:
+        result = []
+        if self._sa_is_local:
+            result += [self._find_binary("sa-learn")]
+            result += ["-u", username, "--sync"]
+        return result
+
     def _get_mailbox_from_rcpt(self, rcpt):
         """Retrieve a mailbox from a recipient address."""
         local_part, domname, extension = split_mailbox(rcpt, return_extension=True)
@@ -199,7 +216,7 @@ def _learn(self, rcpt, msg, mtype):
                     self._setup_cache[username] = True
         if username not in self._username_cache:
             self._username_cache.append(username)
-        cmd = self._learn_cmd.format(mtype, username)
+        cmd = self.get_learn_cmd(mtype, username)
         code, output = exec_cmd(cmd, pinput=smart_bytes(msg), **self._learn_cmd_kwargs)
         if code in self._expected_exit_codes:
             return True
@@ -218,7 +235,7 @@ def done(self):
         """Call this method at the end of the processing."""
         if self._sa_is_local:
             for username in self._username_cache:
-                cmd = self._sync_cmd.format(username)
+                cmd = self.get_sync_cmd(username)
                 exec_cmd(cmd, **self._learn_cmd_kwargs)
 
 

@@ -1,4 +1,4 @@
-""" Utils for password schemes caching/retrieval. """
+"""Utils for password schemes caching/retrieval."""
 
 from django.conf import settings
 from django.core.cache import cache
@@ -30,7 +30,7 @@ def get_dovecot_schemes():
 
     if not schemes:
         try:
-            retcode, schemes = doveadm_cmd("pw -l")
+            retcode, schemes = doveadm_cmd(["pw", "-l"])
         except OSError:
             schemes = default_schemes
             status = 2

@@ -13,7 +13,13 @@
 from django.utils.encoding import force_str
 
 
-def exec_cmd(cmd, sudo_user=None, pinput=None, capture_output=True, **kwargs):
+def exec_cmd(
+    cmd: str | list[str],
+    sudo_user: str | None = None,
+    pinput: str | None = None,
+    capture_output: bool = True,
+    **kwargs,
+):
     """Execute a shell command.
 
     Run a command using the current user. Set :keyword:`sudo_user` if
@@ -28,7 +34,6 @@ def exec_cmd(cmd, sudo_user=None, pinput=None, capture_output=True, **kwargs):
     """
     if sudo_user is not None:
         cmd = f"sudo -u {sudo_user} {cmd}"
-    kwargs["shell"] = True
     if pinput is not None:
         kwargs["stdin"] = subprocess.PIPE
     if capture_output:
@@ -43,7 +48,7 @@ def exec_cmd(cmd, sudo_user=None, pinput=None, capture_output=True, **kwargs):
     return process.returncode, output
 
 
-def doveadm_cmd(params: str, pinput=None, capture_output=True, **kwargs):
+def doveadm_cmd(params: list[str], pinput=None, capture_output=True, **kwargs):
     """Execute doveadm command.
 
     Run doveadm command using the current user. Set :keyword:`sudo_user` if
@@ -57,7 +62,7 @@ def doveadm_cmd(params: str, pinput=None, capture_output=True, **kwargs):
     :return: return code, command output
     """
     dpath = None
-    code, output = exec_cmd("which doveadm")
+    code, output = exec_cmd("which doveadm", shell=True)
     if not code:
         dpath = force_str(output).strip()
     else:
@@ -75,7 +80,7 @@ def doveadm_cmd(params: str, pinput=None, capture_output=True, **kwargs):
     sudo_user = dovecot_user if curuser != dovecot_user else None
     if dpath:
         return exec_cmd(
-            f"{dpath} {params}",
+            [dpath] + params,
             sudo_user=sudo_user,
             pinput=pinput,
             capture_output=capture_output,

@@ -7,7 +7,7 @@
 import os
 
 from django.conf import settings
-from django.utils.encoding import smart_bytes, smart_str
+from django.utils.encoding import smart_str
 from django.utils.translation import gettext as _, gettext_lazy
 
 from modoboa.admin import models as admin_models
@@ -75,7 +75,7 @@ def display_name(self):
     def rrdtool_binary(self):
         """Return path to rrdtool binary."""
         dpath = None
-        code, output = exec_cmd("which rrdtool")
+        code, output = exec_cmd("which rrdtool", shell=True)
         if not code:
             dpath = output.strip()
         else:
@@ -102,9 +102,17 @@ def export(self, rrdfile, start, end):
             cmdargs += curve.to_rrd_command_args(rrdfile)
         code = 0
 
-        cmd = f"{self.rrdtool_binary} xport --json -t --start {str(start)} --end {str(end)} "
-        cmd += " ".join(cmdargs)
-        code, output = exec_cmd(smart_bytes(cmd))
+        cmd = [
+            self.rrdtool_binary,
+            "xport",
+            "--json",
+            "-t",
+            "--start",
+            str(start),
+            "--end",
+            str(end),
+        ] + cmdargs
+        code, output = exec_cmd(cmd)
         if code:
             return []
 

@@ -52,9 +52,17 @@ def delete_imap_copy(self) -> bool:
         # TODO: use doveadm HTTP API when dovecot is not local
         sent_folder = self.account.parameters.get_value("sent_folder")
         code, output = doveadm_cmd(
-            f"move -u {self.account.email} {sent_folder} "
-            f"mailbox {constants.MAILBOX_NAME_SCHEDULED} "
-            f"header {constants.CUSTOM_HEADER_SCHEDULED_ID} {self.id}"
+            [
+                "move",
+                "-u",
+                self.account.email,
+                sent_folder,
+                "mailbox",
+                constants.MAILBOX_NAME_SCHEDULED,
+                "header",
+                constants.CUSTOM_HEADER_SCHEDULED_ID,
+                self.id,
+            ]
         )
         if code:
             self.status = constants.SchedulingState.MOVE_ERROR.value
@@ -64,8 +72,15 @@ def delete_imap_copy(self) -> bool:
 
         # Try to delete mailbox when empty
         doveadm_cmd(
-            f"mailbox delete -u {self.account.email} -s -e "
-            f"{constants.MAILBOX_NAME_SCHEDULED}"
+            [
+                "mailbox",
+                "delete",
+                "-u",
+                self.account.email,
+                "-s",
+                "-e",
+                constants.MAILBOX_NAME_SCHEDULED,
+            ]
         )
 
         return True