fix(taint): resolve lambda bodies against worst-ever var taint (fixes #19 FN)

CHANGELOG.md

@@ -44,24 +44,30 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   Lambda bodies are now traversed as part of the enclosing analyzable scope (lambdas
   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.
+  lambda body in a second pass (after the forward walk) against the **worst**
+  (least-trusted) taint each captured variable holds *anywhere* in the function, in an
+  isolated scope copy (lambda-local params/walrus never leak, and the lambda's own
+  parameters shadow enclosing names of the same id). Whole-function-worst is the
+  fail-closed choice for a closure, which defers execution to an unknown call time and
+  captures free variables by reference: no single program-point snapshot is sound —
+  the definition-site value misses a variable tainted *after* the lambda is defined
+  (`src = "safe"; cb = lambda: eval(src); src = read_raw(p)`), and the final value
+  misses a variable still raw *when the lambda is called* and cleaned only afterwards
+  (`src = read_raw(p); cb = lambda: eval(src); cb(); src = "clean"`). Both are real
+  deferred sinks and now fire. This closes the false-negative (raw →
+  `eval`/`exec`/`pickle.loads`/`subprocess` in a lambda body now fires, including both
+  deferred orderings) and removes the gross over-report where *any* lambda-body sink in
+  a trusted function previously fell to the pessimistic flow-insensitive fallback and
+  fired `UNKNOWN_RAW` regardless of the argument (`lambda: eval("safe")` and a
+  `lambda cmd: eval(cmd)` whose param shadows an enclosing raw `cmd` no longer fire; no
+  `WLN-ENGINE-FLOW-INSENSITIVE-FALLBACK` warning is emitted). The remaining imprecision
+  is a documented, conservative, waivable **false positive**: a variable raw only
+  *before* the lambda captures it (e.g. `x = read_raw(p); x = "clean"; cb = lambda:
+  eval(x)`) is treated tainted, because the analysis joins over the whole function
+  rather than tracking the capture point — the safe direction for a security analyzer,
+  and verified not to fire on wardline's own source (dogfood: 0 new). Regression tests
+  cover discovery (`_own_calls`), both deferred orderings on `PY-WL-107`/`108`, no-fire
+  on a clean local and a shadowing lambda parameter, and the documented conservative FP.
 - **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

@@ -26,7 +26,7 @@
 from dataclasses import dataclass
 from typing import TYPE_CHECKING
 
-from wardline.core.taints import _PROVENANCE_CLASH, TaintState, combine
+from wardline.core.taints import _PROVENANCE_CLASH, TRUST_RANK, TaintState, combine
 
 if TYPE_CHECKING:
     from collections.abc import Iterator
@@ -175,30 +175,53 @@ def _own_scope_lambdas(node: ast.AST) -> Iterator[ast.Lambda]:
         yield from _own_scope_lambdas(child)
 
 
+def _worst_ever_var_taints(
+    call_site_taints: dict[int, dict[str, TaintState]],
+    final_var_taints: dict[str, TaintState],
+) -> dict[str, TaintState]:
+    """The least-trusted (highest ``TRUST_RANK``) taint each variable holds ANYWHERE in
+    the function — joined over every per-statement snapshot plus the final state.
+
+    This is the sound capture taint for a closure free variable: a lambda defers
+    execution to an unknown call time and captures the variable by reference, so it may
+    observe ANY value the variable holds. Picking a single program point is unsound —
+    the definition-site value misses a later raw assignment, and the final value misses
+    a raw value that was cleaned up after the lambda already ran. The whole-function
+    worst guarantees no false-negative (at the cost of a conservative over-approximation
+    when a raw value existed only *before* the capture — the safe direction)."""
+    worst = dict(final_var_taints)
+    for snapshot in call_site_taints.values():
+        for name, taint in snapshot.items():
+            current = worst.get(name)
+            if current is None or TRUST_RANK[taint] > TRUST_RANK[current]:
+                worst[name] = taint
+    return worst
+
+
 def _resolve_lambda_bodies(
     func_node: ast.FunctionDef | ast.AsyncFunctionDef,
     function_taint: TaintState,
     taint_map: dict[str, TaintState],
-    final_var_taints: dict[str, TaintState],
+    worst_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.
+    Runs AFTER the forward walk, against ``worst_var_taints`` — the worst (least-trusted)
+    taint each variable holds anywhere in the function (see :func:`_worst_ever_var_taints`).
+    A lambda defers execution and captures free variables by reference, so the body's
+    free variables must be resolved against the worst value they could carry at call
+    time, not any single program-point snapshot — that is what keeps both a variable
+    tainted *after* the lambda is defined and a variable still raw *when the lambda is
+    called* (cleaned only later) visible to the sink rules.
 
     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)."""
+    inheriting their taint. 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)
+        scope = dict(worst_var_taints)
         args = lam.args
         for param in (*args.posonlyargs, *args.args, *args.kwonlyargs):
             scope[param.arg] = function_taint
@@ -263,10 +286,18 @@ 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)
+        # Second pass: resolve lambda BODIES against the worst taint each variable holds
+        # anywhere in the function, so a deferred lambda that captures a variable tainted
+        # at ANY point — before OR after its definition, or still raw when it is called
+        # and cleaned only afterwards — stays visible to the sink rules
+        # (closure-by-reference soundness; see _worst_ever_var_taints). Requires the
+        # per-statement snapshots; without them (a degraded caller that does not request
+        # flow-sensitive recording) the bodies are left unrecorded so the sink rules fall
+        # back to their pessimistic UNKNOWN_RAW default rather than to a possibly-clean
+        # final value — never silently masking a sink.
+        if call_site_taints is not None:
+            worst = _worst_ever_var_taints(call_site_taints, var_taints)
+            _resolve_lambda_bodies(func_node, function_taint, taint_map, worst)
         return var_taints
     finally:
         if token_clash is not None:

tests/unit/scanner/rules/test_sink_rules.py

@@ -238,8 +238,8 @@ def f(p):
 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.
+    # executes on raw data. The body is resolved against the WORST taint ``src`` holds
+    # anywhere in the function (here: raw), so this real deferred sink stays visible.
     ctx = _analyze(
         tmp_path,
         """
@@ -258,28 +258,55 @@ def f(p):
     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).
+def test_107_lambda_called_before_clean_reassignment_still_fires(tmp_path) -> None:
+    # The symmetric deferred case (the one final-state resolution missed): the lambda
+    # is CALLED while ``src`` is still raw, and the clean reassignment happens AFTER.
+    # At runtime eval() executes on raw data — this MUST fire. Resolving against the
+    # worst taint ``src`` holds anywhere keeps it visible even though the final state
+    # is clean.
     ctx = _analyze(
         tmp_path,
         """
         @trusted(level='ASSURED')
         def f(p):
-            x = read_raw(p)
-            x = "clean"
-            cb = lambda: eval(x)
-            return cb()
+            src = read_raw(p)
+            cb = lambda: eval(src)
+            cb()
+            src = "clean"
+            return 1
         """,
     )
     with warnings.catch_warnings(record=True) as caught:
         warnings.simplefilter("always")
         findings = UntrustedToExec().check(ctx)
-    assert findings == []
+    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_lambda_captures_var_raw_earlier_conservatively_fires(tmp_path) -> None:
+    # DOCUMENTED conservative over-approximation (a known, waivable false positive):
+    # ``x`` is raw, then overwritten clean BEFORE the lambda is defined, so the closure
+    # can never actually observe the raw value. A precise (def-point) analysis would
+    # not fire here. But a lambda defers execution to an unknown call time, and the
+    # engine resolves its body against the worst taint ``x`` holds ANYWHERE in the
+    # function — the fail-closed choice that guarantees no false-NEGATIVE for the
+    # deferred-sink cases above. So this fires. The safe direction for a security
+    # analyzer; pin it so the behaviour is intentional, not accidental.
+    ctx = _analyze(
+        tmp_path,
+        """
+        @trusted(level='ASSURED')
+        def f(p):
+            x = read_raw(p)
+            x = "clean"
+            cb = lambda: eval(x)
+            return cb()
+        """,
+    )
+    findings = UntrustedToExec().check(ctx)
+    assert [(x.rule_id, x.qualname) for x in findings] == [("PY-WL-107", "m.f")]
+
+
 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.