diff --git a/.kiro/specs/corpus-seeded-mechanisms/design.md b/.kiro/specs/corpus-seeded-mechanisms/design.md index 8412b3e..76cd1c0 100644 --- a/.kiro/specs/corpus-seeded-mechanisms/design.md +++ b/.kiro/specs/corpus-seeded-mechanisms/design.md @@ -108,7 +108,7 @@ tests/ ### Modified Files - `semantic/overlay.py` — `OverlayRecord.mechanism_id: str | None = None` (additive). -- `regress/candidate.py` — variant findings eligible after overlay promotions (reuses forced-hotspot path). +- `regress/candidate.py` — exposes `hotspot_from_finding`; variant findings enter the ranked candidate slice as zero-score hotspots from `cli.py` (after every scored promotion, inside the cap, through the safety check). - `config.py` — `[variants] enabled=true, max_mechanisms=500`. ## System Flows @@ -199,8 +199,8 @@ class VariantHit: sink_name: str source_kind: str -def search_tree(root: Path, targets: Sequence[FileTarget], store: MechanismStore, facts: SemanticFacts, *, max_mechanisms: int) -> list[VariantHit]: ... -def hits_to_findings(hits, records: Sequence[OverlayRecord]) -> tuple[list[StaticFinding], list[OverlayRecord]]: ... # merge with overlay flows +def search_tree(root: Path, targets: Sequence[FileTarget], store: MechanismStore, facts: SemanticFacts, *, max_mechanisms: int) -> SearchResult: ... # hits, mechanisms_searched, files_searched, degradations +def hits_to_findings(hits, records: Sequence[OverlayRecord], summaries: Mapping[str, tuple[str, tuple[str, ...], str]]) -> tuple[list[StaticFinding], list[OverlayRecord]]: ... # merge with overlay flows ``` Finding: id `variant:::`, evidence `suspicion`, tags `["variant", "mechanism:"]`, rationale names the mechanism summary and pair provenance. diff --git a/src/openultrasast/config.py b/src/openultrasast/config.py index 39c262a..2a42343 100644 --- a/src/openultrasast/config.py +++ b/src/openultrasast/config.py @@ -343,5 +343,5 @@ def _load_variants(value: object) -> VariantsConfig: data = _section(value) return VariantsConfig( enabled=bool(data.get("enabled", True)), - max_mechanisms=int(str(data.get("max_mechanisms", 500))), + max_mechanisms=_int_value(data.get("max_mechanisms"), 500), ) diff --git a/src/openultrasast/regress/candidate.py b/src/openultrasast/regress/candidate.py index 966a857..84c2b9a 100644 --- a/src/openultrasast/regress/candidate.py +++ b/src/openultrasast/regress/candidate.py @@ -99,19 +99,24 @@ def _hotspot_for_finding(finding: StaticFinding, hotspots: Sequence[Hotspot]) -> return None -def _hotspot_from_forced_finding(finding: StaticFinding) -> Hotspot: +def hotspot_from_finding(finding: StaticFinding, *, rationale: str) -> Hotspot: + """A zero-score hotspot wrapping one finding: forced sev-5 inventory or a variant of a known mechanism.""" return Hotspot( path=finding.path, function_name=finding.function_name, score=0.0, band="low", signals={}, - rationale="forced reachable severity-5 inventory", + rationale=rationale, test_hint=None, inventory_finding_ids=(finding.finding_id,), ) +def _hotspot_from_forced_finding(finding: StaticFinding) -> Hotspot: + return hotspot_from_finding(finding, rationale="forced reachable severity-5 inventory") + + class RegressionRunner: """Run a snippet in the sandbox only after the structural safety check passes.""" diff --git a/src/openultrasast/semantic/seed.py b/src/openultrasast/semantic/seed.py index f1beaff..0e671fe 100644 --- a/src/openultrasast/semantic/seed.py +++ b/src/openultrasast/semantic/seed.py @@ -14,9 +14,9 @@ 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 .ir import FileIR, parse_file from .mechanisms import MechanismStore, append_from_pair -from .variants import derive_shape +from .variants import Shape, derive_shape LANGUAGE_UNSUPPORTED = "variants_language_unsupported" @@ -67,36 +67,22 @@ def export_mechanisms(cases: Sequence[PairCase], store: MechanismStore, *, facts 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 + 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, - shape, - summary=f"{shape.mechanism}: {'/'.join(shape.source_kinds)} into {shape.sink_name} ({row.cwe}); fix adds {shape.guard}", - cwe=row.cwe, + lesson.shape, + summary=lesson.summary, + cwe=lesson.cwe, pair=case.name, provenance=case.provenance, tier=case.review_tier, ) record_ids.add(record.id) - produced += 1 - if produced: + if lessons: seeded.append(case.name) else: skipped.append((case.name, "no_labeled_sink_call_site")) @@ -109,6 +95,54 @@ def export_mechanisms(cases: Sequence[PairCase], store: MechanismStore, *, facts ) +@dataclass(frozen=True) +class Lesson: + """One shape a trusted pair teaches, with the CWE and a text-free summary.""" + + shape: Shape + 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 + 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_reason(case: PairCase) -> str | None: if case.review_tier not in GATING_TIERS: return f"tier:{case.review_tier}" diff --git a/src/openultrasast/semantic/variant_search.py b/src/openultrasast/semantic/variant_search.py index 75dfe4e..8444d7e 100644 --- a/src/openultrasast/semantic/variant_search.py +++ b/src/openultrasast/semantic/variant_search.py @@ -41,16 +41,22 @@ def search_tree( records = corpus_mechanisms(store.load())[: 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 - shape = Shape.from_dict(record.shape) + 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) + return SearchResult(hits=(), mechanisms_searched=0, files_searched=0, degradations=tuple(degradations)) hits: list[VariantHit] = [] files = 0 - degradations: list[dict[str, object]] = [] unsupported: set[str] = set() for target in targets: try: diff --git a/tests/test_variant_search.py b/tests/test_variant_search.py index bb251c8..066871e 100644 --- a/tests/test_variant_search.py +++ b/tests/test_variant_search.py @@ -96,3 +96,29 @@ def test_search_tree_uses_the_store_and_respects_the_mechanism_cap(tmp_path: Pat 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 + + +def test_hits_never_merge_into_demoted_or_flowless_records() -> 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="parameter") + demoted = OverlayRecord( + proposal_id="python-os-command:a.py:7", + path="a.py", + line=7, + disposition="demote", + reason="constant argument", + cwe="CWE-78", + sources=(), + sinks=("os.system",), + sanitizers=(), + evidence_level="suspicion", + dominating_fact="constant", + origin="inventory", + engine="python-ast", + language="python", + ) + findings, records = hits_to_findings([hit], [demoted], {}) + assert records[0].mechanism_id is None and records[0].disposition == "demote" # Req 3.6: never demote, never re-label a demotion + assert [f.finding_id for f in findings] == ["variant:corpus:abc:a.py:7"]