From c1621777f9e752600deb928222d4cebe2a1c57ab Mon Sep 17 00:00:00 2001 From: Marius Ciepluch <11855163+norandom@users.noreply.github.com> Date: Sat, 5 Sep 2026 14:40:15 +0200 Subject: [PATCH] feat(corpus-seeded-mechanisms): structural variant search merged with overlay flows Task 2.2 (Req 3.1-3.3, 3.6). - variants.match_shapes: call sites whose trailing callee name and arity match a shape and whose source positions all carry a source of any kind; constants and arity mismatches never match. - semantic/variant_search.py: search_tree over a tree's parsed files with the store's corpus shapes (bounded by max_mechanisms, language degradations recorded); hits_to_findings merges a hit into an overlay promote/coverage record at the same call site (mechanism_id, no duplicate) and otherwise emits variant::: at suspicion with the pair provenance in the rationale. - OverlayRecord.mechanism_id (additive). Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_01T7XD7x5KonMdmGRkp63EQG --- src/openultrasast/semantic/overlay.py | 1 + src/openultrasast/semantic/variant_search.py | 118 +++++++++++++++++++ tests/test_variant_search.py | 98 +++++++++++++++ 3 files changed, 217 insertions(+) create mode 100644 src/openultrasast/semantic/variant_search.py create mode 100644 tests/test_variant_search.py diff --git a/src/openultrasast/semantic/overlay.py b/src/openultrasast/semantic/overlay.py index 703466d..05c94c3 100644 --- a/src/openultrasast/semantic/overlay.py +++ b/src/openultrasast/semantic/overlay.py @@ -45,6 +45,7 @@ class OverlayRecord: engine: str = "none" language: str = "" function: str | None = None # enclosing function when the file parsed (additive; pair-corpus-honesty) + mechanism_id: str | None = None # known mechanism whose shape matches this call site (additive; corpus-seeded-mechanisms) def __post_init__(self) -> None: if self.disposition not in DISPOSITIONS: diff --git a/src/openultrasast/semantic/variant_search.py b/src/openultrasast/semantic/variant_search.py new file mode 100644 index 0000000..75dfe4e --- /dev/null +++ b/src/openultrasast/semantic/variant_search.py @@ -0,0 +1,118 @@ +"""Variant search over a scanned tree and its merge with overlay flows (corpus-seeded-mechanisms, Req 3). + +Depends on the shapes (``variants``), the store (``mechanisms``), the overlay records and ``StaticFinding``; the shape +module itself stays free of those imports. +""" + +from __future__ import annotations + +from collections.abc import Mapping, Sequence +from dataclasses import dataclass, replace +from pathlib import Path + +from ..findings import StaticFinding +from ..preprocess import FileTarget +from .facts import SemanticFacts +from .ir import parse_file +from .mechanisms import MechanismStore, corpus_mechanisms +from .overlay import OverlayRecord +from .variants import Shape, VariantHit, match_shapes + +LANGUAGE_UNSUPPORTED = "variants_language_unsupported" + + +@dataclass(frozen=True) +class SearchResult: + hits: tuple[VariantHit, ...] + mechanisms_searched: int + files_searched: int + degradations: tuple[dict[str, object], ...] = () + + +def search_tree( + root: Path, + targets: Sequence[FileTarget], + store: MechanismStore, + facts: SemanticFacts, + *, + max_mechanisms: int, +) -> SearchResult: + """Variant search over the parsed files of a tree with the store's corpus shapes (bounded by ``max_mechanisms``).""" + records = corpus_mechanisms(store.load())[: max(max_mechanisms, 0)] + shapes: list[Shape] = [] + ids: dict[str, str] = {} + for record in records: + assert record.shape is not None + shape = Shape.from_dict(record.shape) + shapes.append(shape) + ids[shape.key()] = record.id + if not shapes: + return SearchResult(hits=(), mechanisms_searched=0, files_searched=0) + hits: list[VariantHit] = [] + files = 0 + degradations: list[dict[str, object]] = [] + unsupported: set[str] = set() + for target in targets: + try: + text = (root / target.path).read_text(errors="ignore") + except OSError: + continue + ir = parse_file(target.path, text, target.language) + if not ir.parse_ok: + if ir.reason == "language_unsupported" and target.language not in unsupported: + unsupported.add(target.language) + degradations.append({"stage": "variants", "reason": "variants_language_unsupported", "language": target.language}) + continue + files += 1 + hits.extend(match_shapes(ir, shapes, facts, mechanism_ids=ids)) + return SearchResult(hits=tuple(hits), mechanisms_searched=len(shapes), files_searched=files, degradations=tuple(degradations)) + + +def hits_to_findings( + hits: Sequence[VariantHit], + records: Sequence[OverlayRecord], + summaries: Mapping[str, tuple[str, tuple[str, ...], str]], +) -> tuple[list[StaticFinding], list[OverlayRecord]]: + """Merge hits into overlay records at the same call site (Req 3.3); the rest become ``suspicion`` findings (Req 3.2).""" + by_site: dict[tuple[str, int | None], list[int]] = {} + for index, record in enumerate(records): + if record.disposition in {"promote", "coverage"} and record.sources and record.sinks: + by_site.setdefault((record.path, record.line), []).append(index) + updated = list(records) + findings: list[StaticFinding] = [] + emitted: set[str] = set() + for hit in hits: + indexes = by_site.get((hit.path, hit.line)) + if indexes: + for index in indexes: + if updated[index].mechanism_id is None: + updated[index] = replace(updated[index], mechanism_id=hit.mechanism_id) + continue + finding_id = f"variant:{hit.mechanism_id}:{hit.path}:{hit.line}" + if finding_id in emitted: + continue + emitted.add(finding_id) + summary, pairs, cwe = summaries.get(hit.mechanism_id, (hit.mechanism_id, (), "")) + taught_by = ", ".join(pairs) if pairs else "the mechanism store" + findings.append( + StaticFinding( + finding_id=finding_id, + path=hit.path, + title=f"Variant of known mechanism: {hit.sink_name}", + severity="medium", + confidence="low", + evidence_level="suspicion", + rationale=( + f"{summary}. Structural variant ({hit.source_kind} reaches {hit.sink_name}) of a mechanism learned from " + f"{taught_by}; suspicion until the overlay or the sandbox agrees." + ), + line=hit.line, + function_name=None, + reachability_status="unknown", + reachability_evidence=[], + reachability_conditions=[], + tags=["variant", f"mechanism:{hit.mechanism_id}", *([f"cwe:{cwe}"] if cwe else [])], + ranking_priority=0.0, + ) + ) + return findings, updated diff --git a/tests/test_variant_search.py b/tests/test_variant_search.py new file mode 100644 index 0000000..bb251c8 --- /dev/null +++ b/tests/test_variant_search.py @@ -0,0 +1,98 @@ +"""corpus-seeded-mechanisms task 2.2: structural variant search over a scanned tree, merged with overlay flows.""" + +from __future__ import annotations + +from pathlib import Path + +from openultrasast.semantic.facts import load_facts +from openultrasast.semantic.ir import parse_file +from openultrasast.semantic.overlay import OverlayRecord + +PY_FACTS = load_facts() + + +def _shape(**overrides: object): + from openultrasast.semantic.variants import Shape + + base: dict[str, object] = { + "language": "python", + "sink_name": "system", + "arity": 1, + "source_positions": (0,), + "source_kinds": ("fact_source",), + "guard": "allowlist_test", + "mechanism": "source_reaches_sink", + } + base.update(overrides) + return Shape(**base) # type: ignore[arg-type] + + +VARIANT = "import os\nfrom flask import request\n\n\ndef handler():\n q = request.args.get('q')\n os.system(q)\n" +PARAM_VARIANT = "import os\n\n\ndef handler(q):\n os.system(q)\n" +CONSTANT = "import os\n\n\ndef handler():\n os.system('ls -la')\n" +ARITY = "import os\nfrom flask import request\n\n\ndef handler():\n q = request.args.get('q')\n os.system(q, 1)\n" + + +def test_match_shapes_finds_structural_variants_and_ignores_constants() -> None: + from openultrasast.semantic.variants import match_shapes + + shape = _shape() + hits = match_shapes(parse_file("a.py", VARIANT, "python"), [shape], PY_FACTS, mechanism_ids={shape.key(): "corpus:abc"}) + assert [(h.path, h.line, h.mechanism_id, h.sink_name, h.source_kind) for h in hits] == [ + ("a.py", 7, "corpus:abc", "system", "fact_source") + ] + param_hits = match_shapes(parse_file("b.py", PARAM_VARIANT, "python"), [shape], PY_FACTS, mechanism_ids={shape.key(): "corpus:abc"}) + assert [(h.line, h.source_kind) for h in param_hits] == [(5, "parameter")] # any source kind at the shape's positions matches + assert match_shapes(parse_file("c.py", CONSTANT, "python"), [shape], PY_FACTS, mechanism_ids={shape.key(): "corpus:abc"}) == [] + assert match_shapes(parse_file("d.py", ARITY, "python"), [shape], PY_FACTS, mechanism_ids={shape.key(): "corpus:abc"}) == [] + js_shape = _shape(language="javascript", sink_name="exec") + assert match_shapes(parse_file("a.py", VARIANT, "python"), [js_shape], PY_FACTS, mechanism_ids={js_shape.key(): "x"}) == [] + + +def test_hits_become_suspicion_findings_or_merge_into_an_overlay_flow() -> None: + from openultrasast.semantic.variant_search import hits_to_findings + from openultrasast.semantic.variants import VariantHit + + hit = VariantHit(path="a.py", line=7, mechanism_id="corpus:abc", sink_name="system", source_kind="fact_source") + other = VariantHit(path="z.py", line=3, mechanism_id="corpus:abc", sink_name="system", source_kind="parameter") + coverage = OverlayRecord( + proposal_id="overlay-coverage:a.py:7:os.system", + path="a.py", + line=7, + disposition="coverage", + reason="uninventoried sink os.system reached from request", + cwe="CWE-78", + sources=("request",), + sinks=("os.system",), + sanitizers=(), + evidence_level="static_corroboration", + origin="overlay", + engine="python-ast", + language="python", + ) + summaries = {"corpus:abc": ("request into os.system; fix adds allowlist_test", ("p1", "p2"), "CWE-78")} + findings, records = hits_to_findings([hit, other], [coverage], summaries) + assert len(records) == 1 and records[0].mechanism_id == "corpus:abc" and records[0].disposition == "coverage" + assert [f.finding_id for f in findings] == ["variant:corpus:abc:z.py:3"] # the a.py hit merged, no duplicate + (finding,) = findings + assert finding.evidence_level == "suspicion" and "variant" in finding.tags and "mechanism:corpus:abc" in finding.tags + assert "p1" in finding.rationale and finding.line == 3 and finding.path == "z.py" + + +def test_search_tree_uses_the_store_and_respects_the_mechanism_cap(tmp_path: Path) -> None: + from openultrasast.preprocess import preprocess_repository + from openultrasast.semantic.mechanisms import MechanismStore, append_from_pair + from openultrasast.semantic.variant_search import search_tree + + root = tmp_path / "repo" + root.mkdir() + (root / "views.py").write_text(VARIANT) + (root / "safe.py").write_text(CONSTANT) + store = MechanismStore(tmp_path / "mechanisms.jsonl") + record = append_from_pair(store, _shape(), summary="s", cwe="CWE-78", pair="p1", provenance="human", tier="seeded") + _, targets = preprocess_repository(root) + result = search_tree(root, targets, store, PY_FACTS, max_mechanisms=500) + assert [(h.path, h.line, h.mechanism_id) for h in result.hits] == [("views.py", 7, record.id)] + assert result.mechanisms_searched == 1 and result.files_searched == 2 + capped = search_tree(root, targets, store, PY_FACTS, max_mechanisms=0) + assert capped.hits == () and capped.mechanisms_searched == 0