diff --git a/.kiro/specs/flow-aware-ranking/brief.md b/.kiro/specs/flow-aware-ranking/brief.md index 9a8602d..18c9a78 100644 --- a/.kiro/specs/flow-aware-ranking/brief.md +++ b/.kiro/specs/flow-aware-ranking/brief.md @@ -125,6 +125,18 @@ A phase that cannot show that is not an improvement, it is a story. makes a scan finish, because the job is concentration: the CVE is already inside 500, the question is whether it is inside 50. **If this passes, the model is a refinement, not the plan** — and knowing that is worth more than assuming it. + **Implemented 2026-09-11, measurement in flight.** `Evidence.score` is the published weighted sum, and + every weight is in one table in `model/evidence.py`: per open sink +1.0 (capped at 5), per cleansed or + bound sink −0.5 (capped at 3), local source +2.0, carried +1.5, near source +1.0, entry +0.5, declared + public +1.0. A tier-0 pair scores 0 whatever its sources say, so the weights order but cannot + manufacture. `ScanBudget.order_by_evidence` runs the evidence pass over every region (or + `evidence_regions` of them), takes each region's best pair's (tier, score), and cuts `max_regions` from + that order; `order_by_evidence()` is one function used by the scan and by + `benchmarks/ranking/position.py --evidence`, so the benchmark measures the order the scan spends. Off + by default until the benchmark says where the pinned CVEs land. A region with no taint family sorts + after every region with evidence -- the scan does not know it is safe, it knows nothing -- which is a + known cost of this phase for families like `access_control` and is noted here rather than hidden. + - **Phase 3 — the LLM ranker.** Same evidence, ordered by a model. Gate: beats phase 2 leave-one-repository-out. If it does not, phase 2 ships and phase 3 does not. - **Phase 4 — break the closed loop.** An exploration slice (~10% of budget sampled outside the top-K) and diff --git a/benchmarks/ranking/position.py b/benchmarks/ranking/position.py index 5b9c15b..36bcb1c 100644 --- a/benchmarks/ranking/position.py +++ b/benchmarks/ranking/position.py @@ -38,8 +38,43 @@ def ordered_regions(root: Path) -> list[ScanRegion]: return sorted(regions, key=lambda r: (not r.shipped, -r.rank, r.path, r.function or "")) -def measure(name: str, root: Path, known: list[dict[str, object]]) -> dict[str, object]: +def evidence_ordered_regions(root: Path, regions: list[ScanRegion]) -> tuple[list[ScanRegion], dict[str, object]]: + """The regions in the order phase 2 spends the budget: by each region's best (tier, score), from a real + evidence pass over EVERY region. Needs Joern. The ordering function is the scan's own, imported.""" + import time + + from openultrasast.cpg.backend import resolve_cpg_backend + from openultrasast.model.scan import _collect, _evidence_pass, _hook_callbacks, order_by_evidence + + backend = resolve_cpg_backend() + started = time.monotonic() + cpg = backend.build(root, language=regions[0].language if regions else "") + build_seconds = round(time.monotonic() - started, 1) + if cpg is None: + raise SystemExit(f"no graph for {root}: {getattr(backend, 'last_failure', '')}") + try: + started = time.monotonic() + evidence, failed = _evidence_pass(cpg, _collect(regions), _hook_callbacks(root, regions)) + evidence_seconds = round(time.monotonic() - started, 1) + finally: + if callable(getattr(cpg, "cleanup", None)): + cpg.cleanup() + tiers = Counter(e.tier for e in evidence.values()) + facts = { + "build_seconds": build_seconds, + "evidence_seconds": evidence_seconds, + "pairs": len(evidence), + "pairs_unanswered": failed, + "tier_counts": {str(k): v for k, v in sorted(tiers.items())}, + } + return order_by_evidence(regions, evidence), facts + + +def measure(name: str, root: Path, known: list[dict[str, object]], *, by_evidence: bool = False) -> dict[str, object]: regions = ordered_regions(root) + facts: dict[str, object] = {} + if by_evidence: + regions, facts = evidence_ordered_regions(root, regions) tiers = Counter(r.rank for r in regions) largest = max(tiers.values()) if tiers else 0 rows: list[dict[str, object]] = [] @@ -76,6 +111,7 @@ def measure(name: str, root: Path, known: list[dict[str, object]]) -> dict[str, # The smallest examined-region budget that still holds every in-scope known vulnerability. "budget_at_recall": (max(positioned) + 1) if positioned else None, "known": rows, + **facts, } @@ -83,6 +119,7 @@ def main() -> int: parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) parser.add_argument("--json", action="store_true", help="emit one JSON object per repository") parser.add_argument("--repo", help="only this recipe") + parser.add_argument("--evidence", action="store_true", help="phase 2: order by a real evidence pass over every region (needs Joern)") args = parser.parse_args() results: list[dict[str, object]] = [] @@ -99,7 +136,7 @@ def main() -> int: {"id": k.id, "family": k.family, "file": k.file, "function": k.function, "line": k.line, "in_scope": k.in_scope} for k in recipe.known ] - results.append(measure(recipe.name, root, known)) + results.append(measure(recipe.name, root, known, by_evidence=args.evidence)) if args.json: print(json.dumps(results, indent=2)) @@ -114,6 +151,8 @@ def main() -> int: scope = "" if k["in_scope"] else " [out of scope]" print(f" {k['id']:16} {where:28} {k['site']}{scope}") print(f" budget_at_recall = {r['budget_at_recall']}") + if "tier_counts" in r: + print(f" build {r['build_seconds']}s, evidence {r['evidence_seconds']}s over {r['pairs']} pairs, tiers {r['tier_counts']}") return 0 diff --git a/src/openultrasast/model/evidence.py b/src/openultrasast/model/evidence.py index d553154..1509a4f 100644 --- a/src/openultrasast/model/evidence.py +++ b/src/openultrasast/model/evidence.py @@ -21,6 +21,20 @@ TIER_CLEANSED = 2 # a source exists, but every sink call is bound or cleansed o TIER_OPEN = 3 # a source exists and some sink call is computed and uncleansed: ask first TIER_OPEN_PUBLIC = 4 # tier 3 and declared public, or a source on the same statement: ask very first +# The within-tier weights, published here because they are the whole of what "heuristic ranking" means in +# phase 2 and a reader should be able to see every one of them at once. They order, they never exclude. +WEIGHTS: dict[str, float] = { + "open_sink": 1.0, # per sink call that is computed, uncleansed and unbound, up to OPEN_SINK_CAP + "cleansed_sink": -0.5, # per sink call that is cleansed on the call or bound, up to CLEANSED_SINK_CAP + "source_local": 2.0, # a modelled source in the region's own body + "carried": 1.5, # stage one of the two-stage join says a field or dispatch value arrives tainted + "source_near": 1.0, # a modelled source within the arbiter's callDepth + "entry": 0.5, # parameters untrusted by contract + "access_declared_public": 1.0, +} +OPEN_SINK_CAP = 5 +CLEANSED_SINK_CAP = 3 + @dataclass(frozen=True) class SinkEvidence: @@ -77,6 +91,33 @@ class Evidence: return TIER_OPEN_PUBLIC return TIER_OPEN + @property + def score(self) -> float: + """Order WITHIN a tier, by the published weights below. Never across tiers: the tier is the predicate. + + A weighted sum of the same exact facts the tier is made of, so it can rank but cannot manufacture -- + a pair with no sink scores nothing and stays tier 0 whatever its weights. Counts are capped because + a file-scope region with 346 sinks is not 346 times more likely to hold the bug than one with 5; + the cap is where the evidence stops saying more. Phase 2 of flow-aware-ranking. + """ + if self.tier == TIER_EXCLUDE: + return 0.0 # no sink in reach: nothing to weigh, whatever the sources say + w = WEIGHTS + return ( + w["open_sink"] * min(len(self.open_sinks), OPEN_SINK_CAP) + + w["cleansed_sink"] * min(sum(1 for s in self.sinks if s.cleansed_on_call or self.is_bound(s)), CLEANSED_SINK_CAP) + + (w["source_local"] if self.source_local else 0.0) + + (w["carried"] if self.carried else 0.0) + + (w["source_near"] if self.source_near else 0.0) + + (w["entry"] if self.entry else 0.0) + + (w["access_declared_public"] if self.access_declared_public else 0.0) + ) + + @property + def order_key(self) -> tuple[int, float]: + """Higher sorts first: ``sorted(..., key=lambda e: e.order_key, reverse=True)``.""" + return (self.tier, self.score) + def evidence_from_rows( rows: Sequence[Mapping[str, object]] | None, @@ -139,11 +180,14 @@ def _as_int(value: object) -> int: __all__ = [ + "CLEANSED_SINK_CAP", + "OPEN_SINK_CAP", "TIER_CLEANSED", "TIER_EXCLUDE", "TIER_NO_SOURCE", "TIER_OPEN", "TIER_OPEN_PUBLIC", + "WEIGHTS", "Evidence", "SinkEvidence", "evidence_from_rows", diff --git a/src/openultrasast/model/scan.py b/src/openultrasast/model/scan.py index 1051cde..68d4223 100644 --- a/src/openultrasast/model/scan.py +++ b/src/openultrasast/model/scan.py @@ -26,10 +26,10 @@ from pathlib import Path from typing import Any from ..cpg.backend import CpgResult -from .evidence import TIER_EXCLUDE, evidence_from_rows from .candidates import enumerate_candidates from .config_value import request_params as config_params from .dominance import request_params as dominance_params +from .evidence import TIER_EXCLUDE, Evidence, evidence_from_rows from .ladder import Rung from .pipeline import ModelFinding, scan_region from .regions import MODULE_SCOPE, ScanRegion @@ -75,6 +75,12 @@ class ScanBudget: # language, under any dataflow model. Skipping it cannot lose a finding. It is the only tier that # excludes, which is why it is the only one allowed to act before the ranker is measured. prune_tier0: bool = True + # Phase 2 of flow-aware-ranking: spend the region budget by EVIDENCE, not by static rank. The evidence + # pass runs over `evidence_regions` regions in static order (0 = every region; it costs milliseconds a + # pair), each region takes its best pair's (tier, score), and `max_regions` is then cut from that + # order. Off by default until the position benchmark says where the pinned CVEs land under it. + order_by_evidence: bool = False + evidence_regions: int = 0 @dataclass(frozen=True) @@ -185,50 +191,32 @@ def scan_repository( } ) - # Phase 1: collect the whole scan's questions, for at most `max_regions` regions. Nothing is asked yet. - work: list[tuple[str, ScanRegion, ArbiterSpec]] = [] - for region in ordered_regions[: limits.max_regions]: - for family in region.families: - spec = _spec_for(family, region.language) - if spec is None: - continue - work.append((f"{len(work)}", region, spec)) - # Phase 1b: the evidence tier of every taint pair, before any dataflow is asked. Same query, `evidenceOnly`, - # no `reachableByFlows`. Recorded on the result and NOT used to reorder or skip anything -- that is a - # later phase, and it must be measured against this record before it is allowed to change behaviour. + # no `reachableByFlows`. Phase 0 recorded it; phase 1 prunes tier 0 on it; phase 2 spends the region + # budget by it. Under phase 2 the pass runs over MORE regions than the budget, which is the point. tiers: list[tuple[str, str, str, int]] = [] - tier_by_rid: dict[str, int] = {} per_kind: dict[str, float] = {} + evidence_by_pair: dict[tuple[str, str, str], Evidence] = {} if limits.tiering and callable(getattr(cpg, "run_batch", None)): - grouped_taint = _grouped(work, hook_callbacks=hooks).get("taint", {}) - if grouped_taint: - evidence_requests = {rid: {**dict(params), "evidenceOnly": "true"} for rid, params in grouped_taint.items()} - tier_started = time.monotonic() - answered_evidence = cpg.run_batch("taint", evidence_requests) - per_kind["evidence"] = round(time.monotonic() - tier_started, 2) - if answered_evidence is None: - degradations.append({"stage": "model", "reason": "query_failed", "kind": "evidence", "requests": len(evidence_requests)}) - else: - answered_evidence.pop("__census__", None) - by_rid = {rid: (region, spec) for rid, region, spec in work} - for rid, rows in answered_evidence.items(): - pair = by_rid.get(rid) - if pair is None: - continue - region, spec = pair - evidence = evidence_from_rows( - rows, - entry=_is_entry_point(region), - # Rank 1.0 is the public tier. Provenance -- declared by a route contract versus - # inferred from an absent decorator -- is not yet carried on ScanRegion, so this is - # the proxy phase 0 has; the brief names it as the field to thread through next. - access_declared_public=region.rank >= 1.0, - bound_names=getattr(spec, "safe_shape_sinks", ()), - ) - if evidence is not None: - tiers.append((region.path, region.function or "", getattr(spec, "family", ""), evidence.tier)) - tier_by_rid[rid] = evidence.tier + candidates = ( + ordered_regions[: (limits.evidence_regions or len(ordered_regions))] + if limits.order_by_evidence + else ordered_regions[: limits.max_regions] + ) + tier_started = time.monotonic() + evidence_by_pair, failed = _evidence_pass(cpg, _collect(candidates), hooks) + per_kind["evidence"] = round(time.monotonic() - tier_started, 2) + if failed: + degradations.append({"stage": "model", "reason": "query_failed", "kind": "evidence", "requests": failed}) + tiers = [(path, function, family, evidence.tier) for (path, function, family), evidence in evidence_by_pair.items()] + if limits.order_by_evidence and evidence_by_pair: + ordered_regions = order_by_evidence(ordered_regions, evidence_by_pair) + + # Phase 1: collect the whole scan's questions, for at most `max_regions` regions. Nothing is asked yet. + work = _collect(ordered_regions[: limits.max_regions]) + tier_by_rid = { + rid: evidence_by_pair[_pair_key(region, spec)].tier for rid, region, spec in work if _pair_key(region, spec) in evidence_by_pair + } # Phase 2: ONE invocation per query kind. JVM startup dominates a repository scan -- a call per region # per family put a ten-line file at four minutes and a thousand regions at roughly fifty hours -- while @@ -362,6 +350,85 @@ def scan_repository( ) +def _pair_key(region: ScanRegion, spec: ArbiterSpec) -> tuple[str, str, str]: + return (region.path, region.function or "", getattr(spec, "family", "")) + + +def _collect(regions: Sequence[ScanRegion]) -> list[tuple[str, ScanRegion, ArbiterSpec]]: + """Every (region, family) question with an arbiter, in the order given, with batch-local ids.""" + work: list[tuple[str, ScanRegion, ArbiterSpec]] = [] + for region in regions: + for family in region.families: + spec = _spec_for(family, region.language) + if spec is None: + continue + work.append((f"{len(work)}", region, spec)) + return work + + +def _evidence_pass( + cpg: Any, work: Sequence[tuple[str, ScanRegion, ArbiterSpec]], hooks: str +) -> tuple[dict[tuple[str, str, str], Evidence], int]: + """The evidence vector of every taint pair in ``work``, keyed by (path, function, family). + + Returns the vectors and the number of requests the engine did not answer (0 when it answered), so a + failed pass is recorded rather than read as "no evidence anywhere". + """ + grouped_taint = _grouped(list(work), hook_callbacks=hooks).get("taint", {}) + if not grouped_taint: + return {}, 0 + requests = {rid: {**dict(params), "evidenceOnly": "true"} for rid, params in grouped_taint.items()} + answered = cpg.run_batch("taint", requests) + if answered is None: + return {}, len(requests) + answered.pop("__census__", None) + by_rid = {rid: (region, spec) for rid, region, spec in work} + out: dict[tuple[str, str, str], Evidence] = {} + for rid, rows in answered.items(): + pair = by_rid.get(rid) + if pair is None: + continue + region, spec = pair + evidence = evidence_from_rows( + rows, + entry=_is_entry_point(region), + # Rank 1.0 is the public tier. Provenance -- declared by a route contract versus inferred from + # an absent decorator -- is not yet carried on ScanRegion, so this is the proxy phase 0 has; + # the brief names it as the field to thread through next. + access_declared_public=region.rank >= 1.0, + bound_names=getattr(spec, "safe_shape_sinks", ()), + ) + if evidence is not None: + out[_pair_key(region, spec)] = evidence + return out, 0 + + +def order_by_evidence(regions: Sequence[ScanRegion], evidence: Mapping[tuple[str, str, str], Evidence]) -> list[ScanRegion]: + """Regions by their best pair's (tier, score), then by the static order they arrived in. + + ONE definition, used by the scan to spend its budget and by `benchmarks/ranking/position.py` to say + where a pinned CVE lands under it; if they drifted, the benchmark would measure the wrong thing. A + region with no evidence at all -- no taint family, or a pass that failed -- sorts after every region + that has some, in static order: the scan does not know it is safe, it knows nothing about it. + """ + best: dict[tuple[str, str], tuple[int, float]] = {} + for (path, function, _family), vector in evidence.items(): + key = (path, function) + best[key] = max(best.get(key, (-1, 0.0)), vector.order_key) + indexed = list(enumerate(regions)) + return [ + region + for _, region in sorted( + indexed, + key=lambda item: ( + -best.get((item[1].path, item[1].function or ""), (-1, 0.0))[0], + -best.get((item[1].path, item[1].function or ""), (-1, 0.0))[1], + item[0], + ), + ) + ] + + def _dispose(cpg: object) -> None: """Release a built CPG's scratch tree. Never raises: cleanup must not turn a good scan into a failure.""" cleanup = getattr(cpg, "cleanup", None) diff --git a/tests/test_model_evidence.py b/tests/test_model_evidence.py index 44803d3..133eb39 100644 --- a/tests/test_model_evidence.py +++ b/tests/test_model_evidence.py @@ -97,3 +97,29 @@ def test_a_stringified_boolean_is_read_as_its_value_not_its_truthiness() -> None rows[1]["cleansedOnCall"] = "False" e = evidence_from_rows(rows, entry=True) assert e is not None and e.tier == TIER_OPEN, "a 'False' string must not turn an open sink into a cleansed one" + + +def test_the_score_orders_within_a_tier_and_never_across_one() -> None: + """Phase 2: the published weights rank pairs inside a tier. They cannot lift a pair out of tier 0 -- + a pair with no sink scores nothing whatever the weights say -- and counts are capped so a 346-sink + file-scope region does not outrank everything by volume alone.""" + from openultrasast.model.evidence import OPEN_SINK_CAP, WEIGHTS, Evidence, SinkEvidence + + def sink(n: int, cleansed: bool = False) -> SinkEvidence: + return SinkEvidence(code=f"q({n})", line=n, method="m", arity=1, arg0_literal=False, cleansed_on_call=cleansed) + + nothing = Evidence(sinks=(), source_local=True) + assert nothing.tier == 0 and nothing.score == 0.0, "no sink, no score: the weights cannot manufacture a pair" + + one = Evidence(sinks=(sink(1),), source_near=True) + local = Evidence(sinks=(sink(1),), source_local=True) + assert local.tier > one.tier, "a local source is the higher tier, not merely the higher score" + assert local.score == WEIGHTS["open_sink"] + WEIGHTS["source_local"] + + many = Evidence(sinks=tuple(sink(i) for i in range(50)), source_near=True) + assert many.tier == one.tier + assert many.score == WEIGHTS["open_sink"] * OPEN_SINK_CAP + WEIGHTS["source_near"], "capped: volume stops counting" + + cleansed = Evidence(sinks=(sink(1), sink(2, cleansed=True)), source_near=True) + assert cleansed.score < one.score, "a cleansed sink beside an open one counts against the pair" + assert sorted([many, one, local], key=lambda e: e.order_key, reverse=True) == [local, many, one] diff --git a/tests/test_model_scan.py b/tests/test_model_scan.py index 0b1f94d..f398ca2 100644 --- a/tests/test_model_scan.py +++ b/tests/test_model_scan.py @@ -472,3 +472,67 @@ def test_tier_zero_pairs_are_not_asked_and_nothing_else_is_skipped() -> None: assert taint["count"] == 1, "only the pair with a sink in reach is asked" assert result.requests_pruned == 2, "and the two tier-0 pairs are counted, not silently dropped" assert not any(d.get("reason") == "query_failed" for d in result.degradations) + + +def test_the_budget_is_spent_by_evidence_when_asked() -> None: + """Phase 2 of flow-aware-ranking. With `order_by_evidence`, the evidence pass runs over every region and + the region cap is cut from the (tier, score) order, so a low-ranked region with an open sink and a + local source is examined before a top-ranked region with no sink in reach. Off, the static order holds.""" + from openultrasast.model.regions import ScanRegion + from openultrasast.model.scan import ScanBudget, scan_repository + + class _Backend: + def __init__(self) -> None: + self.taint_files: list[list[str]] = [] + self.evidence_sizes: list[int] = [] + + def available(self) -> bool: + return True + + def build(self, root, *, language=""): # type: ignore[no-untyped-def] + from openultrasast.cpg.backend import CpgResult + + def batch(kind, requests): # type: ignore[no-untyped-def] + if any(p.get("evidenceOnly") == "true" for p in requests.values()): + self.evidence_sizes.append(len(requests)) + out = {} + for rid, params in requests.items(): + hot = str(params.get("file", "")).endswith("low.py") + out[rid] = [{"kind": "summary", "sinks": 1 if hot else 0, "sourceLocal": hot, "familyInRepo": True}] + ( + [ + { + "kind": "sink", + "sink": "os.system(x)", + "sinkLine": "4", + "sinkMethod": "run", + "sinkArity": 1, + "sinkArg0Literal": False, + "cleansedOnCall": False, + } + ] + if hot + else [] + ) + return out + if kind == "taint": + self.taint_files.append(sorted({str(p.get("file", "")) for p in requests.values()})) + return {rid: [] for rid in requests} + + return CpgResult(cpg_path=Path("cpg.bin"), run=lambda q, p: [], run_batch=batch) + + top = ScanRegion(path="top.py", function="handle", language="python", families=("injection",), rank=1.0, source="entry_point") + low = ScanRegion(path="low.py", function="run", language="python", families=("injection",), rank=0.3, source="entry_point") + + static = _Backend() + scan_repository(Path("."), [top, low], backend=static, budget=ScanBudget(max_model_calls=0, max_regions=1)) + assert static.evidence_sizes == [1] and static.taint_files == [], ( + "static order: the budget holds `top`, whose pair is tier 0 and pruned" + ) + + ordered = _Backend() + result = scan_repository( + Path("."), [top, low], backend=ordered, budget=ScanBudget(max_model_calls=0, max_regions=1, order_by_evidence=True) + ) + assert ordered.evidence_sizes == [2], "the evidence pass covers every region, not only the budget" + assert ordered.taint_files == [["low.py"]], "and the budget is spent on the region the evidence points at" + assert result.tier_counts == {0: 1, 4: 1}