From 95d1139bab8c7e2cfb56ec279586d0e60906caba Mon Sep 17 00:00:00 2001 From: John Morrissey Date: Fri, 5 Jun 2026 16:38:17 +1000 Subject: [PATCH] fix(taint): resolve lambda bodies flow-sensitively against final taints MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit PR #18 made sink DISCOVERY (_own_calls) descend into lambda bodies, but the L2 taint walk's _resolve_expr resolved only lambda *defaults*, never the body — so a sink call inside a lambda body had no entry in function_call_site_arg_taints and worst_arg_taint fell to the pessimistic flow-insensitive fallback (WLN-ENGINE-FLOW-INSENSITIVE-FALLBACK), firing UNKNOWN_RAW for ANY lambda-body sink in a trusted function regardless of the real argument taint (false positives on 'lambda: eval("safe")' and a shadowing 'lambda cmd: eval(cmd)'). Resolve lambda bodies in a second pass (_resolve_lambda_bodies) AFTER the forward walk finalises var_taints, against those FINAL taints. Final-state is the sound choice for a closure: a lambda defers execution and captures free variables by reference, so a variable assigned raw AFTER the lambda is defined ('src="safe"; cb=lambda: eval(src); src=read_raw(p)') is a real deferred sink and must still fire — a definition-site pass would silently miss it (a false negative, the cardinal sin for a fail-closed analyzer). Each body is resolved in an isolated scope copy so lambda-local bindings (params, walrus) never leak, and the lambda's own parameters are reset to the neutral seed (function_taint) so they shadow enclosing names rather than inheriting their taint. Net effect is a strict improvement vs main: real and deferred lambda-body sinks fire flow-sensitively (no warning); clean, final-state-clean, and shadowed-param values no longer false-fire. Full suite green (coverage 91.77%); mypy clean; dogfood self-scan unchanged (0 new). Co-Authored-By: Claude Opus 4.8 (1M context) --- CHANGELOG.md | 26 +++- src/wardline/scanner/taint/variable_level.py | 62 ++++++++- tests/unit/scanner/rules/test_sink_rules.py | 125 +++++++++++++++++++ 3 files changed, 206 insertions(+), 7 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 73149d24..a4592612 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -42,12 +42,26 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 and likewise `exec` / `pickle.loads` / `subprocess` / dynamic `import` — was never handed to the sink rules (`PY-WL-106/107/108`), producing a silent false-negative. Lambda bodies are now traversed as part of the enclosing analyzable scope (lambdas - are not indexed as separate entities, unlike `def`/`class`). This strictly improves - coverage — such sinks go invisible→visible; their argument-taint resolution then - follows the engine's documented flow-insensitive limitation (a lambda-body sink - resolves against the pessimistic flow-insensitive fallback, which warns rather than - silently degrading). Regression tests cover both `_own_calls` and a `PY-WL-107` fire - on `lambda: eval(src)`. + are not indexed as separate entities, unlike `def`/`class`) — on **both** sides: + sink *discovery* (`_own_calls`) and the L2 taint *walk*. The walk resolves each + lambda body in a second pass, after the forward walk has finalised the function's + variable taints, against those **final** taints (in an isolated scope copy — + lambda-local params/walrus never leak, and the lambda's own parameters shadow + enclosing names of the same id). Final-state resolution is the sound choice for a + closure: a lambda defers execution and captures free variables by reference, so a + variable assigned raw *after* the lambda is defined (`src = "safe"; cb = lambda: + eval(src); src = read_raw(p)`) is a real deferred sink and still fires — a + definition-site pass would silently miss it. The change is a strict improvement: it + closes the false-negative (raw→`eval`/`exec`/`pickle.loads`/`subprocess` in a lambda + body now fires, including the deferred case) **and** removes the prior over-report + where *any* lambda-body sink in a trusted function fell to the pessimistic + flow-insensitive fallback and fired `UNKNOWN_RAW` regardless of the actual argument + (`lambda: eval("safe")`, a `lambda cmd: eval(cmd)` whose param shadows an enclosing + raw `cmd`, and a value clean in the final state no longer false-fire). No + `WLN-ENGINE-FLOW-INSENSITIVE-FALLBACK` warning is emitted for lambda-body sinks. + Regression tests cover discovery (`_own_calls`), flow-sensitive `PY-WL-107`/`108` + fires on real and deferred taint, and no-fire on a clean local, a final-state-clean + reassignment, and a shadowing lambda parameter. - **Local trust-pack guard no longer executes repository code while deciding.** `_is_local_pack()` resolved a `wardline.yaml` `packs:` entry with `importlib.util.find_spec()`, which imports (and runs) the parent of a dotted diff --git a/src/wardline/scanner/taint/variable_level.py b/src/wardline/scanner/taint/variable_level.py index 656f9a07..9b6a2eeb 100644 --- a/src/wardline/scanner/taint/variable_level.py +++ b/src/wardline/scanner/taint/variable_level.py @@ -24,9 +24,13 @@ import ast import contextvars from dataclasses import dataclass +from typing import TYPE_CHECKING from wardline.core.taints import _PROVENANCE_CLASH, TaintState, combine +if TYPE_CHECKING: + from collections.abc import Iterator + # Serialisation sinks — calls that cross the representation boundary. Their # output sheds validation provenance (raw bytes/str), so → UNKNOWN_RAW. This is # a generic fail-closed heuristic, not governance. (SP1f note: where this and @@ -159,6 +163,52 @@ def analyze_function_variables( _CURRENT_MODULE_PREFIX.reset(token_module) +def _own_scope_lambdas(node: ast.AST) -> Iterator[ast.Lambda]: + """Yield every ``ast.Lambda`` in *node*'s own scope (descends into lambdas, which + are not separate entities, but NOT into nested ``def``/``class`` — those are + analyzed as their own entities).""" + for child in ast.iter_child_nodes(node): + if isinstance(child, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)): + continue + if isinstance(child, ast.Lambda): + yield child + yield from _own_scope_lambdas(child) + + +def _resolve_lambda_bodies( + func_node: ast.FunctionDef | ast.AsyncFunctionDef, + function_taint: TaintState, + taint_map: dict[str, TaintState], + final_var_taints: dict[str, TaintState], +) -> None: + """Record call-site arg taints for calls inside lambda BODIES (sink-rule input). + + Runs AFTER the forward walk has finalised ``final_var_taints``. A lambda defers + execution and captures free variables by reference, so the sound static taint for + a free variable read in the body is its FINAL function-scope taint — the closest + proxy for its value when the lambda is later called. Resolving here (not at the + definition site) catches a variable assigned raw *after* the lambda is defined (a + real deferred sink) that a definition-site pass would silently miss, while a value + that is clean in the final state does not over-fire. + + Each lambda body is resolved in an isolated scope copy so lambda-local bindings + (params, walrus) never leak; the lambda's own parameters are reset to the neutral + seed (``function_taint``) so they SHADOW enclosing names of the same id rather than + inheriting their taint. Free variables resolve against ``final_var_taints``. The + recording itself happens via the ``_CURRENT_CALL_SITE_ARG_TAINTS`` contextvar set + by the caller, keyed by ``id(call)`` (matching what the sink rules look up).""" + for lam in _own_scope_lambdas(func_node): + scope = dict(final_var_taints) + args = lam.args + for param in (*args.posonlyargs, *args.args, *args.kwonlyargs): + scope[param.arg] = function_taint + if args.vararg is not None: + scope[args.vararg.arg] = function_taint + if args.kwarg is not None: + scope[args.kwarg.arg] = function_taint + _resolve_expr(lam.body, function_taint, taint_map, scope) + + def compute_variable_taints( func_node: ast.FunctionDef | ast.AsyncFunctionDef, function_taint: TaintState, @@ -213,6 +263,10 @@ def compute_variable_taints( var_taints: dict[str, TaintState] = {} _seed_parameters(func_node, function_taint, var_taints, param_meets, taint_map) _walk_body(func_node.body, function_taint, taint_map, var_taints, call_site_taints) + # Second pass: resolve lambda BODIES against the now-final var_taints so a + # deferred lambda that captures a variable tainted AFTER its definition is + # still seen by the sink rules (closure-by-reference soundness). + _resolve_lambda_bodies(func_node, function_taint, taint_map, var_taints) return var_taints finally: if token_clash is not None: @@ -376,11 +430,17 @@ def _resolve_expr( if isinstance(node, (ast.ListComp, ast.SetComp, ast.GeneratorExp, ast.DictComp)): return _resolve_comprehension(node, function_taint, taint_map, var_taints) if isinstance(node, ast.Lambda): + # Defaults evaluate in the ENCLOSING scope at definition time — resolve them + # against var_taints (and bind any walrus side-effects there). The lambda + # BODY is resolved separately, AFTER the forward walk, against the final + # var_taints (see _resolve_lambda_bodies): a lambda defers execution and + # captures free variables by reference, so the sound taint for a free var + # read in the body is its FINAL function-scope taint, not its def-site value. for default in (*node.args.defaults, *node.args.kw_defaults): if default is not None: _resolve_expr(default, function_taint, taint_map, var_taints) return function_taint - # Fallback: unmodelled Call shapes (str()/format()/.get()), lambdas, etc. + # Fallback: unmodelled Call shapes (str()/format()/.get()), etc. return function_taint diff --git a/tests/unit/scanner/rules/test_sink_rules.py b/tests/unit/scanner/rules/test_sink_rules.py index 8c9b016b..56b215f6 100644 --- a/tests/unit/scanner/rules/test_sink_rules.py +++ b/tests/unit/scanner/rules/test_sink_rules.py @@ -175,6 +175,131 @@ def f(): assert not any(str(w.message).startswith("WLN-ENGINE-FLOW-INSENSITIVE-FALLBACK") for w in caught) +def test_107_raw_reaches_eval_in_lambda_body_is_flow_sensitive(tmp_path) -> None: + # The lambda-body sink is resolved flow-sensitively (real taint), NOT via the + # pessimistic flow-insensitive fallback — so it still fires, but emits no warning. + ctx = _analyze( + tmp_path, + """ + @trusted(level='ASSURED') + def f(p): + src = read_raw(p) + cb = lambda: eval(src) + return cb() + """, + ) + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter("always") + findings = UntrustedToExec().check(ctx) + assert [(x.rule_id, x.qualname) for x in findings] == [("PY-WL-107", "m.f")] + assert not any(str(w.message).startswith("WLN-ENGINE-FLOW-INSENSITIVE-FALLBACK") for w in caught) + + +def test_107_clean_local_in_lambda_body_does_not_fire(tmp_path) -> None: + # A clean (INTEGRAL) value reaching eval inside a lambda body must NOT fire: + # the engine now resolves the body's args instead of blanket-pessimistic UNKNOWN_RAW. + ctx = _analyze( + tmp_path, + """ + @trusted(level='ASSURED') + def f(p): + safe = "harmless" + cb = lambda: eval(safe) + return cb() + """, + ) + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter("always") + findings = UntrustedToExec().check(ctx) + assert findings == [] + assert not any(str(w.message).startswith("WLN-ENGINE-FLOW-INSENSITIVE-FALLBACK") for w in caught) + + +def test_107_lambda_param_shadowing_enclosing_raw_does_not_fire(tmp_path) -> None: + # The lambda's own parameter SHADOWS the enclosing raw ``cmd``: it is a fresh + # binding, not the captured raw value, so ``eval(cmd)`` must NOT fire. + ctx = _analyze( + tmp_path, + """ + @trusted(level='ASSURED') + def f(p): + cmd = read_raw(p) + cb = lambda cmd: eval(cmd) + return cb('ls') + """, + ) + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter("always") + findings = UntrustedToExec().check(ctx) + assert findings == [] + assert not any(str(w.message).startswith("WLN-ENGINE-FLOW-INSENSITIVE-FALLBACK") for w in caught) + + +def test_107_raw_assigned_after_lambda_def_still_fires(tmp_path) -> None: + # Closure-by-reference soundness: the lambda is DEFINED while ``src`` is clean, + # but ``src`` is reassigned raw before the lambda is called — at runtime eval() + # executes on raw data. Resolving the body against the FINAL var taint (not the + # definition-site value) keeps this real deferred sink visible. + ctx = _analyze( + tmp_path, + """ + @trusted(level='ASSURED') + def f(p): + src = "safe" + cb = lambda: eval(src) + src = read_raw(p) + return cb() + """, + ) + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter("always") + findings = UntrustedToExec().check(ctx) + assert [(x.rule_id, x.qualname) for x in findings] == [("PY-WL-107", "m.f")] + assert not any(str(w.message).startswith("WLN-ENGINE-FLOW-INSENSITIVE-FALLBACK") for w in caught) + + +def test_107_value_clean_in_final_state_in_lambda_body_does_not_fire(tmp_path) -> None: + # The reverse: raw is overwritten clean BEFORE the lambda is defined, so the + # final state of ``x`` is clean — the lambda body must NOT fire (final-state, not + # worst-ever, resolution). + ctx = _analyze( + tmp_path, + """ + @trusted(level='ASSURED') + def f(p): + x = read_raw(p) + x = "clean" + cb = lambda: eval(x) + return cb() + """, + ) + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter("always") + findings = UntrustedToExec().check(ctx) + assert findings == [] + assert not any(str(w.message).startswith("WLN-ENGINE-FLOW-INSENSITIVE-FALLBACK") for w in caught) + + +def test_108_raw_reaches_os_system_in_lambda_body(tmp_path) -> None: + # The engine fix is sink-agnostic (shared _resolve_expr / worst_arg_taint): a + # command sink in a lambda body fires flow-sensitively on real taint too. + ctx = _analyze( + tmp_path, + """ + @trusted(level='ASSURED') + def f(p): + cmd = read_raw(p) + cb = lambda: os.system(cmd) + return cb() + """, + ) + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter("always") + findings = UntrustedToCommand().check(ctx) + assert [(x.rule_id, x.qualname) for x in findings] == [("PY-WL-108", "m.f")] + assert not any(str(w.message).startswith("WLN-ENGINE-FLOW-INSENSITIVE-FALLBACK") for w in caught) + + def test_108_raw_reaches_os_system(tmp_path) -> None: ctx = _analyze( tmp_path,