fix(security): prevent command injection via shell=True (CWE-78)

Replace shell=True with list-based subprocess calls for all git.py
functions that interpolate user-controlled values (tag names, messages,
file paths, git references). This prevents shell injection attacks where
malicious values in pyproject.toml could execute arbitrary commands
during CI/CD runs of 'cz bump'.

Changes:
- cmd.run() now accepts str | Sequence[str]; lists use shell=False
- git.tag() uses list args (fixes primary attack vector)
- git.add() uses list args
- git.commit() uses list args + env= for GIT_COMMITTER_DATE
- git.tag_exist/is_signed_tag/get_tag_message use list args
- git.get_filenames_in_commit() uses list args
- git.get_tags() uses list args
- git._get_log_as_str_list() uses list args

Closes #1918

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

Author: bearomorphism

commitizen/cli.py

@@ -704,7 +704,7 @@ def main() -> None:
             logger.warning(
                 "\nWARN: Incomplete commit command: received -- separator without any following git arguments\n"
             )
-        extra_args = " ".join(unknown_args[1:])
+        extra_args = unknown_args[1:]
         arguments["extra_cli_args"] = extra_args
 
     conf = config.read_cfg(args.config)

commitizen/cmd.py

@@ -2,14 +2,15 @@
 
 import os
 import subprocess
-from typing import TYPE_CHECKING, NamedTuple
+import warnings
+from typing import TYPE_CHECKING, NamedTuple, overload
 
 from charset_normalizer import from_bytes
 
 from commitizen.exceptions import CharacterSetDecodeError
 
 if TYPE_CHECKING:
-    from collections.abc import Mapping
+    from collections.abc import Mapping, Sequence
 
 
 class Command(NamedTuple):
@@ -35,12 +36,18 @@ def _try_decode(bytes_: bytes) -> str:
         raise CharacterSetDecodeError() from e
 
 
-def run(cmd: str, env: Mapping[str, str] | None = None) -> Command:
+def _popen(
+    cmd: str | Sequence[str],
+    *,
+    shell: bool,
+    env: Mapping[str, str] | None = None,
+) -> Command:
     if env is not None:
         env = {**os.environ, **env}
+
     process = subprocess.Popen(
         cmd,
-        shell=True,
+        shell=shell,
         stdout=subprocess.PIPE,
         stderr=subprocess.PIPE,
         stdin=subprocess.PIPE,
@@ -55,3 +62,45 @@ def run(cmd: str, env: Mapping[str, str] | None = None) -> Command:
         stderr,
         return_code,
     )
+
+
+@overload
+def run(cmd: Sequence[str], env: Mapping[str, str] | None = None) -> Command: ...
+
+
+@overload
+def run(cmd: str, env: Mapping[str, str] | None = None) -> Command: ...
+
+
+def run(cmd: str | Sequence[str], env: Mapping[str, str] | None = None) -> Command:
+    """Run a command safely without shell interpretation (shell=False).
+
+    Arguments are passed directly to the OS, preventing shell-injection
+    vulnerabilities (CWE-78).
+
+    Passing a string is deprecated and will be removed in a future version.
+    Use a list of arguments instead, or use run_shell() for shell features.
+    """
+    if isinstance(cmd, str):
+        warnings.warn(
+            "Passing a string to cmd.run() is deprecated and will use shell=True. "
+            "Pass a list of arguments for shell=False (safe), "
+            "or use cmd.run_shell() explicitly for shell features.",
+            DeprecationWarning,
+            stacklevel=2,
+        )
+        return _popen(cmd, shell=True, env=env)
+    return _popen(cmd, shell=False, env=env)
+
+
+def run_shell(cmd: str, env: Mapping[str, str] | None = None) -> Command:
+    """Run a command string via the system shell (shell=True).
+
+    Only use this for cases that intentionally require shell features
+    (e.g., user-defined hooks with pipes/redirects). Never pass
+    untrusted/user-controlled values into *cmd*.
+
+    Related: CWE-78 (OS Command Injection),
+    https://github.com/commitizen-tools/commitizen/issues/1918
+    """
+    return _popen(cmd, shell=True, env=env)

