Fix lambda body sink discovery

CHANGELOG.md

@@ -36,6 +36,18 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   CLI verb shares the same filter core. (WS-B1, WS-B2)
 
 ### Security
+- **Dangerous-sink rules now see lambda bodies (closes a false-green).** `_own_calls`
+  treated `ast.Lambda` as a separate scope and only inspected lambda *default*
+  expressions, so a sink reached inside a lambda *body* — `cb = lambda: eval(src)`,
+  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)`.
 - **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/diagnostics.py

@@ -49,6 +49,7 @@ def _is_native_first_party(mod: str) -> bool:
     """True if ``mod`` is, or is under, a declared native/first-party prefix."""
     return any(mod == prefix or mod.startswith(prefix + ".") for prefix in _NATIVE_FIRST_PARTY_PREFIXES)
 
+
 # code -> (rule_id, severity, kind)
 _DIAG_MAP: dict[str, tuple[str, Severity, Kind]] = {
     "L3_CONVERGENCE_BOUND": ("WLN-L3-CONVERGENCE-BOUND", Severity.WARN, Kind.METRIC),

src/wardline/scanner/rules/_sink_helpers.py

@@ -79,18 +79,16 @@ def canonical_call_name(dotted: str, alias_map: Mapping[str, str]) -> str:
 
 
 def _own_calls(node: ast.AST) -> Iterator[ast.Call]:
-    """Yield every ``ast.Call`` in *node*'s own scope (never descending into nested
-    def/class/lambda — those are separate scopes / separate entities)."""
+    """Yield every ``ast.Call`` in *node*'s own analyzable scope.
+
+    Function, async-function, and class bodies are indexed as their own entities, so
+    they are not traversed here. Lambda bodies are intentionally traversed because
+    the entity index does not emit separate lambda entities; skipping them would hide
+    dangerous calls from sink rules.
+    """
     for child in ast.iter_child_nodes(node):
         if isinstance(child, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)):
             continue
-        if isinstance(child, ast.Lambda):
-            for default in (*child.args.defaults, *child.args.kw_defaults):
-                if default is not None:
-                    if isinstance(default, ast.Call):
-                        yield default
-                    yield from _own_calls(default)
-            continue
         if isinstance(child, ast.Call):
             yield child
         yield from _own_calls(child)

tests/unit/scanner/rules/test_sink_rules.py

@@ -32,16 +32,19 @@ def _analyze(tmp_path: Path, src: str):
     return analyzer.last_context
 
 
-def test_sink_calls_do_not_enter_lambda_body() -> None:
+def test_sink_calls_include_lambda_body() -> None:
     func = ast.parse("def f():\n    cb = lambda: wrapper(eval('1 + 1'))\n    return cb\n").body[0]
-    assert list(_own_calls(func)) == []
-    assert list(sink_calls(func, frozenset({"eval"}), {}, "m")) == []
+    assert [(call.lineno, ast.unparse(call.func)) for call in _own_calls(func)] == [
+        (2, "wrapper"),
+        (2, "eval"),
+    ]
+    assert [(call.lineno, sink) for call, sink in sink_calls(func, frozenset({"eval"}), {}, "m")] == [(2, "eval")]
 
 
 def test_own_calls_preserve_lambda_default_calls() -> None:
     func = ast.parse("def f(raw):\n    cb = lambda value=eval(raw): wrapper(raw)\n    return cb\n").body[0]
     calls = list(sink_calls(func, frozenset({"eval", "wrapper"}), {}, "m"))
-    assert [(call.lineno, sink) for call, sink in calls] == [(2, "eval")]
+    assert [(call.lineno, sink) for call, sink in calls] == [(2, "eval"), (2, "wrapper")]
 
 
 def test_106_raw_reaches_pickle_loads(tmp_path) -> None:
@@ -141,6 +144,20 @@ def f(p):
     assert [(x.rule_id, x.qualname) for x in UntrustedToExec().check(ctx)] == [("PY-WL-107", "m.f")]
 
 
+def test_107_raw_reaches_eval_in_lambda_body(tmp_path) -> None:
+    ctx = _analyze(
+        tmp_path,
+        """
+        @trusted(level='ASSURED')
+        def f(p):
+            src = read_raw(p)
+            cb = lambda: eval(src)
+            return cb()
+        """,
+    )
+    assert [(x.rule_id, x.qualname) for x in UntrustedToExec().check(ctx)] == [("PY-WL-107", "m.f")]
+
+
 def test_107_safe_eval_in_lambda_default_does_not_fallback_or_fire(tmp_path) -> None:
     ctx = _analyze(
         tmp_path,

tests/unit/scanner/test_diagnostics.py

@@ -165,18 +165,14 @@ def test_native_first_party_core_import_resolves_without_project_module() -> Non
 
 def test_native_first_party_decorators_import_resolves() -> None:
     tree = ast.parse("from wardline.decorators import trust_boundary\n")
-    out = diagnose_unknown_imports(
-        tree=tree, module_path="x", project_modules=frozenset(), stdlib_keys=frozenset()
-    )
+    out = diagnose_unknown_imports(tree=tree, module_path="x", project_modules=frozenset(), stdlib_keys=frozenset())
     assert out == []
 
 
 def test_native_allowlist_does_not_suppress_genuine_third_party() -> None:
     # Over-suppression guard: a real unknown third-party import MUST still fire.
     tree = ast.parse("from acme_totally_unknown_pkg import thing\n")
-    out = diagnose_unknown_imports(
-        tree=tree, module_path="x", project_modules=frozenset(), stdlib_keys=frozenset()
-    )
+    out = diagnose_unknown_imports(tree=tree, module_path="x", project_modules=frozenset(), stdlib_keys=frozenset())
     assert len(out) == 1 and "acme_totally_unknown_pkg" in out[0][2]
 
 
@@ -185,9 +181,7 @@ def test_native_allowlist_does_not_suppress_undeclared_wardline_submodule() -> N
     # neither a project module nor a declared native prefix must still report, so
     # the allowlist can't silently swallow a real gap.
     tree = ast.parse("from wardline.experimental.zzz import q\n")
-    out = diagnose_unknown_imports(
-        tree=tree, module_path="x", project_modules=frozenset(), stdlib_keys=frozenset()
-    )
+    out = diagnose_unknown_imports(tree=tree, module_path="x", project_modules=frozenset(), stdlib_keys=frozenset())
     assert len(out) == 1
 
 
@@ -196,7 +190,5 @@ def test_native_allowlist_prefix_boundary_is_dotted() -> None:
     # (wardline.core_helpers vs wardline.core) must NOT be suppressed — guards the
     # ``mod == p or mod.startswith(p + ".")`` boundary against a future ``+ "."`` drop.
     tree = ast.parse("from wardline.core_helpers import q\n")
-    out = diagnose_unknown_imports(
-        tree=tree, module_path="x", project_modules=frozenset(), stdlib_keys=frozenset()
-    )
+    out = diagnose_unknown_imports(tree=tree, module_path="x", project_modules=frozenset(), stdlib_keys=frozenset())
     assert len(out) == 1