diff --git a/docs/system-specs/modules/learn-cron-dashboard.md b/docs/system-specs/modules/learn-cron-dashboard.md index 47c6ac4cc91..c52eea67f2f 100644 --- a/docs/system-specs/modules/learn-cron-dashboard.md +++ b/docs/system-specs/modules/learn-cron-dashboard.md @@ -255,7 +255,7 @@ Deterministic cron jobs that bypass the LLM entirely: - **Script-mode MCP identity**: a script cron runs as the principal `cron:` — the same key `ScriptContext` presents to the gateway over HTTP (`X-Session-Key` / `caller_session`) and the key an agent cron's session runs under, so ownership and audit see one principal per job whichever surface the job uses. It is delivered on two hops: `run_script_sandboxed` exports `KIROCREW_SESSION_KEY=cron:` into the script child, and `McpToolClient` hard-assigns the same key on the env of every MCP server `ctx.call_tool()` spawns — after both the inherited-env and per-server `env` overlays, so a script that rewrites its own `os.environ` cannot present a different principal (best-effort friction on the same footing as the rest of the env-based identity contract, not a containment boundary). This is load-bearing rather than cosmetic: a direct MCP spawn is not routed through gatewayd, so it gets no caller block, and nobody publishes a signed `session_pid_*` sidecar for the launcher pid — leaving a script cron with **none** of the three sources `_resolve_session_key_strict()` accepts, which fails closed for writes only. Every state-mutating tool then returns its refusal as an ordinary result string while read-only calls keep working, so the job reports `ok` and writes nothing. Two consequences to keep in mind: `cloud.aws.assert_human_action` keys on the same env var, so destructive cloud verbs from a script cron are refused exactly as they are from an agent session; and job ownership is unchanged, so a script cron reaches only the jobs recorded under its own key (`cron_trigger` against another job stays refused, as it is for agent crons). The general rule this is one instance of: **a spawn path that hands a child MCP access must hand it an identity too**, because the strict resolver fails closed for writes alone and the omission is therefore silent. - **Command mode**: `command` field specifies a shell command to run (mutually exclusive with `script`). Stdout captured as result. - **Timeout**: configurable per job (default 30s for scripts, 300s for commands). -- **Safety**: scripts must live under `~/.kiro/crew/crons/`. `is_sensitive_path()` blocks credential file access. SEL audit on every invocation. Auto-pause after 5 consecutive failures (`_AUTO_PAUSE_THRESHOLD`, single-sourced in `CronJob.record_failure`/`record_success`). The auto-pause is **persistent**: an execution-owned `auto_paused` flag (distinct from `user_paused`) is written by `_save`, propagated by `_merge_job_result`, and folded into the effective `enabled` derivation in `_load` — so a failing job stays paused across a daemon restart; `enable_job(True)` or a later success clears it (SEL-audited transitions). Concurrent execution guard prevents double-fire. +- **Safety**: scripts must live under `~/.kiro/crew/crons/`. `is_sensitive_path()` blocks credential file access. SEL audit on every invocation. Auto-pause after 5 consecutive failures (`_AUTO_PAUSE_THRESHOLD`, single-sourced in `CronJob.record_failure`/`record_success`). The auto-pause is **persistent**: an execution-owned `auto_paused` flag (distinct from `user_paused`) is written by `_save`, propagated by `_merge_job_result`, and folded into the effective `enabled` derivation in `_load` — so a failing job stays paused across a daemon restart; `enable_job(True)` or a later success clears it (SEL-audited transitions). Concurrent execution guard prevents double-fire. The script **body** is scanned as well — at authoring time and again at every fire, via `_vet_script_file` on the freshly re-resolved path — for credential-path references, protected secret environment variables, exfiltration URLs, and the shared sensitive-path matcher. Two shell-grammar heuristics are deliberately excluded there, because a body is Python source rather than a command line: `is_denied` (tool-name semantics such as `*git*push*`) and the raw-text separator-run collapse (in source a backslash run is an escape, so collapsing it manufactures paths the body never contained). The collapse is **replaced, not dropped** — `security.is_sensitive_source_body()` owns both halves, applying the same fence checks to each decoded string or bytes literal and exonerating one only when it sits provably in the pattern operand of a pattern-consuming call that is the outermost expression it reaches, and the body has neither rebound `re` or one of its attributes nor read a pattern back off an `re` object, so a run that survives into a real path (`open(r"…\\kiro-cli\\c.json")`, which Win32 collapses at open time) is still refused; docstrings are scanned, because Python retains them as `__doc__` where a body can read them back; see [security.md](security.md). A body that does not parse, or one too deep to traverse, keeps the raw-text collapse instead. A fire-time denial keeps the job and does not feed the auto-pause counter, so a body the scan misjudges is denied on every tick until it is edited. - **Kind tag**: `cron_list` labels each job as `script`, `command`, or `agent` based on which mode is configured. #### Auto-pause applies to `agent` (message) crons too diff --git a/docs/system-specs/modules/security.md b/docs/system-specs/modules/security.md index 3bf17c461c5..d663fabe2f8 100644 --- a/docs/system-specs/modules/security.md +++ b/docs/system-specs/modules/security.md @@ -152,6 +152,8 @@ under `(allow default)`, never an edition-resolved or user-writable executable. - **Symlink resolution (CWE-59)**: `is_sensitive_path()` resolves symlinks before matching — it checks multiple candidate forms (`os.path.realpath` + `Path.resolve`, plus the lexically-normalized path as a fail-safe when resolution can't complete) and returns True if ANY lands in a sensitive location, `casefold`-comparing against sensitive dirs anchored at BOTH the logical home and its realpath (defeats a home-prefix OS symlink like macOS `/var`→`/private/var`). So a workspace symlink pointing at `~/.aws/credentials` (absolute or `../../.aws/credentials` traversal) cannot be read through the link - **Relative-traversal block (verb-agnostic)**: home-anchored/absolute references to a sensitive dir are caught by the primary matcher (`_get_sensitive_re()`), but relative-traversal forms (`../../.aws/credentials`) escape it. `is_sensitive_bash_command()` therefore blocks **any** command whose tokens name a sensitive dir via dot-slash traversal (`_RELATIVE_SENSITIVE_RE`), regardless of verb — so `dd`/`base64`/`xxd`/`head`/`tail`/`cp`/`ln` are all covered (it was previously gated on `ln`/`cp` only, letting the others slip past). Returns "command references a sensitive credential path via relative traversal" - `is_sensitive_bash_command(cmd)` — regex matches `cat`, `head`, `tail`, `less`, `cp`, `scp`, `python open()`, pipe redirects targeting sensitive paths +- **Separator-run collapse (shell grammar only)**: a Win32 shell opens the store `%LOCALAPPDATA%\kiro-cli` names when handed `%LOCALAPPDATA%\\kiro-cli`, so a repeated separator carries no meaning there while the matchers spell a single one. `is_sensitive_bash_command()` therefore repeats all three first-pass checks — the path matcher, the trust-root extraction control, and the relative-traversal matcher — over separator-collapsed copies of the subject, covering every run length at linear cost (admitting a run into the patterns instead measures as a watchdog-crossing hang on this gate). All three, never a subset: with the extraction control omitted, `tar -xf evil.tar -C $HOME//.kiro/crew` writes governance files through the doubled separator. The collapse runs only after the unmodified subject misses, so a form needing the run intact (a UNC `\\server\share` anchor) keeps its match. Because it runs only then, the shared helper's own check of the UNMODIFIED value is a provable duplicate on this path — pass 1 put those exact bytes through these same three matchers and missed — so the shell caller opts out of it (`value_already_scanned=True`) and checks only the collapsed copies. That is a cost fix rather than a coverage change: the sensitive-path matcher over a long newline-free line is the most expensive check on this gate, so running it twice doubled the wall time of a path a linearity test guards, taking a 20 KB subject from 2.4s to 4.2s and overshooting that test's ceiling on CI. Detection is unmoved and pinned by test in both directions — the single-separator spelling the skipped check would have caught is still caught by pass 1 itself, and the doubled spelling is still caught by the collapsed copy, as is the trust-root extraction control that travels with them. The default keeps the value-check, so the source-literal path below is unaffected and a new caller is self-sufficient unless it opts out deliberately. `subject_is_shell_grammar=False` skips this pass and only this pass, for a caller whose subject is **source code** rather than a shell command line: in source a backslash run is an escape — `\\` is one backslash, `\.` a literal dot — so collapsing strips the escape and manufactures a path the subject never contained, and a `re` pattern that redacts a fenced store, or a docstring merely naming one, reads as an access to it. Every other pass still applies, and the skip is keyed on the subject rather than on any single check, so a caller either has shell grammar and gets all three or does not and gets none. The cron script-body scan (`mcp_cron._vet_script_contents`) is the one caller that passes it +- **Escape-aware counterpart for source subjects** (`_sensitive_run_in_source_literals`): skipping the collapse on a source body would reopen the doubled-separator bypass *inside* a script, because the run still exists in the DECODED literal — `open(r"…\\kiro-cli\\c.json")` hands the OS two backslashes and Win32 collapses them, while the raw source matches no fence. So the pass is replaced rather than removed: the same three checks run against each decoded string literal. Decoding alone cannot decide, which is why this is **sink-aware** — a regex escape and a path separator are the same character in a decoded value, so `re.compile(r"%LOCALAPPDATA%\\kiro-cli")` and `open(r"%LOCALAPPDATA%\\kiro-cli\\c.json")` are indistinguishable by any transform of the value and differ only in the call that receives it. A literal is exonerated only when it provably flows into the **pattern operand** of a pattern-consuming call (`_SOURCE_PATTERN_SINKS`); an unknown call, a name bound first, or no call at all all keep the deny verdict, so an unenumerated sink over-blocks rather than opening the fence — the direction `_TRUST_ROOT_READ_LISTERS` argues for. The operand matters because the allowlisted calls are not uniformly safe: `re.sub(pattern, repl, string)` returns its subject verbatim and its replacement substantially so, so a fenced path in either slot would flow on to a real sink through a call that merely looks harmless. Only `args[0]` / `pattern=` describes a regex. Occupying that slot means BEING the operand rather than merely reaching it: the position test resolves to the top of the argument subtree, so an expression FEEDING the operand satisfies it while the fenced literal sits underneath, and evaluating that expression runs code before `re` ever receives a pattern — `re.compile(FENCED + Reader())` hands the expanded path to `Reader.__radd__`, `%` formatting reaches `__rmod__`, and every other operator protocol is the same shape, so enumerating operators would be another allow-by-default blocklist. The exoneration therefore requires the literal ITSELF to occupy the slot, positionally or by `pattern=`, which also subsumes the walrus and f-string spellings that previously needed their own reasoning. The pattern slot of a SUBSTITUTING member (`re.sub`, `re.subn`) carries a further hazard: its `repl` may be a FUNCTION, and `re` hands that function the `Match`, which carries `.re` — so `Match.re.pattern` returns the verbatim pattern literal. The compiled-name escape analysis cannot reach it, because that tracks only `_COMPILING_SINK` results while a Match is never bound by the exonerated statement, and the recovery read can be spelled to defeat any enumeration (`getattr(m, "r" + "e")` builds the attribute name from a concatenation, an aliased `g = getattr` hides the callee). Chasing the spelling is therefore the wrong layer, exactly as it was for the compile result: the exoneration is withdrawn at the SLOT unless the replacement PROVABLY cannot be called, meaning a `str` or `bytes` constant. A Name or Attribute may be bound to a function, a lambda plainly is one, a `Starred` puts the replacement at a position nobody can know statically, and an absent argument leaves nothing to prove — each fails closed. The motivating redactor passes a string replacement and is unaffected. Scoping that withdrawal to the substituting members alone was still too narrow, because a `Match` also leaves an exonerated slot as a RETURN VALUE: `re.match`, `re.search`, `re.fullmatch` and `re.finditer` each hand one back, and nothing tracks it — the compiled-name analysis follows only `_COMPILING_SINK` results, so a Match bound to a name, iterated by a `for`, or captured by a comprehension reaches no guard at all, and `_pattern_reextracted` cannot see the recovery because `getattr(m, "r" + "e")` builds the attribute name from a `BinOp` rather than a `Constant`. Those four are therefore simply ABSENT from the sink set, so deny-by-default refuses a fenced literal in their pattern slot outright. A withdrawal rule for them was tried first and removed as dead code: the only shape it could still exonerate was a DISCARDED result — the call in statement position, its value received by nothing — which no real script writes, and no benign-corpus body uses these four at all. `re.findall` and `re.split` remain in the set: they return `str` and `list`, which carry no reference back to the pattern. The COMPILED object still needs the explicit rules, because there the exoneration is earned by `re.compile` and only then is the Match produced, and `_SAFE_COMPILED_PATTERN_METHODS` admitted the entire matching API on the METHOD NAME alone: `p.sub`/`p.subn` pass their replacement the Match (so the provably-non-callable test applies, at position 0 because the pattern is bound in the object rather than passed), and `p.match`/`p.search`/`p.fullmatch`/`p.finditer` return one (so the discarded-result rule applies), while `p.split`/`p.findall` stay safe. The accepted cost is a wider over-block than before — a fenced literal used purely to TEST for the store, `if re.search(FENCED, line):`, is now denied — which is the same direction as the three over-blocks already recorded here. `re.escape` is deliberately NOT in the set even though it is an `re.*` call: it consumes plain TEXT and returns it escaped for onward flow, so it fails the set's own admission rule and would exonerate a literal that continues to a real sink. The exoneration also keys on the *spelling* `re.`, so it is withdrawn for a body that rebinds the name at all — and the spellings do not all reach the AST as a `Name` node: `import evil as re`, `re = …`, `class re`, a parameter or loop variable bind through nodes, while `except E as re:`, `case re:`, `case [*re]` and `case {**re}` bind through plain STRING attributes (`ExceptHandler.name`, `MatchAs`/`MatchStar.name`, `MatchMapping.rest`) that a Name-only walk cannot see. A WILDCARD import carries no alias naming the module yet can bind it anyway: `from evil import *` binds a set this tree cannot enumerate, `re` possibly among them, so matching only the aliases that NAME `re` left the explicit `from evil import thing as re` forfeiting while the wildcard — which can do strictly more — did not, an allow-by-default enumeration inside a deny-first checker. The forfeit condition is therefore an UNKNOWABLE binding set, which closes the class rather than adding a spelling to it; `from re import sub` names `sub` and rebinds nothing, so it stays allowed. It is withdrawn equally for a body that reassigns an ATTRIBUTE of the module (`re.compile = open` leaves the module bound while the call it spells now opens a file), including when that mutation is spelled as a CALL — `setattr(re, "compile", open)` / `delattr(re, "sub")` reach neither the Name nor the Attribute branch, so they are matched on the call itself. Recognising only those two callees was the wrong SHAPE for the same reason the recovery guard was: it is an allow-by-default blocklist inside a deny-by-default checker, so an ordinary `helper(re)` whose body does `m.compile = open` reached no branch at all and the module still read authentic. What the callee does with the object is not readable from this tree, so the rule INVERTS: any ARGUMENT resolving to the module forfeits the exoneration, keyword arguments included. Only the arguments are inspected, never the call's `func` — `re.sub(...)` and `re.compile(...)` name the module there, and those are precisely the calls the exoneration exists to permit. Every guard here is nonetheless a STATIC read of the parse tree, so a body able to run code defeats all of them at once: `exec("re.compile = open")` carries the rebinding inside a STRING, reaching no Name, Attribute, Subscript or argument the tree can be asked about. The string is opaque by construction, so no enumeration closes that class either — the presence of `exec`/`eval`/`compile`/`__import__` as a bare NAME withdraws the exoneration instead, matched on the name rather than the call so that binding it first (`e = exec`) forfeits equally. The builtin `compile` is matched only as a Name; `re.compile` spells its name in an Attribute's `attr` STRING and is not a Name node, so the ordinary redactor is unaffected. Measured on those spellings, the authenticity defect was a layer-contract violation rather than a demonstrated end-to-end leak — nine exploit bodies were already refused by the downstream escape analysis — so this restores defence in depth at the layer that claimed it rather than closing an open path. Mutation is also followed through an ALIAS: `m = re` (or `import re as m`) binds the SAME object, so `m.compile = open` replaces exactly the attribute `re.compile` spells while an `re`-keyed check sees an untouched module. Every name bound from an alias is collected first, and a reference that escapes as a VALUE rather than through such a binding is no longer trackable at all: `holder = [re]` then `m = holder[0]` puts the same object behind a subscript no static walk can resolve, so `m.compile = reader` rebinds exactly what `re.compile` spells while an alias set keyed on bare Name assignments records nothing — and a container display is only one spelling, alongside a tuple, a dict value and a conditional expression. Enumerating those shapes would rebuild the allow-by-default blocklist the call branch already had to abandon, so the rule closes the class instead: a module reference may only be READ through an attribute or bound as a tracked bare alias, and every other mention forfeits. The indirect routes that carry no alias name at all — `vars(re)["compile"] = open`, `re.__dict__[…]`, `sys.modules["re"].compile = open` — are matched on the SUBSCRIPT, deny-first: a subtree naming `sys.modules` cannot be proved not to hand back the module. A namespace MAPPING is the same capability one level further out and is NOT reachable by inspecting the subscript at all: `globals()["re"] = Fake` rebinds the name while its subscript's `value` is the bare `globals()` call, which names no alias, no `modules` attribute and no `"re"` constant, so every binding branch above is bypassed while the module still reads authentic. It is therefore matched on the NAME — `globals`, `locals`, `vars` — because the mapping can be bound first (`g = globals()` then `g["re"] = Fake`) leaving the subscript's own value an unresolvable local; `locals()` at module scope IS the global namespace and `vars()` with no argument is `locals()`, so the three spellings forfeit together. The position must also be PROVABLE: `re.sub(*seq)` makes `args[0]` a starred node whose contents land at unknowable positions, so an unprovable position is not treated as the pattern slot. The two halves are exposed as a single entry point, `is_sensitive_source_body(text)`, which owns the composition — skipping pass 1b is sound only because the literal scan replaces it, so the flag that skips it is internal and a caller cannot take one half without the other. Both `str` and `bytes` constants are inspected, `bytes` decoded latin-1 (total over a byte range, one code point per byte, so a separator run survives unchanged), because `open` and `os.open` accept a bytes path and a `rb\"…\"` literal otherwise reaches the same sinks unexamined. A string constant in statement position is skipped ONLY when it is not a docstring: Python evaluates and discards a bare string expression, so that one reaches no sink, but it RETAINS a module, class or function docstring as `__doc__`, which a body can read back and hand to a sink (`open(f.__doc__)`). Docstrings are therefore scanned. The three checks run against the literal's own value AND its separator-collapsed copies, in that order: `_separator_collapsed_variants` yields nothing when the separators are already single, so iterating it alone left such a value unexamined by this layer and dependent on a later pass — defence in depth sharing the earlier layer's blind spot. Checking the value first makes the layer self-sufficient, and it must stay uniform: exempting docstrings would reopen a single-separator `open(f.__doc__)` path. The cost is that a prose-only docstring naming the store is refused even though it reads nothing, which is recorded as an accepted over-block with its own test rather than left in the benign corpus. **Deferred, same root cause one gate earlier**: `llm_helpers.py`'s tool-input scan runs the shell matcher over every string extracted from `tool_input`, including a file-write tool's `content`, so an agent is still refused *writing* the redactor body that cron may now *run*. A second sibling gate, `skills_script_validator.py`, scans generated skill scripts as RAW TEXT with neither the separator collapse nor the decoded-literal check, so the doubled spelling this change closes for cron bodies still passes there. Both call sites are out of this change's scope and are named here rather than silently carried. Comments never reach the check, the parser having discarded them. Exoneration further requires the matched `re.*` call to be the OUTERMOST expression the literal reaches — a call whose result is consumed by another call or has an attribute read off it hands the verbatim pattern onward (`open(re.sub(FENCED, …))`, `re.compile(FENCED).pattern`) — and it is withdrawn for the whole body when that body reads a `pattern` or `re` attribute anywhere, which is how the cross-statement `p = re.compile(FENCED)` then `open(p.pattern)` escape is closed. That read has a CALL spelling too: `getattr(p, "pattern")` carries the attribute name in a string argument, so it parses to an `ast.Call` and an Attribute-only walk never sees it — the same call-spelled blind spot `setattr`/`delattr` exploited against the authenticity check, and left the dotted and called forms of one read disagreeing. Both attribute names are matched on the call as well, including the dunder-getter forms (`p.__getattribute__("pattern")`, `object.__getattribute__(p, "pattern")`, `operator.attrgetter("pattern")`) whose `func` is itself an Attribute. But enumerating recovery spellings is the wrong SHAPE for a deny-first checker: the ways out of a compiled object are open-ended — an aliased `g = getattr`, and every stringify form (`"%s" % p`, `"{}".format(p)`, `f"{p!r}"`, `str`/`repr`/`vars`) — so a blocklist sits one unenumerated spelling away from reopening the fence. The guard therefore INVERTS: a compiled pattern may be used through its own matching API (`search`, `match`, `fullmatch`, `split`, `findall`, `finditer`, `sub`, `subn`) and nothing else, and any other read — passed to a call, formatted, subscripted, stored in a container, returned — forfeits the exoneration for the whole body. The accepted cost is that a harmless unenumerated attribute read (`p.flags`) withdraws it too. Only `re.compile` results are tracked: the other sinks CONSUME a pattern and return an ordinary value, so tracking them made `redacted = re.sub(F, "", s)` followed by `str(redacted)` read as a re-extraction and refused the very redactor shape this change exists to permit. Because that tracking watches NAMES, a compile result bound anywhere a name cannot follow — a subscript, an attribute, a tuple element — leaves the escape check watching nothing while the literal stays exonerated, so the untrackable binding itself withdraws the exoneration rather than the checker attempting alias analysis into containers. Enumerating binding SHAPES is still not enough, because a result that is never bound reaches no binding node whatsoever: `return re.compile(FENCED)`, `keep(re.compile(FENCED))`, `[re.compile(FENCED)]` and a bare `yield` each leave the tracked set EMPTY, and an escape check handed an empty set cannot fail — so the literal is exonerated with zero tracking behind it and ANY recovery spelling then works, including one that defeats a literal attribute-name match (`getattr(p, "pat" + "tern")` builds the name from a concatenation). Chasing the recovery call is the wrong layer, so the rule inverts: exoneration requires the compile result to be the DIRECT value of an assignment whose every target is a plain Name — precisely what the tracker can follow — and every other position forfeits. The slot is likewise forfeited when a WALRUS binds the literal inside it: `re.compile(p := FENCED)` puts the literal in a genuine pattern operand while also binding it to a name that outlives the call, so a later `open(p)` receives the verbatim fenced spelling and the "a pattern operand goes nowhere else" premise fails. Three over-blocks are accepted deliberately here, on the same reasoning as the rebinding trade: the literal must sit DIRECTLY in the pattern slot, so the idiomatic module-constant-then-compile form is denied; a chained-but-harmless `re.compile(FENCED).search(s)` is denied because an attribute read is indistinguishable from a `.pattern` re-extraction; and a docstring naming a fenced path is denied even where `__doc__` is never read. Each errs toward the false positive rather than toward the fence. A body that does not parse yields no literals, and the caller then runs the raw scan **with** the collapse, so an unparseable body is never quietly exonerated; a body so deeply nested that the traversal exhausts the recursion limit is reported the same way rather than raised, so a legitimate deep expression degrades to the textual scan instead of failing the gate - **Normalizer second pass (verb-independent)**: the regex first-pass matches raw shell text, so it sees only the path spellings it is authored for — two textually different strings naming the same file are not decidable by a regex over an unnormalized command line, and the set of equivalent spellings (dot segments, `..`, repeated slashes, `$HOME` vs the resolved home, quote splitting) is open-ended by construction. `_check_sensitive_via_normalizer()` therefore tokenizes via `normalize_shell_command()` and routes **every** path-like operand through `is_sensitive_path()` — the same normalizing checker the file gate uses — so one implementation is authoritative on both surfaces and a newly registered keystone leaf is protected on both by registration alone. The pass runs regardless of verb, mirroring the verb-independent backstop the sensitive-dir matcher applies: naming a sensitive path is itself the signal, and normalization is the only layer able to decide equivalence, so restricting it to a verb allowlist would leave a spelling such as `~/.kiro/crew/./live_target.json` unchecked for every verb outside that list. `_NORMALIZER_READ_VERBS` / `_LINK_CREATE_VERBS` are consulted only to skip the command name itself, never to decide whether operands are checked. `key=value` operands are split on the first `=` and the value checked as well: `of=/path` does not resolve as a path, and `--output=/path` is otherwise dropped by the flag skip. **Attached redirections** (`>~/path`, `>>~/path`, `2>~/path`) are kept as a single token by `shlex.split`; the leading operator prefix is stripped via `_REDIR_PREFIX_RE` before the path portion is checked, so `printf x >~/.kiro/crew/./live_target.json` is blocked just as `echo x > ~/.kiro/crew/./live_target.json` (with a space) is - **Not covered**: a bare relative operand (`live_target.json` run with the cwd inside the data home). `is_sensitive_path` is called without a `base_dir`, so such a token resolves against the gateway process cwd rather than the command's — a command line inspected before execution does not carry its cwd. Closing it requires a fail-closed decision for relative tokens whose basename matches a keystone leaf - `hooks.on_tool_call` runs **both** `is_sensitive_path` and `is_sensitive_bash_command` on the **normalized** tool title regardless of the kiro-cli `Reading: `/`Running: ` display prefix. The claude-agent-acp adapter sets a file-read tool's title to the bare path and a Bash tool's title to the bare command (no prefix), so gating either check on the prefix would let credential reads through on an alternate ACP backend. `is_sensitive_path` resolves the title as a path (a bare `~/.aws/credentials` matches; a `cat ~/.aws/credentials` command resolves to a non-sensitive path and is caught by `is_sensitive_bash_command` instead). diff --git a/src/kiro_crew/mcp_cron.py b/src/kiro_crew/mcp_cron.py index 34db1f56e70..5f7a73d96fb 100644 --- a/src/kiro_crew/mcp_cron.py +++ b/src/kiro_crew/mcp_cron.py @@ -55,6 +55,7 @@ enabled_rule_ids, is_sensitive_bash_command, is_sensitive_path, + is_sensitive_source_body, scan_exfiltration_urls, ) from kiro_crew.sel import sel @@ -766,6 +767,29 @@ def _vet_script_contents(text: str) -> str | None: covered by the now-required ``cron_add`` approval prompt. Credential 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 + ``is_sensitive_path`` on the resolved path. """ if _CRON_CRED_PATH_RE.search(text): return ( @@ -774,7 +798,10 @@ 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" - reason = is_sensitive_bash_command(text) + # 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 + # ``is_sensitive_source_body``. + reason = is_sensitive_source_body(text) 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 d232dc2595b..891c5e0c5c6 100644 --- a/src/kiro_crew/security.py +++ b/src/kiro_crew/security.py @@ -2,6 +2,7 @@ from __future__ import annotations +import ast import asyncio import base64 import fnmatch @@ -27,6 +28,7 @@ import resource as _resource except ImportError: _resource = None # type: ignore[assignment] # Windows/non-POSIX +from collections.abc import Sequence from datetime import datetime, timezone from pathlib import Path from typing import TYPE_CHECKING, NamedTuple @@ -8821,8 +8823,825 @@ def _replace( return tuple(variants) +# ── The escape-aware counterpart of pass 1b, for SOURCE-CODE subjects ── +# Pass 1b collapses separator runs in a shell command line, where a run is +# redundant. Source code cannot be scanned that way: there a backslash run is an +# ESCAPE, so collapsing the raw text strips escapes and manufactures paths the +# source never contained. But the run still EXISTS once the literal is decoded -- +# ``open(r"%LOCALAPPDATA%\\kiro-cli\\c.json")`` hands the OS a value carrying two +# backslashes, Win32 collapses it, and the fenced store is reached. So the fence +# must be checked against the DECODED value, not the raw text. +# +# Decoding alone is not enough to decide, which is the whole reason this is +# sink-aware. A regex escape and a path separator are the SAME CHARACTER in a +# decoded literal, so these two are indistinguishable by any transform of the +# value: +# +# re.compile(r"%LOCALAPPDATA%\\kiro-cli") <- a pattern; reads nothing +# open(r"%LOCALAPPDATA%\\kiro-cli\\c.json") <- opens the fenced store +# +# Only the SINK differs. So a literal is exonerated ONLY when it provably flows +# into a pattern-consuming call, and the deny verdict is the default for +# everything else -- an unknown call, no call at all, a name bound first. An +# unenumerated sink therefore over-blocks rather than opening the fence, the same +# direction ``_TRUST_ROOT_READ_LISTERS`` argues for a few hundred lines up: naming +# the writers fails OPEN and is the wrong way round. +# +# Every entry below must consume its argument as a PATTERN and never as a path. +# ``re.escape`` is deliberately ABSENT: it consumes plain TEXT and returns it escaped +# for onward flow, so it fails that rule and would exonerate a literal that continues +# to a real sink. Do not reinstate it by symmetry with its neighbours. +_SOURCE_PATTERN_SINKS: frozenset[tuple[str | None, str]] = frozenset( + { + ("re", "compile"), + ("re", "findall"), + ("re", "split"), + ("re", "sub"), + ("re", "subn"), + } +) + + +def _mentions_module_alias(node: ast.AST, aliases: set[str]) -> bool: + """True when ``node``'s subtree can resolve to the ``re`` module object. + + Covers a direct alias name, and the two indirect routes to the same object that + carry no alias name at all: ``sys.modules["re"]`` and a ``vars``/``getattr`` call + over one. Deny-first -- a subtree naming ``sys.modules`` cannot be proved NOT to + hand back the module, so it counts. + """ + for inner in ast.walk(node): + if isinstance(inner, ast.Name) and inner.id in aliases: + return True + if isinstance(inner, ast.Attribute) and inner.attr == "modules": + return True + if isinstance(inner, ast.Constant) and inner.value == "re": + return True + return False + + +def _compile_result_is_untrackable(tree: ast.AST) -> bool: + """True when a compile result is bound somewhere the escape analysis cannot follow. + + ``_compiled_pattern_names`` can only track a plain ``ast.Name`` target, so the + deny-first escape check silently covers nothing when the result is bound to a + SUBSCRIPT, an ATTRIBUTE or a tuple element -- ``d["p"] = re.compile(FENCED)`` is + exonerated in the pattern slot and then ``str(d["p"])`` recovers the literal with + no name for the guard to watch. Rather than attempt alias analysis into containers, + the untrackable binding itself withdraws the exoneration. + + Enumerating binding SHAPES is not sufficient, because a result that is never bound + reaches no binding node at all: ``return re.compile(FENCED)`` hands + ``_compiled_name_escapes`` an EMPTY set, which cannot fail. So the rule is inverted + -- exoneration requires the compile result to be the DIRECT value of an assignment + whose every target is a plain Name, which is exactly what the tracker can follow, + and every other position forfeits. That single inverted rule SUBSUMES the shape + enumeration it replaced: a binding with a non-Name target fails the ``all(...)`` + below, so the compile ``Call``'s id never reaches ``trackable`` and the final walk + already forfeits it. + """ + # An unbound result reaches no binding node at all, so the escape check would be + # handed an EMPTY set and could not fail; only a plain-Name target is trackable. + trackable: set[int] = set() + for node in ast.walk(tree): + if isinstance(node, ast.Assign): + bound, bound_targets = node.value, tuple(node.targets) + elif isinstance(node, ast.AnnAssign) and node.value is not None: + bound, bound_targets = node.value, (node.target,) + elif isinstance(node, ast.NamedExpr): + bound, bound_targets = node.value, (node.target,) + else: + continue + if all(isinstance(t, ast.Name) for t in bound_targets): + trackable.add(id(bound)) + + for node in ast.walk(tree): + if not isinstance(node, ast.Call): + continue + func = node.func + if not (isinstance(func, ast.Attribute) and isinstance(func.value, ast.Name)): + continue + if (func.value.id, func.attr) != _COMPILING_SINK: + continue + if id(node) not in trackable: + return True + return False + + +_DYNAMIC_NAMESPACE_BUILTINS = frozenset({"globals", "locals", "vars"}) +"""Builtins handing back a namespace MAPPING, so a write through one rebinds a name. + +``globals()["re"] = Fake`` rebinds the name ``re`` without producing a Name in +``Store`` context, an Attribute, or any argument mentioning the module -- the subscript's +``value`` is the bare ``globals()`` call, which names nothing this walk recognises. So +every binding branch is bypassed while the module reads authentic and a fenced literal +in a pattern slot stays exonerated. + +Matched on the NAME rather than on the subscript, because the mapping can be bound +first: ``g = globals()`` then ``g["re"] = Fake`` leaves the subscript's value a plain +local, which no amount of inspecting that subscript can resolve back to the namespace. +``locals()`` at module scope IS the global namespace, and ``vars()`` with no argument is +``locals()``, so all three spellings are the same capability and forfeit together. +""" + +_DYNAMIC_EXECUTION_BUILTINS = frozenset({"exec", "eval", "compile", "__import__"}) +"""Builtins that run code this checker cannot see, so their presence forfeits trust. + +Every other guard in ``_re_module_is_authentic`` is a STATIC read of the parse tree, so +a body that builds its rebinding at run time defeats all of them at once: +``exec("re.compile = open")`` carries the mutation inside a STRING, reaching no Name, +Attribute, Subscript or argument the tree can be asked about. Enumerating spellings +cannot close that -- the string is opaque by construction -- so the presence of the +mechanism withdraws the exoneration instead, the same way an untrackable binding does. + +``compile`` is the BUILTIN, matched only as a bare ``ast.Name``. ``re.compile`` spells +its name in an ``ast.Attribute``'s ``attr`` STRING and so is not a Name node at all, +which is why the ordinary redactor this gate exists to permit is unaffected. +""" + + +def _re_module_is_authentic(tree: ast.AST) -> bool: + """True when the name ``re`` in this body is bound ONLY by a plain ``import re``. + + The exoneration set keys on the SPELLING ``re.``, so a body that rebinds the + name could otherwise launder a path read through a call that merely looks + allowlisted. Every binding spelling counts, and they do NOT all reach the AST as a + ``Name`` node: ``import evil as re``, ``re = something``, ``class re``, a parameter + or loop variable named ``re`` bind through nodes, while ``except E as re:``, + ``case re:``, ``case [*re]`` and ``case {**re}`` bind through plain STRING + attributes (``ExceptHandler.name``, ``MatchAs``/``MatchStar.name``, + ``MatchMapping.rest``) that a Name-only walk cannot see at all. + + Reassigning an ATTRIBUTE of the module counts too: ``re.compile = open`` leaves the + NAME bound to the genuine module while the call it spells now opens a file, so the + allowlist would exonerate a literal flowing into ``open``. Passing the module to a + CALL counts as well -- ``setattr(re, "compile", open)``, or an ordinary + ``helper(re)`` whose body does the same -- because the callee's effect on the object + is not readable from this tree, so any argument resolving to the module forfeits + rather than only the callees somebody thought to enumerate. The check is therefore + that the module, its attributes and the name are all untouched, that the module is + never handed to a call, and that the body has no way to run code this walk cannot + see (see ``_DYNAMIC_EXECUTION_BUILTINS``). + + Storing the module through a CONTAINER or a derived expression is the same escape + one level out: ``holder = [re]`` then ``m = holder[0]`` puts the module object + behind a subscript no static walk can resolve, so ``m.compile = reader`` mutates + the module while an alias walk keyed on bare ``m = re`` assignments records + nothing. Enumerating container shapes would rebuild the allow-by-default + blocklist this function already had to abandon once, so the rule closes the class + instead: a module reference may only be READ through an attribute or bound as a + tracked bare alias, and every other mention forfeits. + + Known over-block, by design: this is a WHOLE-BODY verdict, so a local variable named + ``re`` anywhere withdraws exoneration for every literal in the body -- erring toward + the false positive this PR exists to clear rather than toward the fence. Narrowing it + to lexical scope would need full scope resolution; recorded here so it is understood + as a deliberate trade rather than rediscovered as a bug. + """ + imported = False + # A body can hold the SAME module object under another name -- ``m = re``, or + # ``import re as m`` -- and mutate it from there. ``m.compile = open`` rebinds + # exactly the attribute that ``re.compile`` spells, so an ``re``-keyed check sees + # nothing while the call it exonerates now opens a file. Collect every alias first + # and treat mutation through any of them as mutation of the module. + aliases = {"re"} + assigned_pairs: list[tuple[ast.expr, ast.expr]] = [] + for node in ast.walk(tree): + if isinstance(node, ast.Assign): + pairs: list[tuple[ast.expr, ast.expr]] = [ + (t, node.value) for t in node.targets + ] + elif isinstance(node, ast.AnnAssign) and node.value is not None: + pairs = [(node.target, node.value)] + elif isinstance(node, ast.NamedExpr): + pairs = [(node.target, node.value)] + elif isinstance(node, ast.Import): + for alias in node.names: + if alias.name == "re" and alias.asname: + aliases.add(alias.asname) + continue + else: + continue + for target, value in pairs: + assigned_pairs.append((target, value)) + if ( + isinstance(target, ast.Name) + and isinstance(value, ast.Name) + and value.id in aliases + ): + aliases.add(target.id) + + # An attribute READ (``re.compile``) and a tracked bare alias (``m = re``) are the + # only sanctioned mentions; any other reference hands the object somewhere opaque. + sanctioned = { + id(value) + for target, value in assigned_pairs + if isinstance(target, ast.Name) + and isinstance(value, ast.Name) + and value.id in aliases + } + for node in ast.walk(tree): + if isinstance(node, ast.Attribute) and isinstance(node.value, ast.Name): + sanctioned.add(id(node.value)) + for node in ast.walk(tree): + if ( + isinstance(node, ast.Name) + and node.id in aliases + and isinstance(node.ctx, ast.Load) + and id(node) not in sanctioned + ): + return False + + for node in ast.walk(tree): + if isinstance(node, ast.Name): + # Store/Del covers assignment, augmented assignment, walrus, for/with + # targets and comprehension variables -- every rebinding spelling that + # goes through a Name node. + if node.id == "re" and isinstance(node.ctx, (ast.Store, ast.Del)): + return False + # Dynamic execution defeats every static check in this function at once, + # because the mutation travels inside a string the tree cannot be asked + # about. Matched on the NAME rather than the call, so binding it to + # another name first -- ``e = exec`` then ``e("re.compile = open")`` -- + # forfeits just the same. + if node.id in _DYNAMIC_EXECUTION_BUILTINS: + return False + # A namespace mapping is the same problem one level down: a write through + # ``globals()``/``locals()``/``vars()`` rebinds the name with no Name in + # Store context to see. Also matched on the NAME, because the mapping can + # be bound first (``g = globals()`` then ``g["re"] = Fake``), which leaves + # the subscript's own value an unresolvable local. + if node.id in _DYNAMIC_NAMESPACE_BUILTINS: + return False + elif isinstance(node, ast.ExceptHandler) and node.name == "re": + # ``except Exception as re:`` binds the name through a plain STRING + # attribute, not a Name node, so the Name branch above cannot see it. + return False + elif isinstance(node, (ast.MatchAs, ast.MatchStar)): + # ``case re:`` / ``case [*re]`` -- also plain string attributes. + if getattr(node, "name", None) == "re": + return False + elif isinstance(node, ast.MatchMapping) and node.rest == "re": + # ``case {**re}`` -- likewise a string. + return False + elif isinstance(node, ast.Subscript) and isinstance( + node.ctx, (ast.Store, ast.Del) + ): + # ``vars(re)["compile"] = open``, ``re.__dict__["compile"] = open`` and + # ``sys.modules["re"].compile = open`` all reach the module's namespace + # through a SUBSCRIPT, so neither the Attribute nor the Call branch fires. + if _mentions_module_alias(node.value, aliases): + return False + elif isinstance(node, ast.Attribute) and isinstance(node.value, ast.Name): + # ``re.compile = open`` / ``del re.sub`` -- the module survives, the + # function does not. Any ALIAS of the module counts: ``m = re`` then + # ``m.compile = open`` mutates the same object. + if node.value.id in aliases and isinstance(node.ctx, (ast.Store, ast.Del)): + return False + elif isinstance(node, ast.Attribute) and isinstance( + node.ctx, (ast.Store, ast.Del) + ): + # The base is an expression rather than a plain name, e.g. + # ``sys.modules["re"].compile = open``. + if _mentions_module_alias(node.value, aliases): + return False + elif isinstance(node, ast.Call): + # Handing the module OBJECT to a call hands over the ability to mutate it, + # and what the callee does with it is not readable from this tree. The + # branch used to recognise only ``setattr``/``delattr``, which made it an + # allow-by-default blocklist inside a deny-by-default checker: any other + # call taking the module -- ``helper(re)``, whose body does + # ``m.compile = open`` -- reached no branch at all and the module still + # read authentic. So ANY argument resolving to the module forfeits, which + # closes the class instead of enumerating callees. + # + # Only the ARGUMENTS are inspected, never ``node.func``: ``re.sub(...)`` + # and ``re.compile(...)`` name the module in the func position, and those + # are exactly the calls this exoneration exists to permit. + for argument in (*node.args, *(kw.value for kw in node.keywords)): + if _mentions_module_alias(argument, aliases): + return False + elif isinstance(node, ast.Import): + for alias in node.names: + if alias.asname is None and alias.name == "re": + imported = True + elif alias.asname == "re": + return False + elif isinstance(node, ast.ImportFrom): + # ``from evil import *`` can bind ``re`` while no alias in the node names + # it, so an UNKNOWABLE binding set forfeits -- the class, not a spelling. + for alias in node.names: + if alias.name == "*" or (alias.asname or alias.name) == "re": + return False + elif isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)): + if node.name == "re": + return False + elif isinstance(node, ast.arg) and node.arg == "re": + return False + return imported + + +def _replacement_is_provably_non_callable(call: ast.Call, position: int = 1) -> bool: + """True only when a substituting call's ``repl`` argument provably cannot be called. + + ``re.sub``/``re.subn`` accept a replacement that may be a FUNCTION, and ``re`` hands + that function the ``Match`` -- from which ``Match.re.pattern`` returns the verbatim + pattern literal. So a fenced literal is safe in the pattern slot only when nothing + can receive that Match. + + ``position`` is where ``repl`` sits positionally, which differs by spelling: the + module-level ``re.sub(pattern, repl, string)`` puts it at 1, while a COMPILED + ``p.sub(repl, string)`` binds the pattern in the object and puts it at 0. The + keyword spelling is ``repl=`` either way. + + Provable means a ``str`` or ``bytes`` CONSTANT, which cannot be called at all. + Everything else fails closed: a Name or Attribute may be bound to a function, a + lambda plainly is one, a Starred puts the replacement at a position nobody can know + statically, and a missing argument means there is nothing to prove. Deciding on the + argument's shape rather than on the recovery spelling is what closes the class -- + the read itself can be spelled to defeat any enumeration. + """ + replacement: ast.expr | None = None + if len(call.args) > position: + replacement = call.args[position] + for keyword in call.keywords: + if keyword.arg == "repl": + replacement = keyword.value + if replacement is None: + return False + return isinstance(replacement, ast.Constant) and isinstance( + replacement.value, (str, bytes) + ) + + +def _enclosing_call_slot( + chain: "Sequence[ast.AST]", leaf: ast.AST +) -> "tuple[tuple[str | None, str] | None, bool]": + """``((module, attr), literal_is_in_the_pattern_slot)`` for the innermost call. + + Walks outward, so an f-string or a nested expression still resolves to the call + that actually receives the value. A bare name (``open(...)``) yields + ``(None, 'open')``, which is deliberately NOT in the exoneration set. + + The second element is what makes the exoneration argument-position-aware. An + allowlisted ``re.*`` call is only safe for the literal in its PATTERN operand: + ``re.sub(pattern, repl, string)`` returns its SUBJECT verbatim and its + REPLACEMENT substantially so, so a fenced path in either of those slots flows on + to a real sink. Only ``args[0]`` / ``pattern=`` describes a regex. + + Occupying that slot means BEING the operand, not merely reaching it. ``inner`` is + the top of the argument subtree, so an expression feeding the operand satisfies + ``args[0] is inner`` while the literal sits underneath it -- and evaluating that + expression runs code before ``re`` ever sees a pattern. ``re.compile(FENCED + + Reader())`` hands the expanded path to ``Reader.__radd__``, and every other + operator protocol (``__rmod__`` for ``%`` formatting, ``__ror__``, and so on) is + the same shape. Requiring ``inner is leaf`` keeps the exoneration to a literal + that IS the operand, which also subsumes the walrus and f-string spellings. + + The exoneration additionally requires the matched call to be the OUTERMOST + expression the literal reaches. A pattern-consuming call whose RESULT is consumed + by another call, or has an attribute read off it, hands the verbatim literal + onward -- ``open(re.sub(FENCED, ...))`` and ``re.compile(FENCED).pattern`` both + recover the fenced spelling through a call that merely looks safe. + + Known over-block, by design: a chained-but-harmless form such as + ``re.compile(FENCED).search(s)`` is denied too, because an attribute read is not + distinguishable here from a ``.pattern`` re-extraction. That errs toward the + false positive rather than toward the fence, the same trade the rebinding rule + makes. + """ + for index in range(len(chain) - 1, -1, -1): + parent = chain[index] + if isinstance(parent, ast.Call): + # The next node down the chain toward the leaf is the top of whichever + # argument subtree the literal sits in (the leaf itself when the call is + # the literal's direct parent). + inner = chain[index + 1] if index + 1 < len(chain) else leaf + # ``args[0] is inner`` proves the position ONLY when arg 0 is a plain + # operand. ``re.sub(*seq)`` makes args[0] a Starred whose unpacked + # contents land at positions nobody can know statically, so the literal + # may really be the SUBJECT -- treat an unprovable position as not the + # pattern slot and let the deny default stand. + in_pattern_slot = ( + bool(parent.args) + and parent.args[0] is inner + and inner is leaf + and not isinstance(parent.args[0], ast.Starred) + ) + for keyword in parent.keywords: + # ``ast.iter_child_nodes(Call)`` yields the KEYWORD node, so ``inner`` + # is that keyword and never its ``.value``. Comparing against the + # value could therefore never be true, which left this branch dead and + # denied the keyword spelling of a redactor while its positional twin + # was exonerated. + if ( + keyword.arg == "pattern" + and keyword is inner + and keyword.value is leaf + ): + in_pattern_slot = True + if in_pattern_slot: + for ancestor in chain[:index]: + if isinstance(ancestor, (ast.Call, ast.Attribute)): + in_pattern_slot = False + break + if in_pattern_slot: + # The ancestor walk above only sees nodes ABOVE the call. A walrus + # sits BELOW it, inside the argument subtree -- ``re.compile(p := + # FENCED)`` puts the literal in a genuine pattern slot while ALSO + # binding it to a name that outlives the call, so a later + # ``open(p)`` gets the verbatim fenced spelling. The exoneration + # argument is that a pattern operand goes nowhere else; a walrus is + # precisely the counter-example, so withdraw it. + for descendant in (*chain[index + 1 :], leaf): + if isinstance(descendant, ast.NamedExpr): + in_pattern_slot = False + break + func = parent.func + if isinstance(func, ast.Attribute) and isinstance(func.value, ast.Name): + key: tuple[str | None, str] | None = (func.value.id, func.attr) + elif isinstance(func, ast.Name): + key = (None, func.id) + else: + key = None + # A callable replacement receives the Match, whose ``.re.pattern`` hands + # back the verbatim literal, so the slot is safe only without one. + if ( + in_pattern_slot + and key in _SUBSTITUTING_SINKS + and not _replacement_is_provably_non_callable(parent) + ): + in_pattern_slot = False + return (key, in_pattern_slot) + return (None, False) + + +def _compiled_pattern_names(tree: ast.AST) -> set[str]: + """Names bound to the RESULT of a call that returns a COMPILED PATTERN. + + Only ``re.compile`` qualifies. The other members of ``_SOURCE_PATTERN_SINKS`` are + pattern CONSUMERS whose results are ordinary values -- ``re.sub``/``re.subn`` + return a string, ``match``/``search``/``fullmatch`` a Match, ``findall``/``split`` + a list -- and none of them can hand back the verbatim pattern. Tracking them made + ``redacted = re.sub(F, "", s)`` followed by ``str(redacted)`` read as a + re-extraction, which refused the very redactor shape this change exists to allow. + + A Match object does expose the pattern as ``m.re``, but that is an attribute read + and is caught by name in ``_pattern_reextracted`` rather than needing this set. + """ + names: set[str] = set() + for node in ast.walk(tree): + value: ast.expr | None = None + targets: tuple[ast.expr, ...] = () + if isinstance(node, ast.Assign): + value, targets = node.value, tuple(node.targets) + elif isinstance(node, ast.AnnAssign) and node.value is not None: + value, targets = node.value, (node.target,) + elif isinstance(node, ast.NamedExpr): + value, targets = node.value, (node.target,) + if not isinstance(value, ast.Call): + continue + func = value.func + if isinstance(func, ast.Attribute) and isinstance(func.value, ast.Name): + key: tuple[str | None, str] = (func.value.id, func.attr) + elif isinstance(func, ast.Name): + key = (None, func.id) + else: + continue + if key != _COMPILING_SINK: + continue + for target in targets: + if isinstance(target, ast.Name): + names.add(target.id) + return names + + +def _compiled_name_escapes(tree: ast.AST, compiled: set[str]) -> bool: + """True unless every READ of a compiled-pattern name sits in a SAFE position. + + This is the deny-by-default half of the re-extraction guard. Enumerating recovery + spellings cannot work: the ways to get a pattern back out of a compiled object are + open-ended -- ``p.pattern``, ``getattr(p, "pattern")``, + ``p.__getattribute__("pattern")``, ``object.__getattribute__(p, "pattern")``, + ``operator.attrgetter("pattern")(p)``, an aliased ``g = getattr``, and every + stringify form (``"%s" % p``, ``"{}".format(p)``, ``f"{p!r}"``, ``str``/``repr``/ + ``vars``) -- so a blocklist is one unenumerated spelling away from reopening the + fence. That is backwards inside a checker whose stated invariant is deny-first. + + So the direction inverts: a compiled pattern may be USED through its own matching + API and nothing else. Anything that passes the object to a call, formats it, + subscripts it, stores it in a container or returns it forfeits the exoneration for + the whole body, because any of those can recover the literal. + + Known over-block, accepted on the same reasoning as the sibling trades: a read of + a genuinely harmless attribute (``p.flags``, ``p.groupindex``) is not enumerated + and therefore withdraws the exoneration too. That errs toward the false positive + rather than toward the fence. + """ + parents: dict[ast.AST, ast.AST] = {} + for node in ast.walk(tree): + for child in ast.iter_child_nodes(node): + parents[child] = node + + for node in ast.walk(tree): + if not (isinstance(node, ast.Name) and node.id in compiled): + continue + if not isinstance(node.ctx, ast.Load): + # Binding the name is not a read of the object. + continue + parent = parents.get(node) + if ( + isinstance(parent, ast.Attribute) + and parent.attr in _SAFE_COMPILED_PATTERN_METHODS + and isinstance(parents.get(parent), ast.Call) + and getattr(parents.get(parent), "func", None) is parent + ): + call = parents.get(parent) + assert isinstance(call, ast.Call) + # The method NAME is not safety for the two families that hand a Match out: + # `p.sub` passes one to its replacement, `p.search` returns one. + if parent.attr in _SUBSTITUTING_METHODS: + if not _replacement_is_provably_non_callable(call, position=0): + return True + elif parent.attr in _MATCH_RETURNING_METHODS: + if not isinstance(parents.get(call), ast.Expr): + return True + continue + return True + return False + + +def _pattern_reextracted(tree: ast.AST) -> bool: + """True when the body reads a compiled pattern back off an ``re`` object. + + ``re.compile(x).pattern`` and ``m.re.pattern`` hand back the VERBATIM literal, so + a fenced path parked in an exonerated pattern slot can be recovered in a LATER + statement and handed to a real sink -- ``p = re.compile(FENCED)`` then + ``open(p.pattern)``. That crosses statements, so the per-literal slot check cannot + see it; this is a whole-body verdict for the same reason + ``_re_module_is_authentic`` is one. + + An attribute read is not the only spelling, and the set of spellings is open-ended: + ``getattr(p, "pattern")`` carries the name in a string argument, + ``p.__getattribute__("pattern")`` and ``object.__getattribute__(p, "pattern")`` + are Calls whose ``func`` is itself an Attribute, ``operator.attrgetter("pattern")`` + defers the read, an aliased ``g = getattr`` hides the callee's name, and every + stringify form (``"%s" % p``, ``"{}".format(p)``, ``f"{p!r}"``, ``str``/``repr``/ + ``vars``) embeds the pattern in its output. Enumerating them is therefore the wrong + shape for a deny-first checker, so the decision is delegated to + ``_compiled_name_escapes``: a compiled pattern may be used through its own + matching API, and any other use forfeits the exoneration. + + The direct attribute-name check is kept as well, because it also covers reads off + objects this body never bound to a name -- ``m.re`` on a Match, or a chained + ``re.compile(F).pattern``. + + Known over-block, by design: any body reading an attribute named ``pattern`` or + ``re`` loses the exoneration for EVERY literal in it, including where the name + means something unrelated. That errs toward the false positive rather than toward + the fence. + """ + compiled = _compiled_pattern_names(tree) + if _compile_result_is_untrackable(tree): + return True + if compiled and _compiled_name_escapes(tree, compiled): + return True + for node in ast.walk(tree): + if isinstance(node, ast.Attribute) and node.attr in ( + "pattern", + "re", + "__dict__", + ): + return True + # ``getattr(x, "pattern")`` on an object this body never bound -- the escape + # analysis above is keyed on tracked names, so this covers the rest. + if not isinstance(node, ast.Call): + continue + func = node.func + name = func.id if isinstance(func, ast.Name) else getattr(func, "attr", None) + if name in ("getattr", "__getattribute__") and len(node.args) >= 2: + attr = node.args[1] + if isinstance(attr, ast.Constant) and attr.value in ( + "pattern", + "re", + "__dict__", + ): + return True + elif name == "attrgetter" and node.args: + attr = node.args[0] + if isinstance(attr, ast.Constant) and attr.value in ("pattern", "re"): + return True + return False + + +_COMPILING_SINK: tuple[str | None, str] = ("re", "compile") +"""The one sink in ``_SOURCE_PATTERN_SINKS`` whose RESULT is a compiled pattern. + +Every other member consumes a pattern and returns an ordinary value, so binding its +result cannot give a body a route back to the verbatim literal. +""" + +_MATCH_RETURNING_METHODS = frozenset({"match", "search", "fullmatch", "finditer"}) +"""Compiled-object methods whose RESULT is a ``Match``. + +``_SAFE_COMPILED_PATTERN_METHODS`` admits a compiled pattern's whole matching API on +the METHOD NAME alone, which is not safety for these four: ``p.search(s)`` hands back a +Match, and a Match carries ``.re``, so ``Match.re.pattern`` is the verbatim pattern +literal. Nothing tracks a Match -- ``_compiled_pattern_names`` follows only +``_COMPILING_SINK`` results -- and the recovery read cannot be enumerated, since +``getattr(m, "r" + "e")`` builds the attribute name from a `BinOp`. So the literal is +recoverable through the returned object while the compiled name itself never escapes. + +This set has no module-level counterpart, and deliberately so: ``re.match``, +``re.search``, ``re.fullmatch`` and ``re.finditer`` are simply ABSENT from +``_SOURCE_PATTERN_SINKS``, so deny-by-default refuses a fenced literal in their pattern +slot outright. Withdrawing an exoneration they never receive would be dead code. The +compiled route is different because the exoneration is earned by ``re.compile``, which +IS in the set, and only then is the Match produced by a later method call. +""" + +_SUBSTITUTING_METHODS = frozenset({"sub", "subn"}) +"""The compiled-object counterparts of ``_SUBSTITUTING_SINKS``. + +``p.sub(repl, s)`` passes ``repl`` the Match just as the module-level call does, so the +same replacement-is-provably-non-callable test applies -- at position 0, because the +pattern is bound in the object rather than passed. +""" + +_SUBSTITUTING_SINKS: frozenset[tuple[str | None, str]] = frozenset( + {("re", "sub"), ("re", "subn")} +) +"""The members of ``_SOURCE_PATTERN_SINKS`` that accept a CALLABLE replacement. + +These take a ``repl`` that may be a function, and ``re`` hands that function the +``Match``. A Match carries ``.re``, so ``Match.re.pattern`` returns the verbatim +pattern literal -- a route back to a fenced spelling that exists for no other member +of the set. It is not reachable through the compiled-name escape analysis either, +because that tracks only ``_COMPILING_SINK`` results, and a Match is never bound by +the exonerated statement at all. Chasing the recovery spelling is the wrong layer for +the same reason it was for the compile result: the read can be spelled to defeat any +enumeration (``getattr(m, "r" + "e")`` builds the name, ``g = getattr`` hides the +callee), so the exoneration is withdrawn at the SLOT unless the replacement provably +cannot be called. +""" + +_SAFE_COMPILED_PATTERN_METHODS = frozenset( + {"search", "match", "fullmatch", "split", "findall", "finditer", "sub", "subn"} +) +"""The matching API a compiled pattern may be used THROUGH. + +This is an allowlist by design: the re-extraction guard denies by default, so a use +that is not one of these withdraws the exoneration rather than being waved through. +""" + + +def _fence_hit_in_collapsed( + value: str, *, value_already_scanned: bool = False +) -> str | None: + """The three pass-1b checks over ``value`` and its separator-collapsed copies. + + ``_separator_collapsed_variants`` yields NOTHING when the value carries no + separator run, so iterating it alone leaves a value whose separators are already + single completely unchecked. On the shell path that is harmless -- pass 1a checked + the unmodified subject before the collapse ran -- but the source-literal scan has + no such earlier pass over the DECODED literal, so a non-raw + ``"…\\\\kiro-cli\\\\c.json"`` (one separator each once decoded) reached the fence + through no layer of its own and depended on a later pass to catch it. Checking the + value first makes this layer self-sufficient rather than sharing the blind spot of + the pass behind it. + + ``value_already_scanned`` drops only that self-sufficiency check, for a caller that + has ALREADY put the identical bytes through these same three matchers. It exists + because "harmless" above was true of correctness but not of cost: on the shell path + the raw subject is re-scanned in full, and the sensitive-path regex over a long + newline-free line is the most expensive matcher on this gate, so a 20 KB subject + paid for it twice and doubled the wall time of a linearity-guarded path. The + collapsed copies are still checked in every case; what is skipped is a provably + duplicate pass, never a layer. The default stays False so a new caller is + self-sufficient unless it opts out deliberately. + """ + candidates = _separator_collapsed_variants(value) + for candidate in candidates if value_already_scanned else (value, *candidates): + if _get_sensitive_re().search(candidate): + return "Blocked: command accesses sensitive credential path" + if _extracts_into_trust_root(candidate): + return "Blocked: command extracts into the governance trust-root directory" + if _RELATIVE_SENSITIVE_RE.search(candidate): + return ( + "Blocked: command references a sensitive credential path " + "via relative traversal" + ) + return None + + +def _sensitive_run_in_source_literals(source: str) -> tuple[bool, str | None]: + """Fence check for a separator RUN inside a decoded literal of a source body. + + Returns ``(parsed, reason)``. ``parsed`` is False when the body is not valid + Python, in which case there were no literals to inspect and ``reason`` is None — + the caller must then fall back to the raw scan WITH the collapse, so an + unparseable body is never quietly exonerated. Parse status is returned rather + than re-derived so the body is parsed once. + + A bare string constant in STATEMENT position is skipped, but a DOCSTRING is not: + Python evaluates and discards the former, while it retains a module, class or + function docstring as ``__doc__``, which a body can read back and hand to a sink + (``open(f.__doc__)``). So a docstring naming a fenced store IS scanned — which + over-blocks a docstring that merely warns a reader off one, the same direction the + other trades here take. Comments never reach this function at all; the parser + discards them. + + Both ``str`` and ``bytes`` constants are inspected, because ``open`` and + ``os.open`` accept a bytes path. An allowlisted ``re.*`` call exonerates a + literal only in its PATTERN operand, only when that call is the outermost + expression the literal reaches, and only when the body has neither rebound the + name ``re`` nor read a pattern back off an ``re`` object; every other position and + spelling denies. + """ + try: + tree = ast.parse(source) + except (SyntaxError, ValueError, RecursionError): + return (False, None) + + # Statement-position constants are evaluated and thrown away by Python -- EXCEPT a + # DOCSTRING, which Python RETAINS as ``__doc__`` on its module, class or function. + # A fenced literal parked in a docstring is therefore readable at runtime and can + # be handed to a sink (``open(f.__doc__)``), so only a bare string that is NOT a + # docstring is genuinely discarded. + retained: set[int] = set() + for node in ast.walk(tree): + if isinstance( + node, (ast.Module, ast.ClassDef, ast.FunctionDef, ast.AsyncFunctionDef) + ): + body = getattr(node, "body", None) + if body and isinstance(body[0], ast.Expr): + first = body[0].value + if isinstance(first, ast.Constant) and isinstance( + first.value, (str, bytes) + ): + retained.add(id(first)) + + discarded: set[int] = set() + for node in ast.walk(tree): + if isinstance(node, ast.Expr) and isinstance(node.value, ast.Constant): + if ( + isinstance(node.value.value, (str, bytes)) + and id(node.value) not in retained + ): + discarded.add(id(node.value)) + + re_authentic = _re_module_is_authentic(tree) and not _pattern_reextracted(tree) + + def visit(node: ast.AST, parents: list[ast.AST]) -> str | None: + for child in ast.iter_child_nodes(node): + if isinstance(child, ast.Constant) and id(child) not in discarded: + # bytes are decoded because open()/os.open accept a bytes path, so a + # rb"..." literal reaches the same sinks as its str twin. latin-1 is + # total over a byte range and preserves each byte as one code point, + # so a separator run survives the decode unchanged. + if isinstance(child.value, str): + value: str | None = child.value + elif isinstance(child.value, bytes): + value = child.value.decode("latin-1") + else: + value = None + if value is not None: + reason = _fence_hit_in_collapsed(value) + if reason: + key, in_pattern_slot = _enclosing_call_slot( + parents + [node], child + ) + # Every condition here DENIES unless positively cleared: no + # enclosing call (ambiguous), a call outside the allowlist, a + # slot other than the pattern operand, or an ``re`` name this + # body rebound. Spelled out rather than left to ``not in`` so + # the deny direction is explicit. + if ( + key is None + or key not in _SOURCE_PATTERN_SINKS + or not in_pattern_slot + or not re_authentic + ): + return reason + nested = visit(child, parents + [node]) + if nested: + return nested + return None + + try: + return (True, visit(tree, [])) + except RecursionError: + # A legitimately deep expression must not crash the caller. Reporting + # ``parsed=False`` routes it to the raw scan WITH the collapse, which is the + # conservative direction: the body is still fence-checked, just textually. + return (False, None) + + def is_sensitive_bash_command( - command: str, *, enabled_ids: "frozenset[str] | None" = None + command: str, + *, + enabled_ids: "frozenset[str] | None" = None, + _subject_is_shell_grammar: bool = True, ) -> str | None: """Check if a bash command reads sensitive paths, accesses IMDS, or leaks env creds. @@ -8835,6 +9654,23 @@ 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. + Returns denial reason string, or None if clean. """ # ── Pass 1: regex fast-path ── @@ -8860,20 +9696,24 @@ def is_sensitive_bash_command( # ALL THREE pass-1 checks are repeated, not just the path matcher: the # extraction check is a separate control, and omitting it let # ``tar -xf evil.tar -C $HOME//.kiro/crew`` overwrite governance files - # through the doubled separator (found in review). + # through the doubled separator (found in review). That is why the skip below + # is keyed on the SUBJECT and never on a single check: a caller either has + # shell grammar, and gets all three, or does not, and gets none of them. # # Run only after the original missed, so nothing that needs the run intact # (a UNC ``\\server\share`` anchor) loses its match. - for collapsed in _separator_collapsed_variants(command): - if _get_sensitive_re().search(collapsed): - return "Blocked: command accesses sensitive credential path" - if _extracts_into_trust_root(collapsed): - return "Blocked: command extracts into the governance trust-root directory" - if _RELATIVE_SENSITIVE_RE.search(collapsed): - return ( - "Blocked: command references a sensitive credential path " - "via relative traversal" - ) + if _subject_is_shell_grammar: + # ONE spelling of the three collapsed-copy checks, shared with the source + # literal scan -- a second copy here would let the two drift apart. + # + # ``value_already_scanned`` because pass 1 above put THIS command through + # these same three matchers and missed; re-scanning it here bought nothing + # and cost a second sensitive-path regex pass over the whole subject, which + # on a long newline-free line is the dominant cost on this gate. Only the + # duplicate is dropped: every collapsed copy is still checked. + collapsed_reason = _fence_hit_in_collapsed(command, value_already_scanned=True) + if collapsed_reason: + return collapsed_reason # ── Pass 2: normalizer-based sensitive path detection ── normalizer_result = _check_sensitive_via_normalizer(command) @@ -8917,6 +9757,26 @@ def is_sensitive_bash_command( return None +def is_sensitive_source_body(text: str) -> str | None: + """The keystone fence for a subject that is Python SOURCE, not a shell command line. + + THE ONLY supported entry point for a source body, and the reason + ``_subject_is_shell_grammar`` is internal. The two halves are safe only in + composition: skipping pass 1b is sound *because* the literal scan replaces it, so a + caller that reached for the flag alone would silently reopen the doubled-separator + fence inside scripts (#6350). Owning the pairing here makes that a property of the + API rather than a convention a second caller has to know. + + Order matters. The literal scan runs first and its verdict wins; only then does the + shell matcher run, and it keeps pass 1b exactly when the body did NOT parse -- an + uninspected body is scanned as text rather than waved through. + """ + parses, literal_reason = _sensitive_run_in_source_literals(text) + if literal_reason: + return literal_reason + return is_sensitive_bash_command(text, _subject_is_shell_grammar=not parses) + + # `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 443081df93c..a019a881327 100644 --- a/test/test_mcp_cron_security.py +++ b/test/test_mcp_cron_security.py @@ -16,6 +16,7 @@ from __future__ import annotations +import ast import json import time import uuid @@ -446,6 +447,760 @@ def test_vet_script_contents_allows_benign(body): assert _vet_script_contents(body) is None +# 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 +# `%LOCALAPPDATA%\\kiro-cli` and `%LOCALAPPDATA%\kiro-cli` name one store -- +# strips the escapes and manufactures a path the source never contains. Each body +# below READS NOTHING: two only describe or redact a fenced store, and the third +# is a bare docstring. Every one has ZERO pass-1 hits before the collapse. +BENIGN_SCRIPTS_WITH_A_SEPARATOR_RUN = [ + # A redaction pattern over the Windows spelling of a fenced store. + 'import re\nSCRUB = re.compile(r"%LOCALAPPDATA%\\\\kiro-cli")\n', + # Escapes stripped by the collapse turn a REGEX into a literal path: + # `/home/\S*/\.kiro/...` reads as `/home/S*/.kiro/...`. + 'import re\nSCRUB = re.compile(r"/home/\\\\S*/\\\\.kiro/crew/security_policy.json")\n', + # The KEYWORD spelling of the first entry. `pattern=` must be exonerated exactly as + # the positional operand is -- they are the same redactor, and denying one while + # allowing the other is the asymmetry the dead keyword branch produced. + 'import re\nSCRUB = re.compile(pattern=r"%LOCALAPPDATA%\\\\\\\\kiro-cli")\n', + # A redactor that stringifies the RESULT of a consuming call. `re.sub` returns a + # string, so this cannot recover the pattern and must not be refused. + 'import re\n\n\ndef scrub(s):\n redacted = re.sub(r"%LOCALAPPDATA%\\\\\\\\kiro-cli", "", s)\n return str(redacted)\n', + # The motivating redactor, used through the matching API -- the enumerated-safe way. + 'import re\nSCRUB = re.compile(r"%LOCALAPPDATA%\\\\\\\\kiro-cli")\n\n\ndef scrub(s):\n return SCRUB.sub("", s)\n', +] + + +def test_a_docstring_naming_a_fenced_store_is_an_accepted_over_block(): + """A prose-only docstring naming the store is DENIED, deliberately. + + Two rules compose to this outcome and neither can be narrowed safely. Docstrings + are scanned because Python retains them as ``__doc__``, where a body can read one + back into a sink. The fence is checked against the literal's own value, not only + its separator-collapsed copies, because a value whose separators are already single + produces no collapsed copy at all and would otherwise reach this layer unexamined. + + Exempting docstrings from the value check would reopen the single-separator + ``open(f.__doc__)`` path, so the check stays uniform and this shape pays for it. + Recorded as a test rather than left in the benign corpus so the trade is explicit: + the body reads nothing, and it is refused anyway. + """ + body = ( + 'def run(ctx):\n' + ' """Never touch %LOCALAPPDATA%\\\\kiro-cli -- it is the keystone."""\n' + ) + err = _vet_script_contents(body) + assert err is not None and err.startswith("Error:") + + +@pytest.mark.parametrize("body", BENIGN_SCRIPTS_WITH_A_SEPARATOR_RUN) +def test_vet_script_contents_allows_a_separator_run_in_python_source(body): + assert _vet_script_contents(body) is None, f"should allow: {body!r}" + + +# The control for the test above: the run is meaningless only in SHELL grammar, so +# scoping the collapse to that subject must not reach the COMMAND path, where a +# doubled separator still names the store the single spelling names (#6350). A +# carve-out that leaked here would be a hole, not a false-positive fix. +# +# Each payload is reachable ONLY through pass 1b -- verified to be missed when the +# subject flag is flipped -- so this control can actually fail. One per check pass +# 1b repeats, because the collapse is keyed on the subject and never on one check: +# the path matcher, the extraction control, and the relative-traversal matcher. +COMMANDS_WITH_A_SEPARATOR_RUN = [ + r'type "%LOCALAPPDATA%\\kiro-cli\config.json"', + r"cat %USERPROFILE%\\.ssh\id_rsa", + r"tar -xf evil.tar -C $HOME//.kiro/crew", + r"cat ..//.aws/credentials", +] + + +@pytest.mark.parametrize("cmd", COMMANDS_WITH_A_SEPARATOR_RUN) +def test_vet_shell_command_still_blocks_a_separator_run(cmd): + err = _vet_shell_command(cmd) + assert err is not None and err.startswith("Error:"), f"should block: {cmd!r}" + + +# Scoping the collapse away from the script body must not reopen the fence INSIDE a +# script. These bodies hand a path to a filesystem sink whose decoded string VALUE +# carries a separator RUN; Win32 collapses that run when the file is opened, so the +# fenced store is reached — while the raw, uncollapsed source text matches no fence +# pattern. Blocked before the subject scoping, so each is a genuine regression guard. +# +# Note how little separates these from the benign bodies above: the first differs from +# the `re.compile` payload only in the call it wraps. A text- or value-level check +# cannot tell them apart, because a regex escape and a path separator are the same +# character once the literal is decoded — only the SINK differs. +ATTACK_SCRIPTS_WITH_A_SEPARATOR_RUN = [ + # open() on a run-carrying Windows path, raw spelling. + 'f = open(r"%LOCALAPPDATA%\\\\kiro-cli\\\\config.json")\n', + # Same decoded value, non-raw spelling. + 'f = open("%LOCALAPPDATA%\\\\\\\\kiro-cli\\\\\\\\config.json")\n', + # Mixed-separator run — one of the two regressions this collapse has had before. + 'f = open(r"%LOCALAPPDATA%\\/kiro-cli\\/config.json")\n', + # UNC leading pair — the other one; the leading pair must stay meaningful. + 'f = open(r"\\\\\\\\server\\\\share\\\\.kiro\\\\crew\\\\security_policy.json")\n', + # pathlib rather than the open() builtin. + 'from pathlib import Path\nPath(r"%LOCALAPPDATA%\\\\kiro-cli\\\\c.json").read_text()\n', + # The literal is bound to a name first, so no call encloses it. + 'P = r"%LOCALAPPDATA%\\\\kiro-cli\\\\c.json"\nopen(P)\n', + # An f-string, so the literal segment sits under a JoinedStr. + 'import os\nf = open(f"%LOCALAPPDATA%\\\\\\\\kiro-cli\\\\\\\\{os.sep}c.json")\n', + # A sink nobody enumerated: the deny verdict is the default, so this is covered + # without shutil appearing anywhere in the checker. + 'import shutil\nshutil.copy(r"%LOCALAPPDATA%\\\\kiro-cli\\\\c.json", "/tmp/x")\n', + # The two shapes `_separator_collapsed_variants`' own docstring records as prior + # review-found regressions, carried here in their literal form because a run in a + # decoded VALUE is the same hazard the shell path already learned twice. + # + # (1) MIXED run: collapsing to one fixed separator leaves a run matching neither + # spelling. `profiles` is a keystone leaf, so this reaches the trust root. + 'f = open(r"D:/\\\\profiles\\\\u\\\\.kiro\\\\crew\\\\admission_policy.json")\n', + # (2) UNC LEADING PAIR plus an interior run — the case the docstring records as + # having permitted the keystone read, because it matched neither the original + # (interior run) nor the collapsed copy (no UNC prefix left). + 'f = open(r"\\\\\\\\server\\\\share\\\\.kiro\\\\\\\\crew\\\\security_policy.json")\n', + # BYTES twin of the drive-letter case above. open()/os.open accept a bytes path, + # so skipping bytes constants left this reaching the fenced keystone. + 'f = open(rb"D:/\\\\profiles\\\\u\\\\.kiro\\\\crew\\\\admission_policy.json")\n', + # BYTES twin in the relative-traversal spelling — the other form Opus names. + 'f = open(rb"..\\\\..\\\\.kiro\\\\\\\\crew\\\\security_policy.json")\n', + # An allowlisted re.* call, but the fenced literal is in the SUBJECT slot, which + # re.sub returns verbatim to open(). Only the pattern operand is exonerated. + 'import re\nopen(re.sub(r"Q", "", r"%LOCALAPPDATA%\\\\\\\\kiro-cli\\\\\\\\c.json")).read()\n', + # Same call, fenced literal in the REPLACEMENT slot, which re.sub also passes + # through substantially unchanged. + 'import re\nopen(re.sub(r"Q", r"%LOCALAPPDATA%\\\\\\\\kiro-cli\\\\\\\\c.json", "Q")).read()\n', + # The exoneration keys on the SPELLING ``re.compile``, so a rebound ``re`` must + # withdraw it — otherwise the allowlist launders an arbitrary reader. + 'import shutil as re\nre.copy(r"%LOCALAPPDATA%\\\\\\\\kiro-cli\\\\\\\\c.json", "/tmp/x")\n', + # A pattern-slot literal is only exonerated when ``re`` is the imported module; + # here the name is reassigned, so the body loses the exoneration. + 'import re\nre = __import__("builtins")\nre.open(r"%LOCALAPPDATA%\\\\\\\\kiro-cli\\\\\\\\c.json")\n', + # The module NAME survives but its ATTRIBUTE is reassigned, so the call spells an + # allowlisted sink while actually being ``open``. + 'import re\nre.compile = open\nre.compile(r"%LOCALAPPDATA%\\\\\\\\kiro-cli\\\\\\\\c.json").read()\n', + # A STARRED argument: args[0] is the Starred node, so an identity check against it + # cannot prove the literal is the pattern rather than the subject. + 'import re\nopen(re.sub(*[r"Q", "", r"%LOCALAPPDATA%\\\\\\\\kiro-cli\\\\\\\\c.json"])).read()\n', + # The pattern slot is only safe when the re.* call is the OUTERMOST expression the + # literal reaches. Here its result is consumed by open(), so the fenced spelling + # flows on through a call that merely looks allowlisted. + 'import re\nopen(re.sub(r"%LOCALAPPDATA%\\\\\\\\kiro-cli\\\\\\\\c.json", "", "x")).read()\n', + # Compiled, then RE-EXTRACTED verbatim via `.pattern` in a later statement. + 'import re\np = re.compile(r"%LOCALAPPDATA%\\\\\\\\kiro-cli\\\\\\\\c.json")\nopen(p.pattern).read()\n', + # Same escape, bound by a walrus inside the opening call itself. + 'import re\nopen((p := re.compile(r"%LOCALAPPDATA%\\\\\\\\kiro-cli\\\\\\\\c.json")).pattern).read()\n', + # Parked in a DOCSTRING, which Python retains as __doc__, then read back out. + 'def f():\n r"%LOCALAPPDATA%\\\\\\\\kiro-cli\\\\\\\\c.json"\n\n\nopen(f.__doc__).read()\n', + # `except E as re:` binds the name through ExceptHandler.name -- a plain STRING, + # invisible to a Name-node walk -- so the module read as authentic. + 'import re\ntry:\n pass\nexcept Exception as re:\n re.compile(r"%LOCALAPPDATA%\\\\\\\\kiro-cli\\\\\\\\c.json")\n', + # `case re:` binds through MatchAs.name, also a plain string. + 'import re\nmatch object():\n case re:\n re.compile(r"%LOCALAPPDATA%\\\\\\\\kiro-cli\\\\\\\\c.json")\n', + # The attribute mutation spelled as a CALL reaches neither the Name nor the + # Attribute branch. + 'import re\nsetattr(re, "compile", open)\nre.compile(r"%LOCALAPPDATA%\\\\\\\\kiro-cli\\\\\\\\c.json")\n', + 'import re\ndelattr(re, "compile")\nre.compile(r"%LOCALAPPDATA%\\\\\\\\kiro-cli\\\\\\\\c.json")\n', + # Re-extraction spelled as a CALL. `getattr` puts the attribute name in a string + # argument, so it parses to an ast.Call and an Attribute-only walk never sees it -- + # while the dotted twin two entries below IS blocked. + 'import re\np = re.compile(r"%LOCALAPPDATA%\\\\\\\\kiro-cli\\\\\\\\c.json")\nopen(getattr(p, "pattern")).read()\n', + 'import re\np = re.compile(r"%LOCALAPPDATA%\\\\\\\\kiro-cli\\\\\\\\c.json")\nopen(p.pattern).read()\n', + # A dynamic attribute name cannot be proved harmless over a compiled object. + 'import re\nk = "pattern"\np = re.compile(r"%LOCALAPPDATA%\\\\\\\\kiro-cli\\\\\\\\c.json")\nopen(getattr(p, k)).read()\n', + # `repr` of a compiled pattern embeds the verbatim literal. + 'import re\np = re.compile(r"%LOCALAPPDATA%\\\\\\\\kiro-cli\\\\\\\\c.json")\nopen(repr(p)).read()\n', + # A walrus inside the pattern slot binds the literal to a name that OUTLIVES the + # call, so the "a pattern operand goes nowhere else" premise does not hold. + 'import re\nre.compile(p := r"%LOCALAPPDATA%\\\\\\\\kiro-cli\\\\\\\\c.json")\nopen(p).read()\n', + # Reflective reads: each is a Call whose `func` is an ast.Attribute, or hides the + # getter behind a name, so an enumerated bare-Name blocklist never fires. + 'import re\np = re.compile(r"%LOCALAPPDATA%\\\\\\\\kiro-cli\\\\\\\\c.json")\nopen(p.__getattribute__("pattern")).read()\n', + 'import re\np = re.compile(r"%LOCALAPPDATA%\\\\\\\\kiro-cli\\\\\\\\c.json")\nopen(object.__getattribute__(p, "pattern")).read()\n', + 'import operator\nimport re\np = re.compile(r"%LOCALAPPDATA%\\\\\\\\kiro-cli\\\\\\\\c.json")\nopen(operator.attrgetter("pattern")(p)).read()\n', + 'import re\np = re.compile(r"%LOCALAPPDATA%\\\\\\\\kiro-cli\\\\\\\\c.json")\ng = getattr\nopen(g(p, "pattern")).read()\n', + # Stringify forms: the pattern is embedded in the output of the format itself. + 'import re\np = re.compile(r"%LOCALAPPDATA%\\\\\\\\kiro-cli\\\\\\\\c.json")\nopen("%s" % p).read()\n', + 'import re\np = re.compile(r"%LOCALAPPDATA%\\\\\\\\kiro-cli\\\\\\\\c.json")\nopen("{}".format(p)).read()\n', + 'import re\np = re.compile(r"%LOCALAPPDATA%\\\\\\\\kiro-cli\\\\\\\\c.json")\nopen(f"{p!r}").read()\n', + # INDIRECT mutation: the module is held under a second name, or reached through + # sys.modules / vars(), so an `re`-keyed Attribute or Call check never sees it -- + # while `re.compile` still spells the attribute that was replaced. + 'import re\nm = re\nm.compile = open\nre.compile(r"%LOCALAPPDATA%\\\\\\\\kiro-cli\\\\\\\\c.json")\n', + 'import re\nm = re\nsetattr(m, "compile", open)\nre.compile(r"%LOCALAPPDATA%\\\\\\\\kiro-cli\\\\\\\\c.json")\n', + 'import re\nvars(re)["compile"] = open\nre.compile(r"%LOCALAPPDATA%\\\\\\\\kiro-cli\\\\\\\\c.json")\n', + 'import re\nimport sys\nsys.modules["re"].compile = open\nre.compile(r"%LOCALAPPDATA%\\\\\\\\kiro-cli\\\\\\\\c.json")\n', + # UNTRACKABLE binding: the compile result is bound to something that is not a plain + # name, so there is no name for the escape analysis to watch. + 'import re\nd = {}\nd["p"] = re.compile(r"%LOCALAPPDATA%\\\\\\\\kiro-cli\\\\\\\\c.json")\nopen(str(d["p"])).read()\n', + 'import re\np, q = re.compile(r"%LOCALAPPDATA%\\\\\\\\kiro-cli\\\\\\\\c.json"), 1\nopen(str(p)).read()\n', +] + + +@pytest.mark.parametrize("body", ATTACK_SCRIPTS_WITH_A_SEPARATOR_RUN) +def test_vet_script_contents_blocks_a_run_reaching_a_fenced_store(body): + err = _vet_script_contents(body) + assert err is not None and err.startswith("Error:"), f"should block: {body!r}" + + +def test_is_sensitive_source_body_owns_the_pairing(): + """The source entry point must own BOTH halves, so a caller cannot split them. + + Skipping pass 1b is sound only because the literal scan replaces it. A future + source-body caller reaching for the internal flag alone would silently reopen the + doubled-separator fence inside scripts, so the composition belongs to the API: the + public surface is `is_sensitive_source_body`, and the flag is private. + """ + import inspect + + from kiro_crew import security + + assert hasattr(security, "is_sensitive_source_body") + params = inspect.signature(security.is_sensitive_bash_command).parameters + assert "subject_is_shell_grammar" not in params, "the flag must not be public" + assert "_subject_is_shell_grammar" in params + + # The entry point blocks a run that only the literal scan can see... + attack = 'f = open(r"%LOCALAPPDATA%\\\\kiro-cli\\\\config.json")\n' + assert security.is_sensitive_source_body(attack) is not None + # ...and an unparseable body still gets the raw-text collapse. + broken = 'f = open(r"%LOCALAPPDATA%\\\\kiro-cli\\\\c.json"\n' + assert security.is_sensitive_source_body(broken) is not None + + +def test_authenticity_follows_the_module_through_an_alias(): + """Mutating the module under a second name is mutating the module. + + `m = re` binds the SAME object, so `m.compile = open` replaces exactly the + attribute that `re.compile` spells. A check keyed on the literal name `re` sees an + untouched module and exonerates a call that now opens a file. + """ + from kiro_crew.security import _re_module_is_authentic + + assert _re_module_is_authentic(ast.parse('import re\nre.compile("x")\n')) is True + for body in ( + 'import re\nm = re\nm.compile = open\n', + 'import re\nm = re\nsetattr(m, "compile", open)\n', + 'import re\nvars(re)["compile"] = open\n', + 'import re\nimport sys\nsys.modules["re"].compile = open\n', + 'import re as m\nm.compile = open\nimport re\n', + ): + assert _re_module_is_authentic(ast.parse(body)) is False, body + + +def test_a_compile_result_bound_where_it_cannot_be_tracked_forfeits_exoneration(): + """The escape analysis watches NAMES, so a non-name binding must deny instead. + + `_compiled_pattern_names` only tracks a plain `ast.Name` target. Binding the + compile result to a subscript, an attribute or a tuple element left the escape + check watching nothing while the literal stayed exonerated, so `str(d["p"])` + recovered the fenced spelling. + """ + from kiro_crew.security import _compile_result_is_untrackable + + trackable = 'import re\nSCRUB = re.compile("x")\n' + assert _compile_result_is_untrackable(ast.parse(trackable)) is False + for body in ( + 'import re\nd = {}\nd["p"] = re.compile("x")\n', + 'import re\nc.p = re.compile("x")\n', + 'import re\np, q = re.compile("x"), 1\n', + ): + assert _compile_result_is_untrackable(ast.parse(body)) is True, body + + +def test_the_recovery_guard_denies_by_default_rather_than_enumerating(): + """A compiled pattern may be used through its matching API and nothing else. + + The guard was an allow-by-default blocklist inside a deny-by-default checker, so + each unenumerated recovery spelling reopened the fence. Reads through the matching + API stay exonerated; every other use of the object forfeits it, which is what makes + the guard closed against spellings nobody thought of. + """ + from kiro_crew.security import _compiled_name_escapes + + safe = ast.parse('import re\np = re.compile("x")\np.sub("", s)\n') + assert _compiled_name_escapes(safe, {"p"}) is False + + for body in ( + 'import re\np = re.compile("x")\nopen(p.__getattribute__("pattern"))\n', + 'import re\np = re.compile("x")\nopen("%s" % p)\n', + 'import re\np = re.compile("x")\nopen(f"{p!r}")\n', + 'import re\np = re.compile("x")\nsend(p)\n', + 'import re\np = re.compile("x")\nreturn_value = [p]\n', + ): + assert _compiled_name_escapes(ast.parse(body), {"p"}) is True, body + + +def test_only_compile_results_are_tracked_as_compiled_patterns(): + """`re.sub` returns a STRING, so stringifying its result is not a re-extraction. + + Tracking every pattern SINK meant a redactor that returned `str(re.sub(...))` was + refused -- the exact shape this change exists to permit. Only the one sink whose + result is a pattern object can hand the literal back. + """ + from kiro_crew.security import _compiled_pattern_names + + assert _compiled_pattern_names(ast.parse('import re\np = re.compile("x")\n')) == {"p"} + consumed = 'import re\nredacted = re.sub("x", "", s)\n' + assert _compiled_pattern_names(ast.parse(consumed)) == set() + + +def test_call_spelled_reextraction_is_caught_like_the_dotted_spelling(): + """`getattr(p, "pattern")` must count as re-extraction, same as `p.pattern`. + + The attribute name travels in a string argument, so the node is an `ast.Call` and + an Attribute-only walk cannot see it -- the same call-spelled blind spot that + `setattr(re, ...)` exploited against the authenticity check. Two spellings of one + read must not disagree. + """ + from kiro_crew.security import _pattern_reextracted + + dotted = ast.parse('import re\np = re.compile("x")\nopen(p.pattern)\n') + called = ast.parse('import re\np = re.compile("x")\nopen(getattr(p, "pattern"))\n') + assert _pattern_reextracted(dotted) is True + assert _pattern_reextracted(called) is True + + +def test_reextraction_guard_does_not_fire_on_an_unrelated_str_call(): + """`str(count)` must not withdraw the exoneration. + + The verdict is whole-body, so scoping `str`/`repr`/`vars` to names actually bound + from a compiling call is what keeps the guard from denying most real redactor + scripts. Without this control the guard could pass its attack tests by simply + refusing everything. + """ + from kiro_crew.security import _pattern_reextracted + + tree = ast.parse('import re\nSCRUB = re.compile("x")\nn = str(42)\nm = str(n)\n') + assert _pattern_reextracted(tree) is False + + +def test_a_walrus_in_the_pattern_slot_forfeits_the_exoneration(): + """A literal bound by `:=` inside the pattern slot escapes the call. + + The exoneration rests on a pattern operand going nowhere else. A walrus binds the + same literal to a name that outlives the call, so `open(p)` in a later statement + receives the verbatim fenced spelling -- the premise fails and the slot must not + be treated as exonerating. + """ + from kiro_crew.security import _enclosing_call_slot + + tree = ast.parse('import re\nre.compile(p := "x")\n') + literal = next( + n for n in ast.walk(tree) if isinstance(n, ast.Constant) and n.value == "x" + ) + chain: list[ast.AST] = [] + + def walk(node: ast.AST, path: list[ast.AST]) -> bool: + if node is literal: + chain.extend(path) + return True + for child in ast.iter_child_nodes(node): + if walk(child, path + [node]): + return True + return False + + assert walk(tree, []) + key, in_pattern_slot = _enclosing_call_slot(chain, literal) + assert key == ("re", "compile") + assert in_pattern_slot is False + + +def test_fence_layer_checks_the_value_not_only_its_collapsed_copies(): + """The literal scan must catch a fenced path on its own, not lean on a later pass. + + `_separator_collapsed_variants` yields nothing when a value carries no separator + run, so iterating it alone left an already-single-separator decoded literal + unexamined by this layer. The shell path had pass 1a behind it; the source path had + nothing, so the verdict came from a later pass instead — defence in depth that + shared the earlier layer's blind spot. + """ + from kiro_crew.security import _fence_hit_in_collapsed, _separator_collapsed_variants + + single = r"%LOCALAPPDATA%\kiro-cli\config.json" + # Precondition: no run, so the collapsed-variant generator is empty. Without this + # the test could pass for the wrong reason. + assert tuple(_separator_collapsed_variants(single)) == () + assert _fence_hit_in_collapsed(single) is not None + + +def test_shell_path_skips_only_the_scan_pass_1_already_did(): + """The opt-out drops a duplicate pass, never a layer. + + `_fence_hit_in_collapsed` checks the value itself so the SOURCE-literal path is + self-sufficient (see the test above). On the shell path that check is a second full + run of the three pass-1 matchers over bytes pass 1 already rejected, and the + sensitive-path regex over a long newline-free line is the most expensive matcher on + this gate — so a 20 KB subject paid for it twice and doubled the wall time of a + path guarded by a linearity test. `value_already_scanned=True` removes that + duplicate. + + What must NOT change is detection, so this pins both halves: the collapsed copies + are still checked with the opt-out on (a DOUBLED separator is still blocked), and + the single-separator spelling the skipped check would have caught is still blocked + by pass 1 itself, which is why skipping it is sound rather than merely cheaper. + """ + from kiro_crew.security import ( + _fence_hit_in_collapsed, + _separator_collapsed_variants, + is_sensitive_bash_command, + ) + + fenced_single = r"%LOCALAPPDATA%\kiro-cli\config.json" + # Precondition: no separator run, so the variant generator is empty and the + # value-check is the ONLY thing this layer could contribute for this input. + assert tuple(_separator_collapsed_variants(fenced_single)) == () + + # Default is unchanged and still self-sufficient. + assert _fence_hit_in_collapsed(fenced_single) is not None + # With the opt-out, the layer contributes nothing for a no-run value -- that is + # precisely the duplicate being skipped, and it is what makes the gate linear. + assert _fence_hit_in_collapsed(fenced_single, value_already_scanned=True) is None + + # Detection is preserved on both spellings via the shell entry point. + # Single separator: pass 1 catches it, which is why the duplicate was redundant. + assert is_sensitive_bash_command(f"type {fenced_single}") is not None + # Doubled separator (#6350): only pass 1b's COLLAPSED copy catches this, so it + # proves the collapse still runs with the opt-out on. + doubled = r"type %LOCALAPPDATA%\\kiro-cli\\config.json" + assert is_sensitive_bash_command(doubled) is not None + # The extraction control travels with it, for the same reason. + assert is_sensitive_bash_command("tar -xf evil.tar -C $HOME//.kiro/crew") is not None + + +def test_vet_script_contents_refuses_a_fenced_path_through_re_escape(): + """`re.escape` must NOT exonerate a fenced literal — it consumes TEXT, not a pattern. + + `_SOURCE_PATTERN_SINKS` admits an entry only if it "must consume its argument as a + PATTERN and never as a path". `re.escape` takes plain text and returns it escaped + for onward flow, so it fails that rule and was admitted only by symmetry with its + `re.*` neighbours. This pins the removal for the derivation rather than the name, so + the entry cannot be reinstated by that same symmetry argument later. + """ + body = ( + "import re\n" + 'open(re.escape(r"%LOCALAPPDATA%\\\\kiro-cli\\\\c.json")).read()\n' + ) + err = _vet_script_contents(body) + assert err is not None and err.startswith("Error:"), ( + "re.escape must not exonerate a fenced path literal" + ) + + +def test_vet_script_contents_survives_a_deeply_nested_expression(): + """A valid but very deep body must not raise RecursionError out of the gate. + + The traversal is recursive, so a legitimate script with a long chain of operands + could crash `cron_add` into a JSON-RPC internal error. Containing it reports + ``parsed=False``, which routes the body to the raw scan WITH the collapse — still + fence-checked, just textually. + """ + from kiro_crew.security import _sensitive_run_in_source_literals + + deep = "x = " + " + ".join(["1"] * 1200) + "\n" + parsed, reason = _sensitive_run_in_source_literals(deep) + assert reason is None + assert parsed in (True, False) # either path is fine; crashing is not + assert _vet_script_contents(deep) is None # benign body still allowed + + +def test_a_dynamic_namespace_write_forfeits_exoneration(): + """Rebinding through a namespace MAPPING reaches none of the binding branches. + + `globals()["re"] = Fake` rebinds the name `re` while producing no Name in Store + context, no Attribute, and no argument mentioning the module -- the subscript's + `value` is the bare `globals()` call, which names nothing the walk recognises. So + the module read authentic and a fenced literal in a pattern slot stayed exonerated + while `Fake.search` was free to open the path. + + Matched on the NAME, not the subscript, because the mapping can be bound first: + `g = globals()` leaves the subscript's value an unresolvable local, so inspecting + the subscript can never close the class. `locals()` at module scope IS the global + namespace and `vars()` with no argument is `locals()`, so all three forfeit. + """ + from kiro_crew.security import _re_module_is_authentic + + for body in ( + 'import re\nglobals()["re"] = Fake\n', + 'import re\ng = globals()\ng["re"] = Fake\n', + 'import re\nlocals()["re"] = Fake\n', + 'import re\nvars()["re"] = Fake\n', + 'import re\nglobals().update({"re": Fake})\n', + ): + assert _re_module_is_authentic(ast.parse(body)) is False, body + + # NEGATIVE CONTROL: the ordinary redactor names no namespace builtin and must + # stay authentic, or the fix re-breaks the false positive this PR clears. + assert ( + _re_module_is_authentic( + ast.parse('import re\nre.sub(r"/\\\\S*[.]midway/cookie", "", line)\n') + ) + is True + ) + + +def test_an_unbound_compile_result_forfeits_exoneration(): + """A result that is never bound reaches no binding node, so nothing is tracked. + + `_compiled_pattern_names` only collects plain Name targets, so a compile result + that is returned, passed onward, or dropped into a container leaves it EMPTY -- + and `_compiled_name_escapes` handed an empty set cannot fail. The literal was + therefore exonerated with zero tracking behind it, and any recovery spelling then + worked, including one that defeats a literal attribute-name match + (`getattr(p, "pat" + "tern")`). Chasing the recovery call is the wrong layer: the + fix is that exoneration requires the result to be DIRECTLY bound to a Name the + tracker can follow. + """ + from kiro_crew.security import _compile_result_is_untrackable + + for body in ( + # the lane's own shape -- returned from a helper, never bound here + 'import re\ndef build():\n return re.compile("x")\n', + # passed straight into a call + 'import re\np = keep(re.compile("x"))\n', + # dropped into a container literal rather than bound to a bare Name + 'import re\npats = [re.compile("x")]\n', + 'import re\nd = (re.compile("x"), 1)\n', + # yielded + 'import re\ndef gen():\n yield re.compile("x")\n', + # evaluated and discarded + 'import re\nre.compile("x")\n', + ): + assert _compile_result_is_untrackable(ast.parse(body)) is True, body + + # NEGATIVE CONTROL: bound DIRECTLY to a plain Name is the one trackable shape and + # must stay exonerated -- it is what the redactor this PR permits actually writes. + assert ( + _compile_result_is_untrackable(ast.parse('import re\nSCRUB = re.compile("x")\n')) + is False + ) + + +def test_handing_the_module_to_any_call_forfeits_exoneration(): + """A callee's effect on the module is not readable here, so the handover forfeits. + + The call branch recognised only `setattr`/`delattr`, which made it an + allow-by-default blocklist inside a deny-by-default checker: an ordinary + `helper(re)` whose body does `m.compile = open` reached NO branch, so the module + read authentic and a fenced literal in the pattern slot stayed exonerated. + Enumerating callees cannot close that -- the mutation lives in a function this walk + never inspects -- so any ARGUMENT resolving to the module withdraws it instead. + + The func position is deliberately NOT inspected: `re.sub(...)` and `re.compile(...)` + name the module there, and those are the calls the exoneration exists to permit. + """ + from kiro_crew.security import _re_module_is_authentic + + for body in ( + "import re\ndef helper(m):\n m.compile = open\nhelper(re)\n", + "import re\ndef helper(m=None):\n m.compile = open\nhelper(m=re)\n", + "import re\nm = re\nhelper(m)\n", + 'import re\nhelper(getattr(re, "compile"))\n', + ): + assert _re_module_is_authentic(ast.parse(body)) is False, body + + # NEGATIVE CONTROL: naming the module in the FUNC position is the permitted shape + # and must stay authentic, or the fix re-breaks the motivating redactor. + for body in ( + 'import re\nre.sub(r"/\\\\S*[.]midway/cookie", "", line)\n', + 'import re\nSCRUB = re.compile("x")\nSCRUB.sub("", line)\n', + ): + assert _re_module_is_authentic(ast.parse(body)) is True, body + + +def test_dynamic_execution_in_the_body_forfeits_exoneration(): + """Every other guard here is a static read, so a body that runs code defeats them. + + `exec("re.compile = open")` carries the rebinding inside a STRING: it reaches no + Name, Attribute, Subscript or call-argument this tree can be asked about, so the + module read authentic while the call it spells now opens a file. The string is + opaque by construction, so no enumeration of spellings closes the class -- the + presence of the mechanism withdraws the exoneration instead. + """ + from kiro_crew.security import _re_module_is_authentic + + for body in ( + "import re\nexec('re.compile = open')\n", + "import re\ne = exec\ne('re.compile = open')\n", + "import re\neval(\"setattr(re, 'compile', open)\")\n", + "import re\nexec(compile('re.compile = open', '', 'exec'))\n", + "import re\n__import__('os')\n", + ): + assert _re_module_is_authentic(ast.parse(body)) is False, body + + # NEGATIVE CONTROL: `re.compile` spells its name in an Attribute's `attr` STRING, + # not as a Name node, so the builtin-`compile` forfeit must not fire on it. + assert ( + _re_module_is_authentic(ast.parse('import re\nSCRUB = re.compile("x")\n')) is True + ) + + +def test_a_wildcard_import_forfeits_exoneration(): + """`from evil import *` can rebind `re` while naming no alias `re` at all. + + The branch matched only aliases that NAME the module, which made it an + allow-by-default enumeration inside a deny-first checker: the explicit + `from evil import thing as re` forfeited, while the wildcard -- which can bind + strictly more, `re` included -- did not, and the module read authentic. An + unknowable binding set is the forfeit condition, so the class closes rather than + one more spelling. + """ + from kiro_crew.security import _re_module_is_authentic + + redactor = 'SCRUB = re.compile("x")\ndef f(s):\n return SCRUB.sub("y", s)\n' + for body in ( + "import re\nfrom evil import *\n" + redactor, + "import re\nfrom pkg.sub import *\n" + redactor, + "from evil import *\nimport re\n" + redactor, + "import re\nfrom evil import thing as re\n" + redactor, + ): + assert _re_module_is_authentic(ast.parse(body)) is False, body + + # NEGATIVE CONTROLS: neither shape rebinds the module, so both must stay allowed -- + # a bare `import re` redactor, and `from re import sub`, which binds `sub`. + assert _re_module_is_authentic(ast.parse("import re\n" + redactor)) is True + assert ( + _re_module_is_authentic(ast.parse("import re\nfrom re import sub\n" + redactor)) + is True + ) + + +def test_vet_script_contents_still_exonerates_a_pattern_slot_literal(): + """The motivating real-world case must stay allowed after the narrowing. + + A redactor names a fenced store in `re.sub`'s PATTERN operand to strip it out of + a log line. That literal is a regex, reaches no sink, and is the false positive + this PR exists to clear — narrowing the exoneration to the pattern slot must not + take it with it. + """ + body = ( + "import re\n" + 'conf = re.sub(r"/\\\\S*[.]midway/cookie", "", line)[:150]\n' + ) + assert _vet_script_contents(body) is None + + +def test_a_callable_replacement_forfeits_the_pattern_slot(): + """`re.sub`'s replacement may be a FUNCTION, and `re` hands that function the Match. + + A Match carries `.re`, so `Match.re.pattern` returns the verbatim pattern literal -- + a route back to a fenced spelling that no other member of `_SOURCE_PATTERN_SINKS` + offers, and one the compiled-name escape analysis cannot see because it tracks only + `re.compile` results while a Match is never bound by the exonerated statement. + + The recovery read can be spelled to defeat any enumeration (`getattr(m, "r" + "e")` + builds the name from a concatenation, `g = getattr` hides the callee), which is why + the decision is made on the REPLACEMENT's shape rather than on the recovery: only a + str/bytes constant provably cannot be called, and everything else fails closed. + """ + fenced = r"/home/\\S*/\\.kiro/crew/security_policy.json" + recover = 'open(getattr(getattr(m, "r" + "e"), "pat" + "tern")).read()' + for body in ( + 'import re\nconf = re.sub(r"%s", lambda m: %s, line)\n' % (fenced, recover), + 'import re\ng = getattr\nconf = re.sub(r"%s", lambda m: open(g(m)).read(), line)\n' + % fenced, + 'import re\ndef r(m):\n return %s\nconf = re.sub(r"%s", r, line)\n' + % (recover, fenced), + 'import re\nimport helper\nconf = re.sub(r"%s", helper.recover, line)\n' % fenced, + 'import re\nconf = re.sub(pattern=r"%s", repl=lambda m: open(h(m)).read(), string=line)\n' + % fenced, + 'import re\nconf, n = re.subn(r"%s", lambda m: open(k(m)).read(), line)\n' % fenced, + # A Starred puts the replacement at a position nobody can know statically. + 'import re\nconf = re.sub(r"%s", *rest)\n' % fenced, + ): + assert _vet_script_contents(body) is not None, body + + # NEGATIVE CONTROLS: a str/bytes constant can never receive the Match, so the + # motivating redactor survives; `re.findall` returns str, never a Match. + for body in ( + 'import re\nconf = re.sub(r"%s", "", line)[:150]\n' % fenced, + 'import re\nconf = re.sub(rb"%s", b"", line)\n' % fenced, + 'import re\nconf = re.sub(pattern=r"%s", repl="", string=line)\n' % fenced, + 'import re\nhits = re.findall(r"%s", line)\n' % fenced, + ): + assert _vet_script_contents(body) is None, body + + # POSITIVE CONTROL for the fixture: the same literal outside an exonerated slot is + # refused, so the assertions above are not passing on an undetected spelling. + assert _vet_script_contents('f = open(r"%s")\n' % fenced) is not None + + +def test_a_match_returning_sink_forfeits_the_pattern_slot(): + """A `Match` is the OTHER way the verbatim literal leaves an exonerated slot. + + Nothing tracks a Match -- `_compiled_pattern_names` follows only `re.compile` + results -- and the recovery read cannot be enumerated, since `getattr(m, "r" + "e")` + builds the attribute name from a concatenation. So the two families are handled + where the answer is provable, and by different means. + + Module-level `re.match`/`search`/`fullmatch`/`finditer` are simply ABSENT from + `_SOURCE_PATTERN_SINKS`, so deny-by-default refuses a fenced literal in their + pattern slot outright -- no withdrawal rule is needed or wanted. + + The COMPILED route still needs one, because the exoneration is earned by + `re.compile` and only then is the Match produced: `_SAFE_COMPILED_PATTERN_METHODS` + admitted the whole matching API on the METHOD NAME alone, so `p.search` must + discard its result and `p.sub`/`p.subn` must take a provably non-callable + replacement. + """ + fenced = r"/home/\\S*/\\.kiro/crew/security_policy.json" + recover = 'open(getattr(getattr(m, "r" + "e"), "pat" + "tern")).read()' + for body in ( + # Module-level Match-returning sinks, each binding the Match somewhere. + 'import re\nfor m in re.finditer(r"%s", data):\n %s\n' % (fenced, recover), + 'import re\nm = re.search(r"%s", data)\n%s\n' % (fenced, recover), + 'import re\nm = re.match(r"%s", data)\n%s\n' % (fenced, recover), + 'import re\nm = re.fullmatch(r"%s", data)\n%s\n' % (fenced, recover), + 'import re\nif (m := re.search(r"%s", data)):\n %s\n' % (fenced, recover), + 'import re\nxs = [%s for m in re.finditer(r"%s", data)]\n' % (recover, fenced), + # Compiled object: hands a Match to a callable, or returns one. + 'import re\np = re.compile(r"%s")\nout = p.sub(lambda m: %s, data)\n' + % (fenced, recover), + 'import re\np = re.compile(r"%s")\nout, n = p.subn(lambda m: %s, data)\n' + % (fenced, recover), + 'import re\np = re.compile(r"%s")\nm = p.search(data)\n%s\n' % (fenced, recover), + 'import re\np = re.compile(r"%s")\nfor m in p.finditer(data):\n %s\n' + % (fenced, recover), + ): + assert _vet_script_contents(body) is not None, body + + # NEGATIVE CONTROLS: `split`/`findall` return str and list, carrying no reference + # back to the pattern, and a string replacement can never receive a Match. + for body in ( + 'import re\np = re.compile(r"%s")\nout = p.sub("", data)\n' % fenced, + 'import re\np = re.compile(r"%s")\nparts = p.split(data)\n' % fenced, + 'import re\nhits = re.findall(r"%s", data)\n' % fenced, + ): + assert _vet_script_contents(body) is None, body + + # POSITIVE CONTROL for the fixture, so none of the above passes on a spelling the + # fence layer never detects. + assert _vet_script_contents('f = open(r"%s")\n' % fenced) is not None + + +def test_vet_script_contents_keeps_the_collapse_when_the_body_does_not_parse(): + """An unparseable body has no literals to inspect, so it must not be exonerated. + + The literal check needs a parse tree; without one it reports nothing. The caller + therefore falls back to the raw-text scan WITH the collapse, which is the + conservative direction — a body that cannot be understood is scanned as text + rather than waved through. + + The import is local so the attack cases above still COLLECT against a tree without + this fix — otherwise a missing symbol turns their red into a collection error, which + proves the symbol is absent rather than that the bypass is open. + """ + from kiro_crew.security import _sensitive_run_in_source_literals + + broken = 'f = open(r"%LOCALAPPDATA%\\\\kiro-cli\\\\c.json"\n' # unclosed paren + parsed, reason = _sensitive_run_in_source_literals(broken) + assert parsed is False and reason is None + err = _vet_script_contents(broken) + assert err is not None and err.startswith("Error:") + + def test_vet_script_file_reads_and_blocks(tmp_path): f = tmp_path / "evil.py" f.write_text("import os\nopen(os.path.expanduser('~/.aws/credentials')).read()\n") @@ -613,3 +1368,84 @@ def test_vet_script_file_blocks_sensitive_symlink(monkeypatch, tmp_path): assert err is not None and "blocked by security policy" in err # The secret content must NOT leak into the error message. assert "AKIAIOSFODNN7EXAMPLE" not in err + + +def test_an_executable_pattern_expression_forfeits_the_pattern_slot(): + """Occupying the pattern operand means BEING it, not merely reaching it. + + `_enclosing_call_slot` resolves `inner` to the top of the argument subtree, so an + expression feeding the operand satisfies `args[0] is inner` while the fenced literal + sits underneath it. Evaluating that expression runs code BEFORE `re` sees a pattern: + `re.compile(FENCED + Reader())` hands the expanded path to `Reader.__radd__`, and + every other operator protocol is the same shape. So the exoneration requires the + literal itself to occupy the slot, positionally or by `pattern=`. + + The allowed set is counted rather than merely iterated, so a widening that silently + re-refused the redactor this path exists to permit would fail here. + """ + fenced = r"/home/\\S*/\\.kiro/crew/security_policy.json" + radd = "class Reader:\n def __radd__(self, other):\n return open(other).read()\n" + add = "class Reader:\n def __add__(self, other):\n return open(other).read()\n" + rmod = "class Reader:\n def __rmod__(self, other):\n return open(other).read()\n" + for body in ( + 'import re\n%sp = re.compile(r"%s" + Reader())\n' % (radd, fenced), + 'import re\n%sp = re.compile(Reader() + r"%s")\n' % (add, fenced), + 'import re\n%sp = re.compile(pattern=r"%s" + Reader())\n' % (radd, fenced), + 'import re\n%sout = re.sub(r"%s" + Reader(), "", line)\n' % (radd, fenced), + 'import re\n%sp = re.compile(r"%s" %% Reader())\n' % (rmod, fenced), + ): + assert _vet_script_contents(body) is not None, body + + allowed = ( + 'import re\nconf = re.sub(r"%s", "", line)[:150]\n' % fenced, + 'import re\nconf = re.sub(rb"%s", b"", line)\n' % fenced, + 'import re\nout = re.sub(pattern=r"%s", repl="", string=line)\n' % fenced, + 'import re\np = re.compile(r"%s")\nout = p.sub("", data)\n' % fenced, + ) + assert sum(_vet_script_contents(b) is None for b in allowed) == 4 + + assert _vet_script_contents('f = open(r"%s")\n' % fenced) is not None + + +def test_a_module_alias_stored_through_a_container_forfeits_authenticity(): + """A module reference that escapes as a VALUE is no longer statically trackable. + + The alias walk records `m = re`, so mutation through a bare second name is caught. + `holder = [re]` then `m = holder[0]` puts the same object behind a subscript no + static walk can resolve, so `m.compile = reader` rebinds exactly what `re.compile` + spells while an alias set keyed on bare Name assignments records nothing. + + Enumerating container shapes would rebuild the allow-by-default blocklist this + function already had to abandon for calls, so the rule closes the class instead: a + module reference may only be READ through an attribute or bound as a tracked bare + alias, and every other mention forfeits. + + An attribute read off the module is that sanctioned mention, so the allowed set is + counted: were the widening to swallow it, the redactor this PR unblocks is refused + again and this assertion is what says so. + """ + fenced = r"/home/\\S*/\\.kiro/crew/security_policy.json" + reader = "def reader(*a, **k):\n return open(a[0]).read()\n" + for body in ( + 'import re\n%sholder = [re]\nm = holder[0]\nm.compile = reader\nP = re.compile(r"%s")\n' + % (reader, fenced), + 'import re\n%sholder = (re,)\nm = holder[0]\nm.compile = reader\nP = re.compile(r"%s")\n' + % (reader, fenced), + 'import re\n%sholder = {"m": re}\nm = holder["m"]\nm.compile = reader\nP = re.compile(r"%s")\n' + % (reader, fenced), + 'import re\n%sm = re if flag else None\nm.compile = reader\nP = re.compile(r"%s")\n' + % (reader, fenced), + # Already closed by the call branch; pinned so the two rules stay consistent. + 'import re\ndef helper(mod):\n mod.compile = open\nhelper(re)\nP = re.compile(r"%s")\n' + % fenced, + ): + assert _vet_script_contents(body) is not None, body + + allowed = ( + 'import re\nconf = re.sub(r"%s", "", line)[:150]\n' % fenced, + 'import re\np = re.compile(r"%s")\nout = p.sub("", data)\n' % fenced, + 'import re\nhits = re.findall(r"%s", data)\n' % fenced, + ) + assert sum(_vet_script_contents(b) is None for b in allowed) == 3 + + assert _vet_script_contents('f = open(r"%s")\n' % fenced) is not None