fix(taint): resolve lambda bodies flow-sensitively against final taints

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

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
 
 

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,