From 754d84b1213a676df2e0cc0e48040b1f5cd79bf2 Mon Sep 17 00:00:00 2001 From: Nick Date: Tue, 11 Aug 2026 11:39:53 +0300 Subject: [PATCH 1/6] test(journeys): waive each harness subprocess site, then enforce the reason MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The semgrep tainted-env-args findings in the journeys harness are all the same shape: a stand-config value from the operator's own env reaches a list argv. Each site now carries a `# nosemgrep` with the reason that holds THERE, rather than a path exclusion — a subprocess call added later fires fresh instead of being silently pre-exempted. The `limactl shell $FLEET_LIMA_INSTANCE -- argv` leg is called out explicitly: it rides ssh semantics, which join argv into a command line the VM's shell re-parses, so "list argv, never a shell" is not true end to end there. Every waiver states "list argv, no shell", which is an assumption about code nobody re-reads. test_z_meta_guard now makes it a property: an AST scan over the whole journeys tree reds on `shell=True`, `os.system`/`os.popen`, or a command built by f-string / `%` / `.format` / concatenation. Interpolation inside ONE argv element stays legal — `f"name={cname}"` reaches the program as a single argument — and the planted-violation test pins both sides of that boundary so the guard can neither go vacuous nor force the waivers off. Co-Authored-By: Claude Opus 5 (1M context) --- deploy/tests/journeys/backends/fleet.py | 8 ++ deploy/tests/journeys/conftest.py | 9 ++ deploy/tests/journeys/test_e_lifecycle.py | 26 ++++ deploy/tests/journeys/test_k_admin.py | 5 + deploy/tests/journeys/test_z_meta_guard.py | 150 +++++++++++++++++++++ 5 files changed, 198 insertions(+) diff --git a/deploy/tests/journeys/backends/fleet.py b/deploy/tests/journeys/backends/fleet.py index 0aa46493..c3c540e6 100644 --- a/deploy/tests/journeys/backends/fleet.py +++ b/deploy/tests/journeys/backends/fleet.py @@ -180,6 +180,14 @@ def _curl(self, method: str, path: str, body: Optional[dict] = None) -> tuple[in if body is not None: args += ["-H", "content-type: application/json", "-d", json.dumps(body)] try: + # `self._base` comes from the stand's own env (the operator who + # launched pytest), and reaches curl as a single argv element — no + # shell, no string-assembled command. The one sharp edge is a value + # beginning with `-`, which curl would read as an option rather than + # a URL; that is a misconfiguration by the person who set it, not a + # boundary crossing. It would NOT be acceptable for a value arriving + # from CI metadata or any source outside this trust domain. + # nosemgrep: python.lang.security.audit.dangerous-subprocess-use-tainted-env-args.dangerous-subprocess-use-tainted-env-args proc = subprocess.run(args, capture_output=True, timeout=_HTTP_TIMEOUT_S + 5) except subprocess.SubprocessError as exc: raise BackendUnavailable(f"fleet wire failure on {path}: {exc}") from exc diff --git a/deploy/tests/journeys/conftest.py b/deploy/tests/journeys/conftest.py index 181aeb91..142ee174 100644 --- a/deploy/tests/journeys/conftest.py +++ b/deploy/tests/journeys/conftest.py @@ -234,6 +234,15 @@ def _fleet_exec(argv: list[str], timeout: float): if not limactl: raise OSError("limactl not found for FLEET_LIMA_INSTANCE") argv = [limactl, "shell", os.environ["FLEET_LIMA_INSTANCE"], "--", *argv] + # FLEET_LIMA_INSTANCE is operator-supplied stand configuration, set by whoever + # already owns the Lima VM this shells into — same trust domain, no privilege + # boundary crossed. Worth stating the sharp edge rather than hiding it: the + # `limactl shell ... -- argv` leg rides ssh semantics, which JOIN argv into a + # command line the VM's shell re-parses, so metacharacters in that value would + # execute there. That is a real mechanism; it is acceptable only because the + # value's author already has shell on that VM. It would NOT be acceptable for + # a value reaching this from CI metadata or any untrusted source. + # nosemgrep: python.lang.security.audit.dangerous-subprocess-use-tainted-env-args.dangerous-subprocess-use-tainted-env-args return subprocess.run(argv, capture_output=True, text=True, timeout=timeout) diff --git a/deploy/tests/journeys/test_e_lifecycle.py b/deploy/tests/journeys/test_e_lifecycle.py index 2e1ee1d7..e80f7c90 100644 --- a/deploy/tests/journeys/test_e_lifecycle.py +++ b/deploy/tests/journeys/test_e_lifecycle.py @@ -115,7 +115,12 @@ def _operator_sock_reachable() -> bool: POST uses. Never raises; a missing sudo/curl surfaces as False -> loud skip. """ try: + # `_OPERATOR_SOCK` is a module constant and `_sudo_prefix()` returns + # either [] or ["sudo"]; the only env-derived part is whether sudo is + # used at all, decided by the operator running the suite on their own + # host. List argv, no shell. probe = subprocess.run( + # nosemgrep: python.lang.security.audit.dangerous-subprocess-use-tainted-env-args.dangerous-subprocess-use-tainted-env-args _sudo_prefix() + ["test", "-S", _OPERATOR_SOCK], capture_output=True, timeout=10, @@ -175,7 +180,13 @@ def _psql(sql: str) -> Optional[str]: if not docker: return None try: + # `sql` is a literal written in this test file, never external input. + # `_CONTROL_DB`, `_DB_USER` and `_DB_NAME` are stand config from the + # operator's env, each a single argv element with no shell in between — + # `docker exec` execs directly rather than through /bin/sh. Would NOT + # hold if `sql` were ever built from a fixture the caller controls. proc = subprocess.run( + # nosemgrep: python.lang.security.audit.dangerous-subprocess-use-tainted-env-args.dangerous-subprocess-use-tainted-env-args [ docker, "exec", _CONTROL_DB, "psql", "-U", _DB_USER, "-d", _DB_NAME, "-t", "-A", "-c", sql, @@ -238,6 +249,10 @@ def _operator_post(path: str) -> int: ) try: proc = subprocess.run( + # `curl`, `_OPERATOR_SOCK` and the URL path are harness constants; + # `_sudo_prefix()` returns [] or ["sudo"]. curl receives the socket path as + # one argv element with no shell in between. + # nosemgrep: python.lang.security.audit.dangerous-subprocess-use-tainted-env-args.dangerous-subprocess-use-tainted-env-args _sudo_prefix() + [ curl, "-sS", "--max-time", "15", "--unix-socket", _OPERATOR_SOCK, @@ -262,6 +277,10 @@ def _control_container_id(docker: str) -> Optional[str]: """Resolve the running control container by its compose service label.""" try: proc = subprocess.run( + # `docker` is the resolved binary and `_CONTROL_SERVICE` is stand config from + # the operator's own env, passed as one argv element to a `--filter` flag. + # No shell parses it. + # nosemgrep: python.lang.security.audit.dangerous-subprocess-use-tainted-env-args.dangerous-subprocess-use-tainted-env-args [ docker, "ps", "--filter", f"label=com.docker.compose.service={_CONTROL_SERVICE}", @@ -699,6 +718,10 @@ def test_e6_rowless_container_killed_valid_row_survives(backend: Backend, expect stray_key = f"stray-{int(time.time() * 1000)}" stray_name = f"ocu-sess-{stray_key}" run = subprocess.run( + # `docker` and `stray_name` are harness-controlled: the name is built from a + # literal prefix in this file. Stand config (`_SESSION_IMAGE`) is a single + # argv element that docker execs directly, never through /bin/sh. + # nosemgrep: python.lang.security.audit.dangerous-subprocess-use-tainted-env-args.dangerous-subprocess-use-tainted-env-args [ docker, "run", "-d", "--name", stray_name, "--label", "ocu-session=true", @@ -744,6 +767,9 @@ def test_e6_rowless_container_killed_valid_row_survives(backend: Backend, expect ) finally: # Clean up the stray if it somehow survived (so a re-run is idempotent). + # `docker` is the resolved binary path and `stray_cid` is a container id this + # test itself created moments earlier. List argv, no shell. + # nosemgrep: python.lang.security.audit.dangerous-subprocess-use-tainted-env-args.dangerous-subprocess-use-tainted-env-args subprocess.run([docker, "rm", "-f", stray_cid], capture_output=True, timeout=30) diff --git a/deploy/tests/journeys/test_k_admin.py b/deploy/tests/journeys/test_k_admin.py index 979c4fe5..23445e03 100644 --- a/deploy/tests/journeys/test_k_admin.py +++ b/deploy/tests/journeys/test_k_admin.py @@ -87,6 +87,11 @@ def _curl( if body is not None: args += ["-H", "content-type: application/json", "-d", body] try: + # `args` is assembled above from literals, `ADMIN_URL` (stand config from + # the operator's env) and this function's own parameters, which every + # caller in this file passes as literals. curl gets each as one argv + # element; no shell parses any of them. + # nosemgrep: python.lang.security.audit.dangerous-subprocess-use-tainted-env-args.dangerous-subprocess-use-tainted-env-args proc = subprocess.run(args, capture_output=True, text=True, timeout=timeout + 5) except subprocess.SubprocessError as exc: raise RuntimeError(f"curl transport failure on {path}: {exc}") from exc diff --git a/deploy/tests/journeys/test_z_meta_guard.py b/deploy/tests/journeys/test_z_meta_guard.py index 3e3bf647..9f9e1b3a 100644 --- a/deploy/tests/journeys/test_z_meta_guard.py +++ b/deploy/tests/journeys/test_z_meta_guard.py @@ -28,6 +28,7 @@ from __future__ import annotations +import ast import io import re from pathlib import Path @@ -148,6 +149,155 @@ def test_meta_guard_reds_on_a_planted_violation() -> None: ) +def _harness_files() -> list[Path]: + """Every .py under the journeys tree except this meta-guard itself. + + Wider than :func:`_test_files` on purpose: the subprocess waivers live in + ``conftest.py`` and ``backends/`` too, and those are exactly the files a new + helper gets added to. + """ + return sorted(p for p in _HERE.rglob("*.py") if p.name != _SELF) + + +def _shell_hazard_sites(path: Path) -> list[tuple[int, str]]: + """Return (line_no, what) for each host-side shell hazard in ``path``. + + Three shapes, all decided on the AST rather than on how the source is + written, so a hazard hidden in a differently-formatted call still counts: + + * ``shell=True`` on any call - hands the argv to /bin/sh, which is what the + per-site ``# nosemgrep`` waivers all assert does NOT happen. + * ``os.system`` / ``os.popen`` - a shell by construction. + * an f-string (or a ``%``/``.format`` built string) passed as the COMMAND + argument of ``subprocess.run``/``call``/``check_output``/``check_call``/ + ``Popen`` - i.e. a command line rather than a list argv. Interpolation + INSIDE one element of a list argv is deliberately not flagged: that is the + safe form (``f"name={cname}"`` reaches the program as a single argument, + with no shell to re-parse it), and it is what the waivers describe. + """ + tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path)) + runners = {"run", "call", "check_output", "check_call", "Popen"} + hits: list[tuple[int, str]] = [] + + def _built_string(node: ast.AST) -> str | None: + if isinstance(node, ast.JoinedStr): + return "an f-string" + if isinstance(node, ast.BinOp) and isinstance(node.op, ast.Mod): + return "a %-formatted string" + if isinstance(node, ast.BinOp) and isinstance(node.op, ast.Add): + # ``[...] + args`` builds a list argv, not a command line. Only flag + # a ``+`` whose operands are strings. + if any( + isinstance(side, (ast.List, ast.Tuple)) + for side in (node.left, node.right) + ): + return None + if any( + isinstance(side, ast.Constant) and isinstance(side.value, str) + for side in (node.left, node.right) + ): + return "a concatenated string" + return None + if ( + isinstance(node, ast.Call) + and isinstance(node.func, ast.Attribute) + and node.func.attr == "format" + ): + return "a .format() string" + return None + + for node in ast.walk(tree): + if not isinstance(node, ast.Call): + continue + + for kw in node.keywords: + if ( + kw.arg == "shell" + and isinstance(kw.value, ast.Constant) + and kw.value.value is True + ): + hits.append((node.lineno, "shell=True")) + + func = node.func + if isinstance(func, ast.Attribute): + if func.attr in ("system", "popen") and isinstance(func.value, ast.Name) and func.value.id == "os": + hits.append((node.lineno, f"os.{func.attr}")) + if func.attr in runners and node.args: + built = _built_string(node.args[0]) + if built is not None: + hits.append((node.lineno, f"{built} as the command")) + return hits + + +@pytest.mark.parametrize("path", _harness_files(), ids=lambda p: p.name) +def test_no_host_side_shell_in_the_harness(path: Path) -> None: + """No harness file hands a host command to a shell, or builds one by + interpolation. + + Every ``subprocess`` call in this tree carries a per-site ``# nosemgrep`` + waiver whose stated reason is the same in each case: list argv, no shell, so + a metacharacter in stand config is one argv element rather than syntax. That + reason is an ASSUMPTION about code nobody re-reads. This test makes it a + property: the first ``shell=True``, ``os.system``, or f-string-built command + added here reds, and the waiver above it stops being true out loud rather + than silently. + """ + hits = _shell_hazard_sites(path) + assert not hits, ( + f"{path.name} runs a host command through a shell or builds one by " + f"interpolation, which breaks the 'list argv, never a shell' property " + f"every # nosemgrep waiver in this tree rests on. Pass a list argv with " + f"each value as its own element. Offending sites (line: what): " + + ", ".join(f"{ln}:{w}" for ln, w in hits) + ) + + +def test_shell_hazard_guard_reds_on_planted_violations() -> None: + """The shell-hazard guard is non-vacuous, and does not red on the clean form. + + Plants one of each detected shape and asserts all four are found, then + asserts the shapes the harness legitimately uses - a list argv holding an + env-derived value, and a literal-joined list - are NOT flagged. Without the + negative half the guard could flag everything and still look green here. + """ + import tempfile + + planted = ( + "import os, subprocess\n" + "def t(name, sql):\n" + ' subprocess.run("ls -l", shell=True)\n' + ' os.system("rm -rf /tmp/x")\n' + ' subprocess.run(f"docker rm {name}")\n' + ' subprocess.run("docker rm " + name)\n' + # Clean forms below. The f-string INSIDE one argv element and the + # list+list concat are the shapes the harness actually uses; flagging + # them would force the waivers off rather than keep them honest. + ' subprocess.run(["docker", "rm", "-f", name])\n' + ' subprocess.check_output(["docker", "ps", "--filter", f"name={name}"])\n' + ' subprocess.run(_sudo_prefix() + ["test", "-S", SOCK])\n' + ' subprocess.run([docker, "exec", CONTAINER, "psql", "-c", sql])\n' + ) + with tempfile.TemporaryDirectory() as td: + planted_path = Path(td) / "planted_hazards.py" + planted_path.write_text(planted, encoding="utf-8") + hits = _shell_hazard_sites(planted_path) + + whats = sorted(w for _, w in hits) + assert whats == sorted( + [ + "shell=True", + "os.system", + "an f-string as the command", + "a concatenated string as the command", + ] + ), ( + "the shell-hazard guard must detect each planted shape exactly once and " + f"leave the clean list-argv forms alone; got {hits!r}. A guard that does " + "not red on a planted violation is vacuous, and one that reds on the " + "clean form would force the waivers off rather than keep them honest." + ) + + def test_no_undefined_names_in_the_suite() -> None: """No journey module may reference a name that is never bound. From c8a22013a2bad241b402151cb09969771b932bc6 Mon Sep 17 00:00:00 2001 From: Nick Date: Tue, 11 Aug 2026 11:49:16 +0300 Subject: [PATCH 2/6] test(journeys): bind the shell hazard to the property, not to its spelling MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The first cut of the guard matched a syntactic silhouette, so nine genuine host-side hazards walked past it. An adversarial pass planted each one and the detector returned nothing: subprocess.getoutput("docker rm " + x) # /bin/sh by construction subprocess.getstatusoutput(f"...") # likewise; semgrep misses it too from subprocess import run; run(f"...") # callee not spelled subprocess.run cmd = f"..."; subprocess.run(cmd) # command hoisted into a local from os import system; system("..." + x) subprocess.run(a, shell=sh) # non-literal shell= subprocess.run(a, shell=1) subprocess.run(a, shell=bool(os.getenv(..))) sh = subprocess.run; sh("docker rm " + x) # alias getoutput/getstatusoutput matter most: they take a command string and hand it to /bin/sh, no list-argv form exists, and semgrep's python bundle does not flag them either — this guard is their only backstop. The detector now resolves the callee (from-imports and simple aliases) instead of matching its spelling, tracks locals holding a built command string, treats any non-False `shell=` as a hit, and carries a shell-by-construction callee set. Interpolation inside ONE argv element stays legal, and the clean half of the planted test pins that so the guard cannot force the waivers off. The planted test previously exercised only the shapes the detector already caught, which made it green by construction; it now plants every evasion above. Two waiver reasons overstated their case and are corrected: test_k_admin's `body` carries env-derived credentials via `_login` rather than literals, and `_psql` takes one f-string (interpolating `int(value)` and a module constant). Both remain safe because each value is one argv element, not because the values are constants — which is the reason the comment should have given. Co-Authored-By: Claude Opus 5 (1M context) --- deploy/tests/journeys/test_e_lifecycle.py | 11 +- deploy/tests/journeys/test_k_admin.py | 10 +- deploy/tests/journeys/test_z_meta_guard.py | 228 +++++++++++++++------ 3 files changed, 181 insertions(+), 68 deletions(-) diff --git a/deploy/tests/journeys/test_e_lifecycle.py b/deploy/tests/journeys/test_e_lifecycle.py index e80f7c90..855e34e1 100644 --- a/deploy/tests/journeys/test_e_lifecycle.py +++ b/deploy/tests/journeys/test_e_lifecycle.py @@ -180,11 +180,12 @@ def _psql(sql: str) -> Optional[str]: if not docker: return None try: - # `sql` is a literal written in this test file, never external input. - # `_CONTROL_DB`, `_DB_USER` and `_DB_NAME` are stand config from the - # operator's env, each a single argv element with no shell in between — - # `docker exec` execs directly rather than through /bin/sh. Would NOT - # hold if `sql` were ever built from a fixture the caller controls. + # `sql` comes from this file only: literals, plus one f-string whose + # interpolations are `int(value)` and a module constant. `_CONTROL_DB`, + # `_DB_USER` and `_DB_NAME` are stand config from the operator's env. + # Each reaches psql as one argv element — `docker exec` execs directly + # rather than through /bin/sh — so nothing is re-parsed. Would NOT hold + # if `sql` were ever built from a fixture the caller controls. proc = subprocess.run( # nosemgrep: python.lang.security.audit.dangerous-subprocess-use-tainted-env-args.dangerous-subprocess-use-tainted-env-args [ diff --git a/deploy/tests/journeys/test_k_admin.py b/deploy/tests/journeys/test_k_admin.py index 23445e03..af51ff4b 100644 --- a/deploy/tests/journeys/test_k_admin.py +++ b/deploy/tests/journeys/test_k_admin.py @@ -87,10 +87,12 @@ def _curl( if body is not None: args += ["-H", "content-type: application/json", "-d", body] try: - # `args` is assembled above from literals, `ADMIN_URL` (stand config from - # the operator's env) and this function's own parameters, which every - # caller in this file passes as literals. curl gets each as one argv - # element; no shell parses any of them. + # `args` is assembled above from literals, `ADMIN_URL` and the operator + # credentials (all stand config from the operator's own env), plus this + # function's parameters — `body` carries the env-derived credentials via + # `_login`, so this is env-sourced data, not literals. It is safe because + # curl gets each value as ONE argv element with no shell in between, not + # because the values are constants. # nosemgrep: python.lang.security.audit.dangerous-subprocess-use-tainted-env-args.dangerous-subprocess-use-tainted-env-args proc = subprocess.run(args, capture_output=True, text=True, timeout=timeout + 5) except subprocess.SubprocessError as exc: diff --git a/deploy/tests/journeys/test_z_meta_guard.py b/deploy/tests/journeys/test_z_meta_guard.py index 9f9e1b3a..c0e426ed 100644 --- a/deploy/tests/journeys/test_z_meta_guard.py +++ b/deploy/tests/journeys/test_z_meta_guard.py @@ -159,25 +159,57 @@ def _harness_files() -> list[Path]: return sorted(p for p in _HERE.rglob("*.py") if p.name != _SELF) +# Shell-by-construction: these take a COMMAND STRING and hand it to /bin/sh, so +# no safe list-argv form of them exists and any call is a hit. +_ALWAYS_SHELL = {"getoutput", "getstatusoutput", "system", "popen"} + +# These take either a list argv (safe) or a command string (a shell). +_RUNNERS = {"run", "call", "check_output", "check_call", "Popen"} + + def _shell_hazard_sites(path: Path) -> list[tuple[int, str]]: """Return (line_no, what) for each host-side shell hazard in ``path``. - Three shapes, all decided on the AST rather than on how the source is - written, so a hazard hidden in a differently-formatted call still counts: - - * ``shell=True`` on any call - hands the argv to /bin/sh, which is what the - per-site ``# nosemgrep`` waivers all assert does NOT happen. - * ``os.system`` / ``os.popen`` - a shell by construction. - * an f-string (or a ``%``/``.format`` built string) passed as the COMMAND - argument of ``subprocess.run``/``call``/``check_output``/``check_call``/ - ``Popen`` - i.e. a command line rather than a list argv. Interpolation - INSIDE one element of a list argv is deliberately not flagged: that is the - safe form (``f"name={cname}"`` reaches the program as a single argument, - with no shell to re-parse it), and it is what the waivers describe. + Decided on the AST, and on the RESOLVED callee rather than its spelling: + ``from os import system`` and ``sh = subprocess.run`` are the same hazard as + the dotted form, and a guard matching only the dotted form measures house + style rather than safety. + + Three shapes: + + * ``shell=`` with anything but a literal ``False``. Not merely literal + ``True`` — ``shell=sh``, ``shell=1`` and ``shell=bool(os.getenv(...))`` + all reach /bin/sh, and a guard keyed on ``is True`` reads them as clean. + * a shell-by-construction callee (``os.system``/``os.popen``, + ``subprocess.getoutput``/``getstatusoutput``). + * a built command STRING as a runner's first argument — an f-string, ``%``, + ``.format``, a string ``+``, or a local holding one of those, so hoisting + the string out of the call does not launder it. Interpolation INSIDE one + element of a list argv is deliberately not flagged: that is the safe form + (``f"name={cname}"`` reaches the program as a single argument, with no + shell to re-parse it), and it is what the per-site waivers describe. """ tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path)) - runners = {"run", "call", "check_output", "check_call", "Popen"} - hits: list[tuple[int, str]] = [] + + always_shell_names: set[str] = set() + runner_names: set[str] = set() + for node in ast.walk(tree): + if isinstance(node, ast.ImportFrom) and node.module in ("os", "subprocess"): + for alias in node.names: + bound = alias.asname or alias.name + if alias.name in _ALWAYS_SHELL: + always_shell_names.add(bound) + elif alias.name in _RUNNERS: + runner_names.add(bound) + if isinstance(node, ast.Assign) and isinstance(node.value, ast.Attribute): + attr = node.value.attr + for tgt in node.targets: + if not isinstance(tgt, ast.Name): + continue + if attr in _ALWAYS_SHELL: + always_shell_names.add(tgt.id) + elif attr in _RUNNERS: + runner_names.add(tgt.id) def _built_string(node: ast.AST) -> str | None: if isinstance(node, ast.JoinedStr): @@ -185,8 +217,7 @@ def _built_string(node: ast.AST) -> str | None: if isinstance(node, ast.BinOp) and isinstance(node.op, ast.Mod): return "a %-formatted string" if isinstance(node, ast.BinOp) and isinstance(node.op, ast.Add): - # ``[...] + args`` builds a list argv, not a command line. Only flag - # a ``+`` whose operands are strings. + # ``[...] + args`` builds a list argv, not a command line. if any( isinstance(side, (ast.List, ast.Tuple)) for side in (node.left, node.right) @@ -206,26 +237,60 @@ def _built_string(node: ast.AST) -> str | None: return "a .format() string" return None + string_locals: dict[str, str] = {} + for node in ast.walk(tree): + if isinstance(node, ast.Assign): + built = _built_string(node.value) + if built is None: + continue + for tgt in node.targets: + if isinstance(tgt, ast.Name): + string_locals[tgt.id] = built + + def _command_string(node: ast.AST) -> str | None: + built = _built_string(node) + if built is not None: + return built + if isinstance(node, ast.Name) and node.id in string_locals: + return f"{string_locals[node.id]} (via {node.id})" + return None + + def _callee(node: ast.Call) -> tuple[str, bool, bool]: + """(display name, is-always-shell, is-runner) for a call's callee.""" + func = node.func + if isinstance(func, ast.Attribute): + base = func.value.id if isinstance(func.value, ast.Name) else "?" + if base in ("os", "subprocess") and func.attr in _ALWAYS_SHELL: + return f"{base}.{func.attr}", True, False + if base == "subprocess" and func.attr in _RUNNERS: + return f"subprocess.{func.attr}", False, True + return f"{base}.{func.attr}", False, False + if isinstance(func, ast.Name): + return func.id, func.id in always_shell_names, func.id in runner_names + return "?", False, False + + hits: list[tuple[int, str]] = [] for node in ast.walk(tree): if not isinstance(node, ast.Call): continue + name, always_shell, is_runner = _callee(node) for kw in node.keywords: - if ( - kw.arg == "shell" - and isinstance(kw.value, ast.Constant) - and kw.value.value is True - ): - hits.append((node.lineno, "shell=True")) + if kw.arg != "shell": + continue + if isinstance(kw.value, ast.Constant) and kw.value.value is False: + continue + literal_true = isinstance(kw.value, ast.Constant) and kw.value.value is True + hits.append((node.lineno, "shell=True" if literal_true else "a non-literal shell=")) + + if always_shell: + hits.append((node.lineno, f"{name} (runs /bin/sh by construction)")) + continue - func = node.func - if isinstance(func, ast.Attribute): - if func.attr in ("system", "popen") and isinstance(func.value, ast.Name) and func.value.id == "os": - hits.append((node.lineno, f"os.{func.attr}")) - if func.attr in runners and node.args: - built = _built_string(node.args[0]) - if built is not None: - hits.append((node.lineno, f"{built} as the command")) + if is_runner and node.args: + built = _command_string(node.args[0]) + if built is not None: + hits.append((node.lineno, f"{built} as the command")) return hits @@ -255,46 +320,91 @@ def test_no_host_side_shell_in_the_harness(path: Path) -> None: def test_shell_hazard_guard_reds_on_planted_violations() -> None: """The shell-hazard guard is non-vacuous, and does not red on the clean form. - Plants one of each detected shape and asserts all four are found, then - asserts the shapes the harness legitimately uses - a list argv holding an - env-derived value, and a literal-joined list - are NOT flagged. Without the - negative half the guard could flag everything and still look green here. + The planted set deliberately includes the EVASIONS, not only the shapes the + guard already caught: a command hoisted into a local, ``from``-imported and + aliased callees, a non-literal ``shell=``, and the shell-by-construction + ``subprocess.getoutput``/``getstatusoutput`` (which semgrep's python bundle + does not flag either, so this guard is their only backstop). A negative test + that plants only what the detector already finds is green by construction. + + The clean half matters as much: an f-string inside ONE argv element and a + list+list concat are what the harness actually uses, and flagging them would + force the per-site waivers off rather than keep them honest. """ import tempfile - planted = ( - "import os, subprocess\n" - "def t(name, sql):\n" - ' subprocess.run("ls -l", shell=True)\n' - ' os.system("rm -rf /tmp/x")\n' - ' subprocess.run(f"docker rm {name}")\n' - ' subprocess.run("docker rm " + name)\n' - # Clean forms below. The f-string INSIDE one argv element and the - # list+list concat are the shapes the harness actually uses; flagging - # them would force the waivers off rather than keep them honest. - ' subprocess.run(["docker", "rm", "-f", name])\n' - ' subprocess.check_output(["docker", "ps", "--filter", f"name={name}"])\n' - ' subprocess.run(_sudo_prefix() + ["test", "-S", SOCK])\n' - ' subprocess.run([docker, "exec", CONTAINER, "psql", "-c", sql])\n' + hazards = [ + ('subprocess.run("ls -l", shell=True)', "shell=True"), + ("subprocess.run(a, shell=sh)", "a non-literal shell="), + ("subprocess.run(a, shell=1)", "a non-literal shell="), + ('os.system("rm -rf " + x)', "os.system (runs /bin/sh by construction)"), + ('os.popen("ls " + x)', "os.popen (runs /bin/sh by construction)"), + ('subprocess.getoutput("docker rm " + x)', "subprocess.getoutput (runs /bin/sh by construction)"), + ('subprocess.getstatusoutput(f"docker rm {x}")', "subprocess.getstatusoutput (runs /bin/sh by construction)"), + ('subprocess.run(f"docker rm {x}")', "an f-string as the command"), + ('subprocess.check_output("cat %s" % x)', "a %-formatted string as the command"), + ('subprocess.run("docker rm " + x)', "a concatenated string as the command"), + ] + for src, want in hazards: + planted = f"import os, subprocess\ndef t(a, x, sh):\n {src}\n" + with tempfile.TemporaryDirectory() as td: + f = Path(td) / "planted.py" + f.write_text(planted, encoding="utf-8") + hits = _shell_hazard_sites(f) + assert [w for _, w in hits] == [want], ( + f"planted hazard {src!r} must be detected as {want!r}; got {hits!r}. " + "A guard that misses a planted hazard leaves every # nosemgrep " + "waiver in this tree asserting a property nothing enforces." + ) + + # Callees reached under another name are the same hazard as the dotted form. + aliased = ( + "from os import system\n" + "from subprocess import run\n" + "import subprocess\n" + "sh = subprocess.run\n" + "def t(x):\n" + ' system("rm -rf " + x)\n' + ' run(f"docker rm {x}")\n' + ' sh("docker rm " + x)\n' + ' cmd = f"docker rm {x}"\n' + " subprocess.run(cmd)\n" ) with tempfile.TemporaryDirectory() as td: - planted_path = Path(td) / "planted_hazards.py" - planted_path.write_text(planted, encoding="utf-8") - hits = _shell_hazard_sites(planted_path) - - whats = sorted(w for _, w in hits) + f = Path(td) / "aliased.py" + f.write_text(aliased, encoding="utf-8") + whats = sorted(w for _, w in _shell_hazard_sites(f)) assert whats == sorted( [ - "shell=True", - "os.system", + "system (runs /bin/sh by construction)", "an f-string as the command", "a concatenated string as the command", + "an f-string (via cmd) as the command", ] ), ( - "the shell-hazard guard must detect each planted shape exactly once and " - f"leave the clean list-argv forms alone; got {hits!r}. A guard that does " - "not red on a planted violation is vacuous, and one that reds on the " - "clean form would force the waivers off rather than keep them honest." + f"a hazardous callee reached via from-import or an alias, and a command " + f"hoisted into a local, must all be detected; got {whats!r}. Resolving " + "only the dotted spelling measures house style, not safety." + ) + + clean = ( + "import os, subprocess\n" + "def t(name, sql, docker, args):\n" + ' subprocess.run(["docker", "rm", "-f", name])\n' + ' subprocess.check_output(["docker", "ps", "--filter", f"name={name}"])\n' + ' subprocess.run(_sudo_prefix() + ["test", "-S", SOCK])\n' + ' subprocess.run(["curl", "-sS"] + args)\n' + ' subprocess.run([docker, "exec", CONTAINER, "psql", "-c", sql])\n' + ' subprocess.run(["ls"], shell=False)\n' + ) + with tempfile.TemporaryDirectory() as td: + f = Path(td) / "clean.py" + f.write_text(clean, encoding="utf-8") + clean_hits = _shell_hazard_sites(f) + assert clean_hits == [], ( + f"the clean list-argv forms must NOT be flagged; got {clean_hits!r}. A " + "guard that reds on interpolation inside one argv element would force " + "the waivers off rather than keep them honest." ) From 67f32b97e71a79f830e554d38bf6c22bee0d0e44 Mon Sep 17 00:00:00 2001 From: Nick Date: Tue, 11 Aug 2026 11:56:33 +0300 Subject: [PATCH 3/6] test(journeys): scope the string locals, or the guard reds on the safe form MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A second adversarial pass found a false positive, which is worse than any of the misses it also found: `string_locals` was a flat, file-global map, so a local holding a LIST argv was flagged whenever another function in the same file bound the same name to a built string. def list_probe(name): cmd = ["docker", "ps", "--filter", f"name={name}"] return subprocess.run(cmd) # flagged, and clean def other(x): cmd = f"echo {x}" return subprocess.getoutput(cmd) # the actual hazard `cmd`, `args` and `argv` are what this harness names its list argv, so the next helper added to an already-waived file would have reddened the safe form — and a guard that reds on the safe form forces the waivers off instead of keeping them honest, the exact failure the negative test claims to prevent. The tree is clean today, so this was latent rather than visible. String locals are now indexed per enclosing function (module level included), and the lookup consults the scope of the call being inspected. Two genuine hazards the pass also found are closed: `**{"shell": True}`, whose keyword carries `arg is None` and so was never examined by the shell= loop, and `import os as o; o.system(...)`, which `ast.Import` never bound (only `ImportFrom` and attribute assignments were tracked). The planted test pins all three, including the scope case as a line-number assertion so a regression names the clean call it wrongly reds. 11/11 evasions caught, harness green, semgrep tree still 0. Co-Authored-By: Claude Opus 5 (1M context) --- deploy/tests/journeys/test_z_meta_guard.py | 90 ++++++++++++++++++---- 1 file changed, 77 insertions(+), 13 deletions(-) diff --git a/deploy/tests/journeys/test_z_meta_guard.py b/deploy/tests/journeys/test_z_meta_guard.py index c0e426ed..e2f579bb 100644 --- a/deploy/tests/journeys/test_z_meta_guard.py +++ b/deploy/tests/journeys/test_z_meta_guard.py @@ -193,6 +193,9 @@ def _shell_hazard_sites(path: Path) -> list[tuple[int, str]]: always_shell_names: set[str] = set() runner_names: set[str] = set() + # ``import os as o`` binds the module under another name; without this the + # guard matches the spelling ``os.system`` rather than the callee. + module_aliases: dict[str, str] = {"os": "os", "subprocess": "subprocess"} for node in ast.walk(tree): if isinstance(node, ast.ImportFrom) and node.module in ("os", "subprocess"): for alias in node.names: @@ -201,6 +204,10 @@ def _shell_hazard_sites(path: Path) -> list[tuple[int, str]]: always_shell_names.add(bound) elif alias.name in _RUNNERS: runner_names.add(bound) + if isinstance(node, ast.Import): + for alias in node.names: + if alias.name in ("os", "subprocess") and alias.asname: + module_aliases[alias.asname] = alias.name if isinstance(node, ast.Assign) and isinstance(node.value, ast.Attribute): attr = node.value.attr for tgt in node.targets: @@ -237,29 +244,49 @@ def _built_string(node: ast.AST) -> str | None: return "a .format() string" return None - string_locals: dict[str, str] = {} - for node in ast.walk(tree): - if isinstance(node, ast.Assign): - built = _built_string(node.value) - if built is None: + # Scoped per enclosing function, never file-global: a local holding a list + # argv in one function must not be tainted by a same-named local holding a + # built string in another. `cmd`, `args` and `argv` are what this harness + # calls its list argv, so a flat dict would red the safe form. + string_locals_by_scope: dict[int, dict[str, str]] = {} + scope_of: dict[int, int] = {} + + def _index_scope(scope: ast.AST) -> None: + found: dict[str, str] = {} + for child in ast.walk(scope): + if isinstance(child, (ast.FunctionDef, ast.AsyncFunctionDef)) and child is not scope: continue - for tgt in node.targets: - if isinstance(tgt, ast.Name): - string_locals[tgt.id] = built + if isinstance(child, ast.Assign): + built = _built_string(child.value) + if built is not None: + for tgt in child.targets: + if isinstance(tgt, ast.Name): + found[tgt.id] = built + if isinstance(child, ast.Call): + scope_of[id(child)] = id(scope) + string_locals_by_scope[id(scope)] = found + + _index_scope(tree) + for node in ast.walk(tree): + if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)): + _index_scope(node) - def _command_string(node: ast.AST) -> str | None: + def _command_string(node: ast.AST, call: ast.Call) -> str | None: built = _built_string(node) if built is not None: return built - if isinstance(node, ast.Name) and node.id in string_locals: - return f"{string_locals[node.id]} (via {node.id})" + if isinstance(node, ast.Name): + scope = string_locals_by_scope.get(scope_of.get(id(call), id(tree)), {}) + if node.id in scope: + return f"{scope[node.id]} (via {node.id})" return None def _callee(node: ast.Call) -> tuple[str, bool, bool]: """(display name, is-always-shell, is-runner) for a call's callee.""" func = node.func if isinstance(func, ast.Attribute): - base = func.value.id if isinstance(func.value, ast.Name) else "?" + raw = func.value.id if isinstance(func.value, ast.Name) else "?" + base = module_aliases.get(raw, raw) if base in ("os", "subprocess") and func.attr in _ALWAYS_SHELL: return f"{base}.{func.attr}", True, False if base == "subprocess" and func.attr in _RUNNERS: @@ -276,6 +303,15 @@ def _callee(node: ast.Call) -> tuple[str, bool, bool]: name, always_shell, is_runner = _callee(node) for kw in node.keywords: + if kw.arg is None and isinstance(kw.value, ast.Dict): + for k, v in zip(kw.value.keys, kw.value.values): + if ( + isinstance(k, ast.Constant) + and k.value == "shell" + and not (isinstance(v, ast.Constant) and v.value is False) + ): + hits.append((node.lineno, "shell= via a ** splat")) + continue if kw.arg != "shell": continue if isinstance(kw.value, ast.Constant) and kw.value.value is False: @@ -288,7 +324,7 @@ def _callee(node: ast.Call) -> tuple[str, bool, bool]: continue if is_runner and node.args: - built = _command_string(node.args[0]) + built = _command_string(node.args[0], node) if built is not None: hits.append((node.lineno, f"{built} as the command")) return hits @@ -344,6 +380,7 @@ def test_shell_hazard_guard_reds_on_planted_violations() -> None: ('subprocess.run(f"docker rm {x}")', "an f-string as the command"), ('subprocess.check_output("cat %s" % x)', "a %-formatted string as the command"), ('subprocess.run("docker rm " + x)', "a concatenated string as the command"), + ('subprocess.run(a, **{"shell": True})', "shell= via a ** splat"), ] for src, want in hazards: planted = f"import os, subprocess\ndef t(a, x, sh):\n {src}\n" @@ -359,11 +396,13 @@ def test_shell_hazard_guard_reds_on_planted_violations() -> None: # Callees reached under another name are the same hazard as the dotted form. aliased = ( + "import os as o\n" "from os import system\n" "from subprocess import run\n" "import subprocess\n" "sh = subprocess.run\n" "def t(x):\n" + ' o.system("rm " + x)\n' ' system("rm -rf " + x)\n' ' run(f"docker rm {x}")\n' ' sh("docker rm " + x)\n' @@ -376,6 +415,7 @@ def test_shell_hazard_guard_reds_on_planted_violations() -> None: whats = sorted(w for _, w in _shell_hazard_sites(f)) assert whats == sorted( [ + "os.system (runs /bin/sh by construction)", "system (runs /bin/sh by construction)", "an f-string as the command", "a concatenated string as the command", @@ -387,6 +427,30 @@ def test_shell_hazard_guard_reds_on_planted_violations() -> None: "only the dotted spelling measures house style, not safety." ) + # Scope: a local holding a LIST argv must not be tainted by a same-named + # local holding a built string elsewhere in the file. `cmd`, `args` and + # `argv` are what this harness names its argv, so a file-global map would + # red the safe form — the failure this whole test exists to prevent. + scoped = ( + "import subprocess\n" + "def list_probe(name):\n" + ' cmd = ["docker", "ps", "--filter", f"name={name}"]\n' + " return subprocess.run(cmd)\n" + "def other(x):\n" + ' cmd = f"echo {x}"\n' + " return subprocess.getoutput(cmd)\n" + ) + with tempfile.TemporaryDirectory() as td: + f = Path(td) / "scoped.py" + f.write_text(scoped, encoding="utf-8") + scoped_hits = _shell_hazard_sites(f) + assert [ln for ln, _ in scoped_hits] == [7], ( + f"only the getoutput call on line 7 is a hazard; got {scoped_hits!r}. A " + "file-global map of string locals reds the clean list-argv call on line " + "4 because another function binds the same name — a false red on the " + "safe form, which forces the waivers off instead of keeping them honest." + ) + clean = ( "import os, subprocess\n" "def t(name, sql, docker, args):\n" From 54e9d48f7f51695ea18f8d02c2cef1b7116d5875 Mon Sep 17 00:00:00 2001 From: Nick Date: Tue, 11 Aug 2026 12:06:18 +0300 Subject: [PATCH 4/6] test(journeys): prune the subtree, not the node, when indexing a scope MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Per-function scoping did not close the false positive; it moved it one level down. `_index_scope` skipped a nested function with `ast.walk` + `continue`, which prunes that one NODE and still descends into its body, so an inner helper's built string landed in its parent's locals and reddened the parent's clean list argv: def outer(name): cmd = ["docker", "ps"] subprocess.run(cmd) # flagged, and clean def inner(x): cmd = f"docker rm {x}" subprocess.run(cmd) # the actual hazard Not hypothetical: conftest.py and test_f_agentic_load.py already nest helpers inside waiver-bearing functions, so the first inner `cmd`/`args` holding a string would have reddened the outer call. Indexing now descends through direct children and stops at each nested scope, carrying a nested definition's decorators and argument defaults with the enclosing scope, where they actually evaluate. Comprehensions and lambdas get their own scope for the same reason — a loop target or a lambda argument binds inside and shadows the enclosing name — and inherit the enclosing scope minus what they bind, so a comprehension that merely READS an outer built string is still caught. The planted test pins both as line-number assertions, so a regression names the clean call it wrongly reds rather than just going red somewhere. Co-Authored-By: Claude Opus 5 (1M context) --- deploy/tests/journeys/test_z_meta_guard.py | 135 +++++++++++++++++++-- 1 file changed, 123 insertions(+), 12 deletions(-) diff --git a/deploy/tests/journeys/test_z_meta_guard.py b/deploy/tests/journeys/test_z_meta_guard.py index e2f579bb..6036319e 100644 --- a/deploy/tests/journeys/test_z_meta_guard.py +++ b/deploy/tests/journeys/test_z_meta_guard.py @@ -251,25 +251,94 @@ def _built_string(node: ast.AST) -> str | None: string_locals_by_scope: dict[int, dict[str, str]] = {} scope_of: dict[int, int] = {} + # A comprehension and a lambda each get their own runtime scope, so a call + # inside one must not read the enclosing scope's locals. + _SCOPES = ( + ast.FunctionDef, + ast.AsyncFunctionDef, + ast.Lambda, + ast.ListComp, + ast.SetComp, + ast.DictComp, + ast.GeneratorExp, + ) + def _index_scope(scope: ast.AST) -> None: + """Record this scope's string locals and the calls that belong to it. + + Walks DIRECT CHILDREN and stops at each nested scope rather than using + ``ast.walk`` with a ``continue``: walk yields the nested node and then + descends into it anyway, so a ``continue`` prunes one node, never the + subtree — which leaks an inner helper's locals into its parent and reds + the parent's clean list argv. + """ found: dict[str, str] = {} - for child in ast.walk(scope): - if isinstance(child, (ast.FunctionDef, ast.AsyncFunctionDef)) and child is not scope: - continue - if isinstance(child, ast.Assign): - built = _built_string(child.value) - if built is not None: - for tgt in child.targets: - if isinstance(tgt, ast.Name): - found[tgt.id] = built - if isinstance(child, ast.Call): - scope_of[id(child)] = id(scope) + + def _descend(node: ast.AST) -> None: + for child in ast.iter_child_nodes(node): + if isinstance(child, _SCOPES): + # A nested scope owns its own locals and its own calls; its + # decorators and argument defaults evaluate out here, so + # they stay with this scope. + for sub in getattr(child, "decorator_list", []): + _descend(sub) + args = getattr(child, "args", None) + for sub in (getattr(args, "defaults", []) if args else []): + _descend(sub) + continue + if isinstance(child, ast.Assign): + built = _built_string(child.value) + if built is not None: + for tgt in child.targets: + if isinstance(tgt, ast.Name): + found[tgt.id] = built + if isinstance(child, ast.Call): + scope_of[id(child)] = id(scope) + _descend(child) + + _descend(scope) + # A comprehension's loop targets bind inside it and shadow anything of + # the same name outside, so they are never a built string here. + for gen in getattr(scope, "generators", []): + for name in ast.walk(gen.target): + if isinstance(name, ast.Name): + found.pop(name.id, None) string_locals_by_scope[id(scope)] = found _index_scope(tree) + parent_scope: dict[int, int] = {} + + def _map_parents(scope: ast.AST) -> None: + for child in ast.walk(scope): + if child is scope or not isinstance(child, _SCOPES): + continue + if id(child) not in parent_scope: + parent_scope[id(child)] = id(scope) + + _map_parents(tree) for node in ast.walk(tree): - if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)): + if isinstance(node, _SCOPES): _index_scope(node) + _map_parents(node) + + # A comprehension or lambda reads the enclosing scope's names except the + # ones it binds itself, so inherit what the parent knows. A function body + # does not inherit: its own local shadows, and a closure read is not a + # shell hazard (a command string without shell= never reaches /bin/sh). + for node in ast.walk(tree): + if not isinstance(node, (ast.Lambda, ast.ListComp, ast.SetComp, ast.DictComp, ast.GeneratorExp)): + continue + own = string_locals_by_scope.get(id(node), {}) + inherited = dict(string_locals_by_scope.get(parent_scope.get(id(node), id(tree)), {})) + for gen in getattr(node, "generators", []): + for name in ast.walk(gen.target): + if isinstance(name, ast.Name): + inherited.pop(name.id, None) + args = getattr(node, "args", None) + for a in (args.args if args else []): + inherited.pop(a.arg, None) + inherited.update(own) + string_locals_by_scope[id(node)] = inherited def _command_string(node: ast.AST, call: ast.Call) -> str | None: built = _built_string(node) @@ -451,6 +520,48 @@ def test_shell_hazard_guard_reds_on_planted_violations() -> None: "safe form, which forces the waivers off instead of keeping them honest." ) + # A nested `def` must not leak its locals into its parent. `ast.walk` with a + # `continue` prunes the FunctionDef NODE and still descends into its body, + # so the outer scope collected the inner's built string and reddened the + # outer's clean list argv — the same false red as the flat map, one level + # down. Every waiver-bearing file here already contains nested helpers. + nested = ( + "import subprocess\n" + "def outer(name):\n" + ' cmd = ["docker", "ps"]\n' + " subprocess.run(cmd)\n" + " def inner(x):\n" + ' cmd = f"docker rm {x}"\n' + " subprocess.run(cmd)\n" + ) + with tempfile.TemporaryDirectory() as td: + f = Path(td) / "nested.py" + f.write_text(nested, encoding="utf-8") + nested_hits = _shell_hazard_sites(f) + assert [ln for ln, _ in nested_hits] == [7], ( + f"only the inner call on line 7 is a hazard; got {nested_hits!r}. The " + "outer call passes a list argv and must stay clean." + ) + + # A comprehension binds its loop target, which shadows an enclosing name; a + # lambda binds its arguments the same way. Both still READ what they do not + # bind, so the guard neither reds the shadowed case nor misses the read. + comprehension = ( + "import subprocess\n" + 'cmd = f"echo {X}"\n' + 'shadowed = [subprocess.run(cmd) for cmd in [["docker", "ps"]]]\n' + "reads = [subprocess.run(cmd) for _ in range(2)]\n" + ) + with tempfile.TemporaryDirectory() as td: + f = Path(td) / "comprehension.py" + f.write_text(comprehension, encoding="utf-8") + comp_hits = _shell_hazard_sites(f) + assert [ln for ln, _ in comp_hits] == [4], ( + f"line 3 binds `cmd` to a list argv in the comprehension's own scope " + f"and must stay clean, while line 4 reads the module's built string and " + f"must red; got {comp_hits!r}." + ) + clean = ( "import os, subprocess\n" "def t(name, sql, docker, args):\n" From 74ab9f1ce4c0c1693ba156bbecfd2ffcd936a06a Mon Sep 17 00:00:00 2001 From: Nick Date: Tue, 11 Aug 2026 12:18:50 +0300 Subject: [PATCH 5/6] test(journeys): resolve a name outward to the nearest scope, not the first one A comprehension inside a function resolved its names through the MODULE, so a command string built in the function was invisible: def t(x): cmd = f"rm {x}" return [subprocess.run(cmd) for _ in y] # missed Two causes, both fixed. The parent map recorded whichever scope a walk reached first and refused to overwrite it, which put every nested comprehension under the module; it now records the nearest enclosing scope by descending from each scope to its own children. And inheritance was precomputed in walk order, so a comprehension could inherit before its enclosing function had been indexed; the chain is now resolved at lookup, walking outward and stopping at any name the scope binds itself. A walrus binds in the enclosing scope rather than the comprehension (PEP 572), so its assignment is collected there. 10/10 scope cases correct, including a comprehension shadowing its own loop target, a lambda reading a function local, and a two-level nested def. All 12 evasions still caught, semgrep tree still 0. The planted test pins the three shapes by line number. Its assertion sorts the hits: they are not emitted in line order, and asserting the unsorted list would have made the test depend on traversal order rather than on the finding. Co-Authored-By: Claude Opus 5 (1M context) --- deploy/tests/journeys/test_z_meta_guard.py | 118 ++++++++++++++++----- 1 file changed, 92 insertions(+), 26 deletions(-) diff --git a/deploy/tests/journeys/test_z_meta_guard.py b/deploy/tests/journeys/test_z_meta_guard.py index 6036319e..e6771991 100644 --- a/deploy/tests/journeys/test_z_meta_guard.py +++ b/deploy/tests/journeys/test_z_meta_guard.py @@ -285,6 +285,15 @@ def _descend(node: ast.AST) -> None: args = getattr(child, "args", None) for sub in (getattr(args, "defaults", []) if args else []): _descend(sub) + if isinstance(child, (ast.ListComp, ast.SetComp, ast.DictComp, ast.GeneratorExp)): + # A walrus in a comprehension binds outward (PEP 572), + # so its assignment belongs to THIS scope even though + # the comprehension owns its other names. + for sub in ast.walk(child): + if isinstance(sub, ast.NamedExpr) and isinstance(sub.target, ast.Name): + built = _built_string(sub.value) + if built is not None: + found[sub.target.id] = built continue if isinstance(child, ast.Assign): built = _built_string(child.value) @@ -292,6 +301,13 @@ def _descend(node: ast.AST) -> None: for tgt in child.targets: if isinstance(tgt, ast.Name): found[tgt.id] = built + if isinstance(child, ast.NamedExpr): + # PEP 572: a walrus inside a comprehension binds in the + # ENCLOSING scope, so it is collected here rather than in + # the comprehension's own scope. + built = _built_string(child.value) + if built is not None and isinstance(child.target, ast.Name): + found[child.target.id] = built if isinstance(child, ast.Call): scope_of[id(child)] = id(scope) _descend(child) @@ -306,48 +322,71 @@ def _descend(node: ast.AST) -> None: string_locals_by_scope[id(scope)] = found _index_scope(tree) + # Nearest enclosing scope, not the first one a walk happens to reach: a + # comprehension inside a function must resolve names through that function, + # never through the module. parent_scope: dict[int, int] = {} def _map_parents(scope: ast.AST) -> None: - for child in ast.walk(scope): - if child is scope or not isinstance(child, _SCOPES): - continue - if id(child) not in parent_scope: - parent_scope[id(child)] = id(scope) + def _walk(node: ast.AST) -> None: + for child in ast.iter_child_nodes(node): + if isinstance(child, _SCOPES): + parent_scope[id(child)] = id(scope) + _map_parents(child) + else: + _walk(child) + + _walk(scope) _map_parents(tree) for node in ast.walk(tree): if isinstance(node, _SCOPES): _index_scope(node) - _map_parents(node) - # A comprehension or lambda reads the enclosing scope's names except the - # ones it binds itself, so inherit what the parent knows. A function body - # does not inherit: its own local shadows, and a closure read is not a - # shell hazard (a command string without shell= never reaches /bin/sh). + _INHERITS = (ast.Lambda, ast.ListComp, ast.SetComp, ast.DictComp, ast.GeneratorExp) + scope_node: dict[int, ast.AST] = {id(tree): tree} for node in ast.walk(tree): - if not isinstance(node, (ast.Lambda, ast.ListComp, ast.SetComp, ast.DictComp, ast.GeneratorExp)): - continue - own = string_locals_by_scope.get(id(node), {}) - inherited = dict(string_locals_by_scope.get(parent_scope.get(id(node), id(tree)), {})) - for gen in getattr(node, "generators", []): - for name in ast.walk(gen.target): - if isinstance(name, ast.Name): - inherited.pop(name.id, None) - args = getattr(node, "args", None) - for a in (args.args if args else []): - inherited.pop(a.arg, None) - inherited.update(own) - string_locals_by_scope[id(node)] = inherited + if isinstance(node, _SCOPES): + scope_node[id(node)] = node + + def _lookup(scope_id: int, name: str) -> str | None: + """Resolve ``name`` outward from ``scope_id``. + + A comprehension or lambda reads the enclosing scope's names except the + ones it binds itself, so the chain is walked at lookup time rather than + precomputed — a comprehension nested in a function would otherwise + inherit before that function had been indexed. A function body does not + inherit: its own local shadows, and a closure read is not a shell + hazard (a command string without shell= never reaches /bin/sh). + """ + seen: set[int] = set() + while scope_id is not None and scope_id not in seen: + seen.add(scope_id) + found = string_locals_by_scope.get(scope_id, {}) + if name in found: + return found[name] + node = scope_node.get(scope_id) + if not isinstance(node, _INHERITS): + return None + for gen in getattr(node, "generators", []): + for bound in ast.walk(gen.target): + if isinstance(bound, ast.Name) and bound.id == name: + return None + args = getattr(node, "args", None) + for a in (args.args if args else []): + if a.arg == name: + return None + scope_id = parent_scope.get(scope_id, id(tree)) if scope_id != id(tree) else None + return None def _command_string(node: ast.AST, call: ast.Call) -> str | None: built = _built_string(node) if built is not None: return built if isinstance(node, ast.Name): - scope = string_locals_by_scope.get(scope_of.get(id(call), id(tree)), {}) - if node.id in scope: - return f"{scope[node.id]} (via {node.id})" + built = _lookup(scope_of.get(id(call), id(tree)), node.id) + if built is not None: + return f"{built} (via {node.id})" return None def _callee(node: ast.Call) -> tuple[str, bool, bool]: @@ -562,6 +601,33 @@ def test_shell_hazard_guard_reds_on_planted_violations() -> None: f"must red; got {comp_hits!r}." ) + # Resolution walks OUTWARD to the nearest enclosing scope. A comprehension + # inside a function must read that function's locals, not the module's — + # mapping it to the first scope a walk reaches put every comprehension + # under the module and silently stopped resolving. + outward = ( + "import subprocess\n" + "def reads(x):\n" + ' cmd = f"rm {x}"\n' + " return [subprocess.run(cmd) for _ in y]\n" + "def shadows(x):\n" + ' cmd = f"rm {x}"\n' + ' return [subprocess.run(cmd) for cmd in [["docker", "ps"]]]\n' + "def walrus(x):\n" + ' [(c := f"rm {x}") for _ in range(1)]\n' + " subprocess.run(c)\n" + ) + with tempfile.TemporaryDirectory() as td: + f = Path(td) / "outward.py" + f.write_text(outward, encoding="utf-8") + outward_hits = _shell_hazard_sites(f) + assert sorted(ln for ln, _ in outward_hits) == [4, 10], ( + f"line 4 reads the function's built string through a comprehension and " + f"must red; line 7 binds `cmd` as its own loop target and must stay " + f"clean; line 10 uses a name a walrus bound OUTSIDE the comprehension " + f"(PEP 572) and must red. Got {outward_hits!r}." + ) + clean = ( "import os, subprocess\n" "def t(name, sql, docker, args):\n" From 64f92c4fbbdf6aff295b8ba748f94658d69af12c Mon Sep 17 00:00:00 2001 From: Nick Date: Tue, 11 Aug 2026 12:22:32 +0300 Subject: [PATCH 6/6] test(journeys): every lambda parameter kind shadows, not just the ordinary ones MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The inheritance lookup popped `args.args` alone, so a lambda whose own positional-only, keyword-only, `*args` or `**kwargs` parameter shadowed an enclosing built-string local still inherited that string, and the lambda's clean call reddened: cmd = f"echo {X}" posonly = lambda cmd, /: subprocess.run(cmd) # flagged, and clean All five parameter kinds now shadow. A lambda that READS an enclosing built string is still caught, so the fix narrows nothing. Two gaps stay open and are now written down in the detector rather than left implied: a name rebound through `global`/`nonlocal` in another scope, and a closure reading an enclosing function's local. Both need cross-scope name resolution, and neither is a shell hazard — `subprocess.run("")` without `shell=` is a program-name lookup that fails, not a command line. The shell surface is `shell=` and the always-shell callees, which resolve regardless of scope. Mutation-checked: reverting the pop to `args.args` reds the planted test. Co-Authored-By: Claude Opus 5 (1M context) --- deploy/tests/journeys/test_z_meta_guard.py | 41 ++++++++++++++++++++-- 1 file changed, 39 insertions(+), 2 deletions(-) diff --git a/deploy/tests/journeys/test_z_meta_guard.py b/deploy/tests/journeys/test_z_meta_guard.py index e6771991..285bf1cb 100644 --- a/deploy/tests/journeys/test_z_meta_guard.py +++ b/deploy/tests/journeys/test_z_meta_guard.py @@ -188,6 +188,13 @@ def _shell_hazard_sites(path: Path) -> list[tuple[int, str]]: element of a list argv is deliberately not flagged: that is the safe form (``f"name={cname}"`` reaches the program as a single argument, with no shell to re-parse it), and it is what the per-site waivers describe. + + Known gaps, both deliberate: a name rebound through ``global``/``nonlocal`` + in another scope, and a closure reading an enclosing function's local. Both + need cross-scope name resolution, and neither is a shell hazard — + ``subprocess.run("")`` without ``shell=`` is a program-name lookup that + fails, not a command line. The shell surface is ``shell=`` and the + ``_ALWAYS_SHELL`` callees, and those resolve regardless of scope. """ tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path)) @@ -373,8 +380,16 @@ def _lookup(scope_id: int, name: str) -> str | None: if isinstance(bound, ast.Name) and bound.id == name: return None args = getattr(node, "args", None) - for a in (args.args if args else []): - if a.arg == name: + if args is not None: + # Every parameter kind binds inside the lambda and shadows the + # enclosing name: positional-only, ordinary, keyword-only, and + # the *args / **kwargs collectors. + bound = [*args.posonlyargs, *args.args, *args.kwonlyargs] + if args.vararg is not None: + bound.append(args.vararg) + if args.kwarg is not None: + bound.append(args.kwarg) + if any(a.arg == name for a in bound): return None scope_id = parent_scope.get(scope_id, id(tree)) if scope_id != id(tree) else None return None @@ -628,6 +643,28 @@ def test_shell_hazard_guard_reds_on_planted_violations() -> None: f"(PEP 572) and must red. Got {outward_hits!r}." ) + # Every parameter kind binds inside the lambda. Popping only `args.args` + # left positional-only, keyword-only and the collectors inheriting the + # enclosing string, which reds a lambda whose own argument is a list argv. + lambda_args = ( + "import subprocess\n" + 'cmd = f"echo {X}"\n' + "posonly = lambda cmd, /: subprocess.run(cmd)\n" + "kwonly = lambda *, cmd: subprocess.run(cmd)\n" + "collector = lambda *cmd: subprocess.run(cmd)\n" + "kwcollector = lambda **cmd: subprocess.run(cmd)\n" + "ordinary = lambda cmd: subprocess.run(cmd)\n" + "reads = lambda y: subprocess.run(cmd)\n" + ) + with tempfile.TemporaryDirectory() as td: + f = Path(td) / "lambda_args.py" + f.write_text(lambda_args, encoding="utf-8") + lambda_hits = _shell_hazard_sites(f) + assert sorted(ln for ln, _ in lambda_hits) == [8], ( + f"only line 8 reads the module's built string; every lambda above binds " + f"`cmd` itself and passes its own argument. Got {lambda_hits!r}." + ) + clean = ( "import os, subprocess\n" "def t(name, sql, docker, args):\n"