diff --git a/.kiro/specs/learned-decision-engine/tasks.md b/.kiro/specs/learned-decision-engine/tasks.md index 156cd4c..47431fa 100644 --- a/.kiro/specs/learned-decision-engine/tasks.md +++ b/.kiro/specs/learned-decision-engine/tasks.md @@ -38,7 +38,7 @@ a model or embedding call. - _Requirements: 1.1_ - [ ] 6. Program, local memory and compilation (design sections 4-5) - - [ ] 6.1 Excerpts and the example store (no model) + - [x] 6.1 Excerpts and the example store (no model) - `learn/excerpt.py` (function span at the label's pin, bounds 80/4,000 for candidates and 40/2,000 for examples, numbered lines, no path/repository/commit, advisory ids and security wording in comments redacted, delta diff); `learn/examples.py` and `ousast learn memory build` (label + feature record + diff --git a/benchmarks/measurements/2026-09-08-module-audit.json b/benchmarks/measurements/2026-09-08-module-audit.json index 14d1d90..806da5d 100644 --- a/benchmarks/measurements/2026-09-08-module-audit.json +++ b/benchmarks/measurements/2026-09-08-module-audit.json @@ -1,7 +1,7 @@ { - "modules": 143, + "modules": 146, "counts": { - "load_bearing": 136, + "load_bearing": 139, "standalone_capability": 7 }, "orphaned": [], @@ -352,6 +352,9 @@ "reason": "reachable from gate, map_gate, pair_gate or the CLI", "importers": [ "cli", + "learn.cli", + "learn.examples", + "learn.excerpt", "learn.features", "learn.roles", "plane.tasks.features", @@ -359,11 +362,36 @@ "plane.tasks.roles" ] }, + { + "module": "learn.cli", + "classification": "load_bearing", + "reason": "reachable from gate, map_gate, pair_gate or the CLI", + "importers": [ + "cli" + ] + }, + { + "module": "learn.examples", + "classification": "load_bearing", + "reason": "reachable from gate, map_gate, pair_gate or the CLI", + "importers": [ + "learn.cli" + ] + }, + { + "module": "learn.excerpt", + "classification": "load_bearing", + "reason": "reachable from gate, map_gate, pair_gate or the CLI", + "importers": [ + "learn.examples" + ] + }, { "module": "learn.features", "classification": "load_bearing", - "reason": "imported by plane.tasks.features", + "reason": "reachable from gate, map_gate, pair_gate or the CLI", "importers": [ + "learn.excerpt", "plane.tasks.features" ] }, @@ -372,13 +400,16 @@ "classification": "load_bearing", "reason": "reachable from gate, map_gate, pair_gate or the CLI", "importers": [ - "cli" + "cli", + "learn.cli", + "learn.examples", + "learn.excerpt" ] }, { "module": "learn.roles", "classification": "load_bearing", - "reason": "imported by learn.features, plane.tasks.features, plane.tasks.roles", + "reason": "reachable from gate, map_gate, pair_gate or the CLI", "importers": [ "learn.features", "plane.tasks.features", @@ -706,6 +737,7 @@ "improve", "improve.evolve", "improve.memory", + "learn.examples", "learn.labels", "model.candidates", "pair_gate", @@ -719,6 +751,8 @@ "importers": [ "cli", "improve.memory", + "learn.cli", + "learn.examples", "plane.egress", "plane.engine_alerts", "plane.generate", @@ -792,6 +826,8 @@ "reason": "harnessx-removal Req 6.1/6.4: the memory store (file, MinIO), ingest and fact reuse; `ousast plane remember`", "importers": [ "improve.memory", + "learn.cli", + "learn.examples", "plane.generate", "plane.tasks.loop", "plane.tasks.remember" @@ -828,6 +864,7 @@ "reason": "ai-service-plane Req 4.1: the task entrypoint package the runner imports by name; runner integration pending (task 3)", "importers": [ "cli", + "learn.excerpt", "learn.features", "learn.labels", "learn.roles", @@ -884,6 +921,7 @@ "classification": "load_bearing", "reason": "ai-service-plane Req 5.1/5.2: source-only functions and callers per file; runner integration pending", "importers": [ + "learn.excerpt", "learn.features", "learn.labels", "learn.roles", diff --git a/src/openultrasast/cli.py b/src/openultrasast/cli.py index a3f2c98..0bb8d17 100644 --- a/src/openultrasast/cli.py +++ b/src/openultrasast/cli.py @@ -44,6 +44,8 @@ from .harness import HarnessRuntime, HarnessTraceWriter, write_harness_config from .hunter import run_hunter_pool, write_hunter_trajectories from .improve import RoundOutcome, run_improvement from .index import build_code_chunks +from .learn.cli import add_commands as add_learn_commands +from .learn.cli import run as run_learn from .mapping import analyze_entry_points, attach_reachability_hints, ingest_sarif, write_entry_points, write_static_hints from .pair_gate import print_pair_metrics from .pairs import ( @@ -269,6 +271,7 @@ def _main(argv: list[str] | None) -> int: learn_labels.add_argument("--snapshot", type=Path, help="write the counts-only snapshot (the committable record) here") learn_labels.add_argument("--include-title", action="store_true", help="the title-tier pairs as well (a sensitivity arm)") learn_labels.add_argument("--no-assumed-benign", action="store_true", help="skip mining the assumed-benign history source") + add_learn_commands(learn_sub) plane = subparsers.add_parser("plane", help="maintainer: the ax-backed service plane on this host's kind cluster") plane_sub = plane.add_subparsers(dest="plane_command", required=True) plane_run = plane_sub.add_parser("run", help="execute a Run manifest on ax; a rerun skips done tasks") @@ -440,7 +443,7 @@ def _main(argv: list[str] | None) -> int: if args.command == "plane": return _plane(args) if args.command == "learn": - return _learn_labels(args) + return _learn_labels(args) if args.learn_command == "labels" else run_learn(args) if args.command == "mcp": from .mcp import serve # lazy: keeps the import cycle (mcp -> cli) one-directional diff --git a/src/openultrasast/learn/cli.py b/src/openultrasast/learn/cli.py new file mode 100644 index 0000000..5b2f72f --- /dev/null +++ b/src/openultrasast/learn/cli.py @@ -0,0 +1,69 @@ +"""``ousast learn ...`` beyond ``labels``: the local memory and the compiled program (design sections 4-5). + +Host only, maintainer surface. The parser is registered from :mod:`..cli`; every heavy import is lazy. No command +here spends money unless it is given a real client: ``compile``/``evaluate``/``curve`` and ``memory embed`` run under +an explicit ``--budget-usd`` ceiling through the metered clients. +""" + +from __future__ import annotations + +import argparse +import json +import sys +from pathlib import Path +from typing import Any + + +def add_commands(learn_sub: Any) -> None: + memory = learn_sub.add_parser("memory", help="the decision engine's local memory of labelled examples") + memory_sub = memory.add_subparsers(dest="memory_command", required=True) + build = memory_sub.add_parser("build", help="label + feature record + excerpt -> example rows (no model)") + build.add_argument("--labels", type=Path, required=True, help="labels-.jsonl written by `ousast learn labels --out`") + build.add_argument("--features", type=Path, action="append", default=[], help="feature records (JSONL with repo, pin); repeatable") + build.add_argument("--sources", type=Path, help="the source list (default: the packaged learn/sources.toml)") + build.add_argument("--root", type=Path, default=Path("."), help="the repository root the source paths are relative to") + build.add_argument("--cache", type=Path, help="clones and checkouts (default ~/.cache/openultrasast)") + build.add_argument("--memory", help="the store (default OUSAST_MEMORY, then results_root()/memory)") + build.add_argument("--profile", choices=("static", "plane"), default="static") + build.add_argument("--dry-run", action="store_true", help="count, write nothing") + + +def run(args: argparse.Namespace) -> int: + if args.learn_command == "memory" and args.memory_command == "build": + return _memory_build(args) + print(f"learn: unknown command {args.learn_command}", file=sys.stderr) + return 2 + + +def _memory_build(args: argparse.Namespace) -> int: + """Exit 2 on a refused label (``origin: user``, a source outside the list) or a source with >10% unread excerpts.""" + from ..plane.memory import open_store + from .examples import ExampleBuildError, SideReader, build_examples, read_jsonl, records_by_key + from .labels import DEFAULT_CACHE, DEFAULT_SOURCES, load_sources + + sources = load_sources(args.sources or DEFAULT_SOURCES) + labels = read_jsonl(args.labels) + store = None if args.dry_run else open_store(args.memory) + rows = [row for path in args.features for row in read_jsonl(path)] + if store is not None: + rows += [r.row for r in store.rows(kind="features")] + reader = SideReader(args.root, args.cache or DEFAULT_CACHE, sources, labels) + try: + build = build_examples( + labels, records_by_key(rows), reader, sources=sources, store=store, profile=args.profile, license_of=reader.license, + function_of=reader.function, + ) # fmt: skip + except ExampleBuildError as exc: + print(f"learn memory build: {exc}", file=sys.stderr) + return 2 + summary = build.summary() + print(json.dumps(summary, indent=2, sort_keys=True)) + if summary["failing_sources"]: + print( + f"learn memory build: more than 10% of the recorded candidates have no excerpt in {summary['failing_sources']}", file=sys.stderr + ) + return 2 + return 0 + + +__all__ = ["add_commands", "run"] diff --git a/src/openultrasast/learn/examples.py b/src/openultrasast/learn/examples.py new file mode 100644 index 0000000..f3294c6 --- /dev/null +++ b/src/openultrasast/learn/examples.py @@ -0,0 +1,315 @@ +"""The example store: the decision engine's local memory (learned-decision-engine design section 4.2, Req 3.2, 3.6). + +No model. ``ousast learn memory build`` joins each verified label (``ousast learn labels``) with the feature record of +the same candidate and an excerpt of its function at the label's pin, and writes one ``example`` row per candidate +into the memory store, with the excerpt as a content-addressed blob ``excerpts/.txt``:: + + envelope (id, kind=example, repo, pin, run, task, population=, split, image) + + {candidate, family, language, profile, label, source, group, frameworks, unit, direction, pin_role, weight, + license, x, instruments, roles, excerpt_sha, label_set_digest} + +Memory examples carry **no rationale** (code, signals, roles, family and label only). Refusals are loud: + +- a row with ``origin: user`` (the opt-in adaptation layer never enters the generalisation memory, Req 7.5) or a + source that ``sources.toml`` does not list stops the build (:class:`ExampleBuildError`); +- conditional rows (``benign_control``, ``assumed_benign``) are skipped until their condition is evaluated; +- a candidate whose excerpt is empty or whose function is not found is never stored; when more than 10% of a + source's recorded candidates have no excerpt the build fails that source (exit 2 in the CLI): an unread clone is + not a small memory. A fixed side whose file was read but no longer declares the labelled function (a pointer + pair whose fix lives in another function) is a label defect, counted apart as ``absent_fixed_side``. + +Pair labels have no commit pin (the corpus holds excerpts); their envelope pin is :func:`label_pin`, a stable +40-hex digest of the pair and its side, so the vulnerable and the fixed side stay distinct rows. +""" + +from __future__ import annotations + +import hashlib +import json +import re +from collections.abc import Callable, Iterable, Mapping, Sequence +from dataclasses import asdict, dataclass, field +from pathlib import Path +from typing import Any + +from ..plane.memory import MemoryStore, repo_key, row_id +from .excerpt import EXAMPLE, Bounds, delta_diff, excerpt, function_bounds, identities_of +from .labels import Label, Sources + +EXAMPLE_KIND = "example" +RUN = "learn-memory" +TASK = "build" +IMAGE = "host" +NO_EXCERPT_LIMIT = 0.10 +DERIVED_SOURCES = {"benign_control": "population"} # sources the label builder derives from a listed source +_PIN = re.compile(r"[0-9a-f]{40}") + + +class ExampleBuildError(RuntimeError): + """A label the memory must not hold (``origin: user``, a source outside ``sources.toml``) or a source whose + excerpts could not be read.""" + + +def label_pin(label: Mapping[str, Any]) -> str: + """The envelope pin of a label: its commit, or for a pair (no commit) a stable digest of pair and side.""" + pin = str(label.get("pin") or "") + if _PIN.fullmatch(pin): + return pin + subject = f"{label.get('source')}:{label.get('source_ref')}:{label.get('pin_role', 'vulnerable')}" + return hashlib.sha1(subject.encode("utf-8")).hexdigest() # noqa: S324 -- a name, not a security digest + + +def record_key(repo: str, pin: str, candidate: str, family: str, unit: str) -> tuple[str, str, str, str, str]: + return repo_key(repo), pin, candidate, family, unit + + +def _as_dict(label: Label | Mapping[str, Any]) -> dict[str, Any]: + return asdict(label) if isinstance(label, Label) else dict(label) + + +def label_set_digest(labels: Iterable[Label | Mapping[str, Any]]) -> str: + lines = sorted(json.dumps(_as_dict(row), sort_keys=True, default=list) for row in labels) + return hashlib.sha256("\n".join(lines).encode("utf-8")).hexdigest() + + +@dataclass +class SourceCount: + labels: int = 0 + no_record: int = 0 + recorded: int = 0 + no_excerpt: int = 0 # unread side, or the labelled function not found where it must be (counts to the 10%) + unread: int = 0 + absent_fixed_side: int = 0 # file read, the function is gone at the fix: a label defect, reported apart + examples: int = 0 + + +@dataclass +class MemoryBuild: + rows: list[dict[str, Any]] = field(default_factory=list) + sources: dict[str, SourceCount] = field(default_factory=dict) + skipped: dict[str, int] = field(default_factory=dict) + + def skip(self, reason: str) -> None: + self.skipped[reason] = self.skipped.get(reason, 0) + 1 + + def failing(self, limit: float = NO_EXCERPT_LIMIT) -> list[str]: + """Sources where more than ``limit`` of the recorded candidates have no excerpt.""" + return sorted(s for s, c in self.sources.items() if c.recorded and c.no_excerpt / c.recorded > limit) + + def summary(self) -> dict[str, Any]: + return { + "examples": len(self.rows), + "by_source": {s: asdict(c) for s, c in sorted(self.sources.items())}, + "skipped": dict(sorted(self.skipped.items())), + "failing_sources": self.failing(), + } + + +LinesOf = Callable[[Mapping[str, Any], str], Sequence[str] | None] # (label, pin_role) -> file lines at that side + + +def _counter_role(label: Mapping[str, Any]) -> str: + return "vulnerable" if label.get("pin_role") == "fixed" else "fixed" + + +def build_examples( + labels: Iterable[Label | Mapping[str, Any]], + records: Mapping[tuple[str, str, str, str, str], Mapping[str, Any]], + lines_of: LinesOf, + *, + sources: Sources, + store: MemoryStore | None = None, + profile: str = "static", + roles_of: Callable[[Mapping[str, Any]], Sequence[Mapping[str, Any]]] | None = None, + license_of: Callable[[Mapping[str, Any]], str] | None = None, + function_of: Callable[[Mapping[str, Any], str], str | None] | None = None, + bounds: Bounds = EXAMPLE, +) -> MemoryBuild: + """Example rows for the verified labels that have a feature record (``records`` by :func:`record_key`) and an + excerpt; with ``store`` the excerpt blobs and the rows are written. Delta units append the base->head diff.""" + rows = [_as_dict(label) for label in labels] + allowed = {s.id for s in sources.sources} | {d for d, kind in DERIVED_SOURCES.items() if any(s.kind == kind for s in sources.sources)} + for row in rows: + if row.get("origin") == "user": + raise ExampleBuildError(f"label {row.get('candidate')!r} has origin: user -- the adaptation layer never enters the memory") + if row.get("source") not in allowed: + raise ExampleBuildError(f"label source {row.get('source')!r} is not in {sources.path}: it cannot enter the memory") + digest = label_set_digest(rows) + build = MemoryBuild() + for label in rows: + if label.get("conditional") is not None or label.get("source") == "assumed_benign": + build.skip("conditional label (condition not evaluated)") + continue + count = build.sources.setdefault(str(label["source"]), SourceCount()) + count.labels += 1 + pin = label_pin(label) + candidate, family, unit = str(label["candidate"]), str(label["family"]), str(label["unit"]) + record = records.get(record_key(str(label["repo"]), pin, candidate, family, unit)) + if record is None or record.get("profile") != profile: + count.no_record += 1 + continue + count.recorded += 1 + path, _, function = candidate.partition("::") + function = (function_of(label, str(label.get("pin_role") or "vulnerable")) if function_of else None) or function + language = str(record.get("language") or label.get("language") or "other") + identities = identities_of(str(label.get("repo") or ""), path, [str(label.get("pin") or "")]) + lines = lines_of(label, str(label.get("pin_role") or "vulnerable")) + diff = None + if unit == "delta" and lines: + base = lines_of(label, _counter_role(label)) + head_span, base_span = function_bounds(lines, language, function), function_bounds(base or (), language, function) + if base and head_span and base_span: + diff = delta_diff(base[slice(*base_span)], lines[slice(*head_span)], language, identities=identities) + shown = excerpt(lines, language, function, bounds=bounds, identities=identities, diff=diff) + if shown is None: + if not lines: + count.unread += 1 + elif label.get("pin_role") == "fixed": + count.absent_fixed_side += 1 + continue + count.no_excerpt += 1 + continue + if store is not None: + store.put_blob("excerpts", shown.text.encode("utf-8")) + subject = f"{candidate}\x00{family}\x00{unit}\x00{label.get('pin_role')}\x00{label.get('source_ref')}" + build.rows.append( + { + "id": row_id(EXAMPLE_KIND, RUN, TASK, f"{label['repo']}\x00{pin}\x00{subject}"), "kind": EXAMPLE_KIND, + "repo": repo_key(str(label["repo"])), "pin": pin, "run": RUN, "task": TASK, "population": str(label["source"]), + "split": str(label.get("split") or label["source"]), "image": IMAGE, + "candidate": candidate, "family": family, "language": language, "profile": profile, "label": int(label["label"]), + "source": str(label["source"]), "group": str(label.get("group") or ""), + "frameworks": list(label["frameworks"]) if label.get("frameworks") is not None else None, "unit": unit, + "direction": label.get("direction"), "pin_role": label.get("pin_role", "vulnerable"), + "weight": float(label.get("weight", 1.0)), "license": (license_of(label) if license_of else "") or "", + "x": dict(record["x"]), "instruments": dict(record["instruments"]), + "roles": list(roles_of(label)) if roles_of else [], "excerpt_sha": shown.sha, "label_set_digest": digest, + } + ) # fmt: skip + count.examples += 1 + if store is not None and build.rows: + store.put_rows(build.rows) + return build + + +@dataclass(frozen=True) +class Example: + """An example row as retrieval and the program use it (no envelope identity beyond ``id`` and ``group``).""" + + id: str + group: str + source: str + frameworks: tuple[str, ...] + family: str + label: int + language: str + profile: str + unit: str + x: Mapping[str, Any] + instruments: Mapping[str, Mapping[str, Any]] + roles: tuple[Mapping[str, Any], ...] + excerpt_sha: str + license: str = "" + + @classmethod + def from_row(cls, row: Mapping[str, Any]) -> Example: + return cls( + id=str(row["id"]), group=str(row.get("group") or ""), source=str(row["source"]), frameworks=tuple(row.get("frameworks") or ()), + family=str(row["family"]), label=int(row["label"]), language=str(row.get("language") or "other"), + profile=str(row.get("profile") or "static"), unit=str(row.get("unit") or "pin"), x=dict(row["x"]), + instruments={k: dict(v) for k, v in row["instruments"].items()}, roles=tuple(row.get("roles") or ()), + excerpt_sha=str(row["excerpt_sha"]), license=str(row.get("license") or ""), + ) # fmt: skip + + +def load_examples(store: MemoryStore, profile: str | None = None) -> list[Example]: + """Every example row in the store (optionally one profile), in key then id order.""" + where = {"profile": profile} if profile else None + return [Example.from_row(r.row) for r in store.rows(kind=EXAMPLE_KIND, where=where)] + + +# --- reading the label sides -------------------------------------------------------------------------------------------- + + +class SideReader: + """File lines of a label's side, read-only: pairs from the corpus excerpts, populations and adjudications from + their clones (``git show pin:path``), development recipes from their checkouts. ``None`` when unread.""" + + def __init__(self, root: Path, cache: Path, sources: Sources, labels: Sequence[Mapping[str, Any]]) -> None: + from .labels import Clone + + self.root, self.cache, self.sources = Path(root), Path(cache), sources + self._pairs: dict[str, Any] | None = None + self._clones: dict[str, Clone | None] = {} + self._case_of: dict[str, str] = {} + for label in labels: + source = next((s for s in sources.sources if s.id == label.get("source")), None) + if source is not None and source.kind == "population": + self._case_of.setdefault(str(label.get("repo")), f"{source.options.get('cache', 'independent')}/{label.get('source_ref')}") + + def _pair_cases(self) -> dict[str, Any]: + if self._pairs is None: + from ..pairs import load_pair_catalog + + source = next(s for s in self.sources.sources if s.kind == "pairs") + self._pairs = {case.name: case for case in load_pair_catalog(self.root / source.files[0])} + return self._pairs + + def function(self, label: Mapping[str, Any], pin_role: str) -> str | None: + """The function a pair's fixed side holds when it is not the labelled one (``fix_function``).""" + if label.get("source") != "pairs" or pin_role != "fixed": + return None + case = self._pair_cases().get(str(label.get("source_ref"))) + return str(getattr(case, "fix_function", "") or "") or None + + def license(self, label: Mapping[str, Any]) -> str: + if label.get("source") != "pairs": + return "" + case = self._pair_cases().get(str(label.get("source_ref"))) + return str(getattr(case, "license", "") or "") + + def __call__(self, label: Mapping[str, Any], pin_role: str) -> list[str] | None: + from .labels import Clone, LabelSourceError + + source = str(label.get("source")) + path = str(label.get("candidate", "")).partition("::")[0] + if source == "pairs": + case = self._pair_cases().get(str(label.get("source_ref"))) + if case is None: + return None + file = case.vuln_file if pin_role == "vulnerable" else case.fixed_file + return Path(file).read_text(encoding="utf-8", errors="replace").splitlines() if Path(file).is_file() else None + if source == "dev-php": + name = str(label.get("source_ref", "")).partition(":")[0] + file = self.cache / "repos" / name / str(label.get("pin"))[:12] / path + return file.read_text(encoding="utf-8", errors="replace").splitlines() if file.is_file() else None + case_dir = self._case_of.get(str(label.get("repo"))) + if case_dir is None: + return None + if case_dir not in self._clones: + try: + self._clones[case_dir] = Clone(self.cache / case_dir) + except LabelSourceError: + self._clones[case_dir] = None + clone = self._clones[case_dir] + return clone.lines(str(label.get("pin")), path) if clone is not None else None + + +def read_jsonl(path: Path) -> list[dict[str, Any]]: + return [json.loads(line) for line in Path(path).read_text(encoding="utf-8").splitlines() if line.strip()] + + +def records_by_key(rows: Iterable[Mapping[str, Any]]) -> dict[tuple[str, str, str, str, str], dict[str, Any]]: + """Feature records (``features`` rows of the store, or records with ``repo`` and ``pin`` beside them) by + :func:`record_key`.""" + out: dict[tuple[str, str, str, str, str], dict[str, Any]] = {} + for row in rows: + key = record_key(str(row["repo"]), str(row["pin"]), str(row["candidate"]), str(row["family"]), str(row.get("unit", "pin"))) + out[key] = dict(row) + return out + + +__all__ = [ + "EXAMPLE_KIND", "NO_EXCERPT_LIMIT", "Example", "ExampleBuildError", "MemoryBuild", "SideReader", "build_examples", + "label_pin", "label_set_digest", "load_examples", "read_jsonl", "record_key", "records_by_key", +] # fmt: skip diff --git a/src/openultrasast/learn/excerpt.py b/src/openultrasast/learn/excerpt.py new file mode 100644 index 0000000..ef9eb94 --- /dev/null +++ b/src/openultrasast/learn/excerpt.py @@ -0,0 +1,318 @@ +"""Code excerpts for the decision engine's prompts and memory (learned-decision-engine design section 4.2). + +No model. An excerpt is the body of the labelled function at its pin, as numbered lines, bounded: + +- candidates: at most 80 lines and 4,000 characters, centred on the changed lines of a delta or the hit line; +- memory examples: at most 40 lines and 2,000 characters. + +What an excerpt never holds: the path, the repository or the commit (none is rendered; a caller may name identity +strings -- ``owner/name``, the owner, the path, commit ids -- that are replaced wherever the code repeats them). +What it hides: advisory ids (CVE, GHSA) anywhere, and **comments with security wording** (:func:`.labels. +security_reason`, multilingual), which a fixed side often carries ("prevent XSS", "CVE-2021-1234") and which would +otherwise tell the model the label. Both become ````. A delta unit adds the base->head diff of the function +(at most 40 lines), the change under test -- never the later fix. + +The text is normalised (trailing blanks dropped, ``\\n`` endings, one final newline) before it is hashed, so the sha +names the content an embedding and a response cache key are of. +""" + +from __future__ import annotations + +import difflib +import hashlib +import re +from collections.abc import Iterable, Sequence +from dataclasses import dataclass + +from ..plane.tasks.repo_facts import DECLARATION, GLOBAL, _declared_name +from .features import function_span +from .labels import _ADVISORY, security_reason + +REDACTED = "" +LANGUAGE_ALIASES = {"c_cpp": "cpp", "c++": "cpp", "js": "javascript", "ts": "typescript", "py": "python"} +HASH_COMMENTS = frozenset({"python", "php", "ruby", "shell", "perl", "yaml", "r"}) +SLASH_COMMENTS = frozenset( + {"c", "cpp", "java", "javascript", "typescript", "php", "go", "groovy", "csharp", "kotlin", "rust", "swift", "scala"} +) # fmt: skip +MAX_LINE_CHARS = 240 +_HEX = re.compile(r"\b[0-9a-f]{7,40}\b", re.IGNORECASE) + + +@dataclass(frozen=True) +class Bounds: + lines: int + chars: int + + +CANDIDATE = Bounds(80, 4000) +EXAMPLE = Bounds(40, 2000) +DIFF_LINES = 40 + + +@dataclass(frozen=True) +class Excerpt: + """A rendered excerpt: ``text`` (numbered lines, then the diff for a delta), its ``sha``, the first and last + source line shown (1-based), and whether the bounds cut the function.""" + + text: str + sha: str + first: int + last: int + truncated: bool + + @property + def line_numbers(self) -> range: + return range(self.first, self.last + 1) + + +def normalise(text: str) -> str: + lines = [line.rstrip() for line in text.replace("\r\n", "\n").replace("\r", "\n").split("\n")] + while lines and not lines[-1]: + lines.pop() + return "\n".join(lines) + "\n" if lines else "" + + +def excerpt_sha(text: str) -> str: + return hashlib.sha256(normalise(text).encode("utf-8")).hexdigest() + + +# --- comments --------------------------------------------------------------------------------------------------------- + + +def _comment_ranges(line: str, state: str | None, language: str) -> tuple[list[tuple[int, int]], str | None]: + """The comment spans of one line and the block state carried to the next (``block`` inside ``/* */``, a triple + quote inside a Python docstring). Strings are skipped so a ``#`` in a literal is not a comment.""" + hashes, slashes, python = language in HASH_COMMENTS, language in SLASH_COMMENTS, language == "python" + ranges: list[tuple[int, int]] = [] + i, n, quote = 0, len(line), None + if state is not None: + close = "*/" if state == "block" else state + j = line.find(close) + if j < 0: + return [(0, n)], state + ranges.append((0, j + len(close))) + i, state = j + len(close), None + while i < n: + c = line[i] + if quote: + if c == "\\": + i += 2 + continue + if c == quote: + quote = None + i += 1 + continue + if python and line.startswith(('"""', "'''"), i) and not line[:i].strip(): + delim = line[i : i + 3] + j = line.find(delim, i + 3) + if j < 0: + ranges.append((i, n)) + return ranges, delim + ranges.append((i, j + 3)) + i = j + 3 + continue + if c in "\"'`": + quote = c + elif (slashes and line.startswith("//", i)) or (hashes and c == "#"): + ranges.append((i, n)) + break + elif slashes and line.startswith("/*", i): + j = line.find("*/", i + 2) + if j < 0: + ranges.append((i, n)) + return ranges, "block" + ranges.append((i, j + 2)) + i = j + 2 + continue + i += 1 + return ranges, state + + +_OPEN = ("/**", "/*", "//", "#", '"""', "'''", "*") +_CLOSE = ("*/", '"""', "'''") + + +def _redact_comment(text: str) -> str: + if security_reason(text) is None: + return _ADVISORY.sub(REDACTED, text) + stripped = text.strip() + opener = next((o for o in _OPEN if stripped.startswith(o)), "") + closer = next((c for c in _CLOSE if stripped.endswith(c) and len(stripped) > len(opener)), "") + lead = text[: len(text) - len(text.lstrip())] + return lead + " ".join(p for p in (opener, REDACTED, closer) if p) + + +def redact(lines: Sequence[str], language: str, identities: Iterable[str] = ()) -> list[str]: + """``lines`` with security wording in comments and advisory ids anywhere replaced by ````, and every + identity string (case-insensitive, at least 4 characters) replaced too.""" + language = LANGUAGE_ALIASES.get(language, language) + names = sorted({i for i in identities if len(i) >= 4}, key=len, reverse=True) + identity = re.compile("|".join(re.escape(n) for n in names), re.IGNORECASE) if names else None + out: list[str] = [] + state: str | None = None + for line in lines: + ranges, state = _comment_ranges(line, state, language) + text = line + for start, end in reversed(ranges): + text = text[:start] + _redact_comment(text[start:end]) + text[end:] + text = _ADVISORY.sub(REDACTED, text) + if identity is not None: + text = identity.sub(REDACTED, text) + out.append(text) + return out + + +def identities_of(repo: str = "", path: str = "", commits: Iterable[str] = ()) -> tuple[str, ...]: + """The identity strings an excerpt must not repeat: ``owner/name``, the owner, the path, commit ids. The bare + repository name is kept: it is usually the package name the code imports, which is code, not an identity.""" + from .labels import repo_name + + found: list[str] = [] + slug = repo_name(repo) if repo else "" + if "/" in slug: + found += [slug, slug.split("/", 1)[0]] + if path and "/" in path: + found.append(path) + found += [c for c in commits if c and _HEX.fullmatch(c)] + return tuple(found) + + +# --- spans and excerpts ----------------------------------------------------------------------------------------------- + + +def _brace_bounds(lines: Sequence[str], language: str, function: str) -> tuple[int, int] | None: + """A brace language's definition of ``function`` (``Class::name`` and ``a.b`` reduced to the last name): the first + line naming it before ``(`` whose ``{`` comes before any ``;``, to the matching ``}``. Comments are blanked + before braces are counted.""" + name = re.split(r"::|\.|->", function)[-1] + if not re.fullmatch(r"[A-Za-z_]\w*", name): + return None + call = re.compile(r"(?$])" + re.escape(name) + r"\s*\(") + code: list[str] = [] + state: str | None = None + for line in lines: + ranges, state = _comment_ranges(line, state, language if language in SLASH_COMMENTS else "c") + for start, end in reversed(ranges): + line = line[:start] + " " * (end - start) + line[end:] + code.append(line) + for i, line in enumerate(code): + found = call.search(line) + if found is None or re.match(r"\s*(return|if|while|for|switch|else)\b", line): + continue + tail = "\n".join([line[found.end() :], *code[i + 1 : i + 8]]) + brace, semi = tail.find("{"), tail.find(";") + if brace < 0 or (0 <= semi < brace): + continue + depth, opened = 0, False + for j in range(i, len(code)): + for c in line[found.end() :] if j == i else code[j]: + if c == "{": + depth, opened = depth + 1, True + elif c == "}" and opened: + depth -= 1 + if depth == 0: + return i, j + 1 + return i, len(code) + return None + + +def function_bounds(lines: Sequence[str], language: str, function: str) -> tuple[int, int] | None: + """0-based ``[start, end)`` of the first declaration of ``function`` (the whole file for ````), or + ``None`` when the file does not declare it. Languages without a repo-facts declaration pattern (C, C++, ...) and + brace languages whose pattern misses the name use the brace matcher.""" + language = LANGUAGE_ALIASES.get(language, language) + if function == GLOBAL: + return (0, len(lines)) if lines else None + pattern = DECLARATION.get(language) + start = next((i for i, line in enumerate(lines) if _declared_name(pattern, line) == function), None) if pattern else None + if start is None: + return _brace_bounds(lines, language, function) if language in SLASH_COMMENTS or pattern is None else None + span = function_span(lines, language, function) + if not span: + return None + return start, min(start + span, len(lines)) + + +def _window(start: int, end: int, focus: Sequence[int], size: int) -> tuple[int, int]: + if end - start <= size: + return start, end + inside = sorted(f - 1 for f in focus if start <= f - 1 < end) + if not inside: + return start, start + size + centre = inside[len(inside) // 2] + lo = max(start, min(centre - size // 2, end - size)) + return lo, lo + size + + +def _numbered(lines: Sequence[str], first: int) -> list[str]: + out = [] + for offset, line in enumerate(lines): + text = line if len(line) <= MAX_LINE_CHARS else line[: MAX_LINE_CHARS - 3] + "..." + out.append(f"{first + offset:>5} {text}") + return out + + +def excerpt( + lines: Sequence[str] | None, + language: str, + function: str, + *, + bounds: Bounds = CANDIDATE, + focus: Sequence[int] = (), + identities: Iterable[str] = (), + diff: str | None = None, +) -> Excerpt | None: + """The bounded, redacted, numbered excerpt of ``function``, or ``None`` when the file is unread or the function is + not declared there (never an empty excerpt). ``focus`` are 1-based lines to centre on; ``diff`` (a delta's + :func:`delta_diff`) is appended under a ``diff:`` line.""" + if not lines: + return None + span = function_bounds(lines, language, function) + if span is None: + return None + start, end = span + lo, hi = _window(start, end, focus, bounds.lines) + shown = redact(lines[lo:hi], language, identities) + rendered = _numbered(shown, lo + 1) + centre = (len(rendered) - 1) // 2 if focus else 0 + truncated = (lo, hi) != (start, end) + while len(rendered) > 1 and len("\n".join(rendered)) > bounds.chars: + # drop the line farthest from the centre (the end when there is no focus), keeping the window contiguous + if focus and centre > len(rendered) - 1 - centre: + rendered.pop(0) + lo += 1 + centre -= 1 + else: + rendered.pop() + hi = lo + len(rendered) + truncated = True + body = "\n".join(rendered) + if diff: + body += "\ndiff:\n" + diff + text = normalise(body) + if not text.strip(): + return None + return Excerpt(text, excerpt_sha(text), lo + 1, hi, truncated) + + +def delta_diff( + base: Sequence[str], head: Sequence[str], language: str, *, max_lines: int = DIFF_LINES, identities: Iterable[str] = () +) -> str: + """The base->head unified diff of a function (one line of context, no file headers), redacted, at most + ``max_lines`` lines.""" + diff = [line for line in difflib.unified_diff(list(base), list(head), lineterm="", n=1)][2:] + out: list[str] = [] + for line in diff: + if line.startswith("@@"): + out.append(line) + else: + out.append(line[:1] + redact([line[1:]], language, identities)[0]) + if len(out) > max_lines: + out = out[: max_lines - 1] + [f"... ({len(out) - max_lines + 1} more diff lines)"] + return "\n".join(out) + + +__all__ = [ + "CANDIDATE", "DIFF_LINES", "EXAMPLE", "REDACTED", "Bounds", "Excerpt", "delta_diff", "excerpt", "excerpt_sha", + "function_bounds", "identities_of", "normalise", "redact", +] # fmt: skip diff --git a/src/openultrasast/plane/memory.py b/src/openultrasast/plane/memory.py index 841e2b6..dc76941 100644 --- a/src/openultrasast/plane/memory.py +++ b/src/openultrasast/plane/memory.py @@ -8,6 +8,8 @@ engine's own. Layout, identical for both backends:: index.jsonl one row per ingested (run, task): sha256 of its rows, row count facts/.json facts.json by content (identical facts stored once) repos/____/.jsonl the rows of one repository + 40-hex pin, sorted by id + /. blobs (``put_blob``): excerpts/*.txt by content, embeddings//*.json + by excerpt sha, responses/*.json by request sha, programs/*.json ``OUSAST_MEMORY`` selects the backend: ``file:///path`` (:class:`FileStore`, default ``results_root()/memory``) or ``minio://[/]`` (:class:`MinioStore`, the ``minio`` extra). MinIO's endpoint and credentials come @@ -51,11 +53,16 @@ POPULATION_ANNOTATION = "openultrasast.io/population" SPLIT_ANNOTATION = "openultrasast.io/split" SNAPSHOT_ANNOTATION = "openultrasast.io/memory-snapshot" # the loop's guard: {"catalog", "manifest", "populations"} # The decision engine's kinds (learned-decision-engine, Data Models): `features` from the `features` task through -# `remember`; `label`, `decision`, `experiment`, `arm_outcome` and `experiment_result` for the later tasks. +# `remember`; `label`, `decision`, `experiment`, `arm_outcome` and `experiment_result` for the later tasks; `example` +# (a labelled case of the local memory: label + feature record + excerpt sha) from `ousast learn memory build`. KINDS = ( "facts", "verdict", "unit_cost", "alert", "coverage", "proposal_outcome", - "features", "label", "decision", "experiment", "arm_outcome", "experiment_result", + "features", "label", "decision", "experiment", "arm_outcome", "experiment_result", "example", ) # fmt: skip +# Blobs (`put_blob`): a top-level prefix, optionally one sub-directory (`embeddings/`), and a 64-hex name. +_BLOB_PREFIX = re.compile(r"[a-z]+(?:/[A-Za-z0-9._-]+)?") +_BLOB_NAME = re.compile(r"[0-9a-f]{64}") +BLOB_SUFFIX = {"excerpts": ".txt", "labels": ".jsonl"} # every other prefix holds JSON ROW_FIELDS = ("id", "kind", "repo", "pin", "run", "task", "population", "split", "image") TAG_FIELDS = ("repo", "pin", "kind", "family", "run", "population", "split") MIN_FREE_BYTES = 1 << 30 @@ -112,6 +119,15 @@ def parse_memory_key(text: str) -> dict[str, str]: return {k: str(v) for k, v in key.items()} +def blob_key(prefix: str, name: str) -> str: + """The key of a blob: ``/``; a prefix or name outside the layout is refused.""" + if not _BLOB_PREFIX.fullmatch(prefix) or ".." in prefix or prefix.split("/", 1)[0] in ("repos", "facts", "index.jsonl"): + raise MemoryStoreError(f"blob prefix {prefix!r} is not a blob prefix of the layout") + if not _BLOB_NAME.fullmatch(name): + raise MemoryStoreError(f"blob name {name!r} is not a 64-hex sha256") + return f"{prefix}/{name}{BLOB_SUFFIX.get(prefix.split('/', 1)[0], '.json')}" + + def validate_row(row: object, where: str = "row") -> dict[str, Any]: if not isinstance(row, dict): raise MemoryStoreError(f"{where}: a row must be a JSON object") @@ -235,6 +251,20 @@ class MemoryStore(ABC): return None return found[0] + def get_blob(self, prefix: str, name: str, *, verify: bool = True) -> bytes | None: + """The blob ``/``. With ``verify`` (a content-addressed blob) one whose content no longer hashes + to its name is dropped with a warning and reported absent; keyed blobs (an embedding by its excerpt's sha, a + response by its request's sha) pass ``verify=False``.""" + key = blob_key(prefix, name) + found = self._get(key) + if found is None: + return None + if verify and hashlib.sha256(found[0]).hexdigest() != name: + log.warning("memory %s: %s no longer matches its sha256; dropped", self.describe(), key) + self._delete(key) + return None + return found[0] + def rows( self, repo: str | None = None, pin: str | None = None, kind: str | None = None, where: Mapping[str, Any] | None = None ) -> list[Record]: @@ -260,6 +290,23 @@ class MemoryStore(ABC): self._put(key, data) return sha + def put_blob(self, prefix: str, data: bytes, *, name: str | None = None) -> str: + """Store ``data`` under ``/`` (content-addressed; identical data stored once) or under + ``name`` when the blob is keyed by something else (the excerpt an embedding is of). A blob already present is + not rewritten: a key names one content. Returns the name.""" + name = name or hashlib.sha256(data).hexdigest() + key = blob_key(prefix, name) + if self._get(key) is None: + self._check_space() + self._put(key, data) + return name + + def blob_names(self, prefix: str) -> list[str]: + """The names of the blobs under ``prefix``, sorted.""" + blob_key(prefix, "0" * 64) + suffix = BLOB_SUFFIX.get(prefix.split("/", 1)[0], ".json") + return sorted(k.rsplit("/", 1)[1].removesuffix(suffix) for k in self._keys(prefix + "/") if k.count("/") == prefix.count("/") + 1) + def put_row(self, row: Mapping[str, Any]) -> None: self.put_rows([row]) diff --git a/tests/test_learn_examples.py b/tests/test_learn_examples.py new file mode 100644 index 0000000..22407a7 --- /dev/null +++ b/tests/test_learn_examples.py @@ -0,0 +1,157 @@ +"""The example store (learned-decision-engine design 4.2, task 6.1): label + feature record + excerpt -> ``example`` +rows, the refusals, the 10% unread-excerpt failure, and content-addressed blobs on both backends.""" + +from __future__ import annotations + +import hashlib +import json +from collections.abc import Iterator, Mapping, Sequence +from pathlib import Path +from typing import Any + +import pytest +from test_plane_memory import FakeObjects + +from openultrasast.cli import main +from openultrasast.learn.examples import ( + ExampleBuildError, + build_examples, + label_pin, + load_examples, + record_key, + records_by_key, +) +from openultrasast.learn.features import Part, quick_part, record +from openultrasast.learn.labels import Source, Sources +from openultrasast.plane.memory import FileStore, MemoryStore, MemoryStoreError, MinioStore + +PIN = "0123456789abcdef0123456789abcdef01234567" +SOURCES = Sources( + (Source("pairs", "pairs", (), (), "2026-09-30", "development"), Source("population-v1", "population", (), (), "2026-09-25", "spent")), + frozenset(), + {}, + Path("sources.toml"), +) +CODE = { + "vulnerable": ["def handler(request):", " # CVE-2021-1234 fixed later", " return run(request.args['q'])", ""], + "fixed": ["def handler(request):", " return run(quote(request.args['q']))", ""], +} + + +@pytest.fixture(params=["file", "fake-minio"]) +def store(request: pytest.FixtureRequest, tmp_path: Path) -> Iterator[MemoryStore]: + yield FileStore(tmp_path / "memory") if request.param == "file" else MinioStore(FakeObjects("works"), "bucket", "p") + + +def label(**kw: Any) -> dict[str, Any]: + base = { + "candidate": "app.py::handler", "family": "injection", "label": 1, "source": "population-v1", "source_ref": "case-1", + "unit": "pin", "weight": 1.0, "split": "population-v1", "group": "acme/webapp", "repo": "https://github.com/acme/webapp", + "language": "python", "created": "2026-09-30", "pin": PIN, "pin_role": "vulnerable", "frameworks": ["flask"], + "conditional": None, "direction": None, + } # fmt: skip + return {**base, **kw} + + +def feature(lab: Mapping[str, Any], unit: str = "pin") -> dict[str, Any]: + parts = {"quick": quick_part([("r1", "enabled")], {})} + if unit == "delta": + parts["delta"] = Part("ran", {"delta.novelty": "new", "delta.changed_lines": 1, "delta.sanitizer_removed": True}) + out = record(str(lab["candidate"]), str(lab["family"]), "python", "static", parts, unit=unit) + return {**out, "repo": lab["repo"], "pin": label_pin(lab)} + + +def lines_of(lab: Mapping[str, Any], role: str) -> Sequence[str] | None: + return CODE[role] + + +def test_example_rows_join_label_record_and_excerpt(store: MemoryStore) -> None: + pos, neg = label(), label(label=0, pin_role="fixed", pin="f" * 40) + records = records_by_key([feature(pos), feature(neg)]) + build = build_examples([pos, neg], records, lines_of, sources=SOURCES, store=store) + assert build.summary()["examples"] == 2 and not build.failing() + stored = load_examples(store) + assert sorted(e.label for e in stored) == [0, 1] and {e.group for e in stored} == {"acme/webapp"} + row = next(r.row for r in store.rows(kind="example") if r.row["label"] == 1) + assert row["kind"] == "example" and row["frameworks"] == ["flask"] and "rationale" not in row + text = store.get_blob("excerpts", row["excerpt_sha"]) + assert text is not None and hashlib.sha256(text).hexdigest() == row["excerpt_sha"] + assert b"CVE-2021" not in text + assert "acme" not in text.decode() and "app.py" not in text.decode() + + +def test_delta_units_carry_the_diff(store: MemoryStore) -> None: + lab = label(unit="delta", direction="introduce") + build = build_examples([lab], records_by_key([feature(lab, "delta")]), lines_of, sources=SOURCES, store=store) + row = build.rows[0] + text = store.get_blob("excerpts", row["excerpt_sha"]) + assert text is not None and b"\ndiff:\n@@" in text and b"+ return run(request.args['q'])" in text + + +def test_user_origin_and_unlisted_sources_are_refused() -> None: + with pytest.raises(ExampleBuildError, match="origin: user"): + build_examples([label(origin="user")], {}, lines_of, sources=SOURCES) + with pytest.raises(ExampleBuildError, match="not in"): + build_examples([label(source="population-v3")], {}, lines_of, sources=SOURCES) + + +def test_conditional_and_unrecorded_labels_never_become_examples() -> None: + rows = [label(conditional="signal_at_tip_not_base", label=0), label(source="pairs", pin="", source_ref="p1")] + build = build_examples(rows, {}, lines_of, sources=SOURCES) + assert build.rows == [] and build.skipped == {"conditional label (condition not evaluated)": 1} + assert build.sources["pairs"].no_record == 1 + + +def test_more_than_ten_percent_without_excerpt_fails_the_source() -> None: + labs = [label(candidate=f"app.py::f{i}", source_ref=f"c{i}") for i in range(10)] + [label()] + records = records_by_key([feature(lab) for lab in labs]) + build = build_examples(labs, records, lines_of, sources=SOURCES) # f0..f9 are not declared in CODE: no excerpt + assert build.sources["population-v1"].no_excerpt == 10 and build.failing() == ["population-v1"] + ok = build_examples([label()], records, lines_of, sources=SOURCES) + assert ok.failing() == [] + + +def test_pair_pins_are_stable_and_distinct_per_side() -> None: + vuln, fixed = label(source="pairs", pin="", source_ref="p1"), label(source="pairs", pin="", source_ref="p1", pin_role="fixed") + assert label_pin(vuln) != label_pin(fixed) and len(label_pin(vuln)) == 40 and label_pin(vuln) == label_pin(dict(vuln)) + assert label_pin(label()) == PIN + key = record_key(vuln["repo"], label_pin(vuln), "app.py::handler", "injection", "pin") + assert key[0] == "github.com/acme/webapp" + + +def test_blobs_are_content_addressed_on_both_backends(store: MemoryStore) -> None: + sha = store.put_blob("excerpts", b"x = 1\n") + assert sha == hashlib.sha256(b"x = 1\n").hexdigest() and store.get_blob("excerpts", sha) == b"x = 1\n" + assert store.put_blob("excerpts", b"x = 1\n") == sha and store.blob_names("excerpts") == [sha] + keyed = store.put_blob("embeddings/openai__text-embedding-3-small", b'{"v": 1}', name=sha) + assert keyed == sha and store.get_blob("embeddings/openai__text-embedding-3-small", sha, verify=False) == b'{"v": 1}' + assert store.get_blob("excerpts", "0" * 64) is None + for bad in ("repos", "../x", "facts"): + with pytest.raises(MemoryStoreError): + store.put_blob(bad, b"x") + with pytest.raises(MemoryStoreError): + store.get_blob("excerpts", "not-a-sha") + + +def test_a_changed_blob_is_dropped_not_served(tmp_path: Path) -> None: + store = FileStore(tmp_path / "m") + sha = store.put_blob("excerpts", b"a\n") + (tmp_path / "m" / "excerpts" / f"{sha}.txt").write_bytes(b"tampered\n") + assert store.get_blob("excerpts", sha) is None and store.blob_names("excerpts") == [] + + +def test_cli_build_writes_examples_and_exits_2_on_refusal(tmp_path: Path, capsys: pytest.CaptureFixture[str]) -> None: + labels = tmp_path / "labels.jsonl" + labels.write_text(json.dumps(label(origin="user")) + "\n") + assert main(["learn", "memory", "build", "--labels", str(labels), "--dry-run"]) == 2 + assert "origin: user" in capsys.readouterr().err + + +def test_a_fixed_side_without_the_function_is_a_label_defect_not_an_unread_side() -> None: + fixed = label(label=0, pin_role="fixed", pin="f" * 40) + moved = {"vulnerable": CODE["vulnerable"], "fixed": ["def guard(request):", " return True"]} + build = build_examples([fixed], records_by_key([feature(fixed)]), lambda lab, role: moved[role], sources=SOURCES) + count = build.sources["population-v1"] + assert count.absent_fixed_side == 1 and count.no_excerpt == 0 and build.failing() == [] + unread = build_examples([fixed], records_by_key([feature(fixed)]), lambda lab, role: None, sources=SOURCES) + assert unread.sources["population-v1"].unread == 1 and unread.failing() == ["population-v1"] diff --git a/tests/test_learn_excerpt.py b/tests/test_learn_excerpt.py new file mode 100644 index 0000000..df9dbdf --- /dev/null +++ b/tests/test_learn_excerpt.py @@ -0,0 +1,138 @@ +"""Excerpts for the decision engine (learned-decision-engine design 4.2): bounds, centring, no identities, advisory ids +and security wording in comments redacted, the delta diff.""" + +from __future__ import annotations + +from openultrasast.learn.excerpt import ( + CANDIDATE, + EXAMPLE, + REDACTED, + Bounds, + delta_diff, + excerpt, + excerpt_sha, + identities_of, + normalise, + redact, +) + +PY = [ + "import os", + "", + "def handler(request):", + " # CVE-2021-1234: prevent SQL injection by quoting", + " name = request.args['name']", + ' """Fetch the user. Fixes GHSA-abcd-efgh-ijkm."""', + " query = 'SELECT * FROM users WHERE name = ' + name # build the query", + " return db.execute(query)", + "", + "def other():", + " return 1", +] + + +def _long_function(n: int, width: int = 10) -> list[str]: + return ["def big(x):"] + [f" v{i} = x + {i}" + " " * width for i in range(n)] + [" return v0", "", "def after():", " pass"] + + +def test_function_only_numbered_from_its_first_line() -> None: + shown = excerpt(PY, "python", "handler") + assert shown is not None + lines = shown.text.splitlines() + assert lines[0].split()[0] == "3" and "def handler" in lines[0] + assert "def other" not in shown.text and shown.first == 3 and shown.last == 9 + assert shown.sha == excerpt_sha(shown.text) and not shown.truncated + + +def test_unread_file_or_unknown_function_is_none_never_empty() -> None: + assert excerpt(None, "python", "handler") is None + assert excerpt([], "python", "handler") is None + assert excerpt(PY, "python", "missing") is None + assert excerpt(PY, "cobol", "handler") is None + + +def test_security_comments_and_advisory_ids_are_redacted() -> None: + shown = excerpt(PY, "python", "handler") + assert shown is not None + assert "CVE-2021" not in shown.text and "GHSA" not in shown.text and "injection" not in shown.text + assert "prevent" not in shown.text and REDACTED in shown.text + assert "# build the query" in shown.text # an ordinary comment stays + assert "query = 'SELECT * FROM users WHERE name = ' + name" in shown.text # code is never redacted + + +def test_redaction_covers_block_comments_and_other_languages() -> None: + php = ["function f($id) {", " /* sanitize to avoid XSS", " (see report) */", " echo $id; // 修复越权", "}"] + out = redact(php, "php") + assert "XSS" not in "\n".join(out) and "修复越权" not in out[3] and out[3].startswith(" echo $id;") + assert "#" not in redact(["x = '#not a comment'"], "python")[0].replace("'#not a comment'", "") + assert redact(["s = 'CVE-2020-0001'"], "python") == [f"s = '{REDACTED}'"] + + +def test_no_path_repository_or_commit_in_the_excerpt() -> None: + lines = [ + "def handler(request):", + " # see https://github.com/acme-corp/webapp/blob/0123456789abcdef0123456789abcdef01234567/app/views.py", + " log('app/views.py')", + " return request", + ] + ids = identities_of("https://github.com/acme-corp/webapp", "app/views.py", ["0123456789abcdef0123456789abcdef01234567"]) + shown = excerpt(lines, "python", "handler", identities=ids) + assert shown is not None + for identity in ("acme-corp", "0123456789abcdef", "app/views.py"): + assert identity not in shown.text + assert "webapp" in shown.text or REDACTED in shown.text # the bare name is code-level, the owner and slug are not + + +def test_bounds_lines_and_characters() -> None: + lines = _long_function(200) + shown = excerpt(lines, "python", "big", bounds=CANDIDATE) + assert shown is not None and shown.truncated + assert len(shown.text.splitlines()) <= CANDIDATE.lines and len(shown.text) <= CANDIDATE.chars + small = excerpt(lines, "python", "big", bounds=EXAMPLE) + assert small is not None and len(small.text.splitlines()) <= EXAMPLE.lines and len(small.text) <= EXAMPLE.chars + wide = _long_function(60, width=120) + tight = excerpt(wide, "python", "big", bounds=Bounds(80, 2000)) + assert tight is not None and len(tight.text) <= 2000 and tight.truncated + + +def test_window_centres_on_the_focus_lines() -> None: + lines = _long_function(300) + shown = excerpt(lines, "python", "big", bounds=Bounds(40, 100_000), focus=[200]) + assert shown is not None and shown.first <= 200 <= shown.last + assert abs((shown.first + shown.last) / 2 - 200) <= 1 + chars = excerpt(lines, "python", "big", bounds=Bounds(80, 1200), focus=[200]) + assert chars is not None and chars.first <= 200 <= chars.last + + +def test_delta_diff_is_redacted_and_bounded() -> None: + base = ["def f(x):", " return escape(x)"] + head = ["def f(x):", " # CVE-2022-2222 sanitize later", " return x"] + diff = delta_diff(base, head, "python") + assert diff.startswith("@@") and "+ return x" in diff and "- return escape(x)" in diff + assert "CVE-2022" not in diff and "sanitize" not in diff and "---" not in diff + long = delta_diff([f"a{i}" for i in range(100)], [f"b{i}" for i in range(100)], "python", max_lines=40) + assert len(long.splitlines()) == 40 and long.splitlines()[-1].startswith("...") + shown = excerpt(head, "python", "f", diff=diff) + assert shown is not None and "\ndiff:\n@@" in shown.text + + +def test_normalised_text_names_the_content() -> None: + assert normalise("a \r\nb\n\n\n") == "a\nb\n" + assert excerpt_sha("a \nb") == excerpt_sha("a\nb\n") + + +def test_brace_languages_without_a_declaration_pattern() -> None: + c = [ + "/* CVE-2016-0001: heap overflow fixed */", + "static int", + "parse_header(const char *buf, size_t len)", + "{", + " if (len > 4) { memcpy(out, buf, len); }", + " return parse_header_inner(buf); /* not a definition */", + "}", + "int other(void) { return parse_header(0, 0); }", + ] + shown = excerpt(c, "c_cpp", "parse_header") + assert shown is not None and shown.first == 3 and shown.last == 7 and "other(void)" not in shown.text + assert excerpt(c, "cpp", "Parser::parse_header") is not None + assert excerpt(["int f(void);", "int g(void) { return f(); }"], "c", "f") is None # a prototype is not a body