diff --git a/.kiro/specs/contributor-scan/spec.json b/.kiro/specs/contributor-scan/spec.json index 7cac816..e0fc92b 100644 --- a/.kiro/specs/contributor-scan/spec.json +++ b/.kiro/specs/contributor-scan/spec.json @@ -1,9 +1,9 @@ { "feature_name": "contributor-scan", "created_at": "2026-09-08T20:00:00Z", - "updated_at": "2026-09-09T09:15:00Z", + "updated_at": "2026-09-09T09:30:00Z", "language": "en", - "phase": "tasks", + "phase": "implementation", "approvals": { "requirements": { "generated": true, @@ -15,12 +15,12 @@ }, "tasks": { "generated": true, - "approved": false + "approved": true } }, - "ready_for_implementation": false, + "ready_for_implementation": true, "depends_on": [ "model-grounded-detection" ], - "approval_note": "2026-09-08 brief drafted after the maintainer asked what the spec state was and whether anything was forgotten. It was: model-grounded-detection is 16/16 complete and its arbiter is measured at 86.7% pair-correct / 6.7% leak, but the model layer has ZERO production callers -- ousast scan does not use it, so the feature is complete and unshipped at once. That is a scope gap in the predecessor (it had no task saying 'make this reachable', and its byte-identical-gates requirement is satisfied precisely because nothing is wired). This spec covers scan integration, whole-repo scanning against known vulnerable repositories, the deferred memory-family decision, and records the pipeline, ops and corpus work delivered outside any spec. Not approved; requirements not generated. ; 2026-09-08 requirements generated at the maintainer's direction ('we follow the spec and build integration afterwards'). ; 2026-09-09 requirements approved and design generated ('let us continue with these'): the model layer joins as one stage inside MAP beside the obligations checker, one CPG per scan, regions from entry points, a total budget rather than a per-region cap. Phase 2 (the first run on a pinned real repository) can invalidate the rest and comes before shipping." + "approval_note": "2026-09-08 brief drafted after the maintainer asked what the spec state was and whether anything was forgotten. It was: model-grounded-detection is 16/16 complete and its arbiter is measured at 86.7% pair-correct / 6.7% leak, but the model layer has ZERO production callers -- ousast scan does not use it, so the feature is complete and unshipped at once. That is a scope gap in the predecessor (it had no task saying 'make this reachable', and its byte-identical-gates requirement is satisfied precisely because nothing is wired). This spec covers scan integration, whole-repo scanning against known vulnerable repositories, the deferred memory-family decision, and records the pipeline, ops and corpus work delivered outside any spec. Not approved; requirements not generated. ; 2026-09-08 requirements generated at the maintainer's direction ('we follow the spec and build integration afterwards'). ; 2026-09-09 requirements approved and design generated ('let us continue with these'): the model layer joins as one stage inside MAP beside the obligations checker, one CPG per scan, regions from entry points, a total budget rather than a per-region cap. Phase 2 (the first run on a pinned real repository) can invalidate the rest and comes before shipping. ; 2026-09-09 tasks approved by the maintainer ('approve the tasks and start group 1'); implementing group 1 (regions, driver, MAP stage)." } diff --git a/.kiro/specs/contributor-scan/tasks.md b/.kiro/specs/contributor-scan/tasks.md index fca74e1..128b4f4 100644 --- a/.kiro/specs/contributor-scan/tasks.md +++ b/.kiro/specs/contributor-scan/tasks.md @@ -21,7 +21,7 @@ ## Group 1 — Wire it (gate-protected) -- [ ] 1.1 Regions from entry points +- [x] 1.1 Regions from entry points - `model/regions.py`: `ScanRegion(path, function, families, rank, source)` and `regions_for(entries, targets, *, taxonomy, language_of)`. Rank from what `EntryPointRecord` already carries (`access_level`, `trust_boundary`) — do not invent a new signal. File-level fallback when a target @@ -33,7 +33,7 @@ as such; no region carries a family whose spec is absent for its language. - _Requirements: 2.1, 2.2, 2.3, 2.4, 8.5_ -- [ ] 1.2 The repository driver +- [x] 1.2 The repository driver - `model/scan.py`: `ScanBudget`, `ModelScanResult`, `scan_repository(...)`. **One CPG per run**, reused across regions. Regions in rank order; budget decremented per model call and checked before each; the remainder counted into `regions_unjudged`. Findings ordered by rung, then rank. diff --git a/benchmarks/measurements/2026-09-08-module-audit.json b/benchmarks/measurements/2026-09-08-module-audit.json index f9a24f2..e9eea90 100644 --- a/benchmarks/measurements/2026-09-08-module-audit.json +++ b/benchmarks/measurements/2026-09-08-module-audit.json @@ -1,13 +1,13 @@ { "counts": { - "load_bearing": 86, + "load_bearing": 88, "standalone_capability": 7 }, - "generated": "2026-09-08 (regenerated after the detector pipeline was wired)", + "generated": "2026-09-09 (regenerated after contributor-scan group 1 added regions and the repo driver)", "kept_with_a_lever_stripped": { "improve/evolve.py + improve/validator.py": "the design's Removed list named these whole, but each is a mix: the mechanism lever (removed) and the deterministic rule-status improve loop that `ousast improve` still runs from benchmark evidence, which has no LLM in it and is not what this feature supersedes. Deleting the files would have taken a retained capability with them." }, - "modules": 93, + "modules": 95, "note": "Classification of every source module after the group-1 subtraction. load_bearing = reachable from a gate, the CLI, or named by a surviving requirement; standalone_capability = its own entry point or protocol, deliberately off the scan path; orphaned = no importer, no entry point, no surviving requirement -- removed, never carried.", "orphaned": [], "removed_in_this_pass": { @@ -336,6 +336,8 @@ "model.execution", "model.judge", "model.pipeline", + "model.regions", + "model.scan", "model.specs", "model.taint" ], @@ -357,20 +359,25 @@ { "classification": "load_bearing", "importers": [ - "cli" + "cli", + "model.scan" ], "module": "model.candidates", "reason": "reachable from gate, map_gate, pair_gate or the CLI" }, { "classification": "load_bearing", - "importers": [], + "importers": [ + "model.pipeline" + ], "module": "model.config_value", "reason": "model-grounded-detection Req 6.3: constant abstraction for the configuration families" }, { "classification": "load_bearing", - "importers": [], + "importers": [ + "model.pipeline" + ], "module": "model.dominance", "reason": "model-grounded-detection Req 6.2: guard dominance, the arbiter for bugs that never crash" }, @@ -400,6 +407,7 @@ "model.execution", "model.judge", "model.pipeline", + "model.scan", "model.taint" ], "module": "model.ladder", @@ -407,16 +415,32 @@ }, { "classification": "load_bearing", - "importers": [], + "importers": [ + "model.scan" + ], "module": "model.pipeline", "reason": "model-grounded-detection Req 8: the enumerator proposes, the LLM answers, the model disposes" }, + { + "classification": "load_bearing", + "importers": [ + "model.scan" + ], + "module": "model.regions", + "reason": "contributor-scan Req 2: regions from entry points, with the families each admits" + }, { "classification": "load_bearing", "importers": [], "module": "model.report", "reason": "model-grounded-detection Req 9.3/11.1: per-slice reporting with the overfitting gap" }, + { + "classification": "load_bearing", + "importers": [], + "module": "model.scan", + "reason": "contributor-scan Req 3: the repository driver -- one CPG, one budget, ranked spend" + }, { "classification": "load_bearing", "importers": [ @@ -424,6 +448,8 @@ "model.dominance", "model.judge", "model.pipeline", + "model.regions", + "model.scan", "model.taint" ], "module": "model.specs", @@ -443,6 +469,7 @@ "importers": [ "cli", "model.candidates", + "model.scan", "model.specs" ], "module": "model.taxonomy", diff --git a/src/openultrasast/model/audit.py b/src/openultrasast/model/audit.py index 06c92b5..70505b3 100644 --- a/src/openultrasast/model/audit.py +++ b/src/openultrasast/model/audit.py @@ -40,6 +40,8 @@ SPEC_OWNED = { "model.report": "model-grounded-detection Req 9.3/11.1: per-slice reporting with the overfitting gap", "model.execution": "model-grounded-detection Req 10: the deferred execution tier seam, adopted not built", "model.pipeline": "model-grounded-detection Req 8: the enumerator proposes, the LLM answers, the model disposes", + "model.regions": "contributor-scan Req 2: regions from entry points, with the families each admits", + "model.scan": "contributor-scan Req 3: the repository driver -- one CPG, one budget, ranked spend", "cpg": "model-grounded-detection Req 4: the CPG seam package", "cpg.backend": "model-grounded-detection Req 4.1/4.5: the single Joern subprocess boundary", "cpg.capability": "model-grounded-detection Req 4.2/4.3: the engine probe that degrades to suspicion", diff --git a/src/openultrasast/model/regions.py b/src/openultrasast/model/regions.py new file mode 100644 index 0000000..41b05a1 --- /dev/null +++ b/src/openultrasast/model/regions.py @@ -0,0 +1,114 @@ +"""Where to look, and what to look for there (contributor-scan, Req 2). + +Pair scoring could rely on a labelled function: the corpus declares which function carries the bug and which +family it belongs to. A contributor has neither — *which code is risky* is the question they are asking. So a +repository scan derives its regions from the code itself: entry points where the mapper found them, whole +files where it did not, and each region carries only the families its shape and language actually admit. + +Two restraints are deliberate: + +* **The rank comes from what ``EntryPointRecord`` already carries**, not from a new signal invented here. The + mapper has already decided whether a handler is public or local-only and whether it sits at a trust + boundary; a second opinion computed from the same evidence would be a way of disagreeing with ourselves. +* **A family is offered only where its arbiter could answer.** An obligation needs a reachable operation, so + ``access_control`` is offered where there is a handler and withheld from a library file that has none. + Asking every family of every region is the unbounded version of this, and at repository scale the budget is + the thing being spent. +""" + +from __future__ import annotations + +from collections.abc import Sequence +from dataclasses import dataclass + +from .specs import config_specs, dominance_specs, taint_specs + +# From `mapping.AccessLevel`. A publicly reachable handler at a trust boundary is where a missing check costs +# most, so it is looked at first when the budget is finite. +_ACCESS_RANK: dict[str, float] = { + "public": 1.0, + "authenticated": 0.8, + "role-restricted": 0.6, + "contract-only/callback": 0.4, + "review-required": 0.3, + "local-only": 0.2, +} + +# Languages whose facts the model layer carries. Anything else yields no region rather than an empty scan +# that looks like a clean bill of health. +_LANGUAGES = {"python", "javascript", "typescript", "java", "c", "c_cpp"} +_NORMALISE = {"typescript": "javascript", "c_cpp": "c"} + + +@dataclass(frozen=True) +class ScanRegion: + """A place to arbitrate, and the families worth arbitrating there.""" + + path: str + function: str | None # None for a file-level fallback: there is no labelled function on a scan path + language: str + families: tuple[str, ...] + rank: float + source: str # "entry_point" | "file_fallback" + + +def regions_for(entries: Sequence[object], targets: Sequence[object]) -> tuple[ScanRegion, ...]: + """Regions for a repository, strongest first. + + A file with an entry point yields regions for its handlers only — a fallback region beside them would + scan the same code twice and spend the budget on duplicates. + """ + by_path: dict[str, str] = {} + for target in targets: + language = _language_of(target) + if language: + by_path[str(getattr(target, "path", ""))] = language + + regions: list[ScanRegion] = [] + covered: set[str] = set() + for entry in entries: + path = str(getattr(entry, "path", "")) + language = by_path.get(path) + if language is None: + continue + families = _families(language, has_handler=True) + if not families: + continue + covered.add(path) + regions.append( + ScanRegion( + path=path, + function=(getattr(entry, "function_name", None) or None), + language=language, + families=families, + rank=_ACCESS_RANK.get(str(getattr(entry, "access_level", "")), 0.5), + source="entry_point", + ) + ) + + for path, language in by_path.items(): + if path in covered: + continue + families = _families(language, has_handler=False) + if not families: + continue + regions.append(ScanRegion(path=path, function=None, language=language, families=families, rank=0.1, source="file_fallback")) + + # A total order: rank descending, then path and function, so two runs of one repository agree. + return tuple(sorted(regions, key=lambda r: (-r.rank, r.path, r.function or ""))) + + +def _language_of(target: object) -> str | None: + language = str(getattr(target, "language", "") or "") + if language not in _LANGUAGES: + return None + return _NORMALISE.get(language, language) + + +def _families(language: str, *, has_handler: bool) -> tuple[str, ...]: + """The families this language has facts for, minus those whose arbiter needs a handler and has none.""" + families = set(taint_specs(language=language)) + families |= set(config_specs(language=language)) + if has_handler: + families |= set(dominance_specs(language=language)) + return tuple(sorted(families)) diff --git a/src/openultrasast/model/scan.py b/src/openultrasast/model/scan.py new file mode 100644 index 0000000..9bfbccd --- /dev/null +++ b/src/openultrasast/model/scan.py @@ -0,0 +1,183 @@ +"""The repository driver (contributor-scan, Req 3). + +Pair scoring built one CPG per pair and asked about one region. A repository has thousands of regions, and the +two costs that were free at pair scale dominate here: + +* **CPG construction.** A build takes 5-30 seconds on a forty-line excerpt. One per region would make a + repository scan unusable, so this builds **one** and reuses it across every region. +* **Model calls.** The enumerator caps candidates at eight per region, which says nothing useful about a scan + with two thousand regions. The budget is a **total for the run**, spent on the highest-ranked regions first, + and whatever it does not reach is counted into ``regions_unjudged`` rather than quietly dropped. A tool that + silently analyses a tenth of a repository and reports a clean result is worse than one that says what it + did not look at. + +Ordering puts rung before rank: what the model established outranks what the LLM merely proposed, whatever +the region's risk score said beforehand. +""" + +from __future__ import annotations + +import logging +import time +from collections.abc import Mapping, Sequence +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any + +from .candidates import enumerate_candidates +from .ladder import Rung +from .pipeline import ModelFinding, scan_region +from .regions import ScanRegion +from .specs import ConfigSpec, DominanceSpec, TaintSpec, config_specs, dominance_specs, taint_specs +from .taxonomy import load_families + +logger = logging.getLogger(__name__) + +_RUNG_ORDER = {Rung.ENTAILED: 0, Rung.CORROBORATED: 1, Rung.SUSPICION: 2, Rung.EXECUTION_CONFIRMED: -1} + + +@dataclass(frozen=True) +class ScanBudget: + """A total for the run, not a per-region cap.""" + + max_model_calls: int = 200 + max_regions: int = 500 + + +@dataclass(frozen=True) +class ModelScanResult: + findings: tuple[ModelFinding, ...] = () + by_rung: Mapping[str, int] = field(default_factory=dict) + regions_scanned: int = 0 + regions_unjudged: int = 0 + model_calls: int = 0 + cost_usd: float = 0.0 + seconds: float = 0.0 + degradations: tuple[Mapping[str, object], ...] = () + + +def scan_repository( + root: Path, + regions: Sequence[ScanRegion], + *, + backend: Any, + client: Any | None = None, + model: str = "", + budget: ScanBudget | None = None, +) -> ModelScanResult: + """Arbitrate a repository's regions under one CPG and one budget.""" + limits = budget if budget is not None else ScanBudget() + started = time.monotonic() + degradations: list[Mapping[str, object]] = [] + + taxonomy = load_families() + cpg = backend.build(root) + if cpg is None: + return ModelScanResult( + by_rung=_empty_tally(), + regions_unjudged=len(regions), + seconds=round(time.monotonic() - started, 2), + degradations=({"stage": "model", "reason": "cpg_build_failed"},), + ) + + # Count calls here rather than reading a client's own meter: the budget is this driver's contract and + # must hold for any client, including one that keeps no usage log. + counted = _CountingClient(client) if client is not None else None + collected: list[tuple[ModelFinding, float]] = [] + scanned = 0 + + # Sort here rather than trusting the caller. The budget decides what goes unexamined, so the order it is + # spent in belongs to whoever holds the budget. + ordered_regions = sorted(regions, key=lambda r: (-r.rank, r.path, r.function or "")) + + for region in ordered_regions: + if scanned >= limits.max_regions or (counted is not None and counted.calls >= limits.max_model_calls): + degradations.append({"stage": "model", "reason": "budget_exhausted", "regions_unjudged": len(regions) - scanned}) + break + for family in region.families: + spec = _spec_for(family, region.language) + if spec is None: + continue + try: + # Enumerate the region's candidate sites so the judge has something to answer about where the + # model resolves nothing. Without this a repository scan reports only what taint, dominance + # and constant abstraction establish unaided -- and the suspicion band, which supplied 15 of + # 26 findings in the pair measurement, would never be populated at all. + enumerated = enumerate_candidates(root, region, family, taxonomy=taxonomy) + candidates = [{"id": c.id, "text": c.text, "line": c.line} for c in enumerated.candidates] + found = scan_region( + cpg, + spec, + function=region.function or "", + client=counted, + model=model, + candidates=candidates, + parameter_sources=False, # a repository's parameters are not all attacker input + ) + except Exception as exc: # noqa: BLE001 -- one bad region must not end the scan + logger.warning("model scan failed for %s: %s", region.path, exc) + degradations.append({"stage": "model", "reason": "region_failed", "path": region.path}) + continue + collected.extend((finding, region.rank) for finding in found) + scanned += 1 + + findings = _ordered(collected) + return ModelScanResult( + findings=findings, + by_rung=_tally(findings), + regions_scanned=scanned, + regions_unjudged=max(len(regions) - scanned, 0), + model_calls=counted.calls if counted is not None else 0, + cost_usd=round(_cost(counted), 4), + seconds=round(time.monotonic() - started, 2), + degradations=tuple(degradations), + ) + + +def _spec_for(family: str, language: str) -> TaintSpec | DominanceSpec | ConfigSpec | None: + """The spec whose arbiter can decide this family. Routing stays a lookup; dispatch stays in `_arbitrate`.""" + if family == "access_control": + return dominance_specs(language=language).get(family) + if family == "config_secrets": + return config_specs(language=language).get(family) + return taint_specs(language=language).get(family) + + +def _ordered(collected: Sequence[tuple[ModelFinding, float]]) -> tuple[ModelFinding, ...]: + """Rung first, then the region's rank, then the site — a total order, so two runs agree.""" + return tuple(finding for finding, _ in sorted(collected, key=lambda pair: (_RUNG_ORDER.get(pair[0].rung, 9), -pair[1], pair[0].site))) + + +def _empty_tally() -> dict[str, int]: + return {rung.value: 0 for rung in Rung} + + +def _tally(findings: Sequence[ModelFinding]) -> dict[str, int]: + tally = _empty_tally() + for finding in findings: + tally[finding.rung.value] = tally.get(finding.rung.value, 0) + 1 + return tally + + +class _CountingClient: + """Wraps a chat client so the driver counts its own spend, whatever the client records.""" + + def __init__(self, inner: Any) -> None: + self._inner = inner + self.calls = 0 + + def complete(self, **kwargs: Any) -> Any: + self.calls += 1 + return self._inner.complete(**kwargs) + + def cost_usd(self) -> float: + meter = getattr(self._inner, "cost_usd", None) + return float(meter()) if callable(meter) else 0.0 + + +def _cost(client: object | None) -> float: + meter = getattr(client, "cost_usd", None) + return float(meter()) if callable(meter) else 0.0 + + +__all__ = ["ModelScanResult", "ScanBudget", "scan_repository"] diff --git a/tests/test_model_regions.py b/tests/test_model_regions.py new file mode 100644 index 0000000..1a00626 --- /dev/null +++ b/tests/test_model_regions.py @@ -0,0 +1,118 @@ +"""contributor-scan task 1.1: regions come from the code, not from a label (Req 2). + +A contributor cannot tell the tool which function is interesting -- that is the whole question they are +asking. Pair scoring could rely on a labelled function; a repository scan cannot. Regions therefore come from +entry points, with a file-level fallback that says so, and each region carries only the families its shape and +language actually admit. +""" + +from __future__ import annotations + + +def _entry(path="api/users.py", function="update_password", access="public", boundary="network"): # type: ignore[no-untyped-def] + from openultrasast.mapping import EntryPointRecord + + return EntryPointRecord( + path=path, + line=10, + end_line=40, + function_name=function, + name=function, + kind="route", + access_level=access, + trust_boundary=boundary, + access_evidence=[], + conditions=[], + provenance="decorator", + rationale="", + ) + + +def _target(path="api/users.py", language="python"): # type: ignore[no-untyped-def] + from openultrasast.preprocess import FileTarget + + return FileTarget( + path=path, + absolute_path=f"/repo/{path}", + language=language, + loc=100, + tags=[], + has_fuzz_entry_point=False, + static_hints=[], + reachability_hints=[], + ) + + +def test_an_entry_point_becomes_a_region() -> None: + from openultrasast.model.regions import regions_for + + regions = regions_for([_entry()], [_target()]) + assert len(regions) == 1 + assert regions[0].path == "api/users.py" and regions[0].function == "update_password" + assert regions[0].source == "entry_point" + + +def test_a_file_with_no_entry_point_still_gets_a_region_that_says_so() -> None: + """Req 2.2: a fallback is fine; a fallback that pretends to be an entry point is not.""" + from openultrasast.model.regions import regions_for + + regions = regions_for([], [_target(path="lib/util.py")]) + assert len(regions) == 1 + assert regions[0].function is None + assert regions[0].source == "file_fallback" + + +def test_regions_carry_only_families_their_language_has_specs_for() -> None: + """Req 2.3. Asking every family of every region is the unbounded version of this.""" + from openultrasast.model.regions import regions_for + + python = regions_for([_entry()], [_target()])[0] + assert "injection" in python.families + # C has no dominance or config facts; a C region must not claim them. + c_region = regions_for([], [_target(path="src/x.c", language="c")])[0] + assert "access_control" not in c_region.families + assert "config_secrets" not in c_region.families + + +def test_an_absence_family_is_only_offered_where_there_is_a_handler() -> None: + """An obligation needs a reachable operation. A library file with no entry point has no handler to + be inconsistent about, so offering access_control there is asking a question the arbiter cannot answer.""" + from openultrasast.model.regions import regions_for + + handler = regions_for([_entry()], [_target()])[0] + library = regions_for([], [_target(path="lib/util.py")])[0] + assert "access_control" in handler.families + assert "access_control" not in library.families + + +def test_public_entry_points_outrank_internal_ones() -> None: + """Ranking uses what EntryPointRecord already carries rather than inventing a signal.""" + from openultrasast.model.regions import regions_for + + entries = [_entry(path="a.py", function="internal", access="local-only"), _entry(path="b.py", function="exposed", access="public")] + targets = [_target(path="a.py"), _target(path="b.py")] + ranked = regions_for(entries, targets) + assert [r.function for r in ranked] == ["exposed", "internal"], "highest risk first" + assert ranked[0].rank > ranked[1].rank + + +def test_the_order_is_total_so_two_runs_agree() -> None: + from openultrasast.model.regions import regions_for + + entries = [_entry(path=f"{c}.py", function=f"h{i}") for i, c in enumerate("abc")] + targets = [_target(path=f"{c}.py") for c in "abc"] + assert regions_for(entries, targets) == regions_for(entries, targets) + + +def test_a_file_with_an_entry_point_is_not_also_a_fallback_region() -> None: + """Otherwise every handler is scanned twice and the budget is spent on duplicates.""" + from openultrasast.model.regions import regions_for + + regions = regions_for([_entry()], [_target()]) + assert len(regions) == 1 + + +def test_an_unsupported_language_yields_no_region() -> None: + from openultrasast.model.regions import regions_for + + assert regions_for([], [_target(path="x.rb", language="ruby")]) == () diff --git a/tests/test_model_scan.py b/tests/test_model_scan.py new file mode 100644 index 0000000..54e4731 --- /dev/null +++ b/tests/test_model_scan.py @@ -0,0 +1,147 @@ +"""contributor-scan task 1.2: the repository driver (Req 3). + +Pair scoring built one CPG per pair and asked one region. A repository has thousands of regions, and the two +things that were free at pair scale are the two that dominate here: + +* **CPG construction.** A build is 5-30s on a 40-line excerpt. Building one per region would make a + repository scan unusable, so the driver builds ONE and reuses it. +* **Model calls.** The enumerator caps candidates at 8 per region, which says nothing about a scan with two + thousand regions. The budget is therefore a total for the run, spent on the highest-ranked regions first, + and whatever it does not reach is COUNTED rather than quietly dropped. +""" + +from __future__ import annotations + +from pathlib import Path + + +def _region(path="a.py", function="run", families=("injection",), rank=1.0): # type: ignore[no-untyped-def] + from openultrasast.model.regions import ScanRegion + + return ScanRegion(path=path, function=function, language="python", families=families, rank=rank, source="entry_point") + + +class _Backend: + """Counts builds, so the one-CPG-per-scan promise is testable rather than asserted.""" + + def __init__(self, fail: bool = False) -> None: + self.builds = 0 + self._fail = fail + + def available(self) -> bool: + return True + + def build(self, root): # type: ignore[no-untyped-def] + from openultrasast.cpg.backend import CpgResult + + self.builds += 1 + if self._fail: + return None + return CpgResult(cpg_path=Path("cpg.bin"), run=lambda q, p: []) + + +def test_one_cpg_is_built_for_the_whole_repository() -> None: + from openultrasast.model.scan import scan_repository + + backend = _Backend() + regions = tuple(_region(path=f"f{i}.py") for i in range(20)) + scan_repository(Path("/repo"), regions, backend=backend, client=None, model="") + assert backend.builds == 1, "a build per region would make a repository scan unusable" + + +def test_a_failed_build_yields_an_empty_result_not_an_exception() -> None: + from openultrasast.model.scan import scan_repository + + result = scan_repository(Path("/repo"), (_region(),), backend=_Backend(fail=True), client=None, model="") + assert result.findings == () + assert any("cpg_build_failed" in str(d) for d in result.degradations) + + +def test_the_budget_is_a_total_for_the_run_and_the_remainder_is_counted(tmp_path) -> None: + """Req 3.2, 3.3: silently truncating is the failure this exists to prevent. + + Real files on disk, because the driver enumerates candidates from them. A region pointing at a path that + does not exist yields no candidates, no model calls, and a budget that never trips -- which would make + this test pass for the wrong reason. + """ + from openultrasast.model.scan import ScanBudget, scan_repository + + class _Client: + def complete(self, **kw): # type: ignore[no-untyped-def] + from openultrasast.tool_hunter import ChatResponse + + return ChatResponse(content='{"vulnerable": true}') + + source = "import os\n\n\ndef run(cmd):\n os.system(cmd)\n" + for i in range(50): + (tmp_path / f"f{i}.py").write_text(source) + regions = tuple(_region(path=f"f{i}.py") for i in range(50)) + + result = scan_repository( + tmp_path, + regions, + backend=_Backend(), + client=_Client(), + model="m", + budget=ScanBudget(max_model_calls=5, max_regions=50), + ) + assert result.model_calls > 0, "the driver must enumerate candidates and ask the judge about them" + assert result.regions_unjudged > 0 + assert result.regions_scanned + result.regions_unjudged == 50 + assert any("budget_exhausted" in str(d) for d in result.degradations) + + +def test_the_budget_is_spent_on_the_highest_ranked_regions_first() -> None: + from openultrasast.model.scan import ScanBudget, scan_repository + + seen: list[str] = [] + + class _Backend2(_Backend): + def build(self, root): # type: ignore[no-untyped-def] + from openultrasast.cpg.backend import CpgResult + + self.builds += 1 + return CpgResult(cpg_path=Path("c.bin"), run=lambda q, p: (seen.append(str(p.get("function"))), [])[1]) + + regions = (_region(path="low.py", function="low", rank=0.1), _region(path="high.py", function="high", rank=1.0)) + scan_repository(Path("/repo"), regions, backend=_Backend2(), client=None, model="", budget=ScanBudget(max_regions=1)) + assert seen and seen[0] == "high", f"budget spent on the wrong region first: {seen}" + + +def test_findings_are_ordered_by_rung_before_rank() -> None: + """Req 3.1: what the model established outranks what the LLM merely proposed.""" + from openultrasast.model.ladder import Rung + from openultrasast.model.pipeline import ModelFinding + from openultrasast.model.scan import _ordered + + findings = [ + (ModelFinding(site="a", family="injection", rung=Rung.SUSPICION), 1.0), + (ModelFinding(site="b", family="injection", rung=Rung.ENTAILED), 0.1), + (ModelFinding(site="c", family="injection", rung=Rung.CORROBORATED), 0.5), + ] + assert [f.site for f in _ordered(findings)] == ["b", "c", "a"] + + +def test_the_result_reports_what_the_model_arbitrated() -> None: + """Req 1.5: a user should see how much of the result the model established, not have to infer it.""" + from openultrasast.model.scan import scan_repository + + result = scan_repository(Path("/repo"), (_region(),), backend=_Backend(), client=None, model="") + assert set(result.by_rung) >= {"model_entailed", "model_corroborated", "suspicion"} + assert result.regions_scanned == 1 + + +def test_a_region_is_only_asked_about_families_it_carries() -> None: + from openultrasast.model.scan import scan_repository + + asked: list[str] = [] + + class _B(_Backend): + def build(self, root): # type: ignore[no-untyped-def] + from openultrasast.cpg.backend import CpgResult + + self.builds += 1 + return CpgResult(cpg_path=Path("c.bin"), run=lambda q, p: (asked.append(q), [])[1]) + + scan_repository(Path("/repo"), (_region(families=("injection",)),), backend=_B(), client=None, model="") + assert "dominance" not in asked and "config" not in asked