From 344df0d8a2c5c4d0903fd6e737ef61060729f2f5 Mon Sep 17 00:00:00 2001 From: Nick Bowers Date: Fri, 4 Sep 2026 20:31:47 +0000 Subject: [PATCH] fix(cron): vet script bodies as source + literals, not as one shell command MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A cron script body is Python source, but _vet_script_contents fed the whole file to is_sensitive_bash_command, whose shell-grammar passes read their input as ONE command line. The alt-traversal pass walks pipeline stages under a fail-closed budget (_ALT_MAX_STAGES, 512), and a source file's stage count is its line count — so every script past ~512 statements was refused at every fire, forever, with 'command has more pipeline stages than this gate inspects'. With that refusal out of the way, two sibling fabrications surfaced on real scripts: the find delivery analysis resolved cross-line fragments of ordinary Python into a fenced path the file never names, and the env-credential pipeline shapes assembled an env-dump-piped-to-filter verdict from os.environ code plus a detection-regex literal hundreds of lines away. The subject split: is_sensitive_bash_command takes _subject_is_shell_grammar (default True — command lines are unchanged). The cron vet scans the WHOLE BODY with the flag off (text-evidence passes only: regex fences, trust-root extraction, normalizer token scan, IMDS — plus the vet's own credential-path / secret-env / exfil full-text scans) and feeds EVERY NON-DOCSTRING STRING LITERAL back through the full gate at the shell subject, because a literal is exactly the text a shell would receive. That literal scan CLOSES a pre-existing hole rather than preserving parity: Python quoting swallowed embedded payloads from the raw-text scan, so the base revision returned None on a script whose string literal carries a recursive credential-directory traversal handed to shell=True — which matters because the standard-mode script sandbox deliberately leaves the user-level cloud and SSH credential directories readable. Docstrings are excluded from the literal scan: they are prose, and measured on 23 real cron scripts, 3 docstrings drew fabricated traversal verdicts while 3,700+ non-docstring literals drew zero. An unparseable body keeps the old whole-text shell-grammar scan, so it is never quietly exonerated. Dynamic shell sinks are refused outright: os.system / subprocess with shell=True must take a plain string literal (which the literal scan already judged) — a command composed at runtime (concatenation, an f-string, __doc__, a variable) reaches the shell with no individually blocking literal, so literal-or-refused is the only line that leaves nothing between the two scans. Sink recognition is module-qualified (os/subprocess attribute calls, import aliases, from-imports), so an unrelated method that merely shares a sink's name is never misclassified; a run-family call carrying a **kwargs unpacking fails closed, since the unpacking can smuggle shell=True or the command itself; and `shell` is judged POSITIONALLY too — it is Popen's 9th parameter and the run family forwards positionals to Popen, so a 9-positional call is judged on that argument and a *starred unpacking fails closed outright. Assignment aliasing is closed as a CLASS (the same closure #7913 established for module authenticity): a shell-capable value may only be CALLED, and any mention that escapes as a value — r = subprocess.run, x = subprocess, getattr(subprocess, ...), a container element — forfeits the body outright. Non-sink attribute reads (os.environ, os.path) are untouched. Measured cost: zero — none of the 23 real cron scripts uses os.system, shell=True, or an unpacked subprocess call at all. The residual that no static scan of self-referential Python can close (argv-list exec, pure-Python reads, assignment-aliased sinks, source re-read via __file__) is stated in the vet docstring rather than implied, and was equally open before this change. Verified against the real-world reproducer: a 704-line script cron body now vets clean, and 22 of 23 real cron scripts on the reporting host pass end-to-end (the 23rd is refused by the untouched credential-path regex, identically to base). --- src/kiro_crew/mcp_cron.py | 523 ++++++++++++++++++++++++++-- src/kiro_crew/security.py | 104 ++++-- test/test_mcp_cron_security.py | 442 +++++++++++++++++++++++ test/test_security_alt_traversal.py | 77 ++++ 4 files changed, 1103 insertions(+), 43 deletions(-) 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", [