fix(git): pin trusted executables across scan hosts

sdk/typescript/_bundled_plugin/.codex-plugin/plugin.json

@@ -1,6 +1,6 @@
 {
   "name": "codex-security",
-  "version": "0.1.20",
+  "version": "0.1.22",
   "description": "Codex Security workflows for security scans, analysis, and investigation.",
   "author": {
     "name": "OpenAI"

sdk/typescript/_bundled_plugin/.mcp.json

@@ -28,6 +28,8 @@
         "AWS_CONTAINER_AUTHORIZATION_TOKEN",
         "AWS_CONTAINER_AUTHORIZATION_TOKEN_FILE",
         "PYTHON",
+        "CODEX_SECURITY_GIT",
+        "CODEX_SECURITY_RG",
         "CODEX_SECURITY_KNOWLEDGE_BASE",
         "CODEX_SECURITY_DEEP_SCAN_CONFIG_PATH",
         "CODEX_SECURITY_SCAN_ROOT",

sdk/typescript/_bundled_plugin/scripts/generate_in_scope_files.py

@@ -10,6 +10,10 @@
 import tempfile
 from pathlib import Path
 
+# Some plugin hosts launch Python with safe-path isolation enabled.
+sys.path.insert(0, str(Path(__file__).resolve().parent))
+from workbench_target import git_command, ripgrep_command
+
 
 class InventoryError(ValueError):
     """Raised when the repository, scope, or inventory cannot be used safely."""
@@ -70,7 +74,6 @@ def resolve_output(value: str) -> Path:
 def generate_in_scope_files(repository: Path, scope: str, output: Path) -> int:
     """Atomically write the exact ripgrep inventory sorted as ``LC_ALL=C``."""
     command = [
-        "rg",
         "--files",
         "--hidden",
         "--no-ignore",
@@ -83,13 +86,7 @@ def generate_in_scope_files(repository: Path, scope: str, output: Path) -> int:
     ]
     with tempfile.TemporaryFile(mode="w+b") as inventory:
         try:
-            result = subprocess.run(
-                command,
-                cwd=repository,
-                stdout=inventory,
-                stderr=subprocess.PIPE,
-                check=False,
-            )
+            result = ripgrep_command(repository, *command, stdout=inventory)
         except OSError as error:
             raise InventoryError(f"could not run ripgrep: {error}") from error
 
@@ -107,20 +104,16 @@ def generate_in_scope_files(repository: Path, scope: str, output: Path) -> int:
 
 
 def committed_changed_paths(repository: Path, base: str, head: str) -> list[tuple[Path, str]]:
-    result = subprocess.run(
-        [
-            "git",
-            "-C",
-            str(repository),
-            "diff",
-            "--raw",
-            "-z",
-            "--diff-filter=ACMRD",
-            f"{base}..{head}",
-        ],
-        capture_output=True,
-        check=True,
+    result = git_command(
+        repository,
+        "diff",
+        "--raw",
+        "-z",
+        "--diff-filter=ACMRD",
+        f"{base}..{head}",
+        text=False,
     )
+    result.check_returncode()
     fields = result.stdout.split(b"\0")
     changed: list[tuple[Path, str]] = []
     index = 0
@@ -146,7 +139,6 @@ def generate_diff_in_scope_files(
     output: Path,
 ) -> int:
     """Reuse the existing diff selection without generating previews or duplicate worklists."""
-    sys.path.insert(0, str(Path(__file__).resolve().parent))
     from generate_rank_input import git_changed_paths, path_is_excluded
     from rank_preview import (
         DEFAULT_PREVIEW_BYTES,

sdk/typescript/_bundled_plugin/scripts/generate_rank_input.py

@@ -29,7 +29,6 @@
 import json
 import os
 import re
-import subprocess
 import sys
 from collections import Counter
 from collections.abc import Callable
@@ -43,7 +42,12 @@
     preview_for,
     preview_for_bytes,
 )
-from workbench_target import git_blob_bytes, git_directory_snapshot_paths
+from workbench_target import (
+    git_blob_bytes,
+    git_command,
+    git_directory_snapshot_paths,
+    ripgrep_command,
+)
 
 EXCLUDED_DIRS = {
     ".cache",
@@ -521,7 +525,6 @@ def make_repo_scope_input(args: argparse.Namespace) -> None:
                 candidates = git_candidates
             else:
                 command = [
-                    "rg",
                     "--files",
                     "--hidden",
                     "--no-require-git",
@@ -532,7 +535,7 @@ def make_repo_scope_input(args: argparse.Namespace) -> None:
                     str(scope_path.relative_to(repo)),
                 ]
                 try:
-                    result = subprocess.run(command, cwd=repo, capture_output=True, check=False)
+                    result = ripgrep_command(repo, *command)
                 except OSError as exc:
                     ignore_names = (".gitignore", ".ignore", ".rgignore")
                     ancestors = (scope_path, *scope_path.parents)
@@ -608,21 +611,16 @@ def bind_repo_scopes(args: argparse.Namespace) -> None:
 
 
 def run_git_changed_paths(repo: Path, diff_args: list[str]) -> list[tuple[Path, str]]:
-    result = subprocess.run(
-        [
-            "git",
-            "-C",
-            str(repo),
-            "diff",
-            "--name-status",
-            "-z",
-            "--diff-filter=ACMRD",
-            *diff_args,
-        ],
-        check=True,
-        capture_output=True,
+    result = git_command(
+        repo,
+        "diff",
+        "--name-status",
+        "-z",
+        "--diff-filter=ACMRD",
+        *diff_args,
         text=True,
     )
+    result.check_returncode()
     fields = result.stdout.split("\0")
     if fields and not fields[-1]:
         fields.pop()
@@ -646,12 +644,15 @@ def git_changed_paths(repo: Path, base: str, head: str, mode: str) -> list[tuple
     if mode == "local-patch":
         unstaged = run_git_changed_paths(repo, [base])
         staged = run_git_changed_paths(repo, ["--cached", base])
-        untracked = subprocess.run(
-            ["git", "-C", str(repo), "ls-files", "--others", "--exclude-standard", "-z"],
-            capture_output=True,
+        untracked = git_command(
+            repo,
+            "ls-files",
+            "--others",
+            "--exclude-standard",
+            "-z",
             text=True,
-            check=True,
         )
+        untracked.check_returncode()
         combined = dict(staged)
         combined.update(unstaged)
         combined.update(

sdk/typescript/_bundled_plugin/scripts/workbench_target.py

@@ -11,7 +11,7 @@
 import subprocess
 import sys
 from pathlib import Path
-from typing import Any
+from typing import IO, Any
 
 # Some plugin hosts launch Python with safe-path isolation enabled.
 sys.path.insert(0, str(Path(__file__).resolve().parent))
@@ -120,6 +120,137 @@ def _read_sized_nul_field(
     return output[offset:end], end + 1
 
 
+def _protected_git_root(target: Path) -> Path | None:
+    """Return the outermost repository root, or None for a stale target."""
+    try:
+        root = target.resolve(strict=True)
+        if not stat.S_ISDIR(root.stat().st_mode):
+            return None
+        for ancestor in (root, *root.parents):
+            try:
+                (ancestor / ".git").lstat()
+            except FileNotFoundError:
+                continue
+            root = ancestor
+    except (FileNotFoundError, NotADirectoryError):
+        return None
+    return root
+
+
+def _inside_protected_git_root(candidate: Path, root: Path) -> bool:
+    return candidate.is_relative_to(root) or (
+        len(candidate.parts) >= len(root.parts)
+        and Path(*candidate.parts[: len(root.parts)]).samefile(root)
+    )
+
+
+def _is_native_executable(candidate: Path, canonical: Path) -> bool:
+    windows = sys.platform == "win32"
+    return (
+        canonical.is_file()
+        and (
+            not windows
+            or (
+                candidate.suffix.lower() in {".exe", ".com"}
+                and canonical.suffix.lower() not in {".bat", ".cmd"}
+            )
+        )
+        and os.access(canonical, os.F_OK if windows else os.X_OK)
+    )
+
+
+def _trusted_executable(
+    target: Path,
+    environment: dict[str, str],
+    name: str,
+) -> str | None:
+    root = _protected_git_root(target)
+    if root is None:
+        return None
+    setting = f"CODEX_SECURITY_{name.upper()}"
+    configured = environment.get(setting)
+    if configured is not None:
+        if not configured:
+            return None
+        candidate = Path(configured)
+        if not candidate.is_absolute():
+            raise SystemExit(f"{setting} must name an absolute trusted executable.")
+        try:
+            canonical = candidate.resolve(strict=True)
+            if _inside_protected_git_root(canonical, root) or any(
+                _inside_protected_git_root(ancestor.resolve(strict=True), root)
+                for ancestor in candidate.parents
+            ):
+                raise SystemExit(f"{setting} must stay outside the protected repository.")
+        except OSError as error:
+            raise SystemExit(f"{setting} does not name an available executable.") from error
+        if not _is_native_executable(candidate, canonical):
+            raise SystemExit(f"{setting} does not name an available executable.")
+        return configured
+
+    entries: list[str] = []
+    executable: str | None = None
+    names = (f"{name}.exe", f"{name}.com") if sys.platform == "win32" else (name,)
+    if sys.platform == "win32":
+        path_keys = sorted(key for key in environment if key.upper() == "PATH")
+        if path_keys:
+            path = environment[path_keys[0]]
+            for key in path_keys:
+                del environment[key]
+            environment["PATH"] = path
+    for entry in os.get_exec_path(environment):
+        if not entry:
+            continue
+        try:
+            directory = Path(entry).resolve(strict=True)
+            if _inside_protected_git_root(directory, root):
+                continue
+        except OSError:
+            continue
+        candidate: str | None = None
+        safe = True
+        for name in names:
+            path = directory / name
+            try:
+                canonical = path.resolve(strict=True)
+                if _inside_protected_git_root(canonical, root):
+                    safe = False
+                    break
+            except OSError:
+                continue
+            if _is_native_executable(path, canonical):
+                candidate = candidate or str(path)
+        if not safe:
+            continue
+        executable = executable or candidate
+        entries.append(str(directory))
+    environment["PATH"] = os.pathsep.join(entries)
+    return executable
+
+
+def _trusted_git_executable(target: Path, environment: dict[str, str]) -> str | None:
+    return _trusted_executable(target, environment, "git")
+
+
+def ripgrep_command(
+    target: Path,
+    *args: str,
+    stdout: IO[bytes] | int = subprocess.PIPE,
+) -> subprocess.CompletedProcess[bytes]:
+    environment = os.environ.copy()
+    executable = _trusted_executable(target, environment, "rg")
+    if executable is None:
+        raise FileNotFoundError("ripgrep is not available on a trusted PATH.")
+    return subprocess.run(
+        [executable, *args],
+        cwd=target,
+        stdout=stdout,
+        stderr=subprocess.PIPE,
+        env=environment,
+        check=False,
+    )
+
+
 def git_command(
     target: Path,
     *args: str,
@@ -134,25 +265,28 @@ def git_command(
     for name in GIT_REPOSITORY_ENVIRONMENT:
         environment.pop(name, None)
     environment["GIT_LITERAL_PATHSPECS"] = "1"
+    executable = _trusted_git_executable(target, environment)
     # Repository-local config is untrusted; fsmonitor may name an executable hook.
-    command = ["git", "-c", "core.fsmonitor=false", "-C", str(target)]
+    command = [executable or "git", "-c", "core.fsmonitor=false", "-C", str(target)]
     if git_dir is not None and work_tree is not None:
         command.extend(["--git-dir", str(git_dir), "--work-tree", str(work_tree)])
     full_command = [*command, *args]
-    try:
-        return subprocess.run(
-            full_command,
-            check=False,
-            capture_output=True,
-            env=environment,
-            text=text,
-            input=input_data,
-        )
-    except FileNotFoundError:
-        # Git is optional for Codebase scans. Treat an unavailable executable like
-        # any other failed Git probe so the target falls back to a directory snapshot.
-        empty_output = "" if text else b""
-        return subprocess.CompletedProcess(full_command, 127, empty_output, empty_output)
+    if executable is not None:
+        try:
+            return subprocess.run(
+                full_command,
+                check=False,
+                capture_output=True,
+                env=environment,
+                text=text,
+                input=input_data,
+            )
+        except FileNotFoundError:
+            pass
+    # Git is optional for Codebase scans. Treat an unavailable executable like
+    # any other failed Git probe so the target falls back to a directory snapshot.
+    empty_output = "" if text else b""
+    return subprocess.CompletedProcess(full_command, 127, empty_output, empty_output)
 
 
 def update_digest_field(digest: Any, label: bytes, value: bytes) -> None: