mirror of
https://github.com/norandom/OpenUltraSAST.git
synced 2026-10-18 20:49:28 +00:00
feat(model-grounded-detection): group 1, subtract the noise architecture
99 source modules -> 81. 832 tests -> 646. The whole learning/ package (16 modules, ~4,100 lines) and six semantic/ modules removed, with the three gates byte-identical to 2026-09-06-gate-baseline.txt at every step. The sequence is what made it safe: vocabularies ported first (1.1), importers detached second (1.2-1.3), deletion third (1.4). Every import break during the deletion was therefore a detachment miss in the importer and was fixed there -- no module was restored to satisfy one. Both misses were the same shape and were fixed as a class: operations.py and shapes.py each imported trailing_name from the deleted variants.py, so it moved to semantic/ir.py, its natural home, and both importers changed in one edit. Ported before deleting: the closed guard vocabulary (verbatim, nothing else holds it) into model/specs.py, which derives TaintSpec and DominanceSpec from the retained ruleset facts rather than copying them. Re-homed: families -> model/taxonomy, candidates -> model/candidates, endpoint -> model/endpoint, plus Region, labeled_family and labeled_spans. labeled_family no longer falls back to the LLM classifier -- an unlabeled pair is honestly unknown. Four deviations, each recorded in the spec notes with its reason. improve/ evolve.py and validator.py were stripped, not deleted: each mixes the mechanism lever (gone) with the deterministic rule-status loop `ousast improve` still runs, which has no LLM in it; a test now asserts the lever's absence by symbol so it cannot creep back. The learning CLI group became `model candidates`, because deleting it outright would have orphaned the retained enumerator by removing its only entry point. thinking moved from [learning] to [models]. A gap named rather than papered over: no retained sink fact routes by CWE to path, output_encoding, untrusted_destination or prototype -- the same hole the entailment ceiling diagnosed. The test asserts the seeded set and the gap set exactly, so task 4.2 must update it when it authors those sinks. The audit found exactly one genuine orphan: vectorstore.py, a bakeoff harness with no production caller and no CLI, alive only through its own test. The five named subsystems each carry an explicit decision. Manifest committed, with a test asserting the tree still matches it. Group 2 (the Joern go/no-go) has not started. Co-Authored-By: Claude Opus 5 (1M context) <[email protected]> Claude-Session: https://claude.ai/code/session_01T7XD7x5KonMdmGRkp63EQG
This commit is contained in:
Marius Ciepluch
co-authored by
Claude Opus 5
Claude Opus 5
parent
fc4dab97fb
commit
9ed0e45c5f
@@ -3,26 +3,21 @@
|
||||
from .engines import joern_available, tree_sitter_available
|
||||
from .extra import grammar_for, has_semantic_extra
|
||||
from .facts import FactLoadError, SemanticFacts, load_facts
|
||||
from .mechanisms import Mechanism, MechanismStore, append_mechanism, order_promotions
|
||||
from .overlay import OverlayError, OverlayRecord, adjudicate, finding_from_coverage, write_overlay
|
||||
from .prove_filter import filter_promoted_hotspots
|
||||
|
||||
__all__ = [
|
||||
"FactLoadError",
|
||||
"Mechanism",
|
||||
"MechanismStore",
|
||||
"OverlayError",
|
||||
"OverlayRecord",
|
||||
"SemanticFacts",
|
||||
"adjudicate",
|
||||
"append_mechanism",
|
||||
"filter_promoted_hotspots",
|
||||
"finding_from_coverage",
|
||||
"grammar_for",
|
||||
"has_semantic_extra",
|
||||
"joern_available",
|
||||
"load_facts",
|
||||
"order_promotions",
|
||||
"tree_sitter_available",
|
||||
"write_overlay",
|
||||
]
|
||||
|
||||
@@ -3,6 +3,7 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import ast
|
||||
import re
|
||||
from dataclasses import dataclass
|
||||
|
||||
|
||||
@@ -230,3 +231,14 @@ def _unparse(node: ast.AST) -> str:
|
||||
return ast.unparse(node)
|
||||
except Exception:
|
||||
return ""
|
||||
|
||||
|
||||
def trailing_name(callee: str) -> str:
|
||||
"""``os.path.join`` -> ``join``; ``Runtime.getRuntime().exec`` -> ``exec``; keeps the last identifier segment.
|
||||
|
||||
Re-homed here from the deleted ``semantic/variants.py``: it is a pure helper over ``CallSite.name`` and has
|
||||
nothing to do with the mechanism search that used to surround it.
|
||||
"""
|
||||
tail = callee.rsplit(".", 1)[-1].rsplit("::", 1)[-1].rsplit("->", 1)[-1]
|
||||
match = re.search(r"[A-Za-z_$][\w$]*\s*$", tail)
|
||||
return match.group(0).strip() if match else tail
|
||||
|
||||
@@ -1,304 +0,0 @@
|
||||
"""Leave-one-out: the pair corpus's own detection rate (corpus-seeded-mechanisms, Req 4).
|
||||
|
||||
For every scorable pair, a temporary mechanism store is seeded from every *other* trusted pair, both excerpts are
|
||||
materialized and searched for variants, and the pair is scored with the pair rules: detected when a hit lies inside the
|
||||
labeled function and names the labeled mechanism (or sink), silent when the fixed side yields no hit. Only ``seeded`` and
|
||||
``reviewed`` pairs teach; ``advisory`` and ``title`` pairs are scored as held-out targets but never seed. Reported, never
|
||||
gated.
|
||||
|
||||
Obligation-labeled rows (authorization-obligations, Req 7.4) run the obligation checker on both sides with the held-out
|
||||
store's obligation shapes: detected when a finding tagged with the labeled obligation lies inside the labeled function *and*
|
||||
carries a ``known_fix`` supplied by the store (the checker's own facts-only finding is not the corpus teaching), silent when
|
||||
the fixed side carries no such finding; ``found_by`` names the shape record and the pairs that taught it.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import tempfile
|
||||
from collections.abc import Sequence
|
||||
from dataclasses import dataclass, field
|
||||
from pathlib import Path
|
||||
|
||||
from ..findings import StaticFinding
|
||||
from ..pairs import PairCase, _materialize, _targets
|
||||
from .facts import FactLoadError, SemanticFacts, load_facts
|
||||
from .functions import named_function_ranges, spans_named
|
||||
from .mechanisms import Mechanism, MechanismStore, append_from_pair, corpus_mechanisms
|
||||
from .obligations import check_obligations, findings_to_static, load_obligation_facts, obligation_mechanisms
|
||||
from .obligations.dominance import OrderDominance
|
||||
from .seed import Lesson, _parse_language, pair_lessons
|
||||
from .variant_search import search_tree
|
||||
from .variants import Shape, trailing_name
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class LooOutcome:
|
||||
pair: str
|
||||
slice: str
|
||||
provenance: str
|
||||
mechanisms: tuple[str, ...] # labeled mechanism ids of the held-out pair
|
||||
detected: bool
|
||||
silent: bool
|
||||
found_by: tuple[tuple[str, tuple[str, ...]], ...] # (mechanism record id, pairs that taught it) for detecting hits
|
||||
hits_vuln: int
|
||||
hits_fixed: int
|
||||
teaches: bool # the pair contributed at least one shape when others were held out
|
||||
leaked_by: tuple[str, ...] = () # mechanism record ids that hit the fixed side
|
||||
degradations: tuple[dict[str, object], ...] = () # from variant search on either side
|
||||
obligations: tuple[str, ...] = () # labeled obligation kinds of the held-out pair (Req 7.4)
|
||||
|
||||
def to_dict(self) -> dict[str, object]:
|
||||
return {
|
||||
"pair": self.pair,
|
||||
"slice": self.slice,
|
||||
"provenance": self.provenance,
|
||||
"mechanisms": list(self.mechanisms),
|
||||
"detected": self.detected,
|
||||
"silent": self.silent,
|
||||
"found_by": [{"mechanism_id": mechanism_id, "pairs": list(pairs)} for mechanism_id, pairs in self.found_by],
|
||||
"hits_vuln": self.hits_vuln,
|
||||
"hits_fixed": self.hits_fixed,
|
||||
"teaches": self.teaches,
|
||||
"leaked_by": list(self.leaked_by),
|
||||
"obligations": list(self.obligations),
|
||||
}
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class LooResult:
|
||||
outcomes: tuple[LooOutcome, ...]
|
||||
per_slice: dict[str, dict[str, float]]
|
||||
per_profile: dict[str, dict[str, float]]
|
||||
per_mechanism: dict[str, dict[str, float]]
|
||||
skipped: tuple[tuple[str, str], ...] = ()
|
||||
per_obligation_kind: dict[str, dict[str, float]] = field(default_factory=dict)
|
||||
teaching_pairs: int = 0
|
||||
degradations: tuple[dict[str, object], ...] = field(default=())
|
||||
|
||||
def to_dict(self) -> dict[str, object]:
|
||||
return {
|
||||
"per_slice": self.per_slice,
|
||||
"per_profile": self.per_profile,
|
||||
"per_mechanism": self.per_mechanism,
|
||||
"per_obligation_kind": self.per_obligation_kind,
|
||||
"outcomes": [outcome.to_dict() for outcome in self.outcomes],
|
||||
"skipped": [{"pair": pair, "reason": reason} for pair, reason in self.skipped],
|
||||
"teaching_pairs": self.teaching_pairs,
|
||||
"degradations": list(self.degradations),
|
||||
}
|
||||
|
||||
|
||||
def evaluate_loo(cases: Sequence[PairCase], *, facts: SemanticFacts | None = None) -> LooResult:
|
||||
loaded = facts if facts is not None else _facts()
|
||||
targets: list[PairCase] = []
|
||||
skipped: list[tuple[str, str]] = []
|
||||
for case in cases:
|
||||
if case.unscorable or case.known_limit:
|
||||
# `PairCase.unscorable` verbatim when the loader computed one: one spelling per corpus fact, and it
|
||||
# already carries the `known_limit:` prefix when the limit is a maintainer's own word rather than one
|
||||
# of the vocabulary's. A case built by hand may carry only the declaration.
|
||||
skipped.append((case.name, case.unscorable or f"known_limit:{case.known_limit}"))
|
||||
elif not case.vuln_file.is_file() or not case.fixed_file.is_file():
|
||||
skipped.append((case.name, "pointer_pair_not_cached" if not case.vendored else "excerpt_missing"))
|
||||
else:
|
||||
targets.append(case)
|
||||
from ..learning.split import GATING_TIERS, holdout_names, is_teacher, refuse_if_holdout
|
||||
|
||||
# learning-harness Req 5.1: every pair is still scored, but only a train-split teacher may seed a store.
|
||||
lessons = {case.name: pair_lessons(case, loaded) for case in targets if is_teacher(case)}
|
||||
outcomes = [_hold_out(case, targets, lessons, loaded) for case in targets]
|
||||
# Req 5.2: the holdout pairs kept out of the teaching set are named, not silently dropped. Non-teachers held
|
||||
# back for their tier or their unscorable reason are already in `skipped` with that reason.
|
||||
excluded = [case.name for case in targets if case.name not in lessons]
|
||||
refusal = refuse_if_holdout(excluded, targets)
|
||||
refusals = (refusal.degradation(),) if refusal is not None else ()
|
||||
withheld = tuple(
|
||||
(case.name, "tier" if case.review_tier not in GATING_TIERS else "unscorable")
|
||||
for case in targets
|
||||
if case.name in set(excluded) - set(holdout_names(targets))
|
||||
)
|
||||
skipped.extend((name, f"not_a_teacher:{why}") for name, why in withheld)
|
||||
return LooResult(
|
||||
outcomes=tuple(outcomes),
|
||||
per_slice=_group(outcomes, lambda item: (item.slice,)),
|
||||
per_profile=_group(outcomes, lambda item: (item.provenance,)),
|
||||
per_mechanism=_group(outcomes, lambda item: item.mechanisms),
|
||||
per_obligation_kind=_group(outcomes, lambda item: item.obligations),
|
||||
skipped=tuple(skipped),
|
||||
teaching_pairs=sum(1 for shapes in lessons.values() if shapes),
|
||||
degradations=(*refusals, *_unique_degradations(outcomes)),
|
||||
)
|
||||
|
||||
|
||||
def _unique_degradations(outcomes: Sequence[LooOutcome]) -> tuple[dict[str, object], ...]:
|
||||
seen: set[str] = set()
|
||||
unique: list[dict[str, object]] = []
|
||||
for outcome in outcomes:
|
||||
for item in outcome.degradations:
|
||||
key = repr(sorted(item.items()))
|
||||
if key not in seen:
|
||||
seen.add(key)
|
||||
unique.append(item)
|
||||
return tuple(unique)
|
||||
|
||||
|
||||
def _hold_out(case: PairCase, targets: Sequence[PairCase], lessons: dict[str, list[Lesson]], facts: SemanticFacts) -> LooOutcome:
|
||||
with tempfile.TemporaryDirectory(prefix="ousast-loo-") as scratch:
|
||||
root = Path(scratch)
|
||||
store = MechanismStore(root / "mechanisms.jsonl")
|
||||
for teacher in targets:
|
||||
if teacher.name == case.name:
|
||||
continue
|
||||
for lesson in lessons.get(teacher.name, ()):
|
||||
append_from_pair(
|
||||
store,
|
||||
lesson.shape,
|
||||
summary=lesson.summary,
|
||||
cwe=lesson.cwe,
|
||||
pair=teacher.name,
|
||||
provenance=teacher.provenance,
|
||||
tier=teacher.review_tier,
|
||||
)
|
||||
scored = score_pair_with_store(case, store, facts, root=root)
|
||||
return LooOutcome(**{**scored.__dict__, "teaches": bool(lessons.get(case.name))})
|
||||
|
||||
|
||||
def score_pair_with_store(case: PairCase, store: MechanismStore, facts: SemanticFacts, *, root: Path | None = None) -> LooOutcome:
|
||||
"""Search both sides of one pair with a fixed store and score it with the pair rules (shared by LOO and the lever)."""
|
||||
labeled_mechanisms = tuple(sorted({row.mechanism for row in case.expected if row.mechanism}))
|
||||
labeled_obligations = tuple(sorted({str(row.obligation) for row in case.expected if getattr(row, "obligation", None)}))
|
||||
loaded = store.load()
|
||||
records = {record.id: record for record in corpus_mechanisms(loaded)}
|
||||
obligation_vuln: list[StaticFinding] = []
|
||||
obligation_fix: list[StaticFinding] = []
|
||||
with tempfile.TemporaryDirectory(prefix="ousast-loo-pair-") as scratch:
|
||||
base = root if root is not None else Path(scratch)
|
||||
vuln_root = _materialize(base / "vuln", case.vuln_file, case.relpath)
|
||||
fix_root = _materialize(base / "fixed", case.fixed_file, case.relpath)
|
||||
vuln_search = search_tree(vuln_root, _targets(vuln_root), store, facts, max_mechanisms=len(records) or 1)
|
||||
fix_search = search_tree(fix_root, _targets(fix_root), store, facts, max_mechanisms=len(records) or 1)
|
||||
vuln_hits, fix_hits = vuln_search.hits, fix_search.hits
|
||||
if labeled_obligations:
|
||||
shapes = obligation_mechanisms(loaded)
|
||||
obligation_vuln = _obligation_findings(vuln_root, shapes)
|
||||
obligation_fix = _obligation_findings(fix_root, shapes)
|
||||
text = case.vuln_file.read_text(errors="ignore")
|
||||
ranges = named_function_ranges(case.relpath, text, _parse_language(case)) or ()
|
||||
found: list[tuple[str, tuple[str, ...]]] = []
|
||||
for finding in obligation_vuln:
|
||||
record = _known_fix_record(finding, records)
|
||||
if record is None or finding.line is None or not _inside_label(case, finding.line, ranges):
|
||||
continue
|
||||
if any(f"obligation:{kind}" in finding.tags for kind in labeled_obligations):
|
||||
found.append((record.id, tuple(record.pairs)))
|
||||
obligation_leaks = tuple(
|
||||
sorted({record.id for finding in obligation_fix if (record := _known_fix_record(finding, records)) is not None})
|
||||
)
|
||||
for hit in vuln_hits:
|
||||
record = records.get(hit.mechanism_id)
|
||||
if record is None or not _inside_label(case, hit.line, ranges):
|
||||
continue
|
||||
if _names_label(case, record, hit.sink_name):
|
||||
found.append((record.id, tuple(record.pairs)))
|
||||
return LooOutcome(
|
||||
pair=case.name,
|
||||
slice=case.slice,
|
||||
provenance=case.provenance,
|
||||
mechanisms=labeled_mechanisms,
|
||||
detected=bool(found),
|
||||
silent=not fix_hits and not obligation_leaks,
|
||||
found_by=tuple(found),
|
||||
hits_vuln=len(vuln_hits) + len(obligation_vuln),
|
||||
hits_fixed=len(fix_hits) + len(obligation_fix),
|
||||
teaches=False,
|
||||
leaked_by=tuple(sorted({hit.mechanism_id for hit in fix_hits})) + obligation_leaks,
|
||||
degradations=tuple(vuln_search.degradations) + tuple(fix_search.degradations),
|
||||
obligations=labeled_obligations,
|
||||
)
|
||||
|
||||
|
||||
def _obligation_findings(root: Path, shapes: Sequence[object]) -> list[StaticFinding]:
|
||||
"""Function-local obligation findings for one materialized side, with the held-out store's obligation shapes as known fixes."""
|
||||
from ..config import ObligationsConfig
|
||||
from ..mapping import analyze_entry_points
|
||||
from .ir import parse_file
|
||||
|
||||
targets = _targets(root)
|
||||
irs: dict[str, tuple[object, str]] = {}
|
||||
texts: dict[str, str] = {}
|
||||
for target in targets:
|
||||
try:
|
||||
text = (root / target.path).read_text(errors="ignore")
|
||||
except OSError:
|
||||
continue
|
||||
texts[target.path] = text
|
||||
irs[target.path] = (parse_file(target.path, text, target.language), text)
|
||||
result = check_obligations(
|
||||
irs=irs, # type: ignore[arg-type]
|
||||
entries=analyze_entry_points(root, targets),
|
||||
facts=load_obligation_facts(),
|
||||
flow_facts=_facts(),
|
||||
policy=None,
|
||||
paths=(),
|
||||
dominance=OrderDominance(texts=texts),
|
||||
store_shapes=shapes,
|
||||
min_siblings=ObligationsConfig().min_siblings,
|
||||
)
|
||||
return findings_to_static(result)
|
||||
|
||||
|
||||
def _known_fix_record(finding: StaticFinding, records: dict[str, Mechanism]) -> Mechanism | None:
|
||||
for tag in finding.tags:
|
||||
if tag.startswith("mechanism:"):
|
||||
return records.get(tag.split(":", 1)[1])
|
||||
return None
|
||||
|
||||
|
||||
def _inside_label(case: PairCase, line: int, ranges: Sequence[tuple[str, int, int]]) -> bool:
|
||||
functions = {row.function for row in case.expected if row.function}
|
||||
if not functions:
|
||||
return True
|
||||
return any(start <= line <= end for function in functions for start, end in spans_named(ranges, function))
|
||||
|
||||
|
||||
def _names_label(case: PairCase, record: Mechanism, sink_name: str) -> bool:
|
||||
"""The hit names the pair's labeled mechanism, or its labeled sink when the row carries one."""
|
||||
shape = Shape.from_dict(record.shape or {})
|
||||
for row in case.expected:
|
||||
if row.mechanism and row.mechanism == shape.mechanism:
|
||||
return True
|
||||
if row.sink and trailing_name(row.sink) == sink_name:
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
def _group(outcomes: Sequence[LooOutcome], keys: object) -> dict[str, dict[str, float]]:
|
||||
buckets: dict[str, list[LooOutcome]] = {}
|
||||
for outcome in outcomes:
|
||||
for key in keys(outcome): # type: ignore[operator]
|
||||
buckets.setdefault(str(key), []).append(outcome)
|
||||
return {name: _metrics(items) for name, items in sorted(buckets.items())}
|
||||
|
||||
|
||||
def _metrics(outcomes: Sequence[LooOutcome]) -> dict[str, float]:
|
||||
pairs = len(outcomes)
|
||||
detected = sum(1 for item in outcomes if item.detected)
|
||||
silent = sum(1 for item in outcomes if item.silent)
|
||||
recall = detected / pairs if pairs else 0.0
|
||||
silence = silent / pairs if pairs else 0.0
|
||||
return {
|
||||
"pairs": pairs,
|
||||
"detected": detected,
|
||||
"silent": silent,
|
||||
"recall": recall,
|
||||
"silence": silence,
|
||||
"youden": recall + silence - 1.0 if pairs else 0.0,
|
||||
"teaching": sum(1 for item in outcomes if item.teaches),
|
||||
}
|
||||
|
||||
|
||||
def _facts() -> SemanticFacts:
|
||||
try:
|
||||
return load_facts()
|
||||
except FactLoadError:
|
||||
return SemanticFacts(version="", sources=(), sinks=(), sanitizers=())
|
||||
@@ -1,363 +0,0 @@
|
||||
"""Append-only mechanism memory. OpenRouter embeddings rank prove budget; never demote."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import json
|
||||
import math
|
||||
import os
|
||||
import uuid
|
||||
from collections.abc import Sequence
|
||||
from dataclasses import asdict, dataclass
|
||||
from pathlib import Path
|
||||
|
||||
from ..complexity.map import Hotspot
|
||||
from ..findings import StaticFinding
|
||||
from ..index import CodeChunk, EmbeddingClient, VectorIndex, VectorRecord, write_vector_index
|
||||
from .obligations.shapes import ObligationShape
|
||||
from .overlay import OverlayRecord
|
||||
from .variants import Shape
|
||||
|
||||
DEFAULT_MECHANISM_LOG = Path(".openultrasast/mechanisms.jsonl")
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class Mechanism:
|
||||
id: str
|
||||
summary: str
|
||||
cwe: str
|
||||
language: str
|
||||
tags: tuple[str, ...]
|
||||
keywords: tuple[str, ...]
|
||||
what_made_it_exploitable: str
|
||||
source_finding_id: str = ""
|
||||
source_repo: str = ""
|
||||
# corpus-seeded-mechanisms (additive; old rows load with these defaults)
|
||||
origin: str = "sandbox" # "sandbox" (proven finding) | "corpus" (derived from a trusted pair)
|
||||
review_tier: str = ""
|
||||
pairs: tuple[str, ...] = ()
|
||||
shape: dict[str, object] | None = None # Shape.to_dict(); None on sandbox rows (not searchable)
|
||||
guard: str = "none"
|
||||
retracted: bool = False # tombstone row appended by the improve lever; load() folds the record away
|
||||
|
||||
|
||||
class MechanismStore:
|
||||
def __init__(self, log_path: Path, cache_path: Path | None = None) -> None:
|
||||
self.log_path = log_path
|
||||
self.cache_path = cache_path if cache_path is not None else log_path.with_suffix(".index.json")
|
||||
|
||||
def load(self) -> tuple[Mechanism, ...]:
|
||||
if not self.log_path.is_file():
|
||||
return ()
|
||||
by_id: dict[str, Mechanism] = {} # append-only log; the latest row for an id is its current state
|
||||
for line in self.log_path.read_text().splitlines():
|
||||
if not line.strip():
|
||||
continue
|
||||
payload = json.loads(line)
|
||||
record = _mechanism_from_payload(payload)
|
||||
if record.retracted:
|
||||
by_id.pop(record.id, None)
|
||||
else:
|
||||
by_id[record.id] = record
|
||||
return tuple(by_id.values())
|
||||
|
||||
def append(self, mechanism: Mechanism) -> None:
|
||||
self.log_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
with self.log_path.open("a") as handle:
|
||||
handle.write(json.dumps(asdict(mechanism), sort_keys=True) + "\n")
|
||||
|
||||
|
||||
def append_mechanism(
|
||||
store: MechanismStore,
|
||||
*,
|
||||
summary: str,
|
||||
cwe: str,
|
||||
language: str,
|
||||
tags: Sequence[str],
|
||||
what_made_it_exploitable: str,
|
||||
source_finding_id: str = "",
|
||||
source_repo: str = "",
|
||||
keywords: Sequence[str] = (),
|
||||
) -> Mechanism:
|
||||
mechanism = Mechanism(
|
||||
id=str(uuid.uuid4()),
|
||||
summary=summary,
|
||||
cwe=cwe,
|
||||
language=language,
|
||||
tags=tuple(tags),
|
||||
keywords=tuple(keywords),
|
||||
what_made_it_exploitable=what_made_it_exploitable,
|
||||
source_finding_id=source_finding_id,
|
||||
source_repo=source_repo,
|
||||
)
|
||||
store.append(mechanism)
|
||||
return mechanism
|
||||
|
||||
|
||||
_TIER_RANK = {"seeded": 1, "reviewed": 2}
|
||||
|
||||
|
||||
def corpus_mechanism_id(shape: Shape | ObligationShape) -> 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]
|
||||
|
||||
|
||||
def append_from_pair(
|
||||
store: MechanismStore,
|
||||
shape: Shape | ObligationShape,
|
||||
*,
|
||||
summary: str,
|
||||
cwe: str,
|
||||
pair: str,
|
||||
provenance: str,
|
||||
tier: str,
|
||||
) -> Mechanism:
|
||||
"""Second writer (Req 1.3, 1.4): one record per shape, ``origin = "corpus"``, provenance lists every pair that taught it."""
|
||||
mechanism_id = corpus_mechanism_id(shape)
|
||||
existing = next((item for item in store.load() if item.id == mechanism_id), None)
|
||||
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:")))
|
||||
if isinstance(shape, Shape):
|
||||
positions = ", ".join(f"arg{p} ({k})" for p, k in zip(shape.source_positions, shape.source_kinds, strict=True))
|
||||
exploitable = f"{positions} reaches {shape.sink_name}/{shape.arity}; the fix added {shape.guard}"
|
||||
keywords: tuple[str, ...] = (shape.sink_name, shape.guard, *shape.source_kinds)
|
||||
guard = shape.guard
|
||||
else: # obligation shape: the discharger the fix carries is the 'guard' the reports show
|
||||
exploitable = f"{shape.operation_kind} reached without {shape.discharger_kind}; the fix binds {shape.provenance}"
|
||||
keywords = (shape.operation_kind, shape.discharger_kind, shape.provenance, shape.resource_class)
|
||||
guard = shape.discharger_kind
|
||||
record = Mechanism(
|
||||
id=mechanism_id,
|
||||
summary=summary,
|
||||
cwe=cwe,
|
||||
language=shape.language,
|
||||
tags=(
|
||||
shape.mechanism,
|
||||
f"mechanism:{shape.mechanism}",
|
||||
f"guard:{guard}",
|
||||
*sorted({*inherited, f"provenance:{provenance}", f"tier:{tier}"}),
|
||||
),
|
||||
keywords=keywords,
|
||||
what_made_it_exploitable=exploitable,
|
||||
source_finding_id="",
|
||||
source_repo="",
|
||||
origin="corpus",
|
||||
review_tier=tier,
|
||||
pairs=pairs,
|
||||
shape=shape.to_dict(),
|
||||
guard=guard,
|
||||
)
|
||||
store.append(record)
|
||||
return record
|
||||
|
||||
|
||||
def corpus_mechanisms(records: Sequence[Mechanism]) -> tuple[Mechanism, ...]:
|
||||
"""Records that variant search may use: corpus origin with a shape."""
|
||||
return tuple(item for item in records if item.origin == "corpus" and item.shape is not None)
|
||||
|
||||
|
||||
def order_promotions(
|
||||
hotspots: Sequence[Hotspot],
|
||||
records: Sequence[OverlayRecord],
|
||||
findings: Sequence[StaticFinding],
|
||||
*,
|
||||
store: MechanismStore | None = None,
|
||||
client: EmbeddingClient | None = None,
|
||||
model: str | None = None,
|
||||
) -> tuple[tuple[Hotspot, ...], str | None]:
|
||||
"""Heuristic order, then optional OpenRouter cosine boost. Never drops a promotion."""
|
||||
heuristic = tuple(sorted(hotspots, key=lambda item: (-item.score, item.path, item.function_name or "")))
|
||||
if store is None:
|
||||
return heuristic, None
|
||||
mechanisms = store.load()
|
||||
if not mechanisms:
|
||||
return heuristic, None
|
||||
if client is None or not model:
|
||||
return heuristic, "embeddings_unavailable"
|
||||
overlay_by_id = {record.proposal_id: record for record in records}
|
||||
finding_by_id = {finding.finding_id: finding for finding in findings}
|
||||
try:
|
||||
ranked = _rank_with_embeddings(heuristic, overlay_by_id, finding_by_id, mechanisms, store, client, model)
|
||||
except Exception:
|
||||
return heuristic, "embed_failed"
|
||||
return ranked, None
|
||||
|
||||
|
||||
def overlay_text(record: OverlayRecord) -> str:
|
||||
parts = [record.cwe, record.reason, *record.sources, *record.sinks, *record.sanitizers]
|
||||
return " ".join(part for part in parts if part)
|
||||
|
||||
|
||||
def mechanism_text(mechanism: Mechanism) -> str:
|
||||
return " ".join(
|
||||
part
|
||||
for part in (mechanism.summary, mechanism.what_made_it_exploitable, mechanism.cwe, *mechanism.tags, *mechanism.keywords)
|
||||
if part
|
||||
)
|
||||
|
||||
|
||||
def _rank_with_embeddings(
|
||||
hotspots: tuple[Hotspot, ...],
|
||||
overlay_by_id: dict[str, OverlayRecord],
|
||||
finding_by_id: dict[str, StaticFinding],
|
||||
mechanisms: tuple[Mechanism, ...],
|
||||
store: MechanismStore,
|
||||
client: EmbeddingClient,
|
||||
model: str,
|
||||
) -> tuple[Hotspot, ...]:
|
||||
vectors = _mechanism_vectors(mechanisms, store, client, model)
|
||||
scored: list[tuple[float, Hotspot]] = []
|
||||
for hotspot in hotspots:
|
||||
record = _overlay_for_hotspot(hotspot, overlay_by_id)
|
||||
finding = _finding_for_hotspot(hotspot, finding_by_id)
|
||||
language = record.language if record is not None else ""
|
||||
cwe = record.cwe if record is not None else ""
|
||||
tags = tuple(finding.tags) if finding is not None else ()
|
||||
filtered = [item for item in mechanisms if _hard_filter(item, language, cwe, tags)]
|
||||
boost = 0.0
|
||||
if record is not None and filtered:
|
||||
query = client.embed(model=model, inputs=[overlay_text(record)])[0]
|
||||
for mechanism in filtered:
|
||||
vector = vectors.get(mechanism.id)
|
||||
if vector is None:
|
||||
continue
|
||||
boost = max(boost, _cosine(query, vector))
|
||||
scored.append((hotspot.score + boost, hotspot))
|
||||
scored.sort(key=lambda item: (-item[0], item[1].path, item[1].function_name or ""))
|
||||
return tuple(item[1] for item in scored)
|
||||
|
||||
|
||||
def _mechanism_vectors(
|
||||
mechanisms: tuple[Mechanism, ...],
|
||||
store: MechanismStore,
|
||||
client: EmbeddingClient,
|
||||
model: str,
|
||||
) -> dict[str, list[float]]:
|
||||
vectors: dict[str, list[float]] = {}
|
||||
missing: list[Mechanism] = []
|
||||
cached = _read_cache(store.cache_path, model)
|
||||
for mechanism in mechanisms:
|
||||
key = f"{mechanism.id}:{model}"
|
||||
if key in cached:
|
||||
vectors[mechanism.id] = cached[key]
|
||||
else:
|
||||
missing.append(mechanism)
|
||||
if missing:
|
||||
embeddings = client.embed(model=model, inputs=[mechanism_text(item) for item in missing])
|
||||
for mechanism, embedding in zip(missing, embeddings, strict=True):
|
||||
vectors[mechanism.id] = embedding
|
||||
records = [
|
||||
VectorRecord(
|
||||
chunk=CodeChunk(
|
||||
chunk_id=f"{mechanism.id}:{model}",
|
||||
namespace="mechanisms",
|
||||
path=mechanism.id,
|
||||
language=mechanism.language or "text",
|
||||
start_line=1,
|
||||
end_line=1,
|
||||
text=mechanism_text(mechanism),
|
||||
metadata={"mechanism_id": mechanism.id, "model": model, "cwe": mechanism.cwe, "language": mechanism.language},
|
||||
),
|
||||
embedding=vectors[mechanism.id],
|
||||
)
|
||||
for mechanism in mechanisms
|
||||
if mechanism.id in vectors
|
||||
]
|
||||
index = VectorIndex(
|
||||
store="json-local",
|
||||
embedding_model=model,
|
||||
repo_root=str(store.log_path.parent),
|
||||
repo_commit=None,
|
||||
records=records,
|
||||
)
|
||||
write_vector_index(index, store.cache_path)
|
||||
return vectors
|
||||
|
||||
|
||||
def _read_cache(path: Path, model: str) -> dict[str, list[float]]:
|
||||
if not path.is_file():
|
||||
return {}
|
||||
try:
|
||||
payload = json.loads(path.read_text())
|
||||
except json.JSONDecodeError:
|
||||
return {}
|
||||
if str(payload.get("embedding_model")) != model:
|
||||
return {}
|
||||
cached: dict[str, list[float]] = {}
|
||||
for item in payload.get("records", []):
|
||||
chunk = item.get("chunk", {})
|
||||
metadata = chunk.get("metadata", {})
|
||||
mechanism_id = str(metadata.get("mechanism_id", ""))
|
||||
if mechanism_id:
|
||||
cached[f"{mechanism_id}:{model}"] = [float(value) for value in item.get("embedding", [])]
|
||||
return cached
|
||||
|
||||
|
||||
def _hard_filter(mechanism: Mechanism, language: str, cwe: str, tags: Sequence[str]) -> bool:
|
||||
if mechanism.language and language and mechanism.language != language:
|
||||
return False
|
||||
if mechanism.cwe and cwe and mechanism.cwe != cwe:
|
||||
return False
|
||||
tag_ok = not mechanism.tags or not tags or bool(set(mechanism.tags) & set(tags))
|
||||
return tag_ok
|
||||
|
||||
|
||||
def _overlay_for_hotspot(hotspot: Hotspot, overlay_by_id: dict[str, OverlayRecord]) -> OverlayRecord | None:
|
||||
for finding_id in hotspot.inventory_finding_ids:
|
||||
record = overlay_by_id.get(finding_id)
|
||||
if record is not None:
|
||||
return record
|
||||
return None
|
||||
|
||||
|
||||
def _finding_for_hotspot(hotspot: Hotspot, finding_by_id: dict[str, StaticFinding]) -> StaticFinding | None:
|
||||
for finding_id in hotspot.inventory_finding_ids:
|
||||
finding = finding_by_id.get(finding_id)
|
||||
if finding is not None:
|
||||
return finding
|
||||
return None
|
||||
|
||||
|
||||
def _cosine(left: list[float], right: list[float]) -> float:
|
||||
if len(left) != len(right):
|
||||
return 0.0
|
||||
numerator = sum(a * b for a, b in zip(left, right, strict=True))
|
||||
left_norm = math.sqrt(sum(a * a for a in left))
|
||||
right_norm = math.sqrt(sum(b * b for b in right))
|
||||
if left_norm == 0 or right_norm == 0:
|
||||
return 0.0
|
||||
return numerator / (left_norm * right_norm)
|
||||
|
||||
|
||||
def _mechanism_from_payload(payload: dict[str, object]) -> Mechanism:
|
||||
tags = payload.get("tags") or []
|
||||
keywords = payload.get("keywords") or []
|
||||
pairs = payload.get("pairs") or []
|
||||
shape = payload.get("shape")
|
||||
if not isinstance(tags, list) or not isinstance(keywords, list) or not isinstance(pairs, list):
|
||||
raise ValueError("mechanism tags, keywords and pairs must be lists")
|
||||
return Mechanism(
|
||||
id=str(payload["id"]),
|
||||
summary=str(payload.get("summary", "")),
|
||||
cwe=str(payload.get("cwe", "")),
|
||||
language=str(payload.get("language", "")),
|
||||
tags=tuple(str(item) for item in tags),
|
||||
keywords=tuple(str(item) for item in keywords),
|
||||
what_made_it_exploitable=str(payload.get("what_made_it_exploitable", "")),
|
||||
source_finding_id=str(payload.get("source_finding_id", "")),
|
||||
source_repo=str(payload.get("source_repo", "")),
|
||||
origin=str(payload.get("origin", "sandbox")),
|
||||
review_tier=str(payload.get("review_tier", "")),
|
||||
pairs=tuple(str(item) for item in pairs),
|
||||
shape=dict(shape) if isinstance(shape, dict) else None, # validated where it is used (search_tree, the lever)
|
||||
guard=str(payload.get("guard", "none")),
|
||||
retracted=bool(payload.get("retracted", False)),
|
||||
)
|
||||
|
||||
|
||||
def embeddings_configured(model: str | None) -> bool:
|
||||
return bool(model) and bool(os.environ.get("OPENROUTER_API_KEY"))
|
||||
@@ -14,8 +14,7 @@ from collections.abc import Sequence
|
||||
from dataclasses import dataclass
|
||||
|
||||
from ..facts import SemanticFacts
|
||||
from ..ir import FileIR, FunctionIR
|
||||
from ..variants import trailing_name
|
||||
from ..ir import FileIR, FunctionIR, trailing_name
|
||||
from .facts import DischargerFact, ObligationFacts
|
||||
from .shapes import _call_matches, _constraint_provenance, _is_literal, _operations, _permissive
|
||||
|
||||
|
||||
@@ -17,8 +17,7 @@ from dataclasses import dataclass
|
||||
|
||||
from ..facts import SemanticFacts
|
||||
from ..functions import named_ranges_from_ir
|
||||
from ..ir import CallSite, FileIR, FunctionIR
|
||||
from ..variants import trailing_name
|
||||
from ..ir import CallSite, FileIR, FunctionIR, trailing_name
|
||||
from .facts import DISCHARGER_KINDS, OPERATION_KINDS, PROVENANCE_KINDS, DischargerFact, ObligationFacts, OperationFact
|
||||
|
||||
FAMILY = "obligation"
|
||||
|
||||
@@ -1,12 +0,0 @@
|
||||
"""Prove-budget ranker: heuristic order plus optional mechanism recall."""
|
||||
|
||||
from .mechanisms import MechanismStore, append_mechanism, order_promotions
|
||||
from .prove_filter import filter_promoted_hotspots, promoted_findings
|
||||
|
||||
__all__ = [
|
||||
"MechanismStore",
|
||||
"append_mechanism",
|
||||
"filter_promoted_hotspots",
|
||||
"order_promotions",
|
||||
"promoted_findings",
|
||||
]
|
||||
@@ -1,231 +0,0 @@
|
||||
"""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 FileIR, parse_file
|
||||
from .mechanisms import MechanismStore, append_from_pair
|
||||
from .obligations.facts import ObligationFacts, load_obligation_facts
|
||||
from .obligations.shapes import ABSENCE_MECHANISMS, ObligationShape, derive_obligation, explain_skip
|
||||
from .variants import Shape, 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
|
||||
if not any(row.function and row.mechanism for row in case.expected):
|
||||
skipped.append((case.name, "no_labeled_function"))
|
||||
continue
|
||||
lessons = pair_lessons(case, loaded, vuln_ir=vuln_ir, fixed_ir=fixed_ir, vuln_text=vuln_text, fixed_text=fixed_text)
|
||||
for lesson in lessons:
|
||||
record = append_from_pair(
|
||||
store,
|
||||
lesson.shape,
|
||||
summary=lesson.summary,
|
||||
cwe=lesson.cwe,
|
||||
pair=case.name,
|
||||
provenance=case.provenance,
|
||||
tier=case.review_tier,
|
||||
)
|
||||
record_ids.add(record.id)
|
||||
if lessons:
|
||||
seeded.append(case.name)
|
||||
else:
|
||||
skipped.append((case.name, _skip_detail(case, vuln_ir, fixed_ir, vuln_text, fixed_text, loaded)))
|
||||
return ExportReport(
|
||||
seeded=len(seeded),
|
||||
records=len(record_ids),
|
||||
skipped=tuple(skipped),
|
||||
degradations=tuple(degradations),
|
||||
seeded_pairs=tuple(seeded),
|
||||
)
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class Lesson:
|
||||
"""One shape a trusted pair teaches, with the CWE and a text-free summary (a sink shape or an obligation shape)."""
|
||||
|
||||
shape: Shape | ObligationShape
|
||||
cwe: str
|
||||
summary: str
|
||||
|
||||
|
||||
def pair_lessons(
|
||||
case: PairCase,
|
||||
facts: SemanticFacts,
|
||||
*,
|
||||
vuln_ir: FileIR | None = None,
|
||||
fixed_ir: FileIR | None = None,
|
||||
vuln_text: str | None = None,
|
||||
fixed_text: str | None = None,
|
||||
) -> list[Lesson]:
|
||||
"""Shapes derived from every labeled row of a pair (shared by the exporter and leave-one-out). Empty when a side fails to parse."""
|
||||
vuln_text = vuln_text if vuln_text is not None else _read(case.vuln_file)
|
||||
fixed_text = fixed_text if fixed_text is not None else _read(case.fixed_file)
|
||||
language = _parse_language(case)
|
||||
vuln_ir = vuln_ir if vuln_ir is not None else parse_file(case.relpath, vuln_text, language)
|
||||
fixed_ir = fixed_ir if fixed_ir is not None else parse_file(case.relpath, fixed_text, language)
|
||||
if not vuln_ir.parse_ok or not fixed_ir.parse_ok:
|
||||
return []
|
||||
lessons: list[Lesson] = []
|
||||
for row in case.expected:
|
||||
if not row.function or not row.mechanism:
|
||||
continue
|
||||
obligation_kind = getattr(row, "obligation", None)
|
||||
if obligation_kind or row.mechanism in ABSENCE_MECHANISMS:
|
||||
obligation = derive_obligation(
|
||||
vuln_ir,
|
||||
fixed_ir,
|
||||
function=row.function,
|
||||
obligation=obligation_kind,
|
||||
mechanism=row.mechanism,
|
||||
facts=_obligation_facts(),
|
||||
flow_facts=facts,
|
||||
vuln_text=vuln_text,
|
||||
fixed_text=fixed_text,
|
||||
)
|
||||
if obligation is not None:
|
||||
summary = (
|
||||
f"{obligation.mechanism}: {obligation.operation_kind} without {obligation.discharger_kind} ({row.cwe}); "
|
||||
f"fix binds {obligation.provenance}"
|
||||
)
|
||||
lessons.append(Lesson(shape=obligation, cwe=row.cwe, summary=summary))
|
||||
continue
|
||||
# an absence row that teaches no obligation may still teach a sink shape (a weak literal at a sink, say)
|
||||
shape = derive_shape(
|
||||
vuln_ir,
|
||||
fixed_ir,
|
||||
function=row.function,
|
||||
sink=row.sink,
|
||||
line=row.line,
|
||||
mechanism=row.mechanism,
|
||||
facts=facts,
|
||||
vuln_text=vuln_text,
|
||||
fixed_text=fixed_text,
|
||||
cwe=row.cwe or None,
|
||||
)
|
||||
if shape is not None:
|
||||
summary = f"{shape.mechanism}: {'/'.join(shape.source_kinds)} into {shape.sink_name} ({row.cwe}); fix adds {shape.guard}"
|
||||
lessons.append(Lesson(shape=shape, cwe=row.cwe, summary=summary))
|
||||
return lessons
|
||||
|
||||
|
||||
def _skip_detail(case: PairCase, vuln_ir: FileIR, fixed_ir: FileIR, vuln_text: str, fixed_text: str, flow_facts: SemanticFacts) -> str:
|
||||
"""Why a parsing, labeled pair taught nothing: the obligation derivation's reason for absence rows, else the sink reason."""
|
||||
for row in case.expected:
|
||||
if row.function and row.mechanism and (getattr(row, "obligation", None) or row.mechanism in ABSENCE_MECHANISMS):
|
||||
return explain_skip(
|
||||
vuln_ir,
|
||||
fixed_ir,
|
||||
function=row.function,
|
||||
vuln_text=vuln_text,
|
||||
fixed_text=fixed_text,
|
||||
obligation=getattr(row, "obligation", None),
|
||||
facts=_obligation_facts(),
|
||||
flow_facts=flow_facts,
|
||||
)
|
||||
return "no_labeled_sink_call_site"
|
||||
|
||||
|
||||
_OBLIGATION_FACTS: list[ObligationFacts] = []
|
||||
|
||||
|
||||
def _obligation_facts() -> ObligationFacts:
|
||||
if not _OBLIGATION_FACTS:
|
||||
_OBLIGATION_FACTS.append(load_obligation_facts())
|
||||
return _OBLIGATION_FACTS[0]
|
||||
|
||||
|
||||
def _skip_reason(case: PairCase) -> str | None:
|
||||
from ..learning.split import is_teacher
|
||||
|
||||
if case.review_tier not in GATING_TIERS:
|
||||
return f"tier:{case.review_tier}"
|
||||
if case.unscorable or case.known_limit:
|
||||
# `PairCase.unscorable` verbatim when the loader computed one, the same string every other path reports:
|
||||
# `identical_twin` declared in the catalog and `identical_twin` computed from the bytes are one fact, and
|
||||
# calling either of them `split:train` is simply untrue. A case built by hand may carry only the
|
||||
# declaration (Req 5.1, 10.3).
|
||||
return case.unscorable or f"known_limit:{case.known_limit}"
|
||||
if not is_teacher(case) and case.vendored:
|
||||
return f"split:{case.split}"
|
||||
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=())
|
||||
@@ -1,125 +0,0 @@
|
||||
"""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 = [record for record in corpus_mechanisms(store.load()) if (record.shape or {}).get("family", "sink") == "sink"]
|
||||
records = records[: max(max_mechanisms, 0)]
|
||||
shapes: list[Shape] = []
|
||||
ids: dict[str, str] = {}
|
||||
degradations: list[dict[str, object]] = []
|
||||
for record in records:
|
||||
assert record.shape is not None
|
||||
try:
|
||||
shape = Shape.from_dict(record.shape)
|
||||
except (ValueError, TypeError) as exc: # one malformed operator-visible row must not abort the scan
|
||||
degradations.append(
|
||||
{"stage": "variants", "reason": "variants_store_row_invalid", "mechanism_id": record.id, "detail": str(exc)[:120]}
|
||||
)
|
||||
continue
|
||||
shapes.append(shape)
|
||||
ids[shape.key()] = record.id
|
||||
if not shapes:
|
||||
return SearchResult(hits=(), mechanisms_searched=0, files_searched=0, degradations=tuple(degradations))
|
||||
hits: list[VariantHit] = []
|
||||
files = 0
|
||||
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
|
||||
@@ -1,347 +0,0 @@
|
||||
"""Structural mechanism shapes derived from labeled pairs (corpus-seeded-mechanisms, Req 2).
|
||||
|
||||
A shape is the closed, text-free description of how a known vulnerable call looked: the
|
||||
sink's trailing callee name, its arity, which argument positions carried source-derived
|
||||
values and of what kind, and the kind of guard the fix added. Shapes are derived from
|
||||
``FileIR`` (never from regex over code) and compared structurally, so a variant in another
|
||||
project matches even when every identifier differs. Stdlib plus ``.ir``, ``.facts`` and
|
||||
``.functions`` only; never imports the overlay, the scorer, or the store.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
from collections.abc import Mapping, Sequence
|
||||
from dataclasses import dataclass
|
||||
|
||||
from .facts import SemanticFacts, SinkFact
|
||||
from .functions import named_ranges_from_ir
|
||||
from .ir import Bind, CallSite, FileIR, FunctionIR
|
||||
|
||||
GUARD_KINDS = ("null_test", "bounds_test", "allowlist_test", "auth_check", "parameterized_call", "type_change", "none")
|
||||
SOURCE_KINDS = ("fact_source", "parameter", "container_read")
|
||||
_MAX_BIND_DEPTH = 8
|
||||
|
||||
# Closed guard classification over the statements the fix added inside the labeled function.
|
||||
# Order matters: the first matching kind wins. These classify the *fix*, they never detect.
|
||||
_GUARD_PATTERNS: tuple[tuple[str, re.Pattern[str]], ...] = (
|
||||
("parameterized_call", re.compile(r"execute\w*\([^,]+,\s*[\(\[]|prepareStatement\(|\?\s*['\"]?\s*[,)]|\$\d\b|%s['\"]\s*,\s*[\(\[]")),
|
||||
(
|
||||
"auth_check",
|
||||
re.compile(
|
||||
r"\b(request\.user|req\.user|current_user|request\.state|session\[|is_authenticated|authoriz|require_auth"
|
||||
r"|login_required|verify_token|check_permission|owner_id\s*[!=]=)"
|
||||
),
|
||||
),
|
||||
(
|
||||
"allowlist_test",
|
||||
re.compile(
|
||||
r"\bnot in\b|\bin\s*[\(\{\[]|\.includes\(|\.has\(|\bstartswith\(|\bstartsWith\(|\bendswith\(|\bendsWith\("
|
||||
r"|re\.(?:match|fullmatch)\(|\.test\(|allow_?list|whitelist|\bALLOWED\b|\bsafe_join\(|secure_filename\("
|
||||
r"|\brealpath\(|\bnormpath\(|\bresolve\("
|
||||
),
|
||||
),
|
||||
("bounds_test", re.compile(r"\blen\(|\.length\b|\bsizeof\(|\bstrn?len\(|\bsnprintf\(|\bstrlcpy\(|\bmin\(|\bmax\(|\s(?:<|>|<=|>=)\s")),
|
||||
(
|
||||
"null_test",
|
||||
re.compile(
|
||||
r"\bis None\b|\bis not None\b|[!=]= *NULL\b|[!=]==? *null\b|\bnullptr\b|\bif\s*\(\s*!\s*\w|\bif not \w+\s*:"
|
||||
r"|\bif\s+\w+\s+is\s+None|\?\?|\?\."
|
||||
),
|
||||
),
|
||||
(
|
||||
"type_change",
|
||||
re.compile(r"\bint\(|\bparseInt\(|\bNumber\(|\(int\)|\(size_t\)|\(unsigned\b|\bstr\(|\.toString\(|\bstatic_cast<|\bBigInt\("),
|
||||
),
|
||||
)
|
||||
# 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)
|
||||
class Shape:
|
||||
language: str
|
||||
sink_name: str # trailing callee segment
|
||||
arity: int
|
||||
source_positions: tuple[int, ...]
|
||||
source_kinds: tuple[str, ...]
|
||||
guard: str # GUARD_KINDS
|
||||
mechanism: str # closed vocabulary id from the pair label
|
||||
|
||||
def key(self) -> str:
|
||||
"""Stable identity of the shape; carries no path, line, or literal."""
|
||||
positions = ",".join(str(p) for p in self.source_positions)
|
||||
kinds = ",".join(self.source_kinds)
|
||||
return f"{self.language}|{self.sink_name}/{self.arity}|pos={positions}|kinds={kinds}|guard={self.guard}|{self.mechanism}"
|
||||
|
||||
def to_dict(self) -> dict[str, object]:
|
||||
return {
|
||||
"language": self.language,
|
||||
"sink_name": self.sink_name,
|
||||
"arity": self.arity,
|
||||
"source_positions": list(self.source_positions),
|
||||
"source_kinds": list(self.source_kinds),
|
||||
"guard": self.guard,
|
||||
"mechanism": self.mechanism,
|
||||
}
|
||||
|
||||
@classmethod
|
||||
def from_dict(cls, payload: dict[str, object]) -> Shape:
|
||||
positions = payload.get("source_positions") or []
|
||||
kinds = payload.get("source_kinds") or []
|
||||
if not isinstance(positions, list) or not isinstance(kinds, list):
|
||||
raise ValueError("shape positions and kinds must be lists")
|
||||
guard = str(payload.get("guard", "none"))
|
||||
if guard not in GUARD_KINDS:
|
||||
raise ValueError(f"unknown guard kind {guard!r}")
|
||||
return cls(
|
||||
language=str(payload.get("language", "")),
|
||||
sink_name=str(payload.get("sink_name", "")),
|
||||
arity=int(str(payload.get("arity", 0))),
|
||||
source_positions=tuple(int(p) for p in positions),
|
||||
source_kinds=tuple(str(k) for k in kinds),
|
||||
guard=guard,
|
||||
mechanism=str(payload.get("mechanism", "other")),
|
||||
)
|
||||
|
||||
|
||||
def trailing_name(callee: str) -> str:
|
||||
"""``os.path.join`` -> ``join``; ``Runtime.getRuntime().exec`` -> ``exec``; keeps the last identifier segment."""
|
||||
tail = callee.rsplit(".", 1)[-1].rsplit("::", 1)[-1].rsplit("->", 1)[-1]
|
||||
match = re.search(r"[A-Za-z_$][\w$]*\s*$", tail)
|
||||
return match.group(0).strip() if match else tail
|
||||
|
||||
|
||||
def derive_shape(
|
||||
vuln: FileIR,
|
||||
fixed: FileIR,
|
||||
*,
|
||||
function: str,
|
||||
sink: str | None,
|
||||
line: int | None,
|
||||
mechanism: str,
|
||||
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.
|
||||
|
||||
None when either side failed to parse, the function or the labeled call site is not found, or no argument of the
|
||||
call carries a source (a constant-only call teaches nothing).
|
||||
"""
|
||||
if not vuln.parse_ok or not fixed.parse_ok:
|
||||
return None
|
||||
target = _function_named(vuln, function, vuln_text)
|
||||
if target is None:
|
||||
return None
|
||||
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, 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
|
||||
guard = classify_guard(vuln, fixed, function=function, vuln_text=vuln_text, fixed_text=fixed_text)
|
||||
return Shape(
|
||||
language=vuln.language,
|
||||
sink_name=trailing_name(call.name),
|
||||
arity=len(call.arg_texts),
|
||||
source_positions=positions,
|
||||
source_kinds=tuple(kind for kind in kinds if kind is not None),
|
||||
guard=guard,
|
||||
mechanism=mechanism,
|
||||
)
|
||||
|
||||
|
||||
def classify_guard(vuln: FileIR, fixed: FileIR, *, function: str, vuln_text: str, fixed_text: str) -> str:
|
||||
"""Closed guard kind from the statements inside ``function`` that exist only on the fixed side (``none`` when unknown)."""
|
||||
before = _function_lines(vuln, function, vuln_text)
|
||||
after = _function_lines(fixed, function, fixed_text)
|
||||
added = [line for line in after if line not in set(before)]
|
||||
for kind, pattern in _GUARD_PATTERNS:
|
||||
if any(pattern.search(line) for line in added):
|
||||
return kind
|
||||
return "none"
|
||||
|
||||
|
||||
def source_kind_of_name(function: FunctionIR, name: str, facts: SemanticFacts, *, before_line: int, depth: int = 0) -> str | None:
|
||||
"""How ``name`` came to carry attacker data inside ``function``: a fact source, a parameter, or a container read."""
|
||||
if depth > _MAX_BIND_DEPTH:
|
||||
return None
|
||||
bind = _last_bind(function, name, before_line)
|
||||
if bind is not None:
|
||||
if _source_in(bind.value_text, facts):
|
||||
return "fact_source"
|
||||
if _CONTAINER_READ.search(bind.value_text):
|
||||
return "container_read"
|
||||
for inner in bind.names:
|
||||
if inner == name:
|
||||
continue
|
||||
kind = source_kind_of_name(function, inner, facts, before_line=bind.line, depth=depth + 1)
|
||||
if kind is not None:
|
||||
return kind
|
||||
if name in function.params:
|
||||
return "parameter"
|
||||
return None
|
||||
|
||||
|
||||
def _source_kind_of_argument(function: FunctionIR, call: CallSite, index: int, facts: SemanticFacts) -> str | None:
|
||||
text = call.arg_texts[index]
|
||||
if _source_in(text, facts):
|
||||
return "fact_source"
|
||||
if call.arg_is_constant[index] and not call.arg_names[index]:
|
||||
return None
|
||||
if _CONTAINER_READ.search(text) and any(name in function.params for name in call.arg_names[index]):
|
||||
return "container_read"
|
||||
found: list[str] = []
|
||||
for name in call.arg_names[index]:
|
||||
kind = source_kind_of_name(function, name, facts, before_line=call.line + 1)
|
||||
if kind is not None:
|
||||
found.append(kind)
|
||||
for preferred in SOURCE_KINDS:
|
||||
if preferred in found:
|
||||
return preferred
|
||||
return None
|
||||
|
||||
|
||||
def _source_in(text: str, facts: SemanticFacts) -> bool:
|
||||
return any(pattern and pattern in text for source in facts.sources for pattern in source.patterns)
|
||||
|
||||
|
||||
def _last_bind(function: FunctionIR, name: str, before_line: int) -> Bind | None:
|
||||
candidates = [bind for bind in function.binds if bind.name == name and bind.line < before_line]
|
||||
return max(candidates, key=lambda bind: bind.line) if candidates else None
|
||||
|
||||
|
||||
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:
|
||||
return item
|
||||
if text: # labels use declarator names; the parser may name the function differently (or <anon>)
|
||||
for name, start, _end in named_ranges_from_ir(ir, text):
|
||||
if name == function:
|
||||
for item in ir.functions:
|
||||
if item.start_line == start:
|
||||
return item
|
||||
return None
|
||||
|
||||
|
||||
def _function_lines(ir: FileIR, function: str, text: str) -> list[str]:
|
||||
target = _function_named(ir, function, text)
|
||||
if target is None or not text:
|
||||
return []
|
||||
lines = text.splitlines()
|
||||
body = lines[target.start_line - 1 : target.end_line]
|
||||
return [line.strip() for line in body if line.strip()]
|
||||
|
||||
|
||||
def shapes_by_sink(shapes: Sequence[Shape]) -> dict[tuple[str, str], list[Shape]]:
|
||||
"""Index for matching: (language, sink_name) -> shapes."""
|
||||
index: 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
|
||||
Reference in New Issue
Block a user