From a7b3b3602526a4042e768d8683962968f5b5dafd 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): export trusted pairs as mechanism records Task 2.1 (Req 1.1, 1.2, 1.5, 2.5). - semantic/seed.py: export_mechanisms seeds only seeded/reviewed, non-known-limit, parsing pairs with cached excerpts; every skip carries a reason (tier, known_limit, parse_failed, language unsupported with a degradation, pointer not cached, no labeled sink call site); runs offline over the vendored catalog. - variants.derive_shape: labels that name only a function and a mechanism resolve to the fact-sink call inside that function (CWE match and a non-constant argument first); line and sink intersect when both exist; container reads are subscript or .get( only. - append_from_pair keeps the strongest review tier and accumulates provenance tags when a shape is re-seeded. - CLI: `ousast mechanisms export --slice S [--store P] [--json]`. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_01T7XD7x5KonMdmGRkp63EQG --- src/openultrasast/semantic/mechanisms.py | 13 ++- src/openultrasast/semantic/seed.py | 140 +++++++++++++++++++++++ src/openultrasast/semantic/variants.py | 118 ++++++++++++++++--- tests/test_mechanism_seed.py | 112 ++++++++++++++++++ 4 files changed, 368 insertions(+), 15 deletions(-) create mode 100644 src/openultrasast/semantic/seed.py create mode 100644 tests/test_mechanism_seed.py diff --git a/src/openultrasast/semantic/mechanisms.py b/src/openultrasast/semantic/mechanisms.py index 72e0490..e9074b6 100644 --- a/src/openultrasast/semantic/mechanisms.py +++ b/src/openultrasast/semantic/mechanisms.py @@ -89,6 +89,9 @@ def append_mechanism( return mechanism +_TIER_RANK = {"seeded": 1, "reviewed": 2} + + def corpus_mechanism_id(shape: Shape) -> str: """Deterministic id from the shape key, so the same shape gets the same record on every machine (lever-addressable).""" return "corpus:" + hashlib.sha1(shape.key().encode()).hexdigest()[:16] @@ -110,13 +113,21 @@ def append_from_pair( pairs = tuple(existing.pairs) if existing is not None else () if pair not in pairs: pairs = (*pairs, pair) + if existing is not None and _TIER_RANK.get(existing.review_tier, 0) > _TIER_RANK.get(tier, 0): + tier = existing.review_tier # a shape keeps the strongest review it ever had + inherited = tuple(tag for tag in (existing.tags if existing is not None else ()) if tag.startswith(("provenance:", "tier:"))) positions = ", ".join(f"arg{p} ({k})" for p, k in zip(shape.source_positions, shape.source_kinds, strict=True)) record = Mechanism( id=mechanism_id, summary=summary, cwe=cwe, language=shape.language, - tags=(shape.mechanism, f"mechanism:{shape.mechanism}", f"guard:{shape.guard}", f"provenance:{provenance}", f"tier:{tier}"), + tags=( + shape.mechanism, + f"mechanism:{shape.mechanism}", + f"guard:{shape.guard}", + *sorted({*inherited, f"provenance:{provenance}", f"tier:{tier}"}), + ), keywords=(shape.sink_name, shape.guard, *shape.source_kinds), what_made_it_exploitable=f"{positions} reaches {shape.sink_name}/{shape.arity}; the fix added {shape.guard}", source_finding_id="", diff --git a/src/openultrasast/semantic/seed.py b/src/openultrasast/semantic/seed.py new file mode 100644 index 0000000..f1beaff --- /dev/null +++ b/src/openultrasast/semantic/seed.py @@ -0,0 +1,140 @@ +"""Export trusted pairs as mechanism records (corpus-seeded-mechanisms, Req 1). + +Only ``seeded`` and ``reviewed`` pairs seed; ``advisory`` and ``title`` pairs, ``known_limit`` pairs, pointer pairs +without a cached excerpt, and excerpts that fail to parse are skipped with a reason. Runs offline over the vendored +catalog; nothing here fetches, and no code text enters the store. +""" + +from __future__ import annotations + +from collections.abc import Sequence +from dataclasses import dataclass, field +from pathlib import Path + +from ..pairs import GATING_TIERS, PairCase +from ..preprocess import LANGUAGE_BY_EXTENSION +from .facts import FactLoadError, SemanticFacts, load_facts +from .ir import parse_file +from .mechanisms import MechanismStore, append_from_pair +from .variants import derive_shape + +LANGUAGE_UNSUPPORTED = "variants_language_unsupported" + + +@dataclass(frozen=True) +class ExportReport: + seeded: int # pairs that produced at least one record + records: int # distinct records written (deduped by shape) + skipped: tuple[tuple[str, str], ...] = () # (pair, reason) + degradations: tuple[dict[str, object], ...] = () + seeded_pairs: tuple[str, ...] = field(default=()) + + def to_dict(self) -> dict[str, object]: + return { + "seeded": self.seeded, + "records": self.records, + "skipped": [{"pair": pair, "reason": reason} for pair, reason in self.skipped], + "degradations": list(self.degradations), + "seeded_pairs": list(self.seeded_pairs), + } + + +def export_mechanisms(cases: Sequence[PairCase], store: MechanismStore, *, facts: SemanticFacts | None = None) -> ExportReport: + """Derive one shape per labeled expected row of every trusted pair and append it to ``store`` (deduped by shape).""" + loaded = facts if facts is not None else _facts() + skipped: list[tuple[str, str]] = [] + degradations: list[dict[str, object]] = [] + seeded: list[str] = [] + record_ids: set[str] = set() + unsupported_languages: set[str] = set() + for case in cases: # catalog order; the store lists pairs in the order they taught + reason = _skip_reason(case) + if reason is not None: + skipped.append((case.name, reason)) + continue + vuln_text = _read(case.vuln_file) + fixed_text = _read(case.fixed_file) + language = _parse_language(case) + vuln_ir = parse_file(case.relpath, vuln_text, language) + fixed_ir = parse_file(case.relpath, fixed_text, language) + if not vuln_ir.parse_ok or not fixed_ir.parse_ok: + bad = vuln_ir if not vuln_ir.parse_ok else fixed_ir + if bad.reason == "language_unsupported": + skipped.append((case.name, LANGUAGE_UNSUPPORTED)) + if language not in unsupported_languages: + unsupported_languages.add(language) + degradations.append({"stage": "mechanisms", "reason": LANGUAGE_UNSUPPORTED, "language": language}) + else: + skipped.append((case.name, "parse_failed")) + continue + produced = 0 + for row in case.expected: + if not row.function or not row.mechanism: + continue + shape = derive_shape( + vuln_ir, + fixed_ir, + function=row.function, + sink=row.sink, + line=None if row.sink else row.line, + mechanism=row.mechanism, + facts=loaded, + vuln_text=vuln_text, + fixed_text=fixed_text, + cwe=row.cwe or None, + ) + if shape is None: + continue + record = append_from_pair( + store, + shape, + summary=f"{shape.mechanism}: {'/'.join(shape.source_kinds)} into {shape.sink_name} ({row.cwe}); fix adds {shape.guard}", + cwe=row.cwe, + pair=case.name, + provenance=case.provenance, + tier=case.review_tier, + ) + record_ids.add(record.id) + produced += 1 + if produced: + seeded.append(case.name) + else: + skipped.append((case.name, "no_labeled_sink_call_site")) + return ExportReport( + seeded=len(seeded), + records=len(record_ids), + skipped=tuple(skipped), + degradations=tuple(degradations), + seeded_pairs=tuple(seeded), + ) + + +def _skip_reason(case: PairCase) -> str | None: + if case.review_tier not in GATING_TIERS: + return f"tier:{case.review_tier}" + if case.known_limit: + return f"known_limit:{case.known_limit}" + if not case.vuln_file.is_file() or not case.fixed_file.is_file(): + return "pointer_pair_not_cached" if not case.vendored else "excerpt_missing" + return None + + +def _parse_language(case: PairCase) -> str: + by_ext = LANGUAGE_BY_EXTENSION.get(case.vuln_file.suffix.lower()) + if by_ext: + return str(by_ext) + return {"c_cpp": "c"}.get(case.language, case.language) + + +def _read(path: Path) -> str: + try: + return path.read_text(errors="ignore") + except OSError: + return "" + + +def _facts() -> SemanticFacts: + try: + return load_facts() + except FactLoadError: + return SemanticFacts(version="", sources=(), sinks=(), sanitizers=()) diff --git a/src/openultrasast/semantic/variants.py b/src/openultrasast/semantic/variants.py index 65629d6..56a6a2c 100644 --- a/src/openultrasast/semantic/variants.py +++ b/src/openultrasast/semantic/variants.py @@ -11,10 +11,10 @@ project matches even when every identifier differs. Stdlib plus ``.ir``, ``.fact from __future__ import annotations import re -from collections.abc import Sequence +from collections.abc import Mapping, Sequence from dataclasses import dataclass -from .facts import SemanticFacts +from .facts import SemanticFacts, SinkFact from .functions import named_ranges_from_ir from .ir import Bind, CallSite, FileIR, FunctionIR @@ -54,9 +54,8 @@ _GUARD_PATTERNS: tuple[tuple[str, re.Pattern[str]], ...] = ( re.compile(r"\bint\(|\bparseInt\(|\bNumber\(|\(int\)|\(size_t\)|\(unsigned\b|\bstr\(|\.toString\(|\bstatic_cast<|\bBigInt\("), ), ) -_CONTAINER_READ = re.compile( - r"\[[^\]]+\]|\.get\(|\.getParameter\(|\.param\(|\.query\b|\.body\b|\.params\b|\.args\b|\.form\b|\.headers\b|\.cookies\b" -) +# Design §Shape: a container read is a subscript or a `.get(` read; source names themselves live in facts, never here. +_CONTAINER_READ = re.compile(r"\[[^\]]+\]|\.get\(") @dataclass(frozen=True) @@ -124,6 +123,7 @@ def derive_shape( facts: SemanticFacts, vuln_text: str = "", fixed_text: str = "", + cwe: str | None = None, ) -> Shape | None: """The shape of the labeled sink call in ``function`` on the vulnerable side, with the guard the fixed side added. @@ -135,10 +135,11 @@ def derive_shape( target = _function_named(vuln, function, vuln_text) if target is None: return None - call = _labeled_call(target, sink, line) + scoped = facts.for_language(vuln.language) + call = _labeled_call(target, sink, line, scoped, cwe) if call is None: return None - kinds = [_source_kind_of_argument(target, call, index, facts) for index in range(len(call.arg_texts))] + kinds = [_source_kind_of_argument(target, call, index, scoped) for index in range(len(call.arg_texts))] positions = tuple(index for index, kind in enumerate(kinds) if kind is not None) if not positions: return None @@ -214,19 +215,51 @@ def _last_bind(function: FunctionIR, name: str, before_line: int) -> Bind | None return max(candidates, key=lambda bind: bind.line) if candidates else None -def _labeled_call(function: FunctionIR, sink: str | None, line: int | None) -> CallSite | None: - if line is not None: - by_line = [call for call in function.calls if call.line == line] - if by_line: - return by_line[0] - if sink: - wanted = trailing_name(sink) +def _labeled_call(function: FunctionIR, sink: str | None, line: int | None, facts: SemanticFacts, cwe: str | None) -> CallSite | None: + """The call the label names: line and sink intersected when both exist, else either; else the fact sink in the function. + + Labels that carry only a function and a mechanism (Real-Vuln rows) fall back to the fact-sink call inside the labeled + function, preferring the sink whose CWE matches the row and a call with a non-constant argument. + """ + wanted = trailing_name(sink) if sink else None + on_line = [call for call in function.calls if line is not None and call.line == line] + if on_line and wanted: + exact = [call for call in on_line if trailing_name(call.name) == wanted] + if exact: + return exact[0] + if on_line: + return on_line[0] + if wanted: for call in function.calls: if trailing_name(call.name) == wanted: return call + return None + ranked: list[tuple[int, int, CallSite]] = [] + for order, call in enumerate(function.calls): + fact = _sink_fact_for(call.name, facts) + if fact is None: + continue + cwe_match = 0 if (cwe and fact.cwe == cwe) else 1 + non_constant = 0 if any(not const or names for const, names in zip(call.arg_is_constant, call.arg_names, strict=True)) else 1 + ranked.append((cwe_match + non_constant, order, call)) + ranked.sort(key=lambda item: (item[0], item[1])) + return ranked[0][2] if ranked else None + + +def _sink_fact_for(name: str, facts: SemanticFacts) -> SinkFact | None: + for fact in facts.sinks: + if any(_call_matches(name, call) for call in fact.calls): + return fact return None +def _call_matches(name: str, fact_call: str) -> bool: + """Same rule as the taint walker: exact, dotted suffix, or a bare name equal to the fact's last segment.""" + if name == fact_call or name.endswith("." + fact_call): + return True + return "." not in name and name == fact_call.split(".")[-1] + + def _function_named(ir: FileIR, function: str, text: str) -> FunctionIR | None: for item in ir.functions: if item.name == function: @@ -255,3 +288,60 @@ def shapes_by_sink(shapes: Sequence[Shape]) -> dict[tuple[str, str], list[Shape] for shape in shapes: index.setdefault((shape.language, shape.sink_name), []).append(shape) return index + + +@dataclass(frozen=True) +class VariantHit: + path: str + line: int + mechanism_id: str + sink_name: str + source_kind: str + + +def match_shapes( + ir: FileIR, + shapes: Sequence[Shape], + facts: SemanticFacts, + *, + mechanism_ids: Mapping[str, str], +) -> list[VariantHit]: + """Call sites of ``ir`` whose trailing callee name, arity and source-carrying positions match a shape (Req 3.1). + + Any source kind at the shape's positions matches: a variant learned from ``request.args`` still fires when the same + sink takes a parameter. Constant-only arguments never match. ``mechanism_ids`` maps ``Shape.key()`` to the store id. + """ + if not ir.parse_ok: + return [] + index = shapes_by_sink([shape for shape in shapes if shape.language == ir.language]) + if not index: + return [] + scoped = facts.for_language(ir.language) if hasattr(facts, "for_language") else facts + hits: list[VariantHit] = [] + seen: set[tuple[int, str]] = set() + for function in ir.functions: + for call in function.calls: + candidates = index.get((ir.language, trailing_name(call.name))) + if not candidates: + continue + for shape in candidates: + if shape.arity != len(call.arg_texts) or (call.line, shape.key()) in seen: + continue + kinds = [ + _source_kind_of_argument(function, call, position, scoped) + for position in shape.source_positions + if position < len(call.arg_texts) + ] + if len(kinds) != len(shape.source_positions) or any(kind is None for kind in kinds): + continue + seen.add((call.line, shape.key())) + hits.append( + VariantHit( + path=ir.path, + line=call.line, + mechanism_id=mechanism_ids.get(shape.key(), shape.key()), + sink_name=shape.sink_name, + source_kind=str(kinds[0]), + ) + ) + return hits diff --git a/tests/test_mechanism_seed.py b/tests/test_mechanism_seed.py new file mode 100644 index 0000000..9bedb2d --- /dev/null +++ b/tests/test_mechanism_seed.py @@ -0,0 +1,112 @@ +"""corpus-seeded-mechanisms task 2.1: export trusted pairs as mechanism records (offline).""" + +from __future__ import annotations + +import json +from pathlib import Path + +import pytest + +from openultrasast.benchmark import ExpectedFinding +from openultrasast.pairs import PairCase, load_pair_catalog, select_slice +from openultrasast.semantic.mechanisms import MechanismStore + +VULN = "from flask import request\nimport os\n\n\ndef run():\n cmd = request.args.get('cmd')\n os.system(cmd)\n" +FIXED = ( + "from flask import request\nimport os\n\nALLOWED = {'ls', 'date'}\n\n\ndef run():\n cmd = request.args.get('cmd')\n" + " if cmd not in ALLOWED:\n return None\n os.system(cmd)\n" +) + + +def _case(tmp_path: Path, name: str, *, tier: str, known_limit: str | None = None, vuln: str = VULN, language: str = "python") -> PairCase: + ext = {"python": "py", "javascript": "js"}[language] + (tmp_path / f"{name}-v.{ext}").write_text(vuln) + (tmp_path / f"{name}-f.{ext}").write_text(FIXED if language == "python" else vuln) + return PairCase( + name=name, + slice="vibe-py", + language=language, + origin="test", + vuln_file=tmp_path / f"{name}-v.{ext}", + fixed_file=tmp_path / f"{name}-f.{ext}", + relpath=f"app.{ext}", + expected=( + ExpectedFinding( + cwe="CWE-78", + vulnerability_class="command injection", + path=f"app.{ext}", + evidence="", + sink="os.system", + function="run", + mechanism="source_reaches_sink", + ), + ), + min_recall=1.0, + fix_policy="silent", + provenance="human", + review_tier=tier, + reviewer="t" if tier == "reviewed" else "", + known_limit=known_limit, + ) + + +def test_export_seeds_only_trusted_parsing_pairs_and_reports_every_skip(tmp_path: Path) -> None: + from openultrasast.semantic.seed import export_mechanisms + + store = MechanismStore(tmp_path / "mechanisms.jsonl") + cases = [ + _case(tmp_path, "seeded", tier="seeded"), + _case(tmp_path, "reviewed-twin", tier="reviewed"), + _case(tmp_path, "advisory", tier="advisory"), + _case(tmp_path, "title", tier="title"), + _case(tmp_path, "limit", tier="seeded", known_limit="engine cannot see it"), + _case(tmp_path, "broken", tier="seeded", vuln="def run(:\n"), + ] + report = export_mechanisms(cases, store) + records = store.load() + assert len(records) == 1 # seeded and reviewed-twin share one shape: one record, two pairs + (record,) = records + assert record.origin == "corpus" and record.pairs == ("seeded", "reviewed-twin") and record.shape is not None + assert report.seeded == 2 and report.records == 1 + reasons = dict(report.skipped) + assert reasons["advisory"] == "tier:advisory" and reasons["title"] == "tier:title" + assert reasons["limit"].startswith("known_limit") and reasons["broken"] == "parse_failed" + text = (tmp_path / "mechanisms.jsonl").read_text() + assert "request.args" not in text and "app.py" not in text # Req 2.4: no code text in the store + + +def test_export_skips_languages_without_a_grammar_with_a_degradation(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + from openultrasast.semantic.seed import export_mechanisms + + monkeypatch.setenv("OPENULTRASAST_TREE_SITTER_PROBE", "0") + store = MechanismStore(tmp_path / "mechanisms.jsonl") + js = "function run(req) {\n const cmd = req.query.cmd;\n exec(cmd);\n}\n" + report = export_mechanisms([_case(tmp_path, "js", tier="seeded", vuln=js, language="javascript")], store) + assert store.load() == () and dict(report.skipped)["js"] == "variants_language_unsupported" + assert {"stage": "mechanisms", "reason": "variants_language_unsupported", "language": "javascript"} in report.degradations + + +def test_export_over_the_vendored_vibe_py_slice_is_offline_and_deterministic(tmp_path: Path) -> None: + from openultrasast.pairs import select_vendored + from openultrasast.semantic.seed import export_mechanisms + + cases = select_vendored(select_slice(load_pair_catalog(), "vibe-py")) + first = export_mechanisms(cases, MechanismStore(tmp_path / "a.jsonl")) + second = export_mechanisms(cases, MechanismStore(tmp_path / "b.jsonl")) + assert first.records >= 1 and first.records == second.records + assert (tmp_path / "a.jsonl").read_text() == (tmp_path / "b.jsonl").read_text() + assert first.seeded + len(first.skipped) == len(cases) + + +def test_cli_mechanisms_export_writes_the_store_and_prints_the_report(tmp_path: Path) -> None: + import io + from contextlib import redirect_stdout + + from openultrasast.cli import main + + buf = io.StringIO() + with redirect_stdout(buf): + assert main(["mechanisms", "export", "--slice", "vibe-py", "--store", str(tmp_path / "m.jsonl"), "--json"]) == 0 + payload = json.loads(buf.getvalue()) + assert payload["records"] >= 1 and payload["slice"] == "vibe-py" and "skipped" in payload + assert (tmp_path / "m.jsonl").is_file()