From 8eb0f37ebb7494fc1420a669484ab1cd1b66bcdc Mon Sep 17 00:00:00 2001 From: John Morrissey <544926+tachyon-beep@users.noreply.github.com> Date: Fri, 5 Jun 2026 16:05:47 +1000 Subject: [PATCH 1/2] Fix lambda body sink discovery --- src/wardline/scanner/rules/_sink_helpers.py | 16 ++++++------ tests/unit/scanner/rules/test_sink_rules.py | 27 ++++++++++++++++++--- 2 files changed, 30 insertions(+), 13 deletions(-) diff --git a/src/wardline/scanner/rules/_sink_helpers.py b/src/wardline/scanner/rules/_sink_helpers.py index 81291304..602406cd 100644 --- a/src/wardline/scanner/rules/_sink_helpers.py +++ b/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) diff --git a/tests/unit/scanner/rules/test_sink_rules.py b/tests/unit/scanner/rules/test_sink_rules.py index ed32884b..5299a5a6 100644 --- a/tests/unit/scanner/rules/test_sink_rules.py +++ b/tests/unit/scanner/rules/test_sink_rules.py @@ -32,16 +32,21 @@ 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 +146,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, From 61e5f91f842adddee1c61a98b46a708c46bee7d6 Mon Sep 17 00:00:00 2001 From: John Morrissey Date: Fri, 5 Jun 2026 16:19:02 +1000 Subject: [PATCH 2/2] style: ruff-format sweep + CHANGELOG for lambda sink fix Apply the pinned ruff 0.15.15 format (sweeps up pre-existing main drift in diagnostics.py/test_diagnostics.py plus this PR's test file) so the branch is format-clean, and document the lambda-body sink-discovery fix under Security. Co-Authored-By: Claude Opus 4.8 (1M context) --- CHANGELOG.md | 12 ++++++++++++ src/wardline/scanner/diagnostics.py | 1 + tests/unit/scanner/rules/test_sink_rules.py | 4 +--- tests/unit/scanner/test_diagnostics.py | 16 ++++------------ 4 files changed, 18 insertions(+), 15 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 821aa9b9..73149d24 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/src/wardline/scanner/diagnostics.py b/src/wardline/scanner/diagnostics.py index 660bd9bd..323f60e7 100644 --- a/src/wardline/scanner/diagnostics.py +++ b/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), diff --git a/tests/unit/scanner/rules/test_sink_rules.py b/tests/unit/scanner/rules/test_sink_rules.py index 5299a5a6..8c9b016b 100644 --- a/tests/unit/scanner/rules/test_sink_rules.py +++ b/tests/unit/scanner/rules/test_sink_rules.py @@ -38,9 +38,7 @@ def test_sink_calls_include_lambda_body() -> None: (2, "wrapper"), (2, "eval"), ] - assert [(call.lineno, sink) for call, sink in sink_calls(func, frozenset({"eval"}), {}, "m")] == [ - (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: diff --git a/tests/unit/scanner/test_diagnostics.py b/tests/unit/scanner/test_diagnostics.py index c45dd733..b0ec4645 100644 --- a/tests/unit/scanner/test_diagnostics.py +++ b/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