From a767ba80b4f44f7ab6920b430ac7740dadfa157d 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): variant findings in MAP, sandbox-eligible, counted in the manifest Task 2.3 (Req 3.4, 3.5, 3.6, 6.2). - [variants] enabled / max_mechanisms config; MAP runs variant search after the overlay in standard and deep scans only, appends the variant findings, rewrites overlay.json with merged ids, and the manifest records mechanisms_searched, files_searched, findings and merged_into_overlay. - REGRESS: variant findings join the candidate list after the promotions as zero-score hotspots, under the existing safety check and cap. - Spec notes for group 2 and the design synced (variant_search.py, seed.py, derive_shape signature). Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_01T7XD7x5KonMdmGRkp63EQG --- .../specs/corpus-seeded-mechanisms/design.md | 9 +- .kiro/specs/corpus-seeded-mechanisms/tasks.md | 13 +- src/openultrasast/cli.py | 98 +++++++++++++- src/openultrasast/config.py | 17 +++ src/openultrasast/reports.py | 3 + tests/test_variant_scan.py | 124 ++++++++++++++++++ 6 files changed, 257 insertions(+), 7 deletions(-) create mode 100644 tests/test_variant_scan.py diff --git a/.kiro/specs/corpus-seeded-mechanisms/design.md b/.kiro/specs/corpus-seeded-mechanisms/design.md index 38c8f5b..8412b3e 100644 --- a/.kiro/specs/corpus-seeded-mechanisms/design.md +++ b/.kiro/specs/corpus-seeded-mechanisms/design.md @@ -93,6 +93,8 @@ src/openultrasast/ ├── semantic/ │ ├── mechanisms.py # Mechanism += origin, review_tier, pairs, shape, guard; append_from_pair; dedupe_by_shape │ ├── variants.py # Shape, GuardKind, derive_shape(vuln_ir, fixed_ir, label, facts) -> Shape|None; match_shapes(file_ir, shapes, facts) -> list[VariantHit] +│ ├── variant_search.py # search_tree, hits_to_findings, SearchResult (needs store/overlay/findings; variants.py stays free of them) +│ ├── seed.py # export_mechanisms(cases, store) -> ExportReport (maintainer command `mechanisms export`) │ └── loo.py # evaluate_loo(cases, *, store_factory) -> LooResult; payload ├── findings.py # origin "variant" via tags/finding_id prefix "variant:" (no schema change) ├── cli.py # MAP: variant search after overlay; manifest counters; `pairs --loo`; `mechanisms export --slice` @@ -173,8 +175,11 @@ class Shape: def key(self) -> str: ... # stable, text-free -def derive_shape(vuln: FileIR, fixed: FileIR, *, function: str, sink: str | None, line: int | None, mechanism: str, facts: SemanticFacts) -> Shape | None: ... -def match_shapes(ir: FileIR, shapes: Sequence[Shape], facts: SemanticFacts) -> list[VariantHit]: ... +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: ... + # texts: guard statements are neither calls nor binds in FileIR, so the guard kind is classified from the labeled + # function's lines present only on the fixed side (closed set, none of the text enters the shape); cwe: when a label + # names no sink and no line, the fact-sink call in the function whose CWE matches is the labeled call. +def match_shapes(ir: FileIR, shapes: Sequence[Shape], facts: SemanticFacts, *, mechanism_ids: Mapping[str, str]) -> list[VariantHit]: ... ``` Invariants: shapes carry no path, line, or literal beyond identifiers; `derive_shape` returns None when the labeled sink call site is not found. diff --git a/.kiro/specs/corpus-seeded-mechanisms/tasks.md b/.kiro/specs/corpus-seeded-mechanisms/tasks.md index 7bcd065..8a432d2 100644 --- a/.kiro/specs/corpus-seeded-mechanisms/tasks.md +++ b/.kiro/specs/corpus-seeded-mechanisms/tasks.md @@ -15,22 +15,22 @@ - _Boundary: Store_ - _Depends: 1.1_ -- [ ] 2. Core: export and search -- [ ] 2.1 Export mechanisms from trusted pairs as a maintainer command +- [x] 2. Core: export and search +- [x] 2.1 Export mechanisms from trusted pairs as a maintainer command - Only `seeded` and `reviewed`, non-known-limit, parse-ok pairs seed; the command runs offline over the vendored catalog and reports skipped pairs with reasons; extra-absent languages are skipped with a degradation. - Observable: `ousast mechanisms export --slice sast` writes corpus rows and reports the skipped `title`/`advisory`/known-limit pairs. - _Requirements: 1.1, 1.2, 1.5, 2.5_ - _Boundary: Export_ - _Depends: 1.2_ -- [ ] 2.2 Search a tree for variants and merge with overlay flows +- [x] 2.2 Search a tree for variants and merge with overlay flows - For each parsed file within the MAP budget, call sites matching a shape's sink name, arity, and source positions (fact source, parameter, container read) produce `variant` findings at `suspicion` with the mechanism id; when the overlay has a flow at the same call site, its record gains `mechanism_id` and no duplicate finding is emitted. - Observable: the split-sink Python fixture yields a variant finding merged into the overlay coverage record; a constant-argument call does not match; `quick` emits no variants. - _Requirements: 3.1, 3.2, 3.3, 3.6_ - _Boundary: VariantSearch_ - _Depends: 2.1_ -- [ ] 2.3 Make variant findings sandbox-eligible and count them in the manifest +- [x] 2.3 Make variant findings sandbox-eligible and count them in the manifest - Variant findings enter the candidate set after overlay promotions under the existing safety check and caps; the manifest records mechanisms searched, files searched, findings, and merges. - Observable: with the fake sandbox a variant finding runs; the manifest `variants` block is present on standard scans and absent on quick. - _Requirements: 3.4, 3.5, 6.2_ @@ -84,5 +84,10 @@ - Task 1.1 (2026-09-05, RED first: 5 failing tests in `tests/test_variants.py`, `ModuleNotFoundError: openultrasast.semantic.variants`; then green): `semantic/variants.py` with `Shape` (language, sink_name, arity, source_positions, source_kinds, guard, mechanism; `key()`/`to_dict()`/`from_dict()` text-free), closed `GUARD_KINDS`/`SOURCE_KINDS`, `derive_shape(vuln_ir, fixed_ir, *, function, sink, line, mechanism, facts, vuln_text, fixed_text)`. Deviation from the design signature: the two excerpt texts are passed in because guard statements (`if x not in ALLOWED`, `if len(x) > n`) are not call sites or binds in `FileIR`; the guard kind is classified from the lines of the labeled function that exist only on the fixed side, closed set, `none` when unknown; nothing of that text enters the shape. Source kinds trace argument identifiers through binds (fact-source pattern -> `fact_source`; subscript/`.get(`/request container -> `container_read`; parameter -> `parameter`), preferring fact_source, then parameter, then container_read. A constant-only sink call or a missing function/call site yields None. - Task 1.2 (2026-09-05, RED first: 3 failing tests in `tests/test_mechanism_export.py` — `Mechanism` lacked `origin`, `append_from_pair` missing; then green): additive `Mechanism.origin/review_tier/pairs/shape/guard` with defaults so sandbox rows load unchanged; `MechanismStore.load()` folds the append-only log by id (latest row wins) so a re-seeded shape carries the extended `pairs`; `append_from_pair` uses a deterministic id `corpus:` so the improve lever can name records across machines; `corpus_mechanisms()` selects searchable rows. Existing `order_promotions` tests unchanged and green. - Review round 1 (2026-09-05) APPROVED 1.1 and 1.2 (reviewer re-simulated RED, ran six mutations, verified the import boundary and that sandbox rows load unchanged; isolated suite 493 / 480 green, statics and gates clean). Suggestions carried into group 2: trim `_CONTAINER_READ` to the design's subscript/`.get(` definition so no source vocabulary lives outside facts; intersect line and sink in `_labeled_call`; `append_from_pair` keeps the highest tier and accumulates provenance tags when a shape is re-seeded; sync design §Shape to the implemented signature. +- Task 2.1 (2026-09-05, RED first: 4 failing tests in `tests/test_mechanism_seed.py` — `ModuleNotFoundError: semantic.seed` x3, `argparse` rejected `mechanisms`; then green): `semantic/seed.py` `export_mechanisms(cases, store)` seeds only `seeded|reviewed`, non-known-limit, parsing pairs with cached excerpts, skips with reasons (`tier:`, `known_limit:`, `parse_failed`, `variants_language_unsupported` + degradation, `pointer_pair_not_cached`, `no_labeled_sink_call_site`) and keeps catalog order; CLI `ousast mechanisms export --slice S [--store P] [--json]`, default store `.openultrasast/calibration/mechanisms.jsonl` (the path the scan reads; the store module's `DEFAULT_MECHANISM_LOG` differs and is left alone). Discovery: vibe-py labels carry a function and a mechanism but no sink or line, so `derive_shape` gained a facts-driven fallback (the fact-sink call inside the labeled function, CWE-matching and non-constant first) and now intersects line and sink when both exist (round-1 suggestion); `_CONTAINER_READ` trimmed to subscript/`.get(`; `append_from_pair` keeps the highest tier and accumulates provenance tags. Measured export (offline, scratchpad stores): vibe-py 35 pairs -> 16 seeded, 10 records (7 pairs share `execute/1 parameter`; the 19 skips are absence-class labels — IDOR, open redirect, missing guard — with no sink call to shape); local 3 -> 0 (fixtures label no sink call site); sast 11 and vfc-js 17 -> 0, all `tier:advisory` as Req 1.2 requires. Guards are mostly `none` because Real-Vuln fixed twins are separate trap functions, not patched code. +- Task 2.2 (2026-09-05, RED first: 3 failing tests in `tests/test_variant_search.py`, `ImportError: match_shapes|VariantHit|search_tree`; then green): `variants.match_shapes(ir, shapes, facts, mechanism_ids=)` matches trailing callee name + arity and requires every shape source position to carry a source of any kind (a shape learned from `request.args` fires on a parameter); constants and arity mismatches never match. New module `semantic/variant_search.py` (design deviation: the file plan put `search_tree`/`hits_to_findings` in `variants.py`, but they need the store, the overlay records and `StaticFinding`, which `variants.py` must not import): `search_tree(root, targets, store, facts, max_mechanisms)` -> `SearchResult(hits, mechanisms_searched, files_searched, degradations)`; `hits_to_findings` merges a hit into an overlay promote/coverage record at the same path and line (`mechanism_id` set, no duplicate) and otherwise emits `variant:::` at `suspicion` with tags `variant`, `mechanism:` and the pair provenance in the rationale. `OverlayRecord.mechanism_id` added (additive). +- Task 2.3 (2026-09-05, RED first: 3 of 4 tests in `tests/test_variant_scan.py` failing — no variant finding, no overlay merge, variant not offered to the sandbox; the quick-mode test passed trivially and guards the invariant; then green): `[variants] enabled=true, max_mechanisms=500` config; MAP runs `_search_variants` after the overlay (standard/deep only; skipped when facts fail to load), appends variant findings, rewrites overlay.json with merged ids, and the manifest carries `variants = {mechanisms_searched, files_searched, findings, merged_into_overlay}`; in REGRESS variant findings join the candidate list after the promotions with a zero-score hotspot each, under the same safety check and cap. Verified with the fake sandbox: a variant of a project-specific sink with no inventory rule is offered. Suite 504 passed / 4 skipped, extra-free 491 / 17, statics clean, three gates PASS. +- Task 3.1 (2026-09-05, RED first: 3 failing tests in `tests/test_loo.py` — `ModuleNotFoundError: semantic.loo` x2, `pairs --loo` unknown flag; the two API tests went green on the first implementation): `semantic/loo.py` `evaluate_loo(cases)` holds every scorable pair out once (known_limit and uncached pointer pairs skipped with a reason), seeds a temporary store from the *other* `seeded|reviewed` pairs (advisory/title pairs are scored as targets, never teach), materializes both sides, runs `search_tree`, and scores: detected when a hit lies inside the labeled function and names the labeled mechanism or sink; silent when the fixed side yields no hit; `found_by` lists the record id and the pairs that taught it; metrics (pairs, detected, silent, recall, silence, Youden, teaching) per slice, profile and mechanism; `to_dict()` is the `loo.json` artifact. Shape derivation is duplicated from `seed.py` (`_lessons`) pending a shared helper once group 2 is accepted. First measurement, vibe-py vendored (35 pairs, 1.0 s): 11 detected (recall 0.314), 21 silent (0.600), Youden -0.086, 16 teaching pairs; per mechanism source_reaches_sink 27 pairs 9 detected / 19 silent, missing_auth_guard 4 / 2 / 2, permissive_default 3 / 0 / 0, path_join_user_input 1 / 0 / 0. The 14 leaks are the coarse `execute/1 parameter` shape firing on Real-Vuln's false-positive trap twins: the honest over-match rate the `mechanisms` lever (group 4) retracts on. +- Review round 2 (2026-09-05) APPROVED 2.1, 2.2, 2.3 (reviewer re-simulated all three REDs, killed 11 of 12 mutants, ran the export offline in a network namespace and reproduced the counts, confirmed gates byte-identical to HEAD; judged the zero-score-hotspot candidate path a closer fit to Req 3.4 than the forced path). Suggestions carried into round 3: a test that a hit on a `demote`/no-flow overlay record is not merged (the surviving mutant); `search_tree` skips a malformed store row with `variants_store_row_invalid` instead of aborting the scan; design §VariantSearch/§Modified Files synced (`SearchResult`, `regress/candidate.py` untouched); `_load_variants` uses `_int_value`; the exporter passes line and sink together; `no_labeled_function` skip reason; one shared `hotspot_from_finding` helper. - Seeding reads `PairCase.review_tier` from `pair-corpus-honesty` task 8.1; until it lands, sast and vibe-py rows default to `advisory`/`seeded` per that design and only vibe-py seeds. - Guard classification is heuristic and closed; `none` is allowed and reported, never used to demote. diff --git a/src/openultrasast/cli.py b/src/openultrasast/cli.py index 7428e28..b4cd5bd 100644 --- a/src/openultrasast/cli.py +++ b/src/openultrasast/cli.py @@ -29,6 +29,7 @@ from .calibration import ( ) from .complexity import map as complexity_map from .complexity.ledger import persist_verdicts +from .complexity.map import Hotspot from .config import load_config, load_dotenv from .findings import StaticFinding, quick_scan_findings, write_findings from .fusion import FusionDecision, fuse_findings_dispatch @@ -51,9 +52,10 @@ from .pairs import ( select_profile, select_slice, select_split, + select_vendored, ) from .policy import assert_rules_resolve, load_policy -from .preprocess import preprocess_repository, write_preprocess_artifact +from .preprocess import FileTarget, preprocess_repository, write_preprocess_artifact from .provenance import fingerprint from .provider.openrouter import OpenRouterEmbeddingClient, OpenRouterError from .rank import rank_targets, write_rankings @@ -160,6 +162,15 @@ def main(argv: list[str] | None = None) -> int: pairs.add_argument("--json", action="store_true", help="print the pair scoreboard as JSON") subparsers.add_parser("mcp", help="run the narrow MCP server over stdio for OpenCode integration") + mechanisms = subparsers.add_parser("mechanisms", help="maintainer: mechanism memory seeded from trusted pairs") + mechanisms_sub = mechanisms.add_subparsers(dest="mechanisms_command", required=True) + export = mechanisms_sub.add_parser("export", help="derive mechanism records from seeded/reviewed pairs (offline)") + export.add_argument("--slice", choices=SLICE_NAMES, default="all") + export.add_argument("--catalog", type=Path, default=DEFAULT_CATALOG) + export.add_argument( + "--store", type=Path, default=Path(CALIBRATION_DIR) / "mechanisms.jsonl", help="append-only mechanisms.jsonl the scan reads" + ) + export.add_argument("--json", action="store_true") args = parser.parse_args(argv) if args.command == "scan": @@ -193,6 +204,8 @@ def main(argv: list[str] | None = None) -> int: hunter_model=args.hunter_model, pointers=args.pointers, ) + if args.command == "mechanisms": + return _mechanisms_export(args.catalog, args.slice, args.store, json_out=args.json) if args.command == "mcp": from .mcp import serve # lazy: keeps the import cycle (mcp -> cli) one-directional @@ -381,6 +394,8 @@ def _run_scan(path: Path, config_path: Path, mode: str, fail_on: str) -> ScanOut built_map = None overlay_records: list[OverlayRecord] = [] wrote_overlay = False + variants_payload: dict[str, object] | None = None + variant_findings: list[StaticFinding] = [] if Stage.MAP in plan.requested: built_map = runtime.run_stage( "map", @@ -413,6 +428,19 @@ def _run_scan(path: Path, config_path: Path, mode: str, fail_on: str) -> ScanOut if extra: findings = findings + extra write_findings(findings, findings_path) + if config.variants.enabled: + variant_findings, overlay_records, variants_payload = _search_variants( + root=run.target, + targets=targets, + overlay_records=overlay_records, + facts=facts_or_error, + max_mechanisms=config.variants.max_mechanisms, + runtime=runtime, + ) + if variant_findings: + findings = findings + variant_findings + write_findings(findings, findings_path) + write_overlay(overlay_records, overlay_path) plan = record_completed(plan, Stage.MAP) if not hunter_model: runtime.state["degradations"].append(skip_as_degradation(Stage.MAP, "hunter_model_unavailable")) @@ -469,6 +497,10 @@ def _run_scan(path: Path, config_path: Path, mode: str, fail_on: str) -> ScanOut for finding in findings if finding.finding_id in {record.proposal_id for record in overlay_records if record.disposition == "promote"} ] + # Variant findings are candidates after the promotions (Req 3.4): same safety check, same cap, lowest rank. + variant_ids = {finding.finding_id for finding in variant_findings} + promoted = promoted + [finding for finding in findings if finding.finding_id in variant_ids] + hotspots = hotspots + tuple(_hotspot_from_variant(finding) for finding in findings if finding.finding_id in variant_ids) cwe_by_rule_id = {rule.rule_id: rule.cwe for rule in ruleset} rule_cwe_by_finding = {finding.finding_id: cwe_by_rule_id.get(finding.finding_id.split(":", 1)[0], "") for finding in promoted} languages_by_path = {target.path: target.language for target in targets} @@ -541,6 +573,7 @@ def _run_scan(path: Path, config_path: Path, mode: str, fail_on: str) -> ScanOut complexity=complexity_payload, worth_fixing=worth_fixing_payload, provenance=provenance.to_dict(), + variants=variants_payload, ), ) runtime.finish(status="succeeded") @@ -811,6 +844,69 @@ def _pairs( return 0 +def _search_variants( + *, + root: Path, + targets: list[FileTarget], + overlay_records: list[OverlayRecord], + facts: SemanticFacts | FactLoadError, + max_mechanisms: int, + runtime: HarnessRuntime, +) -> tuple[list[StaticFinding], list[OverlayRecord], dict[str, object]]: + """Third MAP proposer (corpus-seeded-mechanisms Req 3): shapes from the mechanism store, merged with overlay flows.""" + from .semantic.mechanisms import corpus_mechanisms + from .semantic.variant_search import hits_to_findings, search_tree + + store = MechanismStore(root / CALIBRATION_DIR / "mechanisms.jsonl") + empty: dict[str, object] = {"mechanisms_searched": 0, "files_searched": 0, "findings": 0, "merged_into_overlay": 0} + if isinstance(facts, FactLoadError): + return [], overlay_records, empty + result = runtime.run_stage("variants", lambda: search_tree(root, targets, store, facts, max_mechanisms=max_mechanisms)) + for degradation in result.degradations: + runtime.state["degradations"].append(degradation) + summaries = {record.id: (record.summary, record.pairs, record.cwe) for record in corpus_mechanisms(store.load())} + findings, records = hits_to_findings(result.hits, overlay_records, summaries) + merged = sum(1 for before, after in zip(overlay_records, records, strict=True) if before.mechanism_id is None and after.mechanism_id) + payload: dict[str, object] = { + "mechanisms_searched": result.mechanisms_searched, + "files_searched": result.files_searched, + "findings": len(findings), + "merged_into_overlay": merged, + } + return findings, records, payload + + +def _hotspot_from_variant(finding: StaticFinding) -> Hotspot: + return Hotspot( + path=finding.path, + function_name=finding.function_name, + score=0.0, + band="low", + signals={}, + rationale="variant of a known mechanism (suspicion; sandbox may raise)", + test_hint=None, + inventory_finding_ids=(finding.finding_id,), + ) + + +def _mechanisms_export(catalog: Path, slice_name: str, store_path: Path, *, json_out: bool) -> int: + from .semantic.mechanisms import MechanismStore + from .semantic.seed import export_mechanisms + + if not catalog.is_file(): + raise SystemExit(f"pair catalog is not a file: {catalog}") + cases = select_vendored(select_slice(load_pair_catalog(catalog), slice_name)) + report = export_mechanisms(cases, MechanismStore(store_path)) + payload = {"slice": slice_name, "store": str(store_path), "pairs": len(cases), **report.to_dict()} + if json_out: + print(json.dumps(payload, indent=2, sort_keys=True)) + return 0 + print(f"mechanisms export slice={slice_name} pairs={len(cases)} seeded={report.seeded} records={report.records} -> {store_path}") + for pair, reason in report.skipped: + print(f" skip {pair}: {reason}") + return 0 + + def _print_improve_outcomes(outcomes: list[RoundOutcome], manifest_path: Path, target: Path, ledger_path: Path, *, dry_run: bool) -> None: print(f"manifest={manifest_path}") print(f"target={target}") diff --git a/src/openultrasast/config.py b/src/openultrasast/config.py index 3b2904d..39c262a 100644 --- a/src/openultrasast/config.py +++ b/src/openultrasast/config.py @@ -128,6 +128,13 @@ class RegressConfig: images: tuple[tuple[str, str], ...] = () +@dataclass(frozen=True) +class VariantsConfig: + # corpus-seeded-mechanisms: structural variant search in MAP (standard/deep only). + enabled: bool = True + max_mechanisms: int = 500 + + @dataclass(frozen=True) class ResolvedConfig: models: ModelConfig = ModelConfig() @@ -143,6 +150,7 @@ class ResolvedConfig: hardening: HardeningConfig = HardeningConfig() complexity: ComplexityConfig = ComplexityConfig() regress: RegressConfig = RegressConfig() + variants: VariantsConfig = VariantsConfig() runs_dir: str = ".openultrasast/runs" @@ -167,6 +175,7 @@ def load_config(config_path: Path | None = None) -> ResolvedConfig: hardening=_load_hardening(data.get("hardening", {})), complexity=_load_complexity(data.get("complexity", {})), regress=_load_regress(data.get("regress", {})), + variants=_load_variants(data.get("variants", {})), runs_dir=os.environ.get("OPENULTRASAST_RUNS_DIR", ".openultrasast/runs"), ) @@ -328,3 +337,11 @@ def _float_value(value: object, default: float) -> float: except ValueError: return default return default + + +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))), + ) diff --git a/src/openultrasast/reports.py b/src/openultrasast/reports.py index fdea267..184414f 100644 --- a/src/openultrasast/reports.py +++ b/src/openultrasast/reports.py @@ -174,6 +174,7 @@ def write_manifest( complexity: dict[str, object] | None = None, worth_fixing: dict[str, object] | None = None, provenance: dict[str, object] | None = None, + variants: dict[str, object] | None = None, ) -> None: path.parent.mkdir(parents=True, exist_ok=True) verification_by_id = _verification_by_id(verifications) @@ -210,6 +211,8 @@ def write_manifest( payload["worth_fixing"] = worth_fixing if provenance is not None: payload["provenance"] = provenance + if variants is not None: + payload["variants"] = variants path.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n") diff --git a/tests/test_variant_scan.py b/tests/test_variant_scan.py new file mode 100644 index 0000000..c6b3f3d --- /dev/null +++ b/tests/test_variant_scan.py @@ -0,0 +1,124 @@ +"""corpus-seeded-mechanisms task 2.3: variant findings in a standard scan, sandbox-eligible, counted in the manifest.""" + +from __future__ import annotations + +import json +from pathlib import Path + +import pytest + +from openultrasast.cli import main +from openultrasast.semantic.mechanisms import MechanismStore, append_from_pair + +VARIANT = "import os\nfrom flask import request\n\n\ndef handler():\n q = request.args.get('q')\n os.system(q)\n" +UNSEEN_SINK = ( + "from shelltools import run_shell\n\n\ndef handler(q):\n run_shell(q)\n" # a project helper: no inventory rule, no fact sink +) + + +def _seed(repo: Path) -> str: + from openultrasast.semantic.variants import Shape + + shape = Shape( + language="python", + sink_name="run_shell", + arity=1, + source_positions=(0,), + source_kinds=("parameter",), + guard="allowlist_test", + mechanism="source_reaches_sink", + ) + store = MechanismStore(repo / ".openultrasast" / "calibration" / "mechanisms.jsonl") + return append_from_pair( + store, shape, summary="parameter into run_shell; fix adds allowlist", cwe="CWE-78", pair="p1", provenance="human", tier="seeded" + ).id + + +def _latest_run(repo: Path) -> Path: + runs = sorted((repo / ".runs").iterdir()) + return runs[-1] + + +def test_standard_scan_emits_variant_findings_and_manifest_counters(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + repo = tmp_path / "repo" + repo.mkdir() + (repo / "app.py").write_text(UNSEEN_SINK) # only the learned shape can find a project-specific sink + mechanism_id = _seed(repo) + monkeypatch.setenv("OPENULTRASAST_RUNS_DIR", ".runs") + monkeypatch.setenv("OPENULTRASAST_SANDBOX_PROBE", "0") + assert main(["scan", str(repo), "--mode", "standard"]) == 0 + run_dir = _latest_run(repo) + findings = json.loads((run_dir / "findings.json").read_text())["findings"] + variant = [f for f in findings if f["finding_id"].startswith("variant:")] + assert len(variant) == 1 and variant[0]["finding_id"] == f"variant:{mechanism_id}:app.py:5" + assert variant[0]["evidence_level"] == "suspicion" and f"mechanism:{mechanism_id}" in variant[0]["tags"] + manifest = json.loads((run_dir / "manifest.json").read_text()) + assert manifest["variants"] == {"mechanisms_searched": 1, "files_searched": 1, "findings": 1, "merged_into_overlay": 0} + + +def test_variant_hit_merges_into_an_overlay_flow_at_the_same_call_site(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + from openultrasast.semantic.variants import Shape + + repo = tmp_path / "repo" + repo.mkdir() + (repo / "app.py").write_text(VARIANT) # os.system is both an inventory rule and a fact sink: the overlay records the flow + shape = Shape( + language="python", + sink_name="system", + arity=1, + source_positions=(0,), + source_kinds=("fact_source",), + guard="none", + mechanism="source_reaches_sink", + ) + record = append_from_pair( + MechanismStore(repo / ".openultrasast" / "calibration" / "mechanisms.jsonl"), + shape, + summary="s", + cwe="CWE-78", + pair="p", + provenance="human", + tier="seeded", + ) + monkeypatch.setenv("OPENULTRASAST_RUNS_DIR", ".runs") + monkeypatch.setenv("OPENULTRASAST_SANDBOX_PROBE", "0") + assert main(["scan", str(repo), "--mode", "standard"]) == 0 + run_dir = _latest_run(repo) + overlay = json.loads((run_dir / "overlay.json").read_text())["records"] + assert any(item["mechanism_id"] == record.id and item["line"] == 7 for item in overlay) + findings = json.loads((run_dir / "findings.json").read_text())["findings"] + assert not any(f["finding_id"].startswith("variant:") for f in findings) # merged, not duplicated + manifest = json.loads((run_dir / "manifest.json").read_text()) + assert manifest["variants"]["merged_into_overlay"] == 1 and manifest["variants"]["findings"] == 0 + + +def test_quick_scan_never_searches_variants(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + repo = tmp_path / "repo" + repo.mkdir() + (repo / "app.py").write_text(UNSEEN_SINK) + _seed(repo) + monkeypatch.setenv("OPENULTRASAST_RUNS_DIR", ".runs") + assert main(["scan", str(repo), "--mode", "quick"]) == 0 + run_dir = _latest_run(repo) + manifest = json.loads((run_dir / "manifest.json").read_text()) + assert "variants" not in manifest + assert not any(f["finding_id"].startswith("variant:") for f in json.loads((run_dir / "findings.json").read_text())["findings"]) + + +def test_deep_scan_offers_variant_findings_to_the_sandbox_after_promotions(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + from openultrasast import cli + from openultrasast.sandbox import FakeSandboxRunner, SandboxResult + + repo = tmp_path / "repo" + repo.mkdir() + (repo / "app.py").write_text(UNSEEN_SINK) + mechanism_id = _seed(repo) + monkeypatch.setenv("OPENULTRASAST_RUNS_DIR", ".runs") + monkeypatch.setenv("OPENULTRASAST_SANDBOX_PROBE", "1") + fake = FakeSandboxRunner(SandboxResult(exit_code=0, stdout="", stderr="", timed_out=False)) + monkeypatch.setattr(cli, "resolve_sandbox_runner", lambda: fake) + assert main(["scan", str(repo), "--mode", "deep"]) == 0 + run_dir = _latest_run(repo) + verdicts = json.loads((run_dir / "verdicts.json").read_text())["verdicts"] + offered = {finding_id for item in verdicts for finding_id in item["inventory_finding_ids"]} + assert f"variant:{mechanism_id}:app.py:5" in offered