github-jwinterm/modoboa
1
0
Fork 0
mirror of https://github.com/jwinterm/modoboa.git synced 2026-10-17 05:41:25 +02:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #3934 from modoboa/fix/prevent-os-command-injection

Browse Source 
Prevent OS command injection using exec_cmd()
This commit is contained in:
Antoine Nguyen
2026-02-21 11:48:14 +01:00
committed by GitHub
parent 6824d963cb bf5310c131
commit abb72fb289
11 changed files with 103 additions and 46 deletions
Download Patch File Download Diff File Expand all files Collapse all files
modoboa/admin/api/v2/serializers.py
+1 -1
View File
@@ -274,7 +274,7 @@ class AdminGlobalParametersSerializer(serializers.Serializer):
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"])
if code:
raise serializers.ValidationError(
_("openssl not found, please make sure it is installed.")
modoboa/admin/app_settings.py
+2 -2
View File
@@ -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"])
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([dpath, "--version"])
except OSError:
hide_fields = True
else:
modoboa/admin/jobs.py
+1 -1
View File
@@ -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
modoboa/admin/management/commands/subcommands/_manage_dkim_keys.py
+6 -2
View File
@@ -44,7 +44,9 @@ class ManageDKIMKeys(BaseCommand):
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 @@ class ManageDKIMKeys(BaseCommand):
)
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)}"
modoboa/admin/models/mailbox.py
+1 -1
View File
@@ -147,7 +147,7 @@ class Mailbox(mixins.MessageLimitMixin, AdminObject):
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
modoboa/amavis/lib.py
+34 -17
View File
@@ -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):
def _find_binary(self, name: str) -> str:
"""Find path to binary."""
code, output = exec_cmd(f"which {name}")
code, output = exec_cmd(["which", name])
if not code:
return smart_str(output).strip()
known_paths = getattr(settings, "SA_LOOKUP_PATH", ("/usr/bin",))
@@ -132,6 +123,32 @@ class SpamassassinClient:
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",
str(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 @@ class SpamassassinClient:
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 @@ class SpamassassinClient:
"""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)
modoboa/core/password_hashers/utils.py
+2 -2
View File
@@ -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
modoboa/lib/sysutils.py
+11 -6
View File
@@ -13,7 +13,13 @@ from django.conf import settings
from django.utils.encoding import force_str
def exec_cmd(cmd, sudo_user=None, pinput=None, capture_output=True, **kwargs):
def exec_cmd(
cmd: 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
@@ -27,8 +33,7 @@ def exec_cmd(cmd, sudo_user=None, pinput=None, capture_output=True, **kwargs):
:return: return code, command output
"""
if sudo_user is not None:
cmd = f"sudo -u {sudo_user} {cmd}"
kwargs["shell"] = True
cmd = ["sudo", "-u", sudo_user] + cmd
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"])
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,
modoboa/maillog/graphics.py
+13 -5
View File
@@ -7,7 +7,7 @@ from itertools import chain
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 @@ class Graphic:
def rrdtool_binary(self):
"""Return path to rrdtool binary."""
dpath = None
code, output = exec_cmd("which rrdtool")
code, output = exec_cmd(["which", "rrdtool"])
if not code:
dpath = output.strip()
else:
@@ -102,9 +102,17 @@ class Graphic:
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 []
modoboa/webmail/models.py
+20 -5
View File
@@ -52,9 +52,17 @@ class ScheduledMessage(models.Model):
# 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,
str(self.id),
]
)
if code:
self.status = constants.SchedulingState.MOVE_ERROR.value
@@ -64,8 +72,15 @@ class ScheduledMessage(models.Model):
# 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
tests.py
+12 -4
View File
@@ -26,10 +26,18 @@ class DeployTest(unittest.TestCase):
def test_silent(self):
dburl = f"default:{self.dbtype}://{self.dbuser}:{self.dbpassword}@{self.dbhost}:{self.dbport}/modoboa"
cmd = (
f"modoboa-admin.py deploy --collectstatic "
f"--dburl {dburl} --domain localhost --admin-username admin {self.projname}"
)
cmd = [
"modoboa-admin.py",
"deploy",
"--collectstatic",
"--dburl",
dburl,
"--domain",
"localhost",
"--admin-username",
"admin",
self.projname,
]
code, output = exec_cmd(cmd, cwd=self.workdir)
self.assertEqual(code, 0)