commitizen/commands/bump.py

@@ -439,8 +439,8 @@ def __call__(self) -> None:
         else:
             out.success("Done!")
 
-    def _get_commit_args(self) -> str:
+    def _get_commit_args(self) -> list[str]:
         commit_args = ["-a"]
         if self.no_verify:
             commit_args.append("--no-verify")
-        return " ".join(commit_args)
+        return commit_args

commitizen/commands/changelog.py

@@ -257,7 +257,7 @@ def __call__(self) -> None:
                     changelog_meta.latest_version_position = None
                     changelog_meta.unreleased_end = latest_full_release_info.index + 1
 
-        commits = git.get_commits(start=start_rev, end=end_rev, args="--topo-order")
+        commits = git.get_commits(start=start_rev, end=end_rev, args=["--topo-order"])
         if (
             not self.allow_no_commit
             and not commits

commitizen/commands/commit.py

@@ -34,7 +34,7 @@ class CommitArgs(TypedDict, total=False):
     all: bool
     dry_run: bool
     edit: bool
-    extra_cli_args: str
+    extra_cli_args: list[str]
     message_length_limit: int
     no_retry: bool
     signoff: bool
@@ -132,7 +132,7 @@ def _get_message(self) -> str:
         return self._get_message_by_prompt_commit_questions()
 
     def __call__(self) -> None:
-        extra_args = self.arguments.get("extra_cli_args", "")
+        extra_args: list[str] = self.arguments.get("extra_cli_args", [])
         dry_run = bool(self.arguments.get("dry_run"))
         write_message_to_file = self.arguments.get("write_message_to_file")
         signoff = bool(self.arguments.get("signoff"))
@@ -167,7 +167,7 @@ def __call__(self) -> None:
             raise DryRunExit()
 
         if self.config.settings["always_signoff"] or signoff:
-            extra_args = f"{extra_args} -s".strip()
+            extra_args = [*extra_args, "-s"]
 
         c = git.commit(commit_message, args=extra_args)
         if c.return_code != 0:

commitizen/commands/init.py

@@ -134,11 +134,12 @@ def __call__(self) -> None:
                     "pre-commit is not installed in current environment."
                 )
 
-            cmd_str = "pre-commit install " + " ".join(
-                f"--hook-type {ty}" for ty in hook_types
-            )
-            c = cmd.run(cmd_str)
+            cmd_args = ["pre-commit", "install"]
+            for ty in hook_types:
+                cmd_args.extend(["--hook-type", ty])
+            c = cmd.run(cmd_args)
             if c.return_code != 0:
+                cmd_str = " ".join(cmd_args)
                 raise InitFailedError(
                     "Failed to install pre-commit hook.\n"
                     f"Error running {cmd_str}."

commitizen/git.py

@@ -5,10 +5,14 @@
 from functools import lru_cache
 from pathlib import Path
 from tempfile import NamedTemporaryFile
+from typing import TYPE_CHECKING
 
 from commitizen import cmd, out
 from commitizen.exceptions import GitCommandError
 
+if TYPE_CHECKING:
+    from collections.abc import Sequence
+
 
 class EOLType(Enum):
     """The EOL type from `git config core.eol`."""
@@ -19,7 +23,7 @@ class EOLType(Enum):
 
     @classmethod
     def for_open(cls) -> str:
-        c = cmd.run("git config core.eol")
+        c = cmd.run(["git", "config", "core.eol"])
         eol = c.out.strip().upper()
         return cls._char_for_open()[cls._safe_cast(eol)]
 
@@ -164,49 +168,47 @@ def tag(
     tag: str, annotated: bool = False, signed: bool = False, msg: str | None = None
 ) -> cmd.Command:
     if not annotated and not signed:
-        return cmd.run(f"git tag {tag}")
+        return cmd.run(["git", "tag", tag])
 
     # according to https://git-scm.com/book/en/v2/Git-Basics-Tagging,
     # we're not able to create lightweight tag with message.
     # by adding message, we make it a annotated tags
     option = "-s" if signed else "-a"  # The else case is for annotated tags
-    return cmd.run(f'git tag {option} {tag} -m "{msg or tag}"')
+    return cmd.run(["git", "tag", option, tag, "-m", msg or tag])
 
 
 def add(*args: str) -> cmd.Command:
-    return cmd.run(f"git add {' '.join(args)}")
+    return cmd.run(["git", "add", *args])
 
 
 def commit(
     message: str,
-    args: str = "",
+    args: Sequence[str] = (),
     committer_date: str | None = None,
 ) -> cmd.Command:
     f = NamedTemporaryFile("wb", delete=False)
     f.write(message.encode("utf-8"))
     f.close()
 
-    command = _create_commit_cmd_string(args, committer_date, f.name)
-    c = cmd.run(command)
-    os.unlink(f.name)
-    return c
+    cmd_args = ["git", "commit"]
+    if args:
+        cmd_args.extend(args)
+    cmd_args.extend(["-F", f.name])
 
+    env: dict[str, str] | None = None
+    if committer_date:
+        env = {"GIT_COMMITTER_DATE": committer_date}
 
-def _create_commit_cmd_string(args: str, committer_date: str | None, name: str) -> str:
-    command = f'git commit {args} -F "{name}"'
-    if not committer_date:
-        return command
-    if os.name != "nt":
-        return f"GIT_COMMITTER_DATE={committer_date} {command}"
-    # Using `cmd /v /c "{command}"` sets environment variables only for that command
-    return f'cmd /v /c "set GIT_COMMITTER_DATE={committer_date}&& {command}"'
+    c = cmd.run(cmd_args, env=env)
+    os.unlink(f.name)
+    return c
 
 
 def get_commits(
     start: str | None = None,
     end: str | None = None,
     *,
-    args: str = "",
+    args: Sequence[str] = (),
 ) -> list[GitCommit]:
     """Get the commits between start and end."""
     if end is None:
@@ -226,7 +228,10 @@ def get_filenames_in_commit(git_reference: str = "") -> list[str]:
 
     :returns: file names committed in the last commit by default or inside the passed git reference
     """
-    c = cmd.run(f"git show --name-only --pretty=format: {git_reference}")
+    cmd_args = ["git", "show", "--name-only", "--pretty=format:"]
+    if git_reference:
+        cmd_args.append(git_reference)
+    c = cmd.run(cmd_args)
     if c.return_code == 0:
         return c.out.strip().split("\n")
     raise GitCommandError(c.err)
@@ -237,15 +242,17 @@ def get_tags(
 ) -> list[GitTag]:
     inner_delimiter = "---inner_delimiter---"
     formatter = (
-        f'"%(refname:strip=2){inner_delimiter}'
+        f"%(refname:strip=2){inner_delimiter}"
         f"%(objectname){inner_delimiter}"
         f"%(creatordate:format:{dateformat}){inner_delimiter}"
-        f'%(object)"'
+        f"%(object)"
     )
-    extra = "--merged" if reachable_only else ""
+    cmd_args = ["git", "tag", f"--format={formatter}", "--sort=-creatordate"]
+    if reachable_only:
+        cmd_args.append("--merged")
     # Force the default language for parsing
     env = {"LC_ALL": "C", "LANG": "C", "LANGUAGE": "C"}
-    c = cmd.run(f"git tag --format={formatter} --sort=-creatordate {extra}", env=env)
+    c = cmd.run(cmd_args, env=env)
     if c.return_code != 0:
         if reachable_only and c.err == "fatal: malformed object name HEAD\n":
             # this can happen if there are no commits in the repo yet
@@ -262,55 +269,55 @@ def get_tags(
 
 
 def tag_exist(tag: str) -> bool:
-    c = cmd.run(f"git tag --list {tag}")
+    c = cmd.run(["git", "tag", "--list", tag])
     return tag in c.out
 
 
 def is_signed_tag(tag: str) -> bool:
-    return cmd.run(f"git tag -v {tag}").return_code == 0
+    return cmd.run(["git", "tag", "-v", tag]).return_code == 0
 
 
 def get_latest_tag_name() -> str | None:
-    c = cmd.run("git describe --abbrev=0 --tags")
+    c = cmd.run(["git", "describe", "--abbrev=0", "--tags"])
     if c.err:
         return None
     return c.out.strip()
 
 
 def get_tag_message(tag: str) -> str | None:
-    c = cmd.run(f"git tag -l --format='%(contents:subject)' {tag}")
+    c = cmd.run(["git", "tag", "-l", "--format=%(contents:subject)", tag])
     if c.err:
         return None
     return c.out.strip()
 
 
 def get_tag_names() -> list[str]:
-    c = cmd.run("git tag --list")
+    c = cmd.run(["git", "tag", "--list"])
     if c.err:
         return []
     return [tag for raw in c.out.split("\n") if (tag := raw.strip())]
 
 
 def find_git_project_root() -> Path | None:
-    c = cmd.run("git rev-parse --show-toplevel")
+    c = cmd.run(["git", "rev-parse", "--show-toplevel"])
     if c.err:
         return None
     return Path(c.out.strip())
 
 
 def is_staging_clean() -> bool:
     """Check if staging is clean."""
-    c = cmd.run("git diff --no-ext-diff --cached --name-only")
+    c = cmd.run(["git", "diff", "--no-ext-diff", "--cached", "--name-only"])
     return not bool(c.out)
 
 
 def is_git_project() -> bool:
-    c = cmd.run("git rev-parse --is-inside-work-tree")
+    c = cmd.run(["git", "rev-parse", "--is-inside-work-tree"])
     return c.out.strip() == "true"
 
 
 def get_core_editor() -> str | None:
-    c = cmd.run("git var GIT_EDITOR")
+    c = cmd.run(["git", "var", "GIT_EDITOR"])
     if c.out:
         return c.out.strip()
     return None
@@ -321,21 +328,31 @@ def smart_open(*args, **kwargs):  # type: ignore[no-untyped-def,unused-ignore] #
     return open(*args, newline=EOLType.for_open(), **kwargs)
 
 
-def _get_log_as_str_list(start: str | None, end: str, args: str) -> list[str]:
+def _get_log_as_str_list(start: str | None, end: str, args: Sequence[str]) -> list[str]:
     """Get string representation of each log entry"""
     delimiter = "----------commit-delimiter----------"
     log_format: str = "%H%n%P%n%s%n%an%n%ae%n%b"
     command_range = f"{start}..{end}" if start else end
-    command = f"git -c log.showSignature=False log --pretty={log_format}{delimiter} {args} {command_range}"
+    cmd_args = [
+        "git",
+        "-c",
+        "log.showSignature=False",
+        "log",
+        f"--pretty={log_format}{delimiter}",
+    ]
+    if args:
+        cmd_args.extend(args)
+    if command_range:
+        cmd_args.append(command_range)
 
-    c = cmd.run(command)
+    c = cmd.run(cmd_args)
     if c.return_code != 0:
         raise GitCommandError(c.err)
     return c.out.split(f"{delimiter}\n")
 
 
 def get_default_branch() -> str:
-    c = cmd.run("git symbolic-ref refs/remotes/origin/HEAD")
+    c = cmd.run(["git", "symbolic-ref", "refs/remotes/origin/HEAD"])
     if c.return_code != 0:
         raise GitCommandError(c.err)
     return c.out.strip()

commitizen/hooks.py

@@ -17,7 +17,7 @@ def run(hooks: str | list[str], _env_prefix: str = "CZ_", **env: object) -> None
     for hook in hooks:
         out.info(f"Running hook '{hook}'")
 
-        c = cmd.run(hook, env=_format_env(_env_prefix, env))
+        c = cmd.run_shell(hook, env=_format_env(_env_prefix, env))
 
         if c.out:
             out.write(c.out)