From 25efdf583e143c029d0e2b7c2857259019c0b4a5 Mon Sep 17 00:00:00 2001 From: Zied Jlassi <6190550+zied-jlassi@users.noreply.github.com> Date: Tue, 23 Jun 2026 23:41:13 +0200 Subject: [PATCH 1/3] feat(behavioral): canonical sink resolver + shared evasion corpus fitness gate Introduce resolve_to_canonical_sink, a single chokepoint reducing every spelling of a dangerous callable -- bare/alias/from/as, builtins.*, importlib.import_module, reflective getattr("lit"), subscript __builtins__["exec"] and .__dict__["lit"], plus importlib/runpy/code sibling machinery -- to one canonical sink id. Wire it into behavioral_ast and behavioral_taint_tracking so the per-idiom ladders stay simple, with a shared parametrized evasion corpus asserting the invariant as a fitness gate: a future missed spelling fails a shared test instead of shipping as a silent blind spot. Reflective subprocess invocations grade AST9-HIGH for parity with reflective os.system; direct subprocess.* keeps baseline AST4-MEDIUM; AST9 and the canonical fallback stay complementary (the _covered_by_ast9 guard prevents double-firing). Benign neighbours never canonicalize to a sink. Stacked on the builtins/importlib sink-evasion fix (reuses resolve_dynamic_import_call); to be rebased onto main once that lands. 108 evasion-corpus tests pass; full suite +108 with zero regressions (ruff 0.15.2 + mypy clean). Signed-off-by: Zied Jlassi <6190550+zied-jlassi@users.noreply.github.com> --- .../nodes/analyzers/behavioral_ast.py | 58 +++- .../analyzers/behavioral_taint_tracking.py | 20 +- .../nodes/analyzers/canonical_sink.py | 289 ++++++++++++++++++ tests/evasion_corpus/__init__.py | 14 + tests/evasion_corpus/corpus.py | 149 +++++++++ .../test_canonical_sink_fitness.py | 209 +++++++++++++ 6 files changed, 728 insertions(+), 11 deletions(-) create mode 100644 src/skillspector/nodes/analyzers/canonical_sink.py create mode 100644 tests/evasion_corpus/__init__.py create mode 100644 tests/evasion_corpus/corpus.py create mode 100644 tests/evasion_corpus/test_canonical_sink_fitness.py diff --git a/src/skillspector/nodes/analyzers/behavioral_ast.py b/src/skillspector/nodes/analyzers/behavioral_ast.py index 78e895221..6d2070869 100644 --- a/src/skillspector/nodes/analyzers/behavioral_ast.py +++ b/src/skillspector/nodes/analyzers/behavioral_ast.py @@ -44,7 +44,13 @@ transitive_remaining_seconds, ) +from .canonical_sink import ( + canonical_sibling_method, + canonical_sibling_sink, + resolve_to_canonical_sink, +) from .common import ( + build_type_map, get_complete_source_segment, get_context_from_lines, resolve_call_name, @@ -291,6 +297,29 @@ def analyzer_exhausted(self) -> bool: return self.total_findings >= MAX_FINDINGS_PER_ANALYZER +def _covered_by_ast9(node: ast.Call) -> bool: + """True when *node* invokes a literal ``getattr`` already flagged by AST9. + + AST9 catches ``getattr(obj, "")(...)`` for ```` in + :data:`_DANGEROUS_GETATTR_NAMES`. The canonical-sink fallback skips exactly those so + the two rules stay complementary rather than double-firing: the canonical layer then + owns only the spellings AST9 does not (subscript ``__builtins__["exec"]`` / + ``vars(builtins)["exec"]`` and getattr targets outside the allowlist such as + ``getattr(subprocess, "Popen")``). + """ + callee = node.func + if not (isinstance(callee, ast.Call) and isinstance(callee.func, ast.Name)): + return False + if callee.func.id != "getattr" or len(callee.args) < 2: + return False + attr = callee.args[1] + return ( + isinstance(attr, ast.Constant) + and isinstance(attr.value, str) + and attr.value in _DANGEROUS_GETATTR_NAMES + ) + + def _is_chain_sink(node: ast.Call, aliases: dict[str, str] | None = None) -> bool: """True if this call is exec(), eval(), or compile() — the outer dangerous call.""" name = resolve_call_name(node, aliases) @@ -429,6 +458,7 @@ def _analyze_python( return [] aliases = python_ast.import_aliases + type_map = build_type_map(tree, aliases) lines = python_ast.lines findings: list[AnalyzerFinding] = [] contexts: dict[tuple[int, int], str] = {} @@ -544,10 +574,32 @@ def _emit( continue call_name = resolve_call_name(ast_node, aliases) + if call_name is not None: + # Dynamic-import / code-exec sibling machinery (importlib.__import__, + # importlib.util.find_spec, runpy.run_module, code.interact, and the + # instance-method tails spec.loader.exec_module / runsource) resolves by name + # but matches no ladder; remap it to the primitive it equals so it re-enters + # the __import__/exec arms below. + call_name = ( + canonical_sibling_sink(call_name) + or canonical_sibling_method(ast_node, type_map) + or call_name + ) if call_name is None: # Dynamic-import chain: importlib.import_module('os').system(...) → # 'os.system', so it re-enters the os./subprocess. sink ladders below. call_name = resolve_dynamic_import_call(ast_node, aliases) + reflective = False + if call_name is None and not _covered_by_ast9(ast_node): + # Reflective invocation whose callee does not resolve by name: + # getattr(subprocess, "Popen")(cmd) (outside AST9's allowlist), + # __builtins__["exec"](src), vars(builtins)["exec"](src). Canonicalize to + # the bare/dotted sink id so it re-enters the ladders below with the correct + # rule + severity. Literal getattr() on an AST9-allowlisted name is left to + # AST9 (see _covered_by_ast9) so the two rules stay complementary, not + # duplicate. + call_name = resolve_to_canonical_sink(ast_node, aliases) + reflective = call_name is not None if call_name is None: continue @@ -586,7 +638,11 @@ def _emit( elif call_name.startswith("subprocess."): attr = call_name.split(".", 1)[1] if attr in _SUBPROCESS_CALLS: - _emit(node_index, "AST4", ast_node) + # A *reflective* subprocess invocation (getattr/subscript) signals the + # same evasion intent as reflective os.system, so it grades AST9-HIGH to + # stay consistent with that case; a direct subprocess.* call keeps its + # baseline AST4-MEDIUM. + _emit(node_index, "AST9" if reflective else "AST4", ast_node) elif call_name.startswith("os."): attr = call_name.split(".", 1)[1] diff --git a/src/skillspector/nodes/analyzers/behavioral_taint_tracking.py b/src/skillspector/nodes/analyzers/behavioral_taint_tracking.py index 4645618b8..2311c72c0 100644 --- a/src/skillspector/nodes/analyzers/behavioral_taint_tracking.py +++ b/src/skillspector/nodes/analyzers/behavioral_taint_tracking.py @@ -53,6 +53,7 @@ transitive_remaining_seconds, ) +from .canonical_sink import resolve_to_canonical_sink from .common import ( apply_import_aliases, build_type_map, @@ -60,7 +61,6 @@ get_context_from_lines, resolve_call_name_typed, resolve_dotted_name, - resolve_dynamic_import_call, ) from .static_runner import ( MAX_FILE_CHARS, @@ -299,12 +299,13 @@ def _resolve_sink_name( type_map: dict[str, str] | None = None, aliases: dict[str, str] | None = None, ) -> str | None: - """Resolve a call to its canonical sink name, including dynamic-import chains. + """Resolve a call to its canonical sink name through the canonicalization chokepoint. - Wraps :func:`resolve_call_name_typed` (type-/alias-aware resolution) and falls back - to :func:`resolve_dynamic_import_call` so that - ``importlib.import_module('subprocess').run(...)`` resolves to ``'subprocess.run'`` - and re-enters ``_EXEC_SINKS`` like the statically-imported form would. + Delegates to :func:`resolve_to_canonical_sink`, which reduces every spelling — + type-/alias-aware names, ``importlib.import_module('subprocess').run(...)``, + reflective ``getattr(os, "system")(...)`` and subscript ``__builtins__["exec"](...)`` + / ``vars(builtins)["exec"](...)`` — to the bare/dotted id so it re-enters + ``_EXEC_SINKS`` like the statically-imported form would. """ if ( isinstance(node.func, ast.Attribute) @@ -313,11 +314,10 @@ def _resolve_sink_name( and resolve_call_name_typed(node.func.value, type_map, aliases) == "urllib.request.build_opener" ): + # ``build_opener(...).open(url)`` has no dotted name of its own; keep this + # special case ahead of the chokepoint so it still maps to its network sink. return "urllib.request.build_opener.open" - name = resolve_call_name_typed(node, type_map, aliases) - if name is None: - name = resolve_dynamic_import_call(node, aliases) - return name + return resolve_to_canonical_sink(node, aliases, type_map) def _classify(name: str, categories: list[tuple[frozenset[str], str]], default: str) -> str: diff --git a/src/skillspector/nodes/analyzers/canonical_sink.py b/src/skillspector/nodes/analyzers/canonical_sink.py new file mode 100644 index 000000000..c79f63ef1 --- /dev/null +++ b/src/skillspector/nodes/analyzers/canonical_sink.py @@ -0,0 +1,289 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Single canonicalization chokepoint for dangerous-callable resolution. + +Every dangerous primitive can be spelled many equivalent ways — bare name, import +alias, ``from`` / ``as`` rebinding, ``builtins.*`` qualification, dynamic import via +``importlib.import_module``, reflective ``getattr(obj, "lit")`` and subscript +reflection ``__builtins__["exec"]``. Detectors that enumerate one idiom at a time +(``import os as o`` here, ``getattr`` there, ``builtins.exec`` elsewhere) leak a new +blind spot with every missed spelling — the recurring regressions behind #114/#115, +#166 and the builtins/importlib gap. + +:func:`resolve_to_canonical_sink` collapses *all* of those spellings to one canonical +sink id (the bare/dotted name the sink sets already match: ``"exec"``, ``"os.system"``, +``"subprocess.run"``). Resolving once, here, lets the sink ladders stay simple and lets +a shared evasion corpus assert the invariant for every new sink. +""" + +from __future__ import annotations + +import ast + +from .common import ( + apply_import_aliases, + resolve_call_name_typed, + resolve_dotted_name, + resolve_dynamic_import_call, +) + + +def _canonical_from_getattr(node: ast.Call, aliases: dict[str, str] | None) -> str | None: + """Resolve ``getattr(obj, "attr")(...)`` to the dotted sink ``".attr"``. + + The dangerous invocation is the *outer* call, so this inspects ``node.func``: when + the callee is itself ``getattr(os, "system")`` the result is ``"os.system"``, and + ``getattr(builtins, "exec")`` collapses to ``"exec"`` (the ``builtins`` prefix is + stripped by :func:`apply_import_aliases`). Only a *constant string* attribute is + resolved; a non-literal attribute is dynamic and left to the caller's + reflective-access rule. Returns ``None`` when ``node.func`` is not a literal + ``getattr`` call. + """ + callee = node.func + if not isinstance(callee, ast.Call): + return None + func_name = resolve_dotted_name(callee.func) + if func_name is not None and aliases: + func_name = apply_import_aliases(func_name, aliases) + if func_name != "getattr" or len(callee.args) < 2: + return None + obj, attr = callee.args[0], callee.args[1] + if not (isinstance(attr, ast.Constant) and isinstance(attr.value, str)): + return None + base = resolve_dotted_name(obj) + if base is None: + return None + if aliases: + base = apply_import_aliases(base, aliases) + if base == "builtins": + return attr.value + return f"{base}.{attr.value}" + + +def _namespace_module(base: ast.expr, aliases: dict[str, str] | None) -> str | None: + """Resolve a subscript base to the module whose namespace dict it exposes. + + Handles the three equivalent namespace handles used to reach a module's callables + by string key: + + - ``__builtins__`` / ``builtins`` → ``"builtins"`` + - ``vars(builtins)`` → ``"builtins"`` (and ``vars(os)`` → ``"os"``) + - ``os.__dict__`` / ``builtins.__dict__`` → ``"os"`` / ``"builtins"`` + + Returns the resolved module name, or ``None`` when the base is not a recognized + namespace handle. + """ + base_name = resolve_dotted_name(base) + if base_name is None and isinstance(base, ast.Call): + # vars() namespace handle. + inner = resolve_dotted_name(base.func) + if inner == "vars" and base.args: + arg = resolve_dotted_name(base.args[0]) + if aliases and arg is not None: + arg = apply_import_aliases(arg, aliases) + return arg + return None + if base_name is None: + return None + if base_name.endswith(".__dict__"): + # ``.__dict__`` exposes the module namespace by string key. Strip the + # trailing ``.__dict__`` *before* alias normalization so the builtins-collapse + # in ``apply_import_aliases`` (which turns ``builtins.__dict__`` into + # ``__dict__``) does not erase the module name. + module = base_name[: -len(".__dict__")] + if aliases: + module = apply_import_aliases(module, aliases) + return "builtins" if module in ("__builtins__", "builtins") else module + # Bare ``__builtins__`` / ``builtins`` are the only other namespace handles; any + # other plain name (a user dict such as ``handlers["exec"]``) is NOT a namespace. + if aliases: + base_name = apply_import_aliases(base_name, aliases) + if base_name in ("__builtins__", "builtins"): + return "builtins" + return None + + +def _canonical_from_subscript(node: ast.Call, aliases: dict[str, str] | None) -> str | None: + """Resolve namespace-dict reflection like ``os.__dict__["system"](...)`` to its sink. + + Recognizes the dict-reflection idiom where ``node.func`` is a subscript whose base is + a module namespace handle (``__builtins__`` / ``vars(builtins)`` / ``os.__dict__``) + and whose index is a string literal: ``__builtins__["exec"]`` → ``"exec"``, + ``os.__dict__["system"]`` → ``"os.system"``. Returns the canonical id, or ``None`` + for any other subscript so a plain user dict (``handlers["exec"]``) is never a sink. + """ + func = node.func + if not isinstance(func, ast.Subscript): + return None + key = func.slice + if not (isinstance(key, ast.Constant) and isinstance(key.value, str)): + return None + module = _namespace_module(func.value, aliases) + if module is None: + return None + if module == "builtins": + return key.value + return f"{module}.{key.value}" + + +# Sibling machinery whose only purpose is dynamic import / code execution. Each maps to +# the canonical sink id of the existing primitive it is equivalent to, so it re-enters +# the established sink ladders rather than introducing a parallel severity scheme: +# +# - ``importlib.__import__`` is literally the ``__import__`` builtin (dynamic import). +# - ``importlib.util.{find_spec,module_from_spec}`` + ``runpy.run_module/run_path`` +# load/execute a module by name → dynamic-import class (``"__import__"``). +# - ``code.interact`` executes arbitrary source → ``exec`` class. +_SIBLING_SINKS: dict[str, str] = { + "importlib.__import__": "__import__", + "importlib.util.find_spec": "__import__", + "importlib.util.module_from_spec": "__import__", + "importlib.util.spec_from_loader": "__import__", + "runpy.run_module": "__import__", + "runpy.run_path": "__import__", + "code.interact": "exec", +} + +# Instance-method tails matched on their canonical code-exec sink, but only when the +# *receiver* statically resolves to the relevant machinery (see :func:`canonical_sibling_method`): +# a loader/interpreter method is typically called on an instance whose class does not +# resolve, so the receiver must be tied to importlib / the ``code`` module to avoid +# flagging an unrelated user method that merely shares the name. +_SIBLING_METHOD_TAILS: dict[str, str] = { + "exec_module": "__import__", + "runsource": "exec", + "runcode": "exec", +} + +# Module roots whose namespace owns the gated instance methods. +_IMPORTLIB_ROOTS: tuple[str, ...] = ("importlib", "importlib.util", "importlib.machinery") +_CODE_INTERPRETER_TYPES: tuple[str, ...] = ( + "code.InteractiveInterpreter", + "code.InteractiveConsole", +) + + +def canonical_sibling_sink(name: str | None) -> str | None: + """Map a resolved sibling-machinery name to the canonical sink it is equivalent to. + + Matches the fully-qualified spelling (``importlib.util.find_spec``); the bare + instance-method tail (``exec_module``, ``runsource``) is resolved separately, with a + receiver check, via :func:`canonical_sibling_method`. Returns the canonical id + (``"__import__"`` / ``"exec"``) or ``None`` when *name* is not a known sibling. + """ + if name is None: + return None + return _SIBLING_SINKS.get(name) + + +def _receiver_is_importlib_loader(receiver: ast.expr, type_map: dict[str, str] | None) -> bool: + """True when *receiver* statically resolves to an importlib spec loader. + + Recognizes the documented ``importlib`` spec-loader chain: ``.loader`` (the + ``ModuleSpec.loader`` attribute, including ``importlib.util.module_from_spec(...)`` + results carried through *type_map*), and bare names whose inferred type is rooted in + ``importlib``. A user object's ``.exec_module`` (receiver not tied to importlib) does + not match. + """ + if isinstance(receiver, ast.Attribute): + # ``.loader`` — the ModuleSpec.loader protocol attribute. + if receiver.attr == "loader": + return True + inferred = None + if isinstance(receiver, ast.Name) and type_map is not None: + inferred = type_map.get(receiver.id) + if inferred is None: + inferred = resolve_dotted_name(receiver) + return bool(inferred and inferred.split(".", 1)[0] == "importlib") + + +def _receiver_is_code_interpreter(receiver: ast.expr, type_map: dict[str, str] | None) -> bool: + """True when *receiver* resolves to ``code.Interactive{Interpreter,Console}``. + + Matches a direct construction (``code.InteractiveInterpreter().runsource(...)``) and a + variable whose inferred constructor type is one of those classes (carried through + *type_map*). A user object defining ``runsource`` / ``runcode`` does not match. + """ + inferred = None + if isinstance(receiver, ast.Call): + inferred = resolve_dotted_name(receiver.func) + elif isinstance(receiver, ast.Name) and type_map is not None: + inferred = type_map.get(receiver.id) + return inferred in _CODE_INTERPRETER_TYPES + + +def canonical_sibling_method(node: ast.Call, type_map: dict[str, str] | None = None) -> str | None: + """Map a gated instance-method call to its canonical code-exec sink id. + + Fires only when *node* is ``.exec_module(...)`` / ``.runsource(...)`` / + ``.runcode(...)`` **and** *recv* statically resolves to the matching machinery — + importlib spec loaders for ``exec_module`` (:func:`_receiver_is_importlib_loader`), + ``code.Interactive{Interpreter,Console}`` for ``runsource`` / ``runcode`` + (:func:`_receiver_is_code_interpreter`). ``exec_module`` → ``"__import__"``, + ``runsource`` / ``runcode`` → ``"exec"``. Returns ``None`` otherwise, so an unrelated + user method sharing the name stays unflagged (documented def-use residual). + """ + func = node.func + if not isinstance(func, ast.Attribute): + return None + canonical = _SIBLING_METHOD_TAILS.get(func.attr) + if canonical is None: + return None + if func.attr == "exec_module": + return canonical if _receiver_is_importlib_loader(func.value, type_map) else None + return canonical if _receiver_is_code_interpreter(func.value, type_map) else None + + +def resolve_to_canonical_sink( + node: ast.Call, + aliases: dict[str, str] | None = None, + type_map: dict[str, str] | None = None, +) -> str | None: + """Reduce any spelling of a call to its canonical sink id. + + Tries, in order of precedence: + + 1. Direct/alias/``from``/``as``/``builtins.*`` resolution via + :func:`resolve_call_name_typed` (also consults *type_map* for instance methods), + then maps dynamic-import / code-exec sibling machinery to the primitive it equals. + 2. Dynamic import chains ``importlib.import_module("os").system`` via + :func:`resolve_dynamic_import_call`. + 3. Reflective ``getattr(obj, "lit")`` via :func:`_canonical_from_getattr`. + 4. Subscript reflection ``__builtins__["exec"]`` / ``os.__dict__["system"]`` via + :func:`_canonical_from_subscript`. + + Returns the canonical id (``"exec"``, ``"os.system"``, …) or ``None`` when the call + cannot be reduced statically (e.g. a non-literal ``getattr`` attribute), which the + caller may still flag through its generic reflective-access rule. + """ + name = resolve_call_name_typed(node, type_map, aliases) + if name is not None: + sibling = canonical_sibling_sink(name) or canonical_sibling_method(node, type_map) + return sibling or name + dynamic = resolve_dynamic_import_call(node, aliases) + if dynamic is not None: + return dynamic + reflective = _canonical_from_getattr(node, aliases) + if reflective is not None: + return reflective + subscript = _canonical_from_subscript(node, aliases) + if subscript is not None: + return subscript + # Instance-method machinery whose receiver does not resolve to a name, e.g. + # ``code.InteractiveInterpreter().runsource(src)`` — match the gated method (the + # receiver must resolve to the interpreter/loader machinery) so the sibling is still + # recognized as a code-exec sink without flagging unrelated same-named user methods. + return canonical_sibling_method(node, type_map) diff --git a/tests/evasion_corpus/__init__.py b/tests/evasion_corpus/__init__.py new file mode 100644 index 000000000..467079831 --- /dev/null +++ b/tests/evasion_corpus/__init__.py @@ -0,0 +1,14 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. diff --git a/tests/evasion_corpus/corpus.py b/tests/evasion_corpus/corpus.py new file mode 100644 index 000000000..aec414416 --- /dev/null +++ b/tests/evasion_corpus/corpus.py @@ -0,0 +1,149 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Shared evasion corpus: equivalent spellings of dangerous sinks + FP neighbours. + +A single source of truth that *every* sink detector must satisfy. ``EQUIVALENT_SPELLINGS`` +lists, per canonical sink id, ≥8 semantically equivalent ways to invoke it (bare name, +import alias, ``from``/``as`` rebinding, ``builtins.*``, ``importlib.import_module``, +``getattr``, subscript reflection). ``FALSE_POSITIVE_NEIGHBOURS`` lists benign code that +merely *resembles* a sink and must never canonicalize to one. New detectors import these +and assert the invariant, so a future regression (a missed spelling) fails a shared gate +rather than shipping as a silent blind spot. +""" + +from __future__ import annotations + +from typing import NamedTuple + + +class Spelling(NamedTuple): + """One concrete way to write a call, with the canonical sink id it must reduce to.""" + + label: str + code: str + canonical: str + + +class Neighbour(NamedTuple): + """A benign call that resembles a sink and must NOT reduce to any dangerous sink id.""" + + label: str + code: str + + +# Each row's ``code`` ends with the sink invocation as the final top-level statement. +EQUIVALENT_SPELLINGS: tuple[Spelling, ...] = ( + # ── exec (bare builtin) ─────────────────────────────────────────── + Spelling("exec_bare", "exec(x)", "exec"), + Spelling("exec_from_builtins", "from builtins import exec\nexec(x)", "exec"), + Spelling("exec_builtins_attr", "import builtins\nbuiltins.exec(x)", "exec"), + Spelling("exec_builtins_alias", "import builtins as b\nb.exec(x)", "exec"), + Spelling("exec_from_builtins_as", "from builtins import exec as e\ne(x)", "exec"), + Spelling("exec_getattr", 'import builtins\ngetattr(builtins, "exec")(x)', "exec"), + Spelling("exec_subscript_dunder", '__builtins__["exec"](x)', "exec"), + Spelling("exec_subscript_vars", 'import builtins\nvars(builtins)["exec"](x)', "exec"), + Spelling("exec_builtins_dunder_dict", 'import builtins\nbuiltins.__dict__["exec"](x)', "exec"), + # ── os.system (dotted module sink) ──────────────────────────────── + Spelling("os_system_direct", "import os\nos.system(x)", "os.system"), + Spelling("os_system_alias", "import os as o\no.system(x)", "os.system"), + Spelling("os_system_from", "from os import system\nsystem(x)", "os.system"), + Spelling("os_system_getattr", 'import os\ngetattr(os, "system")(x)', "os.system"), + Spelling("os_system_dunder_dict", 'import os\nos.__dict__["system"](x)', "os.system"), + Spelling( + "os_system_importlib", + 'import importlib\nimportlib.import_module("os").system(x)', + "os.system", + ), + Spelling( + "os_system_import_module_bare", + 'from importlib import import_module\nimport_module("os").system(x)', + "os.system", + ), + Spelling( + "os_system_importlib_alias", + 'import importlib as il\nil.import_module("os").system(x)', + "os.system", + ), + # ── subprocess.run (dotted module sink) ─────────────────────────── + Spelling("subprocess_run_direct", "import subprocess\nsubprocess.run(x)", "subprocess.run"), + Spelling("subprocess_run_alias", "import subprocess as sp\nsp.run(x)", "subprocess.run"), + Spelling("subprocess_run_from", "from subprocess import run\nrun(x)", "subprocess.run"), + Spelling( + "subprocess_run_getattr", + 'import subprocess\ngetattr(subprocess, "run")(x)', + "subprocess.run", + ), + Spelling( + "subprocess_run_importlib", + 'import importlib\nimportlib.import_module("subprocess").run(x)', + "subprocess.run", + ), + # ── dynamic-import siblings (→ __import__ class) ────────────────── + Spelling( + "importlib_dunder_import", + 'import importlib\nimportlib.__import__("os")', + "__import__", + ), + Spelling( + "importlib_find_spec", + 'import importlib.util\nimportlib.util.find_spec("os")', + "__import__", + ), + Spelling( + "importlib_module_from_spec", + "import importlib.util\nimportlib.util.module_from_spec(spec)", + "__import__", + ), + Spelling("runpy_run_module", 'import runpy\nrunpy.run_module("os")', "__import__"), + Spelling( + "spec_loader_exec_module", + "import importlib.util\n" + "spec = importlib.util.module_from_spec(s)\n" + "spec.loader.exec_module(mod)", + "__import__", + ), + # ── code-exec siblings (→ exec class) ───────────────────────────── + Spelling("code_interact", "import code\ncode.interact()", "exec"), + Spelling( + "code_runsource", + "import code\ncode.InteractiveInterpreter().runsource(x)", + "exec", + ), +) + +# Benign code that resembles a sink; must canonicalize to something OUTSIDE the sink sets. +FALSE_POSITIVE_NEIGHBOURS: tuple[Neighbour, ...] = ( + Neighbour("user_exec_helper", "from mymod import exec_helper\nexec_helper()"), + Neighbour("getattr_benign_attr", 'val = getattr(cfg, "timeout")'), + Neighbour( + "importlib_benign_loads", + 'import importlib\nimportlib.import_module("json").loads(x)', + ), + Neighbour("subscript_user_dict", 'handlers = {}\nhandlers["exec"](x)'), + Neighbour("getattr_non_dangerous_module", 'import os\ngetattr(os, "getcwd")()'), + Neighbour("dunder_dict_benign_attr", 'import os\nos.__dict__["getcwd"]()'), + Neighbour( + "importlib_spec_from_file_benign", + 'import importlib.util\nimportlib.util.spec_from_file_location("m", "/x.py")', + ), + Neighbour("user_object_dunder_dict", 'obj = Thing()\nobj.__dict__["handler"](x)'), + Neighbour( + "user_class_exec_module", + "class L:\n def exec_module(self, m):\n pass\n\n\nL().exec_module(1)", + ), + Neighbour("user_runsource_method", "ci = MyRepl()\nci.runsource(x)"), + Neighbour("user_loader_unrelated", "loader = MyLoader()\nloader.exec_module(m)"), +) diff --git a/tests/evasion_corpus/test_canonical_sink_fitness.py b/tests/evasion_corpus/test_canonical_sink_fitness.py new file mode 100644 index 000000000..32b5f2de2 --- /dev/null +++ b/tests/evasion_corpus/test_canonical_sink_fitness.py @@ -0,0 +1,209 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Fitness gate: the canonical layer maps every evasion spelling to one sink id. + +This is the structural counter-measure to the "enumerate instead of canonicalize" +anti-pattern. Rather than one test per idiom, the shared corpus asserts that *every* +equivalent spelling reduces to the same canonical id through the single chokepoint, and +that benign neighbours never do. Adding a new sink means adding corpus rows, not a new +detector branch. +""" + +from __future__ import annotations + +import ast + +import pytest + +from skillspector.nodes.analyzers import behavioral_ast, behavioral_taint_tracking +from skillspector.nodes.analyzers.behavioral_taint_tracking import _EXEC_SINKS +from skillspector.nodes.analyzers.canonical_sink import resolve_to_canonical_sink +from skillspector.nodes.analyzers.common import build_import_aliases, build_type_map + +from .corpus import EQUIVALENT_SPELLINGS, FALSE_POSITIVE_NEIGHBOURS, Neighbour, Spelling + + +def _outermost_call(code: str) -> ast.Call: + """Return the outermost sink-invocation call from a corpus snippet. + + Corpus rows place the sink call as the last top-level ``Expr``/``Assign`` value; + this returns that call node so the chokepoint is queried on the actual invocation + (e.g. ``getattr(os, "system")(x)`` resolves on the outer ``(x)`` call). + """ + tree = ast.parse(code) + last: ast.Call | None = None + for stmt in tree.body: + value = stmt.value if isinstance(stmt, (ast.Expr, ast.Assign)) else None + if isinstance(value, ast.Call): + last = value + if last is None: # pragma: no cover - corpus rows always end in a call + raise AssertionError(f"corpus snippet has no top-level call: {code!r}") + return last + + +def _canonical_for(code: str) -> str | None: + """Resolve a corpus snippet's outermost call to its canonical sink id.""" + tree = ast.parse(code) + aliases = build_import_aliases(tree) + type_map = build_type_map(tree) + return resolve_to_canonical_sink(_outermost_call(code), aliases, type_map) + + +@pytest.mark.parametrize("row", EQUIVALENT_SPELLINGS, ids=lambda r: r.label) +def test_equivalent_spelling_canonicalizes(row: Spelling) -> None: + """Every equivalent spelling reduces to its single canonical sink id.""" + assert _canonical_for(row.code) == row.canonical + + +@pytest.mark.parametrize("neighbour", FALSE_POSITIVE_NEIGHBOURS, ids=lambda n: n.label) +def test_false_positive_neighbour_not_a_sink(neighbour: Neighbour) -> None: + """Benign look-alikes never canonicalize to a dangerous exec sink id.""" + assert _canonical_for(neighbour.code) not in _EXEC_SINKS + + +def test_corpus_has_min_spellings_per_primitive() -> None: + """Each canonical primitive carries at least eight equivalent spellings. + + Guards the corpus against silently shrinking below the breadth that makes it a + meaningful evasion gate. + """ + counts: dict[str, int] = {} + for row in EQUIVALENT_SPELLINGS: + counts[row.canonical] = counts.get(row.canonical, 0) + 1 + assert counts.get("exec", 0) >= 8 + assert all(count >= 5 for count in counts.values()) + # The sibling-machinery surface (dynamic import / code exec) is represented. + assert "__import__" in counts + + +# ── End-to-end wiring gate ──────────────────────────────────────────── +# +# The fitness checks above prove the resolver *can* canonicalize. These prove the +# resolver is actually *wired into production*: every spelling must reach a real sink +# rule (AST + taint), so an accidental un-wiring or severity downgrade fails loudly +# rather than silently regressing to a missed detection. + +# behavioral_ast rules that constitute a real dangerous-execution sink detection. A +# spelling is "detected" if it produces any of these — the precise rule depends on which +# complementary path fires (the canonical ladders AST1/AST4/AST5/AST6, or AST9 for a +# literal getattr on an allowlisted name, owned by the getattr branch on `main`). +_DANGEROUS_EXECUTION_RULES: frozenset[str] = frozenset( + {"AST1", "AST2", "AST3", "AST4", "AST5", "AST6", "AST9"} +) + +# Minimum severity each canonical primitive must keep — guards against a downgrade +# (e.g. an exec sink silently dropping to AST3-MEDIUM). ``__import__`` is the +# dynamic-import class (AST3-MEDIUM), matching the repo's grading of ``__import__``. +_MIN_SEVERITY: dict[str, str] = { + "exec": "HIGH", + "os.system": "HIGH", + "subprocess.run": "MEDIUM", + "__import__": "MEDIUM", +} + +_SEVERITY_ORDER: dict[str, int] = {"LOW": 0, "MEDIUM": 1, "HIGH": 2, "CRITICAL": 3} + +# Canonical ids the taint analyzer treats as exec sinks (TT5). ``__import__`` is +# deliberately excluded: the repo does not grade bare ``__import__`` as a taint exec +# sink, so its dynamic-import siblings stay consistent with that baseline. +_TAINT_EXEC_CANONICALS: frozenset[str] = frozenset( + {"exec", "eval", "compile", "os.system", "subprocess.run"} +) + + +def _run_ast(code: str) -> list: + """Run the real behavioral_ast analyzer on a one-file skill and return findings.""" + state = {"components": ["s.py"], "file_cache": {"s.py": code}} + return behavioral_ast.node(state)["findings"] + + +def _run_taint(code: str) -> list: + """Run the real behavioral_taint_tracking analyzer; *code* must end in a tainted flow.""" + state = {"components": ["s.py"], "file_cache": {"s.py": code}} + return behavioral_taint_tracking.node(state)["findings"] + + +@pytest.mark.parametrize("row", EQUIVALENT_SPELLINGS, ids=lambda r: r.label) +def test_spelling_reaches_ast_sink_rule(row: Spelling) -> None: + """Every spelling reaches a dangerous-execution rule in production at full severity. + + Proves the chokepoint is actually invoked by production (not merely importable) and + that the spelling is not silently dropped or downgraded. The exact rule may be a + canonical-ladder rule or AST9 (literal getattr, owned by the getattr branch on + ``main``) — both are valid sink detections — so the assertion targets the + dangerous-execution rule set plus a minimum-severity floor. + """ + findings = _run_ast(row.code) + sink_findings = [f for f in findings if f.rule_id in _DANGEROUS_EXECUTION_RULES] + assert sink_findings, f"{row.label}: no sink rule, got {[f.rule_id for f in findings]}" + best = max(_SEVERITY_ORDER[f.severity] for f in sink_findings) + assert best >= _SEVERITY_ORDER[_MIN_SEVERITY[row.canonical]], ( + f"{row.label}: severity downgrade, got {[(f.rule_id, f.severity) for f in sink_findings]}" + ) + + +@pytest.mark.parametrize( + "row", + [r for r in EQUIVALENT_SPELLINGS if r.canonical in _TAINT_EXEC_CANONICALS and "(x)" in r.code], + ids=lambda r: r.label, +) +def test_spelling_reaches_taint_exec_sink(row: Spelling) -> None: + """Routed through a tainted input, every exec-class spelling reaches the TT5 flow. + + Restricted to exec-class canonicals whose sink receives the tainted value ``x``; + ``__import__`` siblings are excluded (the taint analyzer does not grade dynamic + import as an exec sink — baseline parity), as are arg-less sinks like + ``code.interact()`` that cannot carry a tainted argument. + """ + tainted = f"x = input()\n{row.code}" + findings = _run_taint(tainted) + assert any(f.rule_id == "TT5" for f in findings), ( + f"{row.label}: expected TT5, got {[f.rule_id for f in findings]}" + ) + + +# Reflective subprocess invocations must grade HIGH (AST9), consistent with reflective +# os.system — the reflection itself signals evasion intent, unlike a direct AST4-MEDIUM +# subprocess.* call. +_SUBPROCESS_REFLECTION: tuple[str, ...] = ( + 'import subprocess\ngetattr(subprocess, "Popen")(x)', + 'import subprocess\ngetattr(subprocess, "run")(x)', + 'import subprocess\ngetattr(subprocess, "check_output")(x)', +) + + +@pytest.mark.parametrize("code", _SUBPROCESS_REFLECTION) +def test_reflective_subprocess_grades_high(code: str) -> None: + """Reflective subprocess sinks fire AST9-HIGH (not AST4-MEDIUM) for severity parity.""" + findings = _run_ast(code) + ast9 = [f for f in findings if f.rule_id == "AST9"] + assert ast9, f"expected AST9, got {[(f.rule_id, f.severity) for f in findings]}" + assert ast9[0].severity == "HIGH" + assert not any(f.rule_id == "AST4" for f in findings), "should not double-fire AST4" + + +def test_direct_subprocess_stays_medium() -> None: + """A *direct* subprocess.* call keeps its baseline AST4-MEDIUM (no over-escalation).""" + findings = _run_ast("import subprocess\nsubprocess.run(['id'])") + ast4 = [f for f in findings if f.rule_id == "AST4"] + assert ast4 and ast4[0].severity == "MEDIUM" + assert not any(f.rule_id == "AST9" for f in findings) + + +@pytest.mark.parametrize("neighbour", FALSE_POSITIVE_NEIGHBOURS, ids=lambda n: n.label) +def test_neighbour_produces_no_ast_findings(neighbour: Neighbour) -> None: + """Benign look-alikes produce zero behavioral_ast findings in production.""" + assert _run_ast(neighbour.code) == [] From 0c41192c6853ade9ccb5132bd575930b1b671f44 Mon Sep 17 00:00:00 2001 From: Zied Jlassi <6190550+zied-jlassi@users.noreply.github.com> Date: Tue, 30 Jun 2026 21:41:06 +0200 Subject: [PATCH 2/3] fix(canonical): gate exec_module on importlib provenance + alias-normalize code interpreter Addresses the #182 re-review: two resolver precision bugs. exec_module false positive: the `.loader` branch matched ANY object exposing a `.loader` attribute, so an unrelated `self.widget.loader.exec_module(...)` was wrongly classified as a dynamic import. The base must now have importlib provenance -- a spec from importlib.util.module_from_spec / spec_from_loader carried through the type map, or an importlib-rooted dotted name -- via the new `_expr_is_importlib_rooted` helper. The legitimate `spec.loader.exec_module(mod)` chain still fires. code-interpreter false negative: `_receiver_is_code_interpreter` never alias-normalized the inferred constructor type, so `import code as c; c.InteractiveInterpreter().runsource(...)` evaded detection. The inferred name now runs through `apply_import_aliases`, which also covers `from code import InteractiveInterpreter`. Production wiring: `behavioral_ast` now threads import aliases into `canonical_sibling_method` (the taint analyzer already did via `resolve_to_canonical_sink`), so both detection paths apply the gate. Corpus: +2 positive rows (aliased code interpreter, aliased importlib spec loader) and +1 false-positive neighbour (arbitrary `.loader.exec_module`), all exercised end-to-end through the real analyzers as a red->green oracle. Signed-off-by: Zied Jlassi <6190550+zied-jlassi@users.noreply.github.com> --- .../nodes/analyzers/behavioral_ast.py | 2 +- .../nodes/analyzers/canonical_sink.py | 91 +++++++++++++------ tests/evasion_corpus/corpus.py | 18 ++++ 3 files changed, 84 insertions(+), 27 deletions(-) diff --git a/src/skillspector/nodes/analyzers/behavioral_ast.py b/src/skillspector/nodes/analyzers/behavioral_ast.py index 6d2070869..6da807df3 100644 --- a/src/skillspector/nodes/analyzers/behavioral_ast.py +++ b/src/skillspector/nodes/analyzers/behavioral_ast.py @@ -582,7 +582,7 @@ def _emit( # the __import__/exec arms below. call_name = ( canonical_sibling_sink(call_name) - or canonical_sibling_method(ast_node, type_map) + or canonical_sibling_method(ast_node, type_map, aliases) or call_name ) if call_name is None: diff --git a/src/skillspector/nodes/analyzers/canonical_sink.py b/src/skillspector/nodes/analyzers/canonical_sink.py index c79f63ef1..dedf39793 100644 --- a/src/skillspector/nodes/analyzers/canonical_sink.py +++ b/src/skillspector/nodes/analyzers/canonical_sink.py @@ -189,52 +189,91 @@ def canonical_sibling_sink(name: str | None) -> str | None: return _SIBLING_SINKS.get(name) -def _receiver_is_importlib_loader(receiver: ast.expr, type_map: dict[str, str] | None) -> bool: +def _expr_is_importlib_rooted( + expr: ast.expr, + type_map: dict[str, str] | None, + aliases: dict[str, str] | None = None, +) -> bool: + """True when *expr* statically resolves to ``importlib``-rooted machinery. + + A bare name is checked against *type_map* (whose values are already alias-resolved by + :func:`build_type_map`), so a ``spec = importlib.util.module_from_spec(...)`` variable + is recognized as importlib provenance. Otherwise the expression is resolved to a dotted + name and import-alias normalized (``import importlib.util as iu`` → ``iu.x`` reads as + ``importlib.util.x``). Anything not rooted in ``importlib`` returns ``False``. + """ + if isinstance(expr, ast.Name) and type_map is not None: + inferred = type_map.get(expr.id) + if inferred is not None: + return inferred.split(".", 1)[0] == "importlib" + dotted = resolve_dotted_name(expr) + if dotted is not None and aliases: + dotted = apply_import_aliases(dotted, aliases) + return bool(dotted and dotted.split(".", 1)[0] == "importlib") + + +def _receiver_is_importlib_loader( + receiver: ast.expr, + type_map: dict[str, str] | None, + aliases: dict[str, str] | None = None, +) -> bool: """True when *receiver* statically resolves to an importlib spec loader. - Recognizes the documented ``importlib`` spec-loader chain: ``.loader`` (the - ``ModuleSpec.loader`` attribute, including ``importlib.util.module_from_spec(...)`` - results carried through *type_map*), and bare names whose inferred type is rooted in - ``importlib``. A user object's ``.exec_module`` (receiver not tied to importlib) does - not match. + Recognizes the documented ``importlib`` spec-loader chain ``.loader`` (the + ``ModuleSpec.loader`` attribute) **only when the object holding ``.loader`` itself has + importlib provenance** — a spec from ``importlib.util.module_from_spec`` / + ``spec_from_loader`` carried through *type_map*, or an importlib-rooted dotted name. + An unrelated object's ``.loader.exec_module(...)`` (e.g. ``self.widget.loader``) no + longer matches, closing the false positive where any ``.loader`` attribute was treated + as a dynamic import. Bare names whose inferred type is importlib-rooted also match. """ - if isinstance(receiver, ast.Attribute): - # ``.loader`` — the ModuleSpec.loader protocol attribute. - if receiver.attr == "loader": - return True - inferred = None - if isinstance(receiver, ast.Name) and type_map is not None: - inferred = type_map.get(receiver.id) - if inferred is None: - inferred = resolve_dotted_name(receiver) - return bool(inferred and inferred.split(".", 1)[0] == "importlib") + if isinstance(receiver, ast.Attribute) and receiver.attr == "loader": + # ``.loader`` — gated on the base having importlib provenance, not on the + # mere presence of a ``.loader`` attribute (which any user object can expose). + return _expr_is_importlib_rooted(receiver.value, type_map, aliases) + return _expr_is_importlib_rooted(receiver, type_map, aliases) -def _receiver_is_code_interpreter(receiver: ast.expr, type_map: dict[str, str] | None) -> bool: +def _receiver_is_code_interpreter( + receiver: ast.expr, + type_map: dict[str, str] | None, + aliases: dict[str, str] | None = None, +) -> bool: """True when *receiver* resolves to ``code.Interactive{Interpreter,Console}``. Matches a direct construction (``code.InteractiveInterpreter().runsource(...)``) and a variable whose inferred constructor type is one of those classes (carried through - *type_map*). A user object defining ``runsource`` / ``runcode`` does not match. + *type_map*). The inferred name is import-alias normalized, so ``import code as c; + c.InteractiveInterpreter().runsource(...)`` and ``from code import InteractiveInterpreter`` + are recognized rather than evading detection. A user object defining ``runsource`` / + ``runcode`` does not match. """ inferred = None if isinstance(receiver, ast.Call): inferred = resolve_dotted_name(receiver.func) elif isinstance(receiver, ast.Name) and type_map is not None: inferred = type_map.get(receiver.id) + if inferred is not None and aliases: + inferred = apply_import_aliases(inferred, aliases) return inferred in _CODE_INTERPRETER_TYPES -def canonical_sibling_method(node: ast.Call, type_map: dict[str, str] | None = None) -> str | None: +def canonical_sibling_method( + node: ast.Call, + type_map: dict[str, str] | None = None, + aliases: dict[str, str] | None = None, +) -> str | None: """Map a gated instance-method call to its canonical code-exec sink id. Fires only when *node* is ``.exec_module(...)`` / ``.runsource(...)`` / ``.runcode(...)`` **and** *recv* statically resolves to the matching machinery — importlib spec loaders for ``exec_module`` (:func:`_receiver_is_importlib_loader`), ``code.Interactive{Interpreter,Console}`` for ``runsource`` / ``runcode`` - (:func:`_receiver_is_code_interpreter`). ``exec_module`` → ``"__import__"``, - ``runsource`` / ``runcode`` → ``"exec"``. Returns ``None`` otherwise, so an unrelated - user method sharing the name stays unflagged (documented def-use residual). + (:func:`_receiver_is_code_interpreter`). Both receiver checks import-alias normalize the + inferred name, and the ``exec_module`` gate requires the ``.loader`` base to have + importlib provenance. ``exec_module`` → ``"__import__"``, ``runsource`` / ``runcode`` → + ``"exec"``. Returns ``None`` otherwise, so an unrelated user method sharing the name + stays unflagged (documented def-use residual). """ func = node.func if not isinstance(func, ast.Attribute): @@ -243,8 +282,8 @@ def canonical_sibling_method(node: ast.Call, type_map: dict[str, str] | None = N if canonical is None: return None if func.attr == "exec_module": - return canonical if _receiver_is_importlib_loader(func.value, type_map) else None - return canonical if _receiver_is_code_interpreter(func.value, type_map) else None + return canonical if _receiver_is_importlib_loader(func.value, type_map, aliases) else None + return canonical if _receiver_is_code_interpreter(func.value, type_map, aliases) else None def resolve_to_canonical_sink( @@ -271,7 +310,7 @@ def resolve_to_canonical_sink( """ name = resolve_call_name_typed(node, type_map, aliases) if name is not None: - sibling = canonical_sibling_sink(name) or canonical_sibling_method(node, type_map) + sibling = canonical_sibling_sink(name) or canonical_sibling_method(node, type_map, aliases) return sibling or name dynamic = resolve_dynamic_import_call(node, aliases) if dynamic is not None: @@ -286,4 +325,4 @@ def resolve_to_canonical_sink( # ``code.InteractiveInterpreter().runsource(src)`` — match the gated method (the # receiver must resolve to the interpreter/loader machinery) so the sibling is still # recognized as a code-exec sink without flagging unrelated same-named user methods. - return canonical_sibling_method(node, type_map) + return canonical_sibling_method(node, type_map, aliases) diff --git a/tests/evasion_corpus/corpus.py b/tests/evasion_corpus/corpus.py index aec414416..b2292e68f 100644 --- a/tests/evasion_corpus/corpus.py +++ b/tests/evasion_corpus/corpus.py @@ -115,6 +115,13 @@ class Neighbour(NamedTuple): "spec.loader.exec_module(mod)", "__import__", ), + Spelling( + # Aliased importlib must keep its provenance: the spec variable still resolves to + # importlib (build_type_map normalizes the alias), so ``.loader.exec_module`` fires. + "spec_loader_exec_module_alias", + "import importlib.util as iu\nspec = iu.module_from_spec(s)\nspec.loader.exec_module(mod)", + "__import__", + ), # ── code-exec siblings (→ exec class) ───────────────────────────── Spelling("code_interact", "import code\ncode.interact()", "exec"), Spelling( @@ -122,6 +129,13 @@ class Neighbour(NamedTuple): "import code\ncode.InteractiveInterpreter().runsource(x)", "exec", ), + Spelling( + # Aliased ``code`` import must not evade: the receiver type is alias-normalized back + # to ``code.InteractiveInterpreter`` before the gate check. + "code_runsource_alias", + "import code as c\nc.InteractiveInterpreter().runsource(x)", + "exec", + ), ) # Benign code that resembles a sink; must canonicalize to something OUTSIDE the sink sets. @@ -146,4 +160,8 @@ class Neighbour(NamedTuple): ), Neighbour("user_runsource_method", "ci = MyRepl()\nci.runsource(x)"), Neighbour("user_loader_unrelated", "loader = MyLoader()\nloader.exec_module(m)"), + # An arbitrary object's ``.loader.exec_module(...)`` (no importlib provenance on the + # base) must NOT be treated as a dynamic import — the ``.loader`` attribute alone is + # not enough, any plugin object can expose one. + Neighbour("arbitrary_loader_attr_exec_module", "self.widget.loader.exec_module(mod)"), ) From baa3b9a862d3e77d4131a9d50e12f719e3c9d2d2 Mon Sep 17 00:00:00 2001 From: Zied Jlassi <6190550+zied-jlassi@users.noreply.github.com> Date: Sun, 18 Oct 2026 22:38:07 +0200 Subject: [PATCH 3/3] fix(behavioral): don't double-report module-dict reads AST9 already flags Since #517, main reports os.__dict__["system"](x), vars(builtins)["exec"](x) and builtins.__dict__["exec"](x) as AST9 on the subscript. The canonical fallback also resolved the outer call, so the same expression came out a second time as AST5/AST1. _covered_by_ast9 now mirrors every read AST9 owns on main: allowlisted getattr (joined names included, #544), the module-dict subscript, and the .get/.setdefault/.pop reads (#517). The fallback stays quiet on those and keeps the spellings AST9 doesn't see (__builtins__["exec"], getattr on names outside the allowlist). The corpus test now requires exactly one sink finding on the sink line instead of "at least one". It fails on the three rows from the review without the guard and passes with it. Signed-off-by: Zied Jlassi <6190550+zied-jlassi@users.noreply.github.com> --- .../nodes/analyzers/behavioral_ast.py | 66 +++++++++++-------- .../test_canonical_sink_fitness.py | 49 +++++++++++--- 2 files changed, 79 insertions(+), 36 deletions(-) diff --git a/src/skillspector/nodes/analyzers/behavioral_ast.py b/src/skillspector/nodes/analyzers/behavioral_ast.py index 6da807df3..1aaef4c88 100644 --- a/src/skillspector/nodes/analyzers/behavioral_ast.py +++ b/src/skillspector/nodes/analyzers/behavioral_ast.py @@ -297,29 +297,6 @@ def analyzer_exhausted(self) -> bool: return self.total_findings >= MAX_FINDINGS_PER_ANALYZER -def _covered_by_ast9(node: ast.Call) -> bool: - """True when *node* invokes a literal ``getattr`` already flagged by AST9. - - AST9 catches ``getattr(obj, "")(...)`` for ```` in - :data:`_DANGEROUS_GETATTR_NAMES`. The canonical-sink fallback skips exactly those so - the two rules stay complementary rather than double-firing: the canonical layer then - owns only the spellings AST9 does not (subscript ``__builtins__["exec"]`` / - ``vars(builtins)["exec"]`` and getattr targets outside the allowlist such as - ``getattr(subprocess, "Popen")``). - """ - callee = node.func - if not (isinstance(callee, ast.Call) and isinstance(callee.func, ast.Name)): - return False - if callee.func.id != "getattr" or len(callee.args) < 2: - return False - attr = callee.args[1] - return ( - isinstance(attr, ast.Constant) - and isinstance(attr.value, str) - and attr.value in _DANGEROUS_GETATTR_NAMES - ) - - def _is_chain_sink(node: ast.Call, aliases: dict[str, str] | None = None) -> bool: """True if this call is exec(), eval(), or compile() — the outer dangerous call.""" name = resolve_call_name(node, aliases) @@ -448,6 +425,41 @@ def _reflective_module_dict_base(node: ast.Subscript, aliases: dict[str, str]) - return _reflective_module_dict_container(node.value, aliases) +def _covered_by_ast9(node: ast.Call, aliases: dict[str, str]) -> bool: + """True when the callee of *node* is a reflective read that AST9 already reports. + + AST9 fires on the reflective *read* of an allowlisted name + (:data:`_DANGEROUS_GETATTR_NAMES`) in three spellings: ``getattr(obj, "")`` + (including constant-joined names), ``.__dict__[""]`` / + ``vars()[""]``, and the ``.get`` / ``.setdefault`` / ``.pop`` reads of + the same namespaces. When *node* invokes one of those reads, the canonical-sink + fallback must stay silent so one expression yields one sink finding; it then owns + only the spellings AST9 does not report (``__builtins__["exec"]``, an un-imported + ``vars(builtins)``, and getattr targets outside the allowlist such as + ``getattr(subprocess, "Popen")``). + """ + callee = node.func + if isinstance(callee, ast.Subscript): + if _reflective_module_dict_base(callee, aliases) is None: + return False + key = callee.slice + return isinstance(key, ast.Constant) and key.value in _DANGEROUS_GETATTR_NAMES + if not (isinstance(callee, ast.Call) and len(callee.args) >= 1): + return False + if isinstance(callee.func, ast.Name) and callee.func.id == "getattr": + return len(callee.args) >= 2 and ( + _constant_string(callee.args[1]) in _DANGEROUS_GETATTR_NAMES + ) + if ( + isinstance(callee.func, ast.Attribute) + and callee.func.attr in _REFLECTIVE_DICT_READ_METHODS + and _reflective_module_dict_container(callee.func.value, aliases) is not None + ): + key = callee.args[0] + return isinstance(key, ast.Constant) and key.value in _DANGEROUS_GETATTR_NAMES + return False + + def _analyze_python( python_ast: ParsedPythonFile, file_path: str, @@ -590,14 +602,14 @@ def _emit( # 'os.system', so it re-enters the os./subprocess. sink ladders below. call_name = resolve_dynamic_import_call(ast_node, aliases) reflective = False - if call_name is None and not _covered_by_ast9(ast_node): + if call_name is None and not _covered_by_ast9(ast_node, aliases): # Reflective invocation whose callee does not resolve by name: # getattr(subprocess, "Popen")(cmd) (outside AST9's allowlist), # __builtins__["exec"](src), vars(builtins)["exec"](src). Canonicalize to # the bare/dotted sink id so it re-enters the ladders below with the correct - # rule + severity. Literal getattr() on an AST9-allowlisted name is left to - # AST9 (see _covered_by_ast9) so the two rules stay complementary, not - # duplicate. + # rule + severity. A reflective read AST9 already reports (allowlisted getattr + # or module-namespace read) is left to AST9 (see _covered_by_ast9) so the + # two rules stay complementary, not duplicate. call_name = resolve_to_canonical_sink(ast_node, aliases) reflective = call_name is not None if call_name is None: diff --git a/tests/evasion_corpus/test_canonical_sink_fitness.py b/tests/evasion_corpus/test_canonical_sink_fitness.py index 32b5f2de2..aae7f28af 100644 --- a/tests/evasion_corpus/test_canonical_sink_fitness.py +++ b/tests/evasion_corpus/test_canonical_sink_fitness.py @@ -97,9 +97,9 @@ def test_corpus_has_min_spellings_per_primitive() -> None: # rather than silently regressing to a missed detection. # behavioral_ast rules that constitute a real dangerous-execution sink detection. A -# spelling is "detected" if it produces any of these — the precise rule depends on which -# complementary path fires (the canonical ladders AST1/AST4/AST5/AST6, or AST9 for a -# literal getattr on an allowlisted name, owned by the getattr branch on `main`). +# spelling is "detected" if it produces exactly one of these — the precise rule depends on +# which complementary path fires (the canonical ladders AST1/AST4/AST5/AST6, or AST9 for a +# literal getattr / module-namespace read of an allowlisted name, owned by `main`). _DANGEROUS_EXECUTION_RULES: frozenset[str] = frozenset( {"AST1", "AST2", "AST3", "AST4", "AST5", "AST6", "AST9"} ) @@ -138,17 +138,25 @@ def _run_taint(code: str) -> list: @pytest.mark.parametrize("row", EQUIVALENT_SPELLINGS, ids=lambda r: r.label) def test_spelling_reaches_ast_sink_rule(row: Spelling) -> None: - """Every spelling reaches a dangerous-execution rule in production at full severity. + """Every spelling reaches exactly one dangerous-execution rule at full severity. Proves the chokepoint is actually invoked by production (not merely importable) and that the spelling is not silently dropped or downgraded. The exact rule may be a - canonical-ladder rule or AST9 (literal getattr, owned by the getattr branch on - ``main``) — both are valid sink detections — so the assertion targets the - dangerous-execution rule set plus a minimum-severity floor. + canonical-ladder rule or AST9 (literal getattr / module-namespace reflection, owned + by the reflective branches on ``main``) — both are valid sink detections — but never + both: one expression must yield one sink finding, so the canonical fallback and AST9 + stay complementary rather than double-reporting. Only the sink invocation's own line + is counted, so a snippet whose setup line is itself a sink (``module_from_spec`` ahead + of ``spec.loader.exec_module``) is not mistaken for a double report. """ findings = _run_ast(row.code) - sink_findings = [f for f in findings if f.rule_id in _DANGEROUS_EXECUTION_RULES] - assert sink_findings, f"{row.label}: no sink rule, got {[f.rule_id for f in findings]}" + sink_line = _outermost_call(row.code).lineno + sink_findings = [ + f for f in findings if f.rule_id in _DANGEROUS_EXECUTION_RULES and f.start_line == sink_line + ] + assert len(sink_findings) == 1, ( + f"{row.label}: expected exactly one sink rule, got {[f.rule_id for f in findings]}" + ) best = max(_SEVERITY_ORDER[f.severity] for f in sink_findings) assert best >= _SEVERITY_ORDER[_MIN_SEVERITY[row.canonical]], ( f"{row.label}: severity downgrade, got {[(f.rule_id, f.severity) for f in sink_findings]}" @@ -207,3 +215,26 @@ def test_direct_subprocess_stays_medium() -> None: def test_neighbour_produces_no_ast_findings(neighbour: Neighbour) -> None: """Benign look-alikes produce zero behavioral_ast findings in production.""" assert _run_ast(neighbour.code) == [] + + +# Every reflective read AST9 reports on `main` (getattr incl. constant-joined names, +# module-namespace subscript, and its .get/.setdefault/.pop reads). The canonical +# fallback must stay silent on all of them, even for spellings it does not resolve +# today, so the one-expression/one-finding invariant survives the layer growing. +_AST9_OWNED_READS: tuple[str, ...] = ( + 'import os\ngetattr(os, "system")(x)', + 'import os\ngetattr(os, "".join(["sys", "tem"]))(x)', + 'import os\nos.__dict__["system"](x)', + 'import os as o\nvars(o)["system"](x)', + 'import builtins\nbuiltins.__dict__["exec"](x)', + 'import os\nos.__dict__.get("system")(x)', + 'import builtins\nvars(builtins).pop("exec")(x)', +) + + +@pytest.mark.parametrize("code", _AST9_OWNED_READS) +def test_ast9_owned_read_reports_once(code: str) -> None: + """An AST9-owned reflective read yields AST9 alone, never a second ladder finding.""" + findings = _run_ast(code) + sink_rules = [f.rule_id for f in findings if f.rule_id in _DANGEROUS_EXECUTION_RULES] + assert sink_rules == ["AST9"], f"expected only AST9, got {sink_rules}"