diff --git a/src/kiro_crew/mcp_cron.py b/src/kiro_crew/mcp_cron.py index 2db6b8a63be..72c9f11fad5 100644 --- a/src/kiro_crew/mcp_cron.py +++ b/src/kiro_crew/mcp_cron.py @@ -17,6 +17,7 @@ from __future__ import annotations +import ast import fnmatch import logging import os @@ -64,6 +65,7 @@ is_sensitive_bash_command, is_sensitive_path, is_sensitive_source_body, + is_shell_payload_literal, scan_exfiltration_urls, ) from kiro_crew.sel import sel @@ -754,6 +756,411 @@ def _unquote(s: str) -> str: return None +def _shell_scannable_literals(text: str) -> tuple[bool, list[str]]: + """(parses, the string literals of *text* worth scanning as shell commands). + + A shell payload embedded in Python lives in a string literal, and a literal + is exactly the text a shell would receive — so it is the one part of a source + body the execution-model shell passes are SOUND on. Docstrings are excluded + because they are prose and the shell modeling fabricates on prose (see + :func:`_vet_script_contents`); a docstring is the first statement of a + module, class or function body, per the compiler's own definition. F-string + fragments are plain ``ast.Constant`` strings inside a ``JoinedStr`` and are + included, so a payload split around an interpolation still shows its parts. + Literals are deduplicated, order-preserving. ``parses=False`` means the body + is not valid Python and yields no literals — the caller falls back to + scanning the raw text. + """ + try: + tree = ast.parse(text) + except (SyntaxError, ValueError, RecursionError): + return False, [] + docstrings: set[int] = set() + for node in ast.walk(tree): + if isinstance(node, (ast.Module, ast.ClassDef, ast.FunctionDef, ast.AsyncFunctionDef)): + body = getattr(node, "body", []) + if ( + body + and isinstance(body[0], ast.Expr) + and isinstance(body[0].value, ast.Constant) + and isinstance(body[0].value.value, str) + ): + docstrings.add(id(body[0].value)) + seen: set[str] = set() + literals: list[str] = [] + for node in ast.walk(tree): + if ( + isinstance(node, ast.Constant) + and isinstance(node.value, str) + and id(node) not in docstrings + and node.value.strip() + and node.value not in seen + ): + seen.add(node.value) + literals.append(node.value) + return True, literals + + +#: Callable names that hand their command to a shell. The ``run`` family is a +#: sink only with ``shell=True``; ``system``/``popen``/``getoutput``/ +#: ``getstatusoutput``/``create_subprocess_shell`` always are. Matched on the +#: call's LAST name segment so both ``subprocess.run`` and a +#: ``from subprocess import run`` spelling hit. ``create_subprocess_exec`` is +#: deliberately absent: it takes an argv list and starts no shell, the same +#: reason the subprocess argv-list form is a residual, not a sink. +_SHELL_SINK_ALWAYS = frozenset( + {"system", "popen", "getoutput", "getstatusoutput", "create_subprocess_shell"} +) +_SHELL_SINK_WITH_FLAG = frozenset({"run", "call", "check_call", "check_output", "Popen"}) +#: Modules whose sink attributes the walk tracks. ``asyncio`` is here for +#: ``create_subprocess_shell`` (also importable from ``asyncio.subprocess``); +#: ``asyncio.run`` is NOT a shell sink -- it shares a name with subprocess's +#: flag-gated sink, and with no ``shell`` spelling and fewer than 9 positionals +#: the flag rule already answers False, so ordinary async crons stay vettable. +_SHELL_SINK_MODULES = ("os", "subprocess", "asyncio") + +#: Characters no shell REWRITES: a sink literal built solely from these is +#: executed as written on POSIX shells and cmd.exe alike, so the textual gate's +#: verdict on it is a verdict on what runs. Everything else -- `$`/backtick +#: (POSIX expansion/substitution), `%`/`!`/`^` (cmd.exe variables, delayed +#: expansion, escapes), `*`/`?`/`[` (pathname expansion: `cat ~/.ss*/id_rsa` +#: reads the key while naming no fenced path), quotes and backslashes +#: (dequoting splices: `~/.s"s"h`), and redirection/pipe/separator operators -- +#: lets the executed text differ from the scanned text. An ALLOWLIST, not a +#: denylist of rewrite operators: shells differ and grow, so enumerating the +#: dangerous set is one unenumerated operator away from reopening the class, +#: while the safe set is small and closed. `~` is included: tilde expansion is +#: a rewrite, but to a path the fence matchers already model (they match +#: home-anchored spellings), so it cannot CONCEAL -- excluding it would refuse +#: nearly every benign command for no coverage. +_SHELL_VERBATIM_CHARS = frozenset( + "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789" " -_./:=,+@~" +) + + +def _shell_executes_verbatim(text: str) -> bool: + """True when every character is one no shell rewrites (see the set).""" + return all(ch in _SHELL_VERBATIM_CHARS for ch in text) + + +def _attr_chain_root(node: ast.Attribute) -> str | None: + """The base Name of a pure attribute chain, or None. + + ``asyncio.subprocess.create_subprocess_shell`` reaches the sink through a + NESTED attribute whose ``value`` is itself an Attribute; resolving the + chain to its root Name lets both the call matcher and the escape-as-value + forfeits see it without enumerating chain depths. + """ + value: ast.expr = node.value + while isinstance(value, ast.Attribute): + value = value.value + return value.id if isinstance(value, ast.Name) else None + + +def _dynamic_shell_sink(text: str) -> bool: + """True when a shell-execution call takes a command this scan cannot read. + + The literal scan (:func:`_shell_scannable_literals`) judges every string + the source can hand to a shell — but only strings that exist in the source. + A command COMPOSED at runtime (``verb + tail``, an f-string, ``__doc__``, a + variable) reaches ``shell=True`` with no individually-blocking literal, so + a sink whose command argument is not a plain string literal is refused + outright: it cannot be statically vetted, and the honest answer is the same + fail-closed one the shell gate gives its own analysis budgets. A literal + argument is fine — the literal scan already judged it. An ``args`` LIST + (no shell) is not a sink here; see the residual note in + :func:`_vet_script_contents`. + + Recognition is MODULE-QUALIFIED: attribute calls on the os/subprocess/ + asyncio modules (``asyncio.create_subprocess_shell`` is an always-shell + sink; nested chains like ``asyncio.subprocess.create_subprocess_shell`` + resolve through :func:`_attr_chain_root`) and bare names + imported FROM those modules (``from subprocess import run``). An unrelated + method that merely shares a sink's name (``renderer.run(job, shell=theme)``) + is not a sink. ASSIGNMENT aliasing is closed as a CLASS rather than by + chasing spellings (the closure #7913 established for module authenticity): + a sink-capable value may only be CALLED, and any reference that ESCAPES as + a value forfeits the whole body — ``r = subprocess.run``, ``x = subprocess``, + ``getattr(subprocess, "run")``, ``holder = [os.system]`` and every container/ + argument/return spelling all put a shell-capable callable somewhere no + static walk can follow, so each fails closed on the mention itself. Plain + attribute READS on the modules (``os.environ``, ``os.path.join``) are + unaffected: only the module object itself as a bare value, a SINK attribute + outside call position, and a from-imported SINK name outside call position + forfeit. The remaining residual is acquisition that never mentions a watched + name at all — ``importlib.import_module("subprocess")``, ``sys.modules`` + string routes, ``exec`` — which is invisible to every static text scan (an + ``exec`` body needs no shell sink to read files in pure Python) and was + equally open on base. An unparseable body answers False: the caller's + fallback scans it raw, with shell grammar, instead. + """ + try: + tree = ast.parse(text) + except (SyntaxError, ValueError, RecursionError): + return False + # Sink recognition is MODULE-QUALIFIED, not name-shaped: `renderer.run(job, + # shell=theme)` and `client.system(payload)` are ordinary application calls + # and must not be misread as shell sinks. A call counts only when its target + # is an attribute of a tracked shell module (through `import + # subprocess as sp` aliases) or a bare name imported FROM one of them + # (`from subprocess import run`, `from os import system`, aliased or not). + # Assignment aliasing is closed by the escape-as-value forfeits below. + module_aliases: set[str] = set() + imported_names: dict[str, str] = {} + import_roots: set[str] = set() + for node in ast.walk(tree): + if isinstance(node, ast.Import): + for alias in node.names: + import_roots.add(alias.asname or alias.name.split(".")[0]) + if alias.name in _SHELL_SINK_MODULES: + module_aliases.add(alias.asname or alias.name) + elif alias.name.split(".")[0] in _SHELL_SINK_MODULES: + # `import os.path` (no asname) binds the TOP-LEVEL name + # `os`, so `os.system` is reachable through it; with an + # asname (`import os.path as p`) only the submodule is + # bound and the top-level name is not. `asyncio.subprocess` + # is the one submodule that CARRIES a sink, so its asname + # binds a sink-carrying module and is tracked too. + if alias.asname is None: + module_aliases.add(alias.name.split(".")[0]) + elif alias.name == "asyncio.subprocess": + module_aliases.add(alias.asname) + elif isinstance(node, ast.ImportFrom): + for alias in node.names: + if alias.name == "*": + # A wildcard import binds a set this walk cannot + # enumerate -- and for ANY source module, not only the + # tracked ones: a module without ``__all__`` re-exports + # every imported module it holds (``from glob import *`` + # would bind ``os`` on such a module), so the sink + # carriers can arrive under their own names, recorded + # nowhere. The forfeit condition is the UNKNOWABLE + # binding set itself (#7913's wildcard closure), so every + # wildcard import fails closed outright. + return True + if alias.name in _SHELL_SINK_MODULES: + # A from-imported name that IS a tracked module name binds + # (presumptively) that MODULE, whatever module re-exports + # it: `from asyncio import subprocess [as asp]`, + # `from subprocess import os as o` (subprocess imports + # os), `from shutil import os` -- the source module is + # irrelevant to what arrives. Fail closed on the NAME: a + # same-named non-module attribute only makes vetting + # stricter, never looser. + module_aliases.add(alias.asname or alias.name) + continue + if node.module in _SHELL_SINK_MODULES or node.module == "asyncio.subprocess": + imported_names[alias.asname or alias.name] = alias.name + # ESCAPE-AS-VALUE forfeits (the aliasing closure — see the docstring). A + # shell-capable value may only be CALLED; any other mention hands it to a + # place no static walk can follow, so the mention itself fails closed: + # * a SINK attribute of the module outside call position + # (`r = subprocess.run`, `keep(os.system)`, `[sp.Popen]`); + # * the MODULE itself as a bare value (`x = subprocess`) — with it goes + # every sink it carries; + # * a from-imported SINK name outside call position (`r = run`); + # * `getattr(, ...)` — the attribute name is a string this walk + # will not chase, so the read is treated as an escape. + # Plain non-sink attribute reads (`os.environ`, `os.path.join(...)`) are + # untouched: the attribute must itself be a sink name for the first rule, + # and an Attribute node's `value` mention of the module is not a bare-value + # mention of it. + sink_names = _SHELL_SINK_ALWAYS | _SHELL_SINK_WITH_FLAG + sink_imported = {local for local, orig in imported_names.items() if orig in sink_names} + call_funcs = {id(n.func) for n in ast.walk(tree) if isinstance(n, ast.Call)} + attr_values = {id(n.value) for n in ast.walk(tree) if isinstance(n, ast.Attribute)} + for node in ast.walk(tree): + if isinstance(node, ast.Attribute): + if ( + _attr_chain_root(node) in module_aliases + and node.attr in sink_names + and id(node) not in call_funcs + ): + return True + if _attr_chain_root(node) in module_aliases and ( + node.attr.startswith("__") and node.attr.endswith("__") + ): + # A DUNDER attribute of a tracked module is reflection + # surface, not a named capability: `subprocess.__dict__["run"]` + # hands back the namespace and `__getattribute__("ru" + "n")` + # builds the name at runtime, so neither can be checked + # against the sink set. The Attribute node is what shields the + # module Name from the bare-value forfeit below (`vars(sp)` + # already forfeits there), so the dunder read forfeits as the + # same class: the module's capability set escaping whole. + return True + if ( + _attr_chain_root(node) in module_aliases + and node.attr in _SHELL_SINK_MODULES + and id(node) not in attr_values + and id(node) not in call_funcs + ): + # `x = asyncio.subprocess` -- or `x = subprocess.os`, since + # tracked modules re-export each other -- binds a + # sink-CARRYING module to an untracked name: the + # module-as-value escape one attribute deep. A CHAINED read + # (`asyncio.subprocess.create_subprocess_shell(...)`) is this + # node in attribute-value position and stays allowed; any + # other mention hands the submodule somewhere the walk cannot + # follow, so it forfeits like the bare module value. + return True + if ( + node.attr in _SHELL_SINK_MODULES + and isinstance(node.value, ast.Name) + and node.value.id in import_roots + and node.value.id not in module_aliases + ): + # A tracked module reached as an ATTRIBUTE of some other + # imported module -- `shutil.os`, `glob.os.system(...)`: + # most stdlib modules re-export the modules they import, so + # any import is a carrier. The attribute IS (presumptively) + # the sink-carrying module arriving under an untracked chain + # root, which this walk's module_aliases mechanism cannot + # follow, so the mention itself fails closed -- read, call + # chain, or value escape alike. Same-named non-module + # attributes only make vetting stricter. + return True + elif isinstance(node, ast.Name) and not isinstance(node.ctx, ast.Store): + if node.id in sink_imported and id(node) not in call_funcs: + return True + if ( + node.id in module_aliases + and id(node) not in attr_values + and id(node) not in call_funcs + ): + # The module as a bare value. Import statements bind via + # ast.alias (not Name), so `import os` itself never trips this. + return True + elif isinstance(node, ast.Call): + func = node.func + if ( + isinstance(func, ast.Name) + and func.id == "getattr" + and node.args + and isinstance(node.args[0], ast.Name) + and node.args[0].id in module_aliases + ): + return True + for node in ast.walk(tree): + if not isinstance(node, ast.Call): + continue + func = node.func + if isinstance(func, ast.Attribute): + if _attr_chain_root(func) not in module_aliases: + continue + name = func.attr + elif isinstance(func, ast.Name): + name = imported_names.get(func.id, "") + if not name: + continue + else: + continue + has_unpacking = any(kw.arg is None for kw in node.keywords) + has_starred_args = any(isinstance(a, ast.Starred) for a in node.args) + if name in _SHELL_SINK_ALWAYS: + is_sink = True + elif name in _SHELL_SINK_WITH_FLAG: + # `shell` is also REACHABLE POSITIONALLY: it is Popen's 9th + # parameter, and the run/call/check_* wrappers forward their + # positionals to Popen — so `run(cmd, -1, None, None, None, None, + # None, True, True)` runs a shell with no `shell=` keyword on the + # call. A call with 9+ positionals is judged on that argument + # (non-literal-False fails closed, same rule as the keyword), and + # a *starred unpacking puts every position at an unknowable index, + # so it fails closed outright. + shell_kw = next((kw for kw in node.keywords if kw.arg == "shell"), None) + if shell_kw is not None: + # shell=False is not a sink; shell= cannot be ruled out, so it is treated as one. + is_sink = not ( + isinstance(shell_kw.value, ast.Constant) and shell_kw.value.value is False + ) + elif has_starred_args: + is_sink = True + elif len(node.args) >= 9: + shell_pos = node.args[8] + is_sink = not (isinstance(shell_pos, ast.Constant) and shell_pos.value is False) + elif has_unpacking: + # No explicit shell= — but a `**kwargs` unpacking can carry + # `shell: True` invisibly (`subprocess.run(cmd, **{"shell": + # True})` runs a shell), so an unpacked call cannot be ruled + # out and is treated as a sink. Without an unpacking, no + # shell= means no shell. + is_sink = True + else: + continue + else: + continue + if not is_sink: + continue + command = node.args[0] if node.args else None + if command is not None and isinstance(command, ast.Starred): + # The command position itself is unknowable under *unpacking. + return True + if command is None: + # `command` is os.system/os.popen's keyword; `args` the subprocess + # family's; `cmd` getoutput/getstatusoutput's. + command = next( + (kw.value for kw in node.keywords if kw.arg in ("args", "cmd", "command")), + None, + ) + if command is None: + # A sink with no visible command: fine when the call simply has + # none, but a `**kwargs` unpacking can hide it (`subprocess.run( + # shell=True, **{"args": payload})`), so that shape fails closed. + if has_unpacking: + return True + continue + if isinstance(command, ast.Constant) and isinstance(command.value, str): + # A literal command at a shell sink is judged by the FULL + # command-line gate here, not left to the pass-through literal + # scan alone: this is the one place a literal is PROVABLY a shell + # command, so the whole-command analyses apply at full strength. + if is_sensitive_bash_command(command.value) is not None: + return True + # Even a clean scan cannot vouch for text the shell REWRITES -- + # POSIX `${UNSET}` splices, cmd.exe `%VAR%`/`!VAR!`/`^` escapes, + # glob expansion (`~/.ss*/id_rsa` reads the key naming no fenced + # path) -- so the literal must consist solely of characters no + # shell rewrites (see `_SHELL_VERBATIM_CHARS`): then the scanned + # text IS the executed text. Anything else fails closed, the same + # rule as a composed command, which such a literal is, just + # spelled inside one string. + if not _shell_executes_verbatim(command.value): + return True + continue + if isinstance(command, (ast.List, ast.Tuple)): + literal_elements = [ + el.value + for el in command.elts + if isinstance(el, ast.Constant) and isinstance(el.value, str) + ] + if len(literal_elements) == len(command.elts): + # An argv list/tuple of ALL string literals -- the exact + # shape the refusal message recommends -- is statically + # vettable even when a `**kwargs` unpacking might hide + # `shell=True`: every element is a literal the literal + # scan already judged. But element-wise judgment is not + # enough under a shell: Windows JOINS the list into one + # cmd.exe command line, so a traversal split across + # elements (`["rg", "'AKIA'", "~"]`) executes combined + # while each fragment scans clean. The JOINED form is + # therefore judged as a command line too (space-join is + # exact for Windows and conservative for POSIX, where a + # shell would execute only args[0] -- a judged literal). + # The verbatim-character rule applies equally: joined text + # a shell would rewrite cannot be vetted. Any non-literal + # element keeps the refusal. + joined = " ".join(literal_elements) + if is_sensitive_bash_command(joined) is not None: + return True + if not _shell_executes_verbatim(joined): + return True + continue + return True + return False + + def _vet_script_contents(text: str) -> str | None: """Scan a cron SCRIPT body for credential-exfiltration patterns. @@ -772,28 +1179,79 @@ def _vet_script_contents(text: str) -> str | None: exfiltration — which a human rubber-stamping the prompt would not catch — is the threat this gate closes. - ``is_sensitive_source_body`` is the same carve-out for the same reason, - one pass further in: ``is_sensitive_bash_command``'s pass 1b collapses - separator RUNS because a Win32 shell treats them as redundant, but in Python - source a backslash run is an ESCAPE. Collapsing strips it, so a body that - merely REDACTS or NAMES a fenced store — a ``re`` pattern, a docstring — - reads as an access to it and the job is denied at every fire, permanently - (the fire-time gate deliberately does not auto-pause). - - Dropping that pass outright would reopen the doubled-separator fence bypass - INSIDE a script, so it is REPLACED rather than removed: - ``is_sensitive_source_body`` owns that pairing in ``security.py`` — it applies - the same three checks to each - DECODED string literal, which is where the run still exists — - ``open(r"...\\\\kiro-cli\\\\c.json")`` hands the OS two backslashes and Win32 - collapses them. A literal is exonerated only when it provably flows into the - PATTERN operand of a pattern-consuming call, so an unknown sink over-blocks. A - body that does not - parse yields no literals to inspect, and then the raw shell scan runs WITH the - collapse, so an unparseable body is never quietly exonerated. - - Every other pass still runs, and ``_vet_script_file`` keeps its own + THE SUBJECT SPLIT. A script body is Python SOURCE, but a shell payload it + carries lives in a STRING LITERAL (``subprocess.run("…", shell=True)``, + ``os.system("…")``). So the body and its literals are scanned as different + subjects: + + * The WHOLE BODY goes through ``is_sensitive_source_body``: every + text-evidence pass over the full text, with the shell-grammar heuristics + off. Pass 1b's separator collapse is REPLACED (not dropped) by that + function's decoded-literal fence scan — in source a backslash run is an + ESCAPE, so collapsing raw source manufactures paths, while the run still + exists in the DECODED literal, which is where Win32 would collapse it + (``open(r"...\\\\kiro-cli\\\\c.json")``). The execution-model passes + (pipeline walk, ``find`` delivery analysis, env pipeline shapes) are off + for raw source because they judge fiction there: the pipeline walk's + fail-closed stage budget is exceeded by LINE COUNT alone (a ~700-line + script was refused with "more pipeline stages than this gate inspects" + at every fire, forever — the fire-time gate deliberately does not + auto-pause), and the ``find`` analysis has resolved cross-line fragments + of ordinary Python into a fenced path the file never names. + * EACH NON-DOCSTRING STRING LITERAL runs the shell-EXECUTION analyses + (``is_shell_payload_literal``: native entry, alt-traversal, ``find`` + delivery, env pipeline shapes): a literal is exactly the text a shell + would receive, so the modeling is sound there, and this is what catches a + traversal payload (``rg 'AKIA' ~``) that raw-text scanning never caught — + Python quoting swallows the payload, so the pre-split scan returned None + on ``subprocess.run("rg 'AKIA' ~", shell=True)`` too. The NAMING passes + are deliberately not re-run on literals: ``is_sensitive_source_body`` + already judged the values, with its redactor exonerations — re-asking + would re-deny the #7912 class. This matters because + script crons run in the ``standard`` sandbox, which deliberately leaves + ``~/.aws``/``~/.ssh`` readable (user scripts may legitimately use creds); + only the crew-fenced leaves (``_CREW_HIDDEN_LEAVES``) are masked at every + sandbox level. Docstrings are excluded from THIS payload scan: they are + prose, and the execution modeling fabricates on prose (measured on this + box's real cron scripts: 3 of 23 scripts' docstrings drew a traversal + verdict from sentences like "Find commits on main…", while 3,700+ + non-docstring literals drew zero). Docstring VALUES still go through + ``is_sensitive_source_body``'s fence scan, which is naming-based and + prose-safe. + * EVERY SHELL EXECUTION SINK must take a literal. A command COMPOSED at + runtime (``verb + tail``, an f-string, ``__doc__`` — which is how an + excluded docstring would become executable — or any variable) reaches + ``os.system``/``shell=True`` with no individually-blocking literal, so + :func:`_dynamic_shell_sink` refuses a sink whose command argument is not + a plain string literal: literal-or-refused, nothing dynamic slips between + the two scans. Measured cost of the rule: zero — none of the 23 real + cron scripts on the reporting host uses ``os.system`` or ``shell=True`` + at all. + * A body that does NOT parse yields no literals to inspect, and + ``is_sensitive_source_body`` then scans the raw text WITH full shell + grammar, so an unparseable body is never quietly exonerated (it could + not run as a cron script anyway — the runner imports it as Python). + + Every other check still runs, and ``_vet_script_file`` keeps its own ``is_sensitive_path`` on the resolved path. + + THE RESIDUAL, stated plainly rather than implied: this vet is a lexical + gate against what a human rubber-stamping the ``cron_add`` prompt would not + catch; it does not claim to defeat obfuscated Python. An argv-list exec + (``subprocess.run(["rg", …])``), a pure-Python read (``open``/``os.walk`` + — no shell shape exists for any pass to see), source re-read via + ``__file__``, and sink acquisition that never mentions a watched name + (``importlib.import_module("subprocess")``, ``sys.modules`` string routes, + ``exec``) are all outside static text analysis — and were equally outside + it BEFORE this change, when the raw-text scan returned None even on a + direct ``shell=True`` call whose literal carried a credential-directory + traversal, because Python quoting swallowed the payload. (Assignment + aliasing and wildcard imports are NOT residual: the escape-as-value and + unknowable-binding forfeits in ``_dynamic_shell_sink`` fail closed on + those.) For the residual classes the controls are the runtime sandbox + (crew-fenced leaves masked at every level) and the ``standard`` mode's + deliberate posture on user cloud/SSH credential directories — a product + decision, not a property this scan can supply. """ if _CRON_CRED_PATH_RE.search(text): return ( @@ -802,10 +1260,29 @@ def _vet_script_contents(text: str) -> str | None: ) if _CRON_SECRET_ENV_RE.search(text) or _CRON_SECRET_NAME_RE.search(text): return "Error: cron script blocked: references a protected secret environment variable" - # One entry point owns the pairing: the literal scan replaces pass 1b for a source - # subject, and a body that did not parse keeps the raw-text collapse. See + # One entry point owns the pass-1b pairing: the decoded-literal fence scan + # replaces the separator collapse for a source subject, and a body that did + # not parse keeps the raw-text scan WITH full shell grammar. See # ``is_sensitive_source_body``. reason = is_sensitive_source_body(text) + if reason is None: + parses, literals = _shell_scannable_literals(text) + if parses: + # PAYLOAD scan: each non-docstring literal through the full gate at + # the shell subject -- a literal is exactly the text a shell would + # receive, so the execution modeling is sound there and only there. + for literal in literals: + literal_reason = is_shell_payload_literal(literal) + if literal_reason: + reason = f"{literal_reason} (in a string literal)" + break + if reason is None and _dynamic_shell_sink(text): + return ( + "Error: cron script blocked: a shell execution call (os.system / " + "subprocess with shell=True) takes a command that is not a plain " + "string literal, so it cannot be statically vetted. Use a literal " + "command string, or an argv list without shell=True." + ) if reason: safe_reason = redact(reason) return f"Error: cron script blocked by security policy: {safe_reason}" diff --git a/src/kiro_crew/security.py b/src/kiro_crew/security.py index 2cf43037922..89201bcc559 100644 --- a/src/kiro_crew/security.py +++ b/src/kiro_crew/security.py @@ -11029,22 +11029,43 @@ def is_sensitive_bash_command( ``is_sensitive_path()`` to catch obfuscation (e.g. ``ca""t ~/.aws/credentials``, ``awk '{print}' $HOME/.ssh/id_rsa``, ``sed -n p ~/../../etc/shadow``). - Between them runs **pass 1b**, which repeats the pass-1 matchers over - separator-run-COLLAPSED copies of the subject. That is a Win32 *shell grammar* - heuristic: a shell opens the store ``%LOCALAPPDATA%\\kiro-cli`` names when - handed ``%LOCALAPPDATA%\\\\kiro-cli``, so the run carries no meaning and - collapsing it closes the doubled spelling (#6350). - - ``_subject_is_shell_grammar=False`` skips ONLY pass 1b, for a caller scanning a - subject that is not a shell command line -- a **source-code body**, where a - backslash run is an ESCAPE rather than a redundant separator. There ``\\\\`` is - one backslash and ``\\.`` is a literal dot, so collapsing strips the escapes and - manufactures a path the subject never contained: a ``re`` pattern that redacts a - fenced store, or a docstring merely naming one, reads as an access to it. Every - other pass still runs, so a source body keeps the path matcher, the extraction - control, the relative-traversal matcher, the normalizer, IMDS and env-credential - detection -- and its own caller keeps its ``is_sensitive_path`` check on the - resolved file path. + ``_subject_is_shell_grammar`` says whether *command* is a shell COMMAND LINE + (the default) or a larger text — a source-code body — being scanned as + defense-in-depth. It keys the shell-grammar heuristics that have NO + per-string subject form — pass 1b and pass 3 below — while the remaining + execution-model passes (4, 5, and the env-credential shapes) are governed + by the SUBJECT parameters further down: they always run, re-pointed at the + command strings a source body contains rather than switched off. The two + mechanisms compose; neither alone describes a source caller: + + * **Pass 1b** (separator-run collapse) repeats the pass-1 matchers over + separator-run-COLLAPSED copies of the subject. That is a Win32 shell + heuristic: a shell opens the store ``%LOCALAPPDATA%\\kiro-cli`` names when + handed ``%LOCALAPPDATA%\\\\kiro-cli``, so the run carries no meaning and + collapsing it closes the doubled spelling (#6350). In SOURCE a backslash + run is an ESCAPE — ``\\\\`` is one backslash, ``\\.`` a literal dot — so + collapsing manufactures a path the subject never contained: a ``re`` + pattern that redacts a fenced store reads as an access to it. A source + caller replaces this pass with the decoded-literal scan + (``is_sensitive_source_body``). + * **Pass 3** (native-shell entry-then-read) models ``cd`` state and + variable resolution across statements — one command line's execution, + which raw source text is not. A shell payload embedded in source lives + in a string literal, and a literal IS shell-grammar subject matter, so + the caller extracts the literals and feeds each one back through this + function at the default subject (see + ``mcp_cron._shell_scannable_literals``). + + The TEXT-EVIDENCE passes run for both subjects: the pass-1 regex fences, + the trust-root extraction control, the relative-traversal matcher, the + normalizer token scan, and IMDS all refuse only when the text itself names + something sensitive, which is as meaningful in source as in a command — and + a source caller keeps its own ``is_sensitive_path`` check on the resolved + file path. The runtime sandbox is a partial backstop, not the control: + script subprocesses get the crew-fenced leaves masked at every sandbox + level (``_CREW_HIDDEN_LEAVES``), but the ``standard`` mode they run under + deliberately leaves ``~/.aws``/``~/.ssh`` readable — which is exactly why + the literal scan is required. ``_traversal_subjects`` and ``_env_subject`` re-point the passes that REQUIRE a command line at the @@ -11123,10 +11144,16 @@ def is_sensitive_bash_command( if normalizer_result: return normalizer_result - # ── Pass 3: native-shell entry-then-relative-read scan ── - native_result = _check_native_home_entry_then_fenced_read(command) - if native_result: - return native_result + # ── Passes 3-5 model shell EXECUTION, so they run only when the subject IS a + # shell command line. On a source file they resolve variables, `cd` state and + # pipeline delivery that no shell will ever perform on that text, and their + # fail-closed analysis budgets refuse on the file's sheer length — see the + # docstring. The text-evidence passes above and below run for every subject. + if _subject_is_shell_grammar: + # ── Pass 3: native-shell entry-then-relative-read scan ── + native_result = _check_native_home_entry_then_fenced_read(command) + if native_result: + return native_result # ── Passes 4 and 5: the two TRAVERSAL analyses ── # Both walk shell STRUCTURE under a fail-closed budget, so unlike every pass above @@ -11302,6 +11329,43 @@ def is_sensitive_source_body(text: str) -> str | None: ) +def is_shell_payload_literal(literal: str) -> str | None: + """The shell-EXECUTION verdict on ONE string literal from a source body. + + The companion to :func:`is_sensitive_source_body`, owning the other half of + the source-subject split: that function answers the NAMING question for a + body and its literal VALUES (fence hits, with pattern-slot exoneration for + redactors), while this one answers the EXECUTION question for a literal — + is this text, handed to a shell, a traversal or credential-dump payload? + A literal is exactly the text a shell would receive, so the execution-model + analyses that judge fiction on raw source (see + ``is_sensitive_bash_command``'s subject flag) are sound here and only here. + + Deliberately does NOT re-run the naming passes: they already ran, with + their exonerations, in ``is_sensitive_source_body`` — re-asking them here + would re-deny the redaction literals that scan deliberately allows (the + #7912 class). Callers pair this with that function, never use it alone. + + Returns a denial reason, or None when clean. + """ + native = _check_native_home_entry_then_fenced_read(literal) + if native: + return native + alt = _check_alt_traversal_reaches_fence(literal) + if alt: + return alt + find_result = _check_find_traversal_reaches_fence(literal) + if find_result: + return find_result + # IMDS runs here as well as on the raw body: an ESCAPE-spelled endpoint + # (`"http://\x31\x36\x39.254.169.254/…"`) exists only in the DECODED + # literal, so the raw-text IMDS pass never sees it. + imds = _check_imds_access(literal) + if imds: + return imds + return _check_env_credential_access(literal) + + # `NAME=value` prefix. `normalize_shell_command` keeps it as a single token, and # the value is already $HOME-expanded by the time we see it. #: ``NAME=value`` and ``NAME+=value``. The append form is a separate group so the diff --git a/test/test_mcp_cron_security.py b/test/test_mcp_cron_security.py index a019a881327..83be5a61321 100644 --- a/test/test_mcp_cron_security.py +++ b/test/test_mcp_cron_security.py @@ -447,6 +447,448 @@ def test_vet_script_contents_allows_benign(body): assert _vet_script_contents(body) is None +# A script body is PYTHON SOURCE, not a shell command line. The execution-model +# passes inside `is_sensitive_bash_command` (native-shell entry scan, the +# alt-traversal pipeline walk, the `find` delivery analysis) model what a shell +# would DO with the text — but a source file's stage count is its LINE count, so +# every body past ~512 statements exhausted the pipeline walk's fail-closed +# budget by construction and was refused at every fire, forever ("command has +# more pipeline stages than this gate inspects"). The vet now scans script +# bodies with `_subject_is_shell_grammar=False`, which skips the execution-model +# passes and keeps every text-evidence pass on for the whole body. +LONG_BENIGN_SCRIPT = ( + "def run(ctx):\n" + + "".join(f" x{i} = {i}\n" for i in range(600)) + + " ctx.notify('done')\n" +) + + +def test_vet_script_contents_allows_a_long_python_body(): + """Red before the fix: the stage-budget refusal fired on line count alone.""" + assert _vet_script_contents(LONG_BENIGN_SCRIPT) is None + + +def test_vet_script_contents_still_blocks_credentials_in_a_long_body(): + """Length must not dilute the full-text scans: the credential-path detector + reads the whole body regardless of how many stages the traversal walk saw. + """ + body = LONG_BENIGN_SCRIPT + "data = open('/home/u/.aws/credentials').read()\n" + err = _vet_script_contents(body) + assert err is not None and err.startswith("Error:") + + +def test_vet_script_contents_still_blocks_secret_env_in_a_long_body(): + body = LONG_BENIGN_SCRIPT + "import os\nt = os.getenv('SLACK_BOT_TOKEN')\n" + err = _vet_script_contents(body) + assert err is not None and err.startswith("Error:") + + +# A shell payload embedded in Python lives in a STRING LITERAL, and Python +# quoting swallowed it from the raw-text scan: on the pre-split code, +# is_sensitive_bash_command over the whole body returned None for every case +# below (measured), so these pin a hole the literal scan CLOSES, not parity +# with old behavior. Each literal is fed back through the full gate at the +# default (shell) subject, where `rg 'AKIA' ~` is a recursive traversal rooted +# at a directory holding credential paths -- reachable, because script crons run +# in the `standard` sandbox, which deliberately leaves ~/.aws and ~/.ssh +# readable. +SHELL_PAYLOAD_SCRIPTS = [ + # The reviewer's vector: payload in a multiline literal, run via shell=True. + "import subprocess\n" + 'CMD = """rg \'AKIA\' ~"""\n' + "subprocess.run(CMD, shell=True)\n", + # Payload literal directly at the call site. + "import subprocess\nsubprocess.run(\"rg 'AKIA' ~\", shell=True)\n", + # os.system spelling. + "import os\nos.system(\"grep -r AKIA ~\")\n", + # The literal exists but never visibly flows to a shell call -- still + # refused: the scan judges literals, not dataflow, so hiding the call + # behind getattr()/aliasing does not exonerate the payload. + "PAYLOAD = \"rg 'AKIA' ~\"\nprint('nothing to see')\n", +] + + +@pytest.mark.parametrize("body", SHELL_PAYLOAD_SCRIPTS) +def test_vet_script_contents_blocks_shell_payload_literals(body): + err = _vet_script_contents(body) + assert err is not None and err.startswith("Error:"), body + # Which layer answers is composition detail: the source-body scan's + # per-literal traversal subjects (#8550) or the vet's own literal scan. + # Both name the traversal; the invariant is the DENIAL, not the reporter. + assert "traversal" in err or "string literal" in err + + +def test_vet_script_contents_blocks_a_payload_literal_in_a_long_body(): + """Length must not dilute the literal scan either.""" + body = LONG_BENIGN_SCRIPT + "CMD = \"rg 'AKIA' ~\"\n" + err = _vet_script_contents(body) + assert err is not None and ("traversal" in err or "string literal" in err) + + +def test_vet_script_contents_does_not_scan_docstrings_as_shell(): + """Docstrings are prose, and the shell modeling fabricates on prose. + + Measured on 23 real cron scripts: 3 docstrings drew traversal verdicts from + sentences like "Find commits on main...", while 3,700+ non-docstring + literals drew zero. The path that would EXECUTE a docstring -- + ``subprocess.run(__doc__, shell=True)`` -- is closed by the dynamic-sink + rule (below), not by scanning the prose: a sink's command must be a plain + string literal, and ``__doc__`` is not one. + """ + body = ( + '"""Find commits on main that belong to no pull request.\n\n' + "A commit whose message names no PR is delivered to the operator.\n" + '"""\n' + "def run(ctx):\n" + ' """Find and deliver a match to a command channel."""\n' + " ctx.notify('ok')\n" + ) + assert _vet_script_contents(body) is None + + +# A command COMPOSED at runtime reaches a shell sink with no individually +# blocking literal, so the sink itself is gated: os.system / subprocess with +# shell=True must take a PLAIN STRING LITERAL (which the literal scan already +# judged) or the script is refused. Every vector below carries no blocking +# literal -- the payload only exists assembled. +DYNAMIC_SHELL_SINK_SCRIPTS = [ + # The verifier's __doc__ vector: the excluded docstring becomes executable. + '"""rg AKIA in the home directory, recursively."""\n' + "import subprocess\n" + "subprocess.run(__doc__, shell=True)\n", + # Concatenated fragments. + "import subprocess\n" + 'verb = "rg "\n' + "tail = \"'AKIA' ~\"\n" + "subprocess.run(verb + tail, shell=True)\n", + # f-string composition. + "import subprocess\n" + 'pat = "AKIA"\n' + "subprocess.run(f\"rg '{pat}' ~\", shell=True)\n", + # A variable at the sink -- refused even when the literal it carries is + # benign, because what a NAME holds at runtime is not statically readable + # (one indirection re-opens the concat vector otherwise). The error tells + # the author the two accepted shapes. + "import subprocess\n" + 'CMD = "echo hi"\n' + "subprocess.run(CMD, shell=True)\n", + # os.system with a composed command. + "import os\n" + 'home = "~"\n' + 'os.system("grep -r AKIA " + home)\n', + # os.system's own keyword spelling -- `command=`, not `args=`. + "import os\n" + '"""payload docstring"""\n' + "os.system(command=__doc__)\n", + # shell= smuggled through a **kwargs unpacking: no explicit shell keyword + # exists on the call, so an unpacked run-family call fails closed. + "import subprocess\n" + 'verb = "rg "\n' + "tail = \"'AKIA' ~\"\n" + 'subprocess.run(args=verb + tail, **{"shell": True})\n', + # The command itself hidden in the unpacking. + "import subprocess\n" + 'p = "x"\n' + 'subprocess.run(shell=True, **{"args": p})\n', + # Module-alias spelling is still recognized. + "import subprocess as sp\n" + 'c = "x"\n' + "sp.run(c, shell=True)\n", + # from-import spelling is still recognized. + "from subprocess import run\n" + 'c = "x"\n' + "run(c, shell=True)\n", + # `shell` reached POSITIONALLY: it is Popen's 9th parameter, and the + # run-family forwards positionals to Popen -- no shell= keyword appears. + "import subprocess\n" + 'verb = "rg "\n' + "tail = \"'AKIA' ~\"\n" + "subprocess.Popen(verb + tail, -1, None, None, None, None, None, True, True)\n", + "import subprocess\n" + 'verb = "rg "\n' + "tail = \"'AKIA' ~\"\n" + "subprocess.run(verb + tail, -1, None, None, None, None, None, True, True)\n", + # *starred positional unpacking puts every argument at an unknowable + # position, so the call fails closed. + "import subprocess\n" + "argv = ['whatever']\n" + "subprocess.Popen(*argv, shell=True)\n", + "import subprocess\n" + "everything = ['cmd', -1, None, None, None, None, None, True, True]\n" + "subprocess.Popen(*everything)\n", + # ESCAPE-AS-VALUE forfeits: a shell-capable value may only be CALLED. + # Each body moves one somewhere no static walk can follow, so the mention + # itself fails closed (the reviewer's aliased-docstring vector first). + '"""rg AKIA in the home directory."""\n' + "import subprocess\n" + "r = subprocess.run\n" + "r(__doc__, shell=True)\n", + "import subprocess\n" + "x = subprocess\n" + "x.run('anything', shell=True)\n", + "import os\n" + "keep = [os.system]\n", + "import subprocess\n" + "f = getattr(subprocess, 'r' + 'un')\n", + "from subprocess import run\n" + "r = run\n", + # A wildcard import binds a set the walk cannot enumerate (run possibly + # among it, under its own name, recorded nowhere) -- unknowable binding + # set, fails closed outright. + '"""rg AKIA in the home directory."""\n' + "from subprocess import *\n" + "run(__doc__, shell=True)\n", + "from os import *\n" + "x = 1\n", + # `import os.path` binds the top-level `os`, so `os.system` is reachable + # through it -- the tracker records the top-level name. + '"""grep -r AKIA in the home dir."""\n' + "import os.path\n" + "os.system(__doc__)\n", + # asyncio's shell sink, every direct spelling: module attribute, nested + # submodule chain, from-imports (both module paths), module alias, aliased + # sink-carrying submodule, and the escape-as-value forfeit. + "import asyncio\n" + 'V = "rg " + "\'AKIA\' ~"\n' + "async def m():\n" + " await asyncio.create_subprocess_shell(V)\n", + "import asyncio.subprocess\n" + "async def m(v):\n" + " await asyncio.subprocess.create_subprocess_shell(v)\n", + "from asyncio import create_subprocess_shell\n" + "async def m(v):\n" + " await create_subprocess_shell(v)\n", + "from asyncio.subprocess import create_subprocess_shell\n" + "async def m(v):\n" + " await create_subprocess_shell(v)\n", + "import asyncio as aio\n" + "async def m(v):\n" + " await aio.create_subprocess_shell(v)\n", + "import asyncio.subprocess as asp\n" + "async def m(v):\n" + " await asp.create_subprocess_shell(v)\n", + # `from asyncio import subprocess` binds the sink-carrying MODULE under a + # bare (or aliased) name -- the third first-class import spelling. + "from asyncio import subprocess as asp\n" + "async def m(v):\n" + " await asp.create_subprocess_shell(v)\n", + "from asyncio import subprocess\n" + "async def m(v):\n" + " await subprocess.create_subprocess_shell(v)\n", + # A NON-literal element inside an unpacked argv list keeps the refusal: + # the composed element is exactly what the scan cannot read. + "import subprocess\n" + "opts = {'capture_output': True}\n" + "tail = 'sta' + 'tus'\n" + "subprocess.run(['git', tail], **opts)\n", + "import asyncio\n" + "f = asyncio.create_subprocess_shell\n", + # Dunder reflection on a tracked module: the namespace (or an arbitrary + # attribute) escapes whole, with the sink name built at runtime. + "import subprocess\n" + "v = 'rg ' + \"'AKIA' ~\"\n" + 'subprocess.__dict__["run"](v, shell=True)\n', + "import os\n" + 'os.__getattribute__("sys" + "tem")("id")\n', + # A traversal SPLIT across argv-list literals: each element scans clean, + # but Windows joins the list into one cmd.exe command line under + # shell=True, so the JOINED form is judged as a command line. + "import subprocess\n" + "subprocess.run(['rg', \"'AKIA'\", '~'], shell=True)\n", + "import subprocess\n" + "opts = {'shell': True}\n" + "subprocess.run(['rg', \"'AKIA'\", '~'], **opts)\n", + # Shell EXPANSION in a sink literal: the executed text differs from the + # scanned text (`${UNSET}` collapses), so it cannot be statically vetted. + "import os\n" + 'os.system("cat ~/.ss${UNSET}h/id_rsa")\n', + "import subprocess\n" + "subprocess.run('echo `rg AKIA ~`', shell=True)\n", + # The Windows rewrite operators: cmd.exe variables, escapes -- and glob + # expansion, which both shells perform (`~/.ss*` matches `.ssh` while + # naming no fenced path). The rule is a verbatim-character ALLOWLIST, so + # every rewrite family fails closed without being enumerated. + "import os\n" + 'os.system("cat ~/.s%EMPTY%sh/id_rsa")\n', + "import os\n" + 'os.system("cat ~/.s^sh/id_rsa")\n', + "import os\n" + 'os.system("cat ~/.ss*/id_rsa")\n', + "import subprocess\n" + "subprocess.run(['cat', '~/.ss*/id_rsa'], shell=True)\n", + # The sink-carrying submodule escaping as a VALUE: `x = asyncio.subprocess` + # binds it to an untracked name, so the mention forfeits. + "import asyncio\n" + "x = asyncio.subprocess\n", + # A tracked module arriving as a RE-EXPORT of another module: most stdlib + # modules re-export the modules they import, so `from subprocess import + # os`, `from shutil import os`, and `carrier.os.system(...)` all hand the + # sink carrier over under a chain the alias tracker cannot follow. The + # NAME is the forfeit condition, whatever the source module. + "from subprocess import os as o\n" + "v = 'rg ' + \"'AKIA' ~\"\n" + "o.system(v)\n", + "from shutil import os\n" + "v = 'rg ' + \"'AKIA' ~\"\n" + "os.system(v)\n", + "import shutil\n" + "v = 'rg ' + \"'AKIA' ~\"\n" + "shutil.os.system(v)\n", + "import glob\n" + "x = glob.os\n", + # A wildcard import from ANY module can bind `os`/`subprocess` under + # their own names (no `__all__` means every imported module re-exports), + # so the unknowable-binding-set forfeit applies to every wildcard. + "from glob import *\n" + "x = 1\n", + # The tracked-module-valued attribute of a TRACKED root escaping as a + # value: `subprocess.os` is the os module itself. + "import subprocess\n" + "x = subprocess.os\n", +] + + +@pytest.mark.parametrize("body", DYNAMIC_SHELL_SINK_SCRIPTS) +def test_vet_script_contents_blocks_dynamic_shell_sinks(body): + err = _vet_script_contents(body) + assert err is not None and err.startswith("Error:"), body + assert "statically vetted" in err + + +# Sink recognition is module-qualified, so an unrelated method that merely +# shares a sink's NAME is never a sink -- rejecting these at registration (and +# again at every fire) would be the same permanent-false-positive class this +# PR exists to remove. +NOT_SHELL_SINK_SCRIPTS = [ + "class R:\n def run(self, job, **kw):\n return job\n" + "renderer = R()\n" + "theme = 'dark'\n" + "renderer.run('job', shell=theme)\n", + "class C:\n def system(self, payload):\n return payload\n" + "client = C()\n" + "data = {'a': 1}\n" + "client.system(data)\n", + # A local function named like a sink, not imported from os/subprocess. + "def run(cmd, shell=False):\n return cmd\n" + "x = ['a']\n" + "run(x, shell=True)\n", + # Non-sink module attribute reads and argv-list sink CALLS are untouched + # by the escape forfeits: `os.environ` / `os.path` are attribute reads + # (not sink names), and a called sink is exactly the allowed mention. + "import os\n" + "region = os.environ.get('AWS_REGION', 'us-east-1')\n" + "p = os.path.join(os.getcwd(), 'x')\n" + "os.makedirs(p, exist_ok=True)\n", + "import subprocess\n" + "result = subprocess.run(['git', 'status'], capture_output=True)\n" + "print(result.returncode)\n", + "from subprocess import run\n" + "run(['ls', '-l'])\n", + # Ordinary async idioms: `asyncio.run` shares a name with subprocess's + # flag-gated sink, but with no shell spelling and fewer than 9 positionals + # the flag rule answers False -- an async cron's entry point stays + # vettable, as do gather/sleep and the argv-form exec sink. + "import asyncio\n" + "async def main():\n" + " await asyncio.sleep(1)\n" + " await asyncio.gather(asyncio.sleep(0))\n" + "asyncio.run(main())\n", + "import asyncio\n" + "async def main():\n" + " p = await asyncio.create_subprocess_exec('git', 'status')\n" + " await p.wait()\n" + "asyncio.run(main())\n", + # A literal command through the async shell sink: the literal scan judged + # the string, so the sink rule allows the call. + "import asyncio\n" + "async def main():\n" + " p = await asyncio.create_subprocess_shell('echo ok')\n" + " await p.wait()\n" + "asyncio.run(main())\n", + # The submodule-tracking rule is scoped to `from asyncio import + # subprocess` alone: `from os import path` binds a sink-FREE submodule, + # so a bare mention of `path` must not forfeit. + "from os import path\n" + "p = path\n" + "print(p.join('a', 'b'))\n", + # An argv list of ALL string literals under `**kwargs` unpacking: the + # exact shape the refusal message recommends. Every element is a literal + # the literal scan judged, and under a hidden shell=True it is args[0] + # (a judged literal) that reaches the shell -- statically vettable. + "import subprocess\n" + "opts = {'capture_output': True}\n" + "subprocess.run(['git', 'status'], **opts)\n", +] + + +@pytest.mark.parametrize("body", NOT_SHELL_SINK_SCRIPTS) +def test_vet_script_contents_module_qualifies_sink_recognition(body): + assert _vet_script_contents(body) is None, body + + +def test_vet_script_contents_allows_literal_shell_sinks_and_argv_lists(): + """The two shapes the sink rule's error message points authors at: a plain + literal command (already judged by the literal scan) and an argv list + without a shell. The argv-list residual is documented in the vet docstring: + no shell shape exists for any static pass to see, before or after this + change. + """ + assert ( + _vet_script_contents( + 'import subprocess\nsubprocess.run("echo hi", shell=True)\n' + ) + is None + ) + assert ( + _vet_script_contents( + "import subprocess\nsubprocess.run(['git', 'push'])\n" + ) + is None + ) + # Positional boundaries: shell as Popen's 9th positional literally False + # is not a sink; a literal command with positional shell True is a sink + # whose command the literal scan already judged. + assert ( + _vet_script_contents( + "import subprocess\n" + "p = subprocess.Popen(['x'], -1, None, None, None, None, None, " + "True, False)\n" + ) + is None + ) + assert ( + _vet_script_contents( + "import subprocess\n" + 'subprocess.run("echo hi", -1, None, None, None, None, None, ' + "True, True)\n" + ) + is None + ) + + +def test_vet_script_contents_unparseable_body_keeps_the_raw_shell_scan(): + """A body that is not Python yields no literals to judge, so it keeps the + whole-text scan WITH shell grammar -- never quietly exonerated. (It could + not run as a cron script anyway; the runner imports it as Python.) + """ + body = "this is not python (\nrg 'AKIA' ~\n" + err = _vet_script_contents(body) + assert err is not None and err.startswith("Error:") + + +def test_vet_script_contents_blocks_escape_spelled_imds_in_a_literal(): + """An escape-spelled IMDS endpoint exists only in the DECODED literal, so + the raw-text IMDS pass never sees it -- the literal scan runs IMDS too. + """ + body = ( + "import urllib.request\n" + 'u = "http://\\x31\\x36\\x39.254.169.254/latest/meta-data/"\n' + "urllib.request.urlopen(u)\n" + ) + err = _vet_script_contents(body) + assert err is not None and err.startswith("Error:") + + # A cron script body is PYTHON SOURCE, not a shell command line. In Python source # a backslash run is an ESCAPE (`\\` is one backslash, `\.` is a literal dot), so # collapsing separator runs -- correct for a Win32 shell string, where diff --git a/test/test_security_alt_traversal.py b/test/test_security_alt_traversal.py index 1bd14ebe93e..ebad0e72a46 100644 --- a/test/test_security_alt_traversal.py +++ b/test/test_security_alt_traversal.py @@ -1102,6 +1102,83 @@ def test_an_ordinary_pipeline_is_nowhere_near_the_budget() -> None: assert len(stages) < security._ALT_MAX_STAGES +def test_a_source_body_past_the_stage_budget_is_not_refused_for_length() -> None: + """A script FILE's stage count is its line count, so the fail-closed stage + cap -- right for a command line, where the uninspected suffix is executable + -- refused every long legitimate script, forever. The source-body entry + point re-points the traversal passes at the command strings the body + CONTAINS (none here), so length alone can no longer refuse. + """ + body = "\n".join(f"x{i} = {i}" for i in range(security._ALT_MAX_STAGES + 100)) + # The command-line default keeps its pinned refusal. + reason = is_sensitive_bash_command(body) + assert reason is not None and "pipeline stages" in reason + # The same text as a source body is clean: nothing in it names anything. + assert security.is_sensitive_source_body(body) is None + + +def test_a_source_body_keeps_every_text_evidence_pass() -> None: + """The subject flag turns off the execution model, never the text evidence: + a body that NAMES a credential path is refused at any length. + """ + padding = "\n".join(f"x{i} = {i}" for i in range(security._ALT_MAX_STAGES + 100)) + body = f"{padding}\ndata = open('/home/u/.aws/credentials').read()" + reason = is_sensitive_bash_command(body, _subject_is_shell_grammar=False) + assert reason is not None and "credential path" in reason + + +def test_a_source_body_skips_the_execution_model_passes_by_design() -> None: + """The documented trade, recorded so it is explicit rather than implied. + + A traversal SHAPE (a reader rooted above the fenced leaves, naming no leaf) + is judged only for a shell subject. For a source body the RAW TEXT is not + modeled: the text is never handed to a shell as a unit, and the modeling has + been observed to fabricate verdicts from cross-line fragments of ordinary + Python. The traversal passes instead receive the command strings the body + CONTAINS (#8550's per-string subjects) -- so a LITERAL carrying the same + payload is still judged at the shell subject, including traversals reaching + ~/.aws and ~/.ssh, which the standard-mode script sandbox deliberately + leaves readable. A body that NAMES a fenced leaf or credential path is + still refused by the text-evidence passes (previous test). + """ + # The traversal spelling lives in a COMMENT: present in the raw text, never + # executed, and invisible to the literal subjects (the parser discards it). + # A bare `rg . ~` line would be a SyntaxError, and an unparseable body + # deliberately keeps the full shell scan -- so the comment spelling is the + # one that isolates "raw text is not modeled" from the unparseable fallback. + body = f"# rg . {CREW}\n" + "\n".join(f"x{i} = {i}" for i in range(20)) + # Judged as a command line, the same traversal spelling refuses. + assert is_sensitive_bash_command(f"rg . {CREW}") is not None + # Judged as source: the raw text is not modeled and no LITERAL carries a + # payload, so nothing refuses (a literal carrying the same payload is + # caught at the shell subject -- pinned in test_mcp_cron_security.py). + assert security.is_sensitive_source_body(body) is None + + +def test_a_source_body_env_credential_shapes_are_shell_grammar_only() -> None: + """The env-credential pass is regex-based but its shared rules are ordered + PIPELINE shapes (`dump .* | .* filter .* selector`). Over a source file `|` + is regex alternation, so the shape assembles from fragments of unrelated + lines: here `env = dict(os.environ)` plus a detection-regex literal reads + as `env | grep AWS_SECRET`. Benign scanner code -- exactly the script most + likely to name these patterns on purpose. Env-secret NAMING in a script + body stays covered by the cron vet's own full-text `_CRON_SECRET_ENV_RE` / + `_CRON_SECRET_NAME_RE` scans (pinned in test_mcp_cron_security.py), and + string literals come back through this pass at the shell subject. + """ + body = ( + "import os, re\n" + "env = dict(os.environ)\n" + 'PAT = re.compile(r"(rm|del)|grep .*AWS_SECRET")\n' + ) + # As a command line the assembled shape refuses (unchanged default). + assert is_sensitive_bash_command(body) is not None + # As source it is recognized as fragments, not a pipeline: the env rules + # receive the body's LITERALS (concatenated in source order, #8550), and + # the regex literal alone carries no env accessor, so no shape assembles. + assert security.is_sensitive_source_body(body) is None + + @pytest.mark.parametrize( "command", [