"""coding-agent-security-prompt-injection-mcp-secrets.py — audit coding-agent tool configs
for the patterns that widen an agent's blast radius.

Reads any of: Claude Code `.mcp.json`, `.claude/settings.json`, `.claude/settings.local.json`,
`~/.claude.json`; Gemini CLI `settings.json` (its `mcpServers` block); Codex `config.toml`
(`approval_policy`, `sandbox_mode`, `web_search`, `[mcp_servers.*]`, `[features.network_proxy]`).

Flags (severity in brackets):
  UNPINNED   [high]   stdio server fetched by npx/uvx/pipx without a version, or docker without a tag/digest
  SHELL      [medium] server launched through a shell or a curl|sh pipeline
  SECRET     [high]   a literal secret in env/headers instead of a ${VAR} reference or bearer_token_env_var
  PLAINTEXT  [high]   http:// endpoint that is not loopback
  BYPASS     [high]   Gemini trust:true; Claude enableAllProjectMcpServers / bypassPermissions;
                      Codex approval_policy=never with danger-full-access; dangerously_* proxy flags
  BROAD      [medium] bare Bash, Bash(*), bare WebFetch, WebFetch(domain:*), curl/wget allow rules,
                      whole-server MCP allow rules
  NETWORK    [medium] Codex network_access=true; [low] web_search="live"
  SANDBOX    [low]    Codex danger-full-access is [high]; Claude sandbox missing/off is [low]
  ADVISORY   [low]    no deny rule for secret files; no includeTools allowlist; SSE transport; hooks in
                      a shared project file; a host secret forwarded to a server

Usage:
    python coding-agent-security-prompt-injection-mcp-secrets.py --demo
    python coding-agent-security-prompt-injection-mcp-secrets.py --write-demo <dir>
    python coding-agent-security-prompt-injection-mcp-secrets.py <file-or-project-dir> [...]

Given a directory it looks for .mcp.json, .claude/settings.json, .claude/settings.local.json,
.gemini/settings.json and .codex/config.toml. Exit status: 2 if any high finding, 1 if any medium,
else 0. Standard library only; TOML needs Python 3.11+ (tomllib). It reads configuration and never
executes anything from it.
"""
from __future__ import annotations

import json
import os
import re
import sys
from urllib.parse import urlsplit

try:
    import tomllib  # Python 3.11+
except ImportError:  # pragma: no cover
    tomllib = None

SECRET_NAME = re.compile(r"TOKEN|SECRET|PASSW|KEY|AUTH|CREDENTIAL|PRIVATE", re.I)
SECRET_HEADER = re.compile(r"^(authorization|x-api-key|api-key|cookie|proxy-authorization)$", re.I)
REFERENCE = re.compile(r"^\s*(\$\{[A-Za-z_][A-Za-z0-9_]*(:-[^}]*)?\}|\$[A-Za-z_][A-Za-z0-9_]*|%[A-Za-z_][A-Za-z0-9_]*%)\s*$")
LOOPBACK = {"localhost", "127.0.0.1", "::1", "[::1]", "0.0.0.0"}
NPX_LIKE = {"npx", "bunx", "pnpx", "pnpm dlx", "yarn dlx"}
PY_RUNNERS = {"uvx", "pipx"}
SHELLS = {"sh", "bash", "zsh", "cmd", "cmd.exe", "powershell", "pwsh", "curl", "wget"}
LEVEL_ORDER = {"high": 0, "medium": 1, "low": 2}


class Audit:
    def __init__(self):
        self.findings: list[tuple[str, str, str, str]] = []

    def add(self, level, code, where, msg):
        self.findings.append((level, code, where, msg))

    # ---------------------------------------------------------------- servers ---
    def server(self, where, s: dict, flavour: str):
        cmd = str(s.get("command", "") or "")
        args = [str(a) for a in (s.get("args") or [])]
        if cmd:
            self.launch(where, cmd, args)
        for field in ("env",):
            for k, v in (s.get(field) or {}).items():
                self.secret_value(where + f".env.{k}", k, v, name_based=True)
        for field in ("headers", "http_headers"):
            for k, v in (s.get(field) or {}).items():
                self.secret_value(where + f".{field}.{k}", k, v, name_based=False)
        for name in s.get("env_vars") or []:  # Codex: host variables whitelisted through to the server
            if SECRET_NAME.search(str(name)):
                self.add("low", "ADVISORY", where, f"host variable {name} is forwarded to the server process")
        for field in ("url", "httpUrl"):
            u = s.get(field)
            if u:
                self.endpoint(where + f".{field}", str(u))
        if str(s.get("type", "")).lower() == "sse" or ("url" in s and "httpUrl" not in s and flavour == "gemini"):
            self.add("low", "ADVISORY", where, "SSE transport; the Claude Code docs call it deprecated, prefer streamable HTTP")
        if s.get("trust") is True:
            self.add("high", "BYPASS", where, "trust: true bypasses every tool-call confirmation for this server")
        if flavour == "gemini" and not s.get("includeTools"):
            self.add("low", "ADVISORY", where, "no includeTools allowlist; every tool the server exposes is available")

    def launch(self, where, cmd, args):
        base = os.path.basename(cmd).lower()
        flags_done = [a for a in args if not a.startswith("-")]
        if base in NPX_LIKE:
            spec = flags_done[0] if flags_done else ""
            if spec and not re.search(r"[^@]@[^/]+$", spec):
                self.add("high", "UNPINNED", where,
                         f"{base} resolves \"{spec}\" to whatever version is newest at launch; pin it ({spec}@x.y.z)")
            if "-y" in args or "--yes" in args:
                self.add("low", "ADVISORY", where, f"{base} -y installs the package without a confirmation")
        elif base in PY_RUNNERS:
            spec = flags_done[0] if flags_done else ""
            if "--from" in args:
                i = args.index("--from")
                spec = args[i + 1] if i + 1 < len(args) else spec
            if spec and not re.search(r"(==|@[^/]+$)", spec):
                self.add("high", "UNPINNED", where,
                         f"{base} resolves \"{spec}\" to the newest release at launch; pin it ({spec}==x.y.z)")
        elif base == "docker" and "run" in args:
            image = next((a for a in args[args.index("run") + 1:]
                          if not a.startswith("-") and "=" not in a and ("/" in a or ":" in a or "." in a)), "")
            if image and "@sha256:" not in image and (":" not in image.rsplit("/", 1)[-1] or image.endswith(":latest")):
                self.add("medium", "UNPINNED", where, f"docker image \"{image}\" has no tag or digest")
        elif base in SHELLS:
            self.add("medium", "SHELL", where, f"server is launched through {base}; the command line is arbitrary code")
        joined = " ".join(args)
        if re.search(r"(curl|wget)[^|]*\|\s*(sh|bash|zsh|pwsh|powershell)", joined):
            self.add("high", "SHELL", where, "a download is piped straight into a shell at server start")

    def secret_value(self, where, key, value, name_based):
        v = str(value)
        looks_secret = SECRET_NAME.search(key) if name_based else SECRET_HEADER.match(key)
        if not looks_secret or not v.strip():
            return
        if REFERENCE.match(v) or REFERENCE.match(re.sub(r"^\s*bearer\s+", "", v, flags=re.I)):
            self.add("low", "ADVISORY", where, f"{key} is passed through to the server by reference (fine if it needs it)")
        else:
            self.add("high", "SECRET", where,
                     f"{key} holds a literal value in the config file; use a ${{VAR}} reference or bearer_token_env_var")

    def endpoint(self, where, url):
        parts = urlsplit(url)
        host = (parts.hostname or "").lower()
        if parts.scheme == "http" and host not in LOOPBACK and not url.startswith("${"):
            self.add("high", "PLAINTEXT", where, f"http:// endpoint {host or url} is not loopback; tokens and tool results cross the wire in clear")

    # ------------------------------------------------------- Claude settings ---
    def claude_settings(self, where, d: dict, shared_project: bool):
        if d.get("enableAllProjectMcpServers") is True:
            self.add("high", "BYPASS", where, "enableAllProjectMcpServers approves every server in .mcp.json without a prompt")
        perms = d.get("permissions") or {}
        mode = perms.get("defaultMode")
        if mode == "bypassPermissions":
            self.add("high", "BYPASS", where + ".permissions.defaultMode", "bypassPermissions: nothing asks; the docs reserve it for isolated containers and VMs")
        for rule in perms.get("allow") or []:
            r = str(rule).strip()
            tool, _, spec = r.partition("(")
            spec = spec.rstrip(")")
            loc = where + f".permissions.allow[{r}]"
            if tool == "Bash" and spec in ("", "*"):
                self.add("medium", "BROAD", loc, "every shell command runs without a prompt")
            elif tool == "Bash" and re.match(r"^(curl|wget|Invoke-WebRequest|iwr)\b", spec):
                self.add("medium", "BROAD", loc, "network fetch commands are pre-approved; fetched text becomes tool output the model acts on")
            elif tool == "WebFetch" and spec in ("", "domain:*"):
                self.add("medium", "BROAD", loc, "any URL can be fetched without a prompt")
            elif tool.startswith("mcp__") and (tool.count("__") == 1 or tool.endswith("__*")):
                self.add("medium", "BROAD", loc, "approves every tool of that server, including tools it adds later")
        deny = [str(x) for x in perms.get("deny") or []]
        if not any(re.search(r"\.env|secret|credential|\.pem|id_rsa", x, re.I) for x in deny):
            self.add("low", "ADVISORY", where + ".permissions.deny", "no deny rule for secret files (the docs' example is Read(./.env))")
        sb = d.get("sandbox")
        if sb is None:
            self.add("low", "SANDBOX", where, "no sandbox block; the permission rules are the only boundary for Bash")
        elif sb.get("enabled") is False:
            self.add("low", "SANDBOX", where + ".sandbox.enabled", "sandbox explicitly off")
        if shared_project and d.get("hooks"):
            self.add("low", "ADVISORY", where + ".hooks", "hooks in a shared project file run on every clone, including -p sessions without --bare")

    # ---------------------------------------------------------------- Codex ---
    def codex(self, where, d: dict):
        ap, sm = d.get("approval_policy"), d.get("sandbox_mode")
        if sm == "danger-full-access":
            lvl = "high"
            self.add(lvl, "BYPASS" if ap == "never" else "SANDBOX", where,
                     "approval_policy=never with danger-full-access: no prompt and no sandbox" if ap == "never"
                     else "danger-full-access: full filesystem and network access during command execution")
        elif ap == "never":
            self.add("medium", "BYPASS", where + ".approval_policy", "never: nothing prompts; safe only inside the sandbox it runs in")
        if (d.get("sandbox_workspace_write") or {}).get("network_access") is True:
            self.add("medium", "NETWORK", where + ".sandbox_workspace_write.network_access", "network on for sandboxed commands; the docs keep it off by default")
        if d.get("web_search") == "live":
            self.add("low", "NETWORK", where + ".web_search", "live web results reach the model; the docs default to a cache and say to treat results as untrusted")
        proxy = (d.get("features") or {}).get("network_proxy") or {}
        for k in ("dangerously_allow_non_loopback_proxy", "dangerously_allow_all_unix_sockets"):
            if proxy.get(k) is True:
                self.add("high", "BYPASS", where + f".features.network_proxy.{k}", "the docs say to use this only in tightly controlled environments")
        for name, s in (d.get("mcp_servers") or {}).items():
            self.server(where + f".mcp_servers.{name}", s, "codex")


# ------------------------------------------------------------------ drivers ---
def audit_json(a: Audit, label: str, d: dict):
    low = label.replace("\\", "/").lower()
    flavour = "gemini" if "/.gemini/" in low or low.endswith("/.gemini/settings.json") else "claude"
    for name, s in (d.get("mcpServers") or {}).items():
        a.server(f"{label}:mcpServers.{name}", s, flavour)
    for proj, pd in (d.get("projects") or {}).items():  # ~/.claude.json per-project servers
        for name, s in (pd.get("mcpServers") or {}).items():
            a.server(f"{label}:projects[{proj}].mcpServers.{name}", s, "claude")
    if any(k in d for k in ("permissions", "sandbox", "enableAllProjectMcpServers", "hooks")):
        a.claude_settings(label, d, shared_project=low.endswith("/.claude/settings.json") or low.endswith("settings.json") and "local" not in low)


def audit_path(a: Audit, path: str, label: str | None = None):
    label = label or path
    if path.endswith(".toml"):
        if tomllib is None:
            a.add("low", "ADVISORY", label, "TOML needs Python 3.11+; file skipped")
            return
        with open(path, "rb") as fh:
            a.codex(label, tomllib.load(fh))
        return
    with open(path, encoding="utf-8") as fh:
        audit_json(a, label, json.load(fh))


KNOWN = [".mcp.json", ".claude/settings.json", ".claude/settings.local.json", ".gemini/settings.json", ".codex/config.toml"]

DEMO = {
    ".mcp.json": json.dumps({"mcpServers": {
        "docs": {"type": "stdio", "command": "uvx", "args": ["acme-docs-mcp==2.1.0"]},
        "scraper": {"type": "stdio", "command": "npx", "args": ["-y", "acme-scraper-mcp"],
                    "env": {"ACME_API_KEY": "ak_live_9f3c2e7d1b4a"}},
        "tracker": {"type": "http", "url": "http://tracker.corp.example:8080/mcp",
                    "headers": {"Authorization": "Bearer tr_8d1e5c0a"}},
        "issues": {"type": "http", "url": "https://mcp.example.com/mcp",
                   "headers": {"Authorization": "Bearer ${ISSUES_TOKEN}"}}}}, indent=2),
    ".claude/settings.json": json.dumps({
        "enableAllProjectMcpServers": True,
        "permissions": {"defaultMode": "acceptEdits",
                        "allow": ["Bash(python -m pytest *)", "Bash(curl *)", "WebFetch", "mcp__scraper__*"],
                        "deny": []}}, indent=2),
    ".gemini/settings.json": json.dumps({"mcpServers": {
        "browser": {"command": "npx", "args": ["-y", "acme-browser-mcp@0.9.3"], "trust": True,
                    "env": {"BROWSER_AUTH_TOKEN": "$BROWSER_AUTH_TOKEN"}}}}, indent=2),
    ".codex/config.toml": (
        'approval_policy = "never"\n'
        'sandbox_mode = "danger-full-access"\n'
        'web_search = "live"\n\n'
        '[mcp_servers.ci]\n'
        'command = "npx"\n'
        'args = ["-y", "acme-ci-mcp"]\n'
        'env = { CI_TOKEN = "ci_4b7e2a9d" }\n\n'
        '[mcp_servers.wiki]\n'
        'url = "https://wiki.example.com/mcp"\n'
        'bearer_token_env_var = "WIKI_TOKEN"\n'),
}


def run_demo(a: Audit):
    for name, text in DEMO.items():
        label = "demo/" + name
        if name.endswith(".toml"):
            a.codex(label, tomllib.loads(text))
        else:
            audit_json(a, label, json.loads(text))


def main(argv):
    a = Audit()
    if not argv or argv[0] in ("-h", "--help"):
        print(__doc__)
        return 0
    if argv[0] == "--write-demo":
        root = argv[1] if len(argv) > 1 else "agent-config-demo"
        for name, text in DEMO.items():
            p = os.path.join(root, *name.split("/"))
            os.makedirs(os.path.dirname(p), exist_ok=True)
            with open(p, "w", encoding="utf-8") as fh:
                fh.write(text + "\n")
            print("wrote", p)
        return 0
    if argv[0] == "--demo":
        run_demo(a)
    else:
        for target in argv:
            if os.path.isdir(target):
                found = [os.path.join(target, *k.split("/")) for k in KNOWN if os.path.isfile(os.path.join(target, *k.split("/")))]
                if not found:
                    print(f"{target}: none of {', '.join(KNOWN)} present")
                for p in found:
                    audit_path(a, p)
            else:
                audit_path(a, target)
    a.findings.sort(key=lambda f: (LEVEL_ORDER[f[0]], f[2]))
    for level, code, where, msg in a.findings:
        print(f"{level.upper():<7}{code:<10}{where}\n        {msg}")
    counts = {lvl: sum(1 for f in a.findings if f[0] == lvl) for lvl in LEVEL_ORDER}
    print(f"\n{len(a.findings)} findings: {counts['high']} high, {counts['medium']} medium, {counts['low']} low")
    return 2 if counts["high"] else 1 if counts["medium"] else 0


if __name__ == "__main__":
    sys.exit(main(sys.argv[1:]))