diff --git a/src/openultrasast/learning/candidates.py b/src/openultrasast/learning/candidates.py new file mode 100644 index 0000000..01908fd --- /dev/null +++ b/src/openultrasast/learning/candidates.py @@ -0,0 +1,257 @@ +"""What a detector is asked to judge, enumerated deterministically (constrained-detector, Req 1). + +The detector stops searching. Measured on 2026-09-06, 120 runs of the unconstrained loop over the same twelve pairs +produced 120 distinct tool-call trajectories, and outcome disagreement tracked trajectory divergence rather than +judgement: the model was spending its budget guessing grep patterns to find sinks the semantic IR already knows. + +So something deterministic decides what the model looks at, and that enumerator becomes a hard ceiling on recall. +Which is why it was measured before any of this was built: the pattern ruleset finds a candidate inside the labeled +function for **12.1%** of the corpus — below the 64.9% the unconstrained loop already reaches — while the IR's call +sites and bindings reach **96.6%**, a median of ten sites per function. The ruleset is never the generator here. + +Ordering is `(path, line, name)` and the bound is a constant, so the batches the judgment step forms, and therefore +the sequence of model calls, are determined by the input alone. +""" + +from __future__ import annotations + +from collections.abc import Sequence +from dataclasses import dataclass, field +from pathlib import Path +from typing import Literal + +CandidateKind = Literal["call", "bind", "operation", "config"] + +# What can carry each family's bug. An injection lives at a call; an access-control bug is the *absence* of a guard +# around an operation, so its candidate is the operation and what the model needs is the guards in scope. +FAMILY_SHAPE: dict[str, tuple[CandidateKind, ...]] = { + "injection": ("call",), + "path": ("call",), + "deserialization": ("call",), + "untrusted_destination": ("call",), + "output_encoding": ("call",), + "memory": ("call",), + "prototype": ("call", "bind"), + "access_control": ("operation",), + "config_secrets": ("bind", "config"), + "unknown": ("call", "bind"), +} + +# The p90 of the measured distribution is 22 sites per labeled function, so two batches cover all but the tail. +MAX_CANDIDATES_PER_CALL = 12 +_MAX_TEXT = 400 +_LANGUAGE = {"python": "python", "javascript": "javascript", "typescript": "typescript", "c_cpp": "c", "java": "java"} + + +@dataclass(frozen=True) +class Candidate: + """One site the model will be asked a single bounded question about.""" + + path: str + line: int + kind: CandidateKind + name: str + text: str + function: str + arg_texts: tuple[str, ...] = () + arg_names: tuple[tuple[str, ...], ...] = () + # What the IR resolved for the arguments, nearest binding first. This is the work the model used to do by + # grepping, and the reason it no longer has to. + binding_chain: tuple[str, ...] = () + guards: tuple[str, ...] = () + + @property + def id(self) -> str: + return f"{self.path}:{self.line}:{self.name}" + + def to_dict(self) -> dict[str, object]: + return { + "id": self.id, + "kind": self.kind, + "name": self.name, + "line": self.line, + "function": self.function, + "text": self.text, + "arguments": list(self.arg_texts), + "binding_chain": list(self.binding_chain), + "guards": list(self.guards), + } + + +@dataclass(frozen=True) +class CandidateSet: + """The sites to judge, what was dropped past the bound, and why there is nothing when there is nothing.""" + + candidates: tuple[Candidate, ...] = () + discarded: int = 0 + reason: str = "" # "", unsupported_language, no_candidate + function_found: bool = True + + +def enumerate_candidates( + root: Path, + region: object, + family: str, + *, + taxonomy: object, + limit: int = MAX_CANDIDATES_PER_CALL, +) -> CandidateSet: + """Every site in the region that could carry ``family``'s bug, ordered and bounded.""" + del taxonomy # the family id is the key; the taxonomy validates it elsewhere + from ..semantic.ir import parse_file + + path = str(getattr(region, "path", "") or "") + wanted = str(getattr(region, "function", "") or "") + target = root / path + if not target.is_file(): + return CandidateSet(reason="unsupported_language", function_found=False) + text = target.read_text(errors="ignore") + ir = parse_file(path, text, _language_of(path)) + if not ir.parse_ok: + return CandidateSet(reason="unsupported_language", function_found=False) + functions = [item for item in ir.functions if item.name != "" and (not wanted or item.name == wanted)] + if wanted and not functions: + return CandidateSet(reason="no_candidate", function_found=False) + kinds = FAMILY_SHAPE.get(family, ("call", "bind")) + lines = text.splitlines() + found: list[Candidate] = [] + for function in functions: + guards = _guards(ir, function, text, path) if "operation" in kinds else () + found.extend(_from_function(path, function, kinds, lines, guards)) + found.sort(key=lambda item: (item.path, item.line, item.name)) + # One question per site: a chained expression reaches the IR as several calls at the same line and name — one + # bare, one carrying the arguments — and asking twice would spend budget and double-count the answer. Keep the + # richest, because `filter_by(user_id=owner)` and a bare `filter_by` are the same site and only one of them + # says what the constraint was. + best: dict[str, Candidate] = {} + for item in found: + current = best.get(item.id) + if current is None or (len(item.arg_texts), len(item.binding_chain)) > (len(current.arg_texts), len(current.binding_chain)): + best[item.id] = item + found = sorted(best.values(), key=lambda item: (item.path, item.line, item.name)) + if not found: + return CandidateSet(reason="no_candidate") + return CandidateSet(candidates=tuple(found[:limit]), discarded=max(len(found) - limit, 0)) + + +def _from_function(path: str, function: object, kinds: Sequence[str], lines: Sequence[str], guards: tuple[str, ...]): # type: ignore[no-untyped-def] + binds = tuple(getattr(function, "binds", ()) or ()) + name = str(getattr(function, "name", "")) + out: list[Candidate] = [] + if "call" in kinds or "operation" in kinds: + kind: CandidateKind = "operation" if "operation" in kinds else "call" + for site in getattr(function, "calls", ()) or (): + out.append( + Candidate( + path=path, + line=site.line, + kind=kind, + name=site.name, + text=_line_text(lines, site.line), + function=name, + arg_texts=tuple(site.arg_texts), + arg_names=tuple(tuple(names) for names in site.arg_names), + binding_chain=_chain(binds, [n for names in site.arg_names for n in names], site.line), + guards=guards, + ) + ) + if "bind" in kinds or "config" in kinds: + for bind in binds: + if "config" in kinds and "bind" not in kinds and not bind.is_constant: + continue + out.append( + Candidate( + path=path, + line=bind.line, + kind="config" if (bind.is_constant and "config" in kinds) else "bind", + name=bind.name, + text=_line_text(lines, bind.line), + function=name, + arg_texts=(bind.value_text[:_MAX_TEXT],), + arg_names=(tuple(bind.names),), + binding_chain=_chain(binds, list(bind.names), bind.line), + guards=guards, + ) + ) + return out + + +def _chain(binds: Sequence[object], names: Sequence[str], before: int, depth: int = 4) -> tuple[str, ...]: + """Walk the bindings backwards from the names an argument mentions, nearest first. + + This is exactly the work the unconstrained loop was doing with `grep_repo`, and it is deterministic.""" + chain: list[str] = [] + seen: set[str] = set() + wanted = [name for name in names if name] + for _ in range(depth): + step = None + for bind in sorted(binds, key=lambda item: -int(getattr(item, "line", 0))): + line = int(getattr(bind, "line", 0)) + bound = str(getattr(bind, "name", "")) + if line >= before or bound not in wanted or bound in seen: + continue + step = bind + break + if step is None: + break + seen.add(str(step.name)) # type: ignore[attr-defined] + chain.append(f"{step.name} = {str(step.value_text)[:_MAX_TEXT]}") # type: ignore[attr-defined] + before = int(step.line) # type: ignore[attr-defined] + wanted = [name for name in getattr(step, "names", ()) or () if name] + if not wanted: + break + return tuple(chain) + + +def _guards(ir: object, function: object, text: str, path: str) -> tuple[str, ...]: + """Decorators and in-body witnesses that discharge an obligation on this function, by identifier only.""" + try: + from ..semantic.facts import load_facts + from ..semantic.obligations import load_obligation_facts + from ..semantic.obligations.operations import find_discharges + except ImportError: # pragma: no cover - the obligations facts are shipped, this is belt and braces + return () + name = str(getattr(function, "name", "")) + try: + witnesses = find_discharges(ir, load_obligation_facts(), load_facts(), text=text) # type: ignore[arg-type] + except Exception: # noqa: BLE001 - a missing fact file must not stop enumeration + return () + del path + return tuple(sorted({item.detail or item.fact_id for item in witnesses if item.function == name})) + + +def _line_text(lines: Sequence[str], line: int) -> str: + return lines[line - 1].strip()[:_MAX_TEXT] if 0 < line <= len(lines) else "" + + +def _language_of(path: str) -> str: + from ..preprocess import detect_language + + return _LANGUAGE.get(detect_language(Path(path)), detect_language(Path(path))) + + +@dataclass(frozen=True) +class CandidateReport: + """The ceiling: what a candidate-driven detector could find before any model is asked anything.""" + + per_family: dict[str, dict[str, object]] = field(default_factory=dict) + per_slice: dict[str, dict[str, object]] = field(default_factory=dict) + gaps: tuple[tuple[str, str, str], ...] = () # (slice, pair, reason) + + def to_dict(self) -> dict[str, object]: + return { + "per_family": {name: dict(block) for name, block in sorted(self.per_family.items())}, + "per_slice": {name: dict(block) for name, block in sorted(self.per_slice.items())}, + "gaps": [list(row) for row in self.gaps], + } + + +__all__ = [ + "FAMILY_SHAPE", + "MAX_CANDIDATES_PER_CALL", + "Candidate", + "CandidateKind", + "CandidateReport", + "CandidateSet", + "enumerate_candidates", +] diff --git a/tests/test_candidate_enumeration.py b/tests/test_candidate_enumeration.py new file mode 100644 index 0000000..521d812 --- /dev/null +++ b/tests/test_candidate_enumeration.py @@ -0,0 +1,156 @@ +"""Candidates come from the semantic IR, shaped by family (constrained-detector 1.1, Req 1.1, 1.2, 1.4-1.6). + +The detector stops searching. Something deterministic decides what it looks at, which makes that enumerator a hard +ceiling on recall — measured before this was built at 12.1% for the pattern ruleset against 96.6% for the IR. The +ordering is part of the contract: the batches a later task forms, and therefore the sequence of model calls, must +be determined by the input alone. +""" + +from __future__ import annotations + +import sys +from pathlib import Path + +import pytest + +sys.path.insert(0, str(Path(__file__).parent)) +from test_obligation_operations import APP # noqa: E402 + +from openultrasast.learning.families import load_families # noqa: E402 + +INJECTION = ( + "from flask import request\n" + "import os\n" + "import sqlite3\n\n\n" + "@app.route('/run')\n" + "def run():\n" + " raw = request.args.get('cmd')\n" + " command = 'echo ' + raw\n" + " os.system(command)\n" + " return 'ok'\n" +) + + +def _region(path: str = "app.py", function: str | None = "run"): # type: ignore[no-untyped-def] + from openultrasast.learning.detectors import Region + + return Region(path=path, function=function) + + +def _write(tmp_path: Path, name: str, text: str) -> Path: + (tmp_path / name).write_text(text) + return tmp_path + + +def test_a_call_site_candidate_carries_what_the_model_needs_to_judge_it(tmp_path: Path) -> None: + from openultrasast.learning.candidates import enumerate_candidates + + root = _write(tmp_path, "app.py", INJECTION) + result = enumerate_candidates(root, _region(), "injection", taxonomy=load_families()) + assert result.reason == "" + sites = {item.name: item for item in result.candidates} + assert "os.system" in sites or "system" in sites, sorted(sites) + sink = sites.get("os.system") or sites["system"] + assert sink.kind == "call" and sink.function == "run" and sink.path == "app.py" + assert sink.line == 10 # `from flask` 1, imports 2-3, blanks, decorator 6, def 7, raw 8, command 9 + assert sink.arg_texts == ("command",) + # the chain the IR resolved, nearest first: command <- 'echo ' + raw <- request.args.get('cmd') + assert any("raw" in step for step in sink.binding_chain), sink.binding_chain + assert any("request" in step for step in sink.binding_chain), sink.binding_chain + assert sink.id in {"app.py:10:os.system", "app.py:10:system"} + + +def test_the_family_decides_which_kinds_are_candidates() -> None: + from openultrasast.learning.candidates import FAMILY_SHAPE + + taxonomy = load_families() + assert set(FAMILY_SHAPE) == {family.id for family in taxonomy.families}, "every family says what carries its bug" + assert FAMILY_SHAPE["injection"] == ("call",) + assert FAMILY_SHAPE["access_control"] == ("operation",) + assert "bind" in FAMILY_SHAPE["config_secrets"] + assert "call" in FAMILY_SHAPE["prototype"] + + +def test_access_control_candidates_are_operations_with_the_guards_in_scope(tmp_path: Path) -> None: + """An absence bug has no sink. Its candidate is the operation, and what the model needs is what guards it.""" + from openultrasast.learning.candidates import enumerate_candidates + + root = _write(tmp_path, "app.py", APP) + guarded = enumerate_candidates(root, _region(function="guarded"), "access_control", taxonomy=load_families()) + assert guarded.reason == "" and guarded.candidates + assert all(item.kind == "operation" for item in guarded.candidates) + assert any("login_required" in guard for item in guarded.candidates for guard in item.guards) + # `leaky` and `constrained` both constrain by `filter_by`; what separates them is where the constraining value + # came from, and that is the binding chain the IR resolved — the work the model used to do by grepping. + leaky = enumerate_candidates(root, _region(function="leaky"), "access_control", taxonomy=load_families()) + constrained = enumerate_candidates(root, _region(function="constrained"), "access_control", taxonomy=load_families()) + leaky_site = next(item for item in leaky.candidates if item.name.endswith("filter_by")) + good_site = next(item for item in constrained.candidates if item.name.endswith("filter_by")) + assert "user_id=owner" in leaky_site.arg_texts and "user_id=user_id" in good_site.arg_texts + assert any("request.args.get" in step for step in leaky_site.binding_chain), leaky_site.binding_chain + assert any("request.user.id" in step for step in good_site.binding_chain), good_site.binding_chain + + +def test_a_side_the_parser_rejects_yields_no_candidates_and_says_why(tmp_path: Path) -> None: + from openultrasast.learning.candidates import enumerate_candidates + + root = _write(tmp_path, "app.py", "def broken(:\n pass\n") + result = enumerate_candidates(root, _region(function="broken"), "injection", taxonomy=load_families()) + assert result.candidates == () and result.reason == "unsupported_language" + + +def test_a_labeled_function_with_nothing_to_judge_says_no_candidate(tmp_path: Path) -> None: + from openultrasast.learning.candidates import enumerate_candidates + + root = _write(tmp_path, "app.py", "def empty():\n return 1\n") + result = enumerate_candidates(root, _region(function="empty"), "injection", taxonomy=load_families()) + assert result.candidates == () and result.reason == "no_candidate" + + +def test_the_bound_is_a_constant_and_the_overflow_is_counted(tmp_path: Path) -> None: + from openultrasast.learning.candidates import MAX_CANDIDATES_PER_CALL, enumerate_candidates + + # one call per line, so the arithmetic is the bound and not the fixture + body = "\n".join(f" sink_{index}(value)" for index in range(MAX_CANDIDATES_PER_CALL + 5)) + root = _write(tmp_path, "app.py", f"def many(value):\n{body}\n") + result = enumerate_candidates(root, _region(function="many"), "injection", taxonomy=load_families()) + assert len(result.candidates) == MAX_CANDIDATES_PER_CALL + assert result.discarded == 5 + + +def test_enumerating_twice_returns_an_identical_sequence(tmp_path: Path) -> None: + """The batches, and therefore the sequence of model calls, are determined by the input alone (Req 2.4).""" + from openultrasast.learning.candidates import enumerate_candidates + + root = _write(tmp_path, "app.py", APP) + taxonomy = load_families() + first = enumerate_candidates(root, _region(function="constrained"), "injection", taxonomy=taxonomy) + second = enumerate_candidates(root, _region(function="constrained"), "injection", taxonomy=taxonomy) + assert [item.id for item in first.candidates] == [item.id for item in second.candidates] + assert [item.id for item in first.candidates] == sorted({item.id for item in first.candidates}, key=_order(first)) + + +def _order(result): # type: ignore[no-untyped-def] + index = {item.id: (item.path, item.line, item.name) for item in result.candidates} + return lambda identifier: index[identifier] + + +def test_a_region_without_a_function_enumerates_the_whole_file(tmp_path: Path) -> None: + from openultrasast.learning.candidates import enumerate_candidates + + root = _write(tmp_path, "app.py", APP) + whole = enumerate_candidates(root, _region(function=None), "injection", taxonomy=load_families()) + named = enumerate_candidates(root, _region(function="leaky"), "injection", taxonomy=load_families()) + assert len(whole.candidates) >= len(named.candidates) + assert {item.function for item in whole.candidates} > {"leaky"} + + +@pytest.mark.parametrize("family", ["injection", "path", "deserialization", "untrusted_destination", "config_secrets"]) +def test_every_family_shape_enumerates_something_on_a_file_that_has_it(tmp_path: Path, family: str) -> None: + from openultrasast.learning.candidates import enumerate_candidates + + root = _write(tmp_path, "app.py", INJECTION) + result = enumerate_candidates(root, _region(), family, taxonomy=load_families()) + assert result.reason in {"", "no_candidate"} + if family in {"injection", "path", "deserialization", "untrusted_destination"}: + assert result.candidates, f"{family} takes call sites and this file has three"