diff --git a/.github/workflows/engine-image.yml b/.github/workflows/engine-image.yml index dca961c..764f092 100644 --- a/.github/workflows/engine-image.yml +++ b/.github/workflows/engine-image.yml @@ -7,6 +7,7 @@ on: - 'src/**' - 'benchmarks/learn/engine_trace*.py' - 'benchmarks/learn/engine_task_entry.py' + - 'benchmarks/unseen/replay_entry.py' - 'benchmarks/learn/engine_worker_service.py' - 'Dockerfile' - 'plane/Dockerfile.engine-task' @@ -30,6 +31,10 @@ jobs: dockerfile: Dockerfile - package: ousast-engine-task dockerfile: plane/Dockerfile.engine-task + target: engine + - package: ousast-replay-task + dockerfile: plane/Dockerfile.engine-task + target: replay steps: - uses: actions/checkout@v5 - uses: docker/setup-buildx-action@v3 @@ -50,6 +55,7 @@ jobs: with: context: . file: ${{ matrix.dockerfile }} + target: ${{ matrix.target }} platforms: linux/amd64 push: true tags: | diff --git a/.gitignore b/.gitignore index 9dd8031..82df916 100644 --- a/.gitignore +++ b/.gitignore @@ -36,3 +36,6 @@ workspace/ # MkDocs build output (mkdocs build) site/ + +# Unseen evaluation pointers for non-permissive repositories. +benchmarks/unseen/private/ diff --git a/.kiro/specs/unseen-repo-evaluation/tasks.md b/.kiro/specs/unseen-repo-evaluation/tasks.md index 124060e..d18b477 100644 --- a/.kiro/specs/unseen-repo-evaluation/tasks.md +++ b/.kiro/specs/unseen-repo-evaluation/tasks.md @@ -29,8 +29,8 @@ Execution rules: - **Instrument first** (AGENTS.md). A zero, an empty draft or a fast run is treated as an instrument failure until bytes read and wall time prove otherwise. -- [ ] 1. Used set and guard (design 1.2, 6; increment 1) -- [ ] 1.1 Used-set builder +- [x] 1. Used set and guard (design 1.2, 6; increment 1) +- [x] 1.1 Used-set builder - `benchmarks/unseen/eligibility.py`: normalises names (lower case, no `.git`, no trailing slash) and builds the used set from every source of design table 1.2: pair catalogs, recipes and datasets under `benchmarks/pairs/` plus the local pointer-pair manifests as `benchmarks/pairs/catalog_gen.py` reads them; `population-v1.toml` @@ -43,14 +43,14 @@ Execution rules: and trailing-slash variants collapse; stdout and the output JSON contain no name; a v3-shaped file planted in the synthetic tree is not opened (open is patched to fail on it). - _Requirements: 1.1_ -- [ ] 1.2 Canonical resolution, renames and forks +- [x] 1.2 Canonical resolution, renames and forks - `eligibility.py`: `resolve(candidate, api)` with an injected GitHub client returns the canonical `full_name`, former names (redirects) and the fork network (`parent`, `source`). A candidate is ineligible when any of them is in the used set; one candidate per fork network is kept. - Tests with a fake client: a planted used repository, a renamed one and a fork of a used one are each rejected; two forks of one network keep one; the rejection reason is a source label, not a name. - _Requirements: 1.1_ -- [ ] 1.3 Guard extended to pools +- [x] 1.3 Guard extended to pools - `tests/test_independent_population.py`: the sweep becomes a list of (reservation directory, names): `benchmarks/independent/` for populations, `benchmarks/unseen/` for pools (public manifests, plus `benchmarks/unseen/private/*.toml` when present). The sweep also covers `plane/`. New tests: pool and @@ -62,13 +62,13 @@ Execution rules: - Evidence: the guard is green with an empty `benchmarks/unseen/`, and red when a test fixture names a pool repository in `benchmarks/measurements/` (shown once in the record). - _Requirements: 1.2, 1.3, 1.4_ -- [ ] 1.4 Usage ledger +- [x] 1.4 Usage ledger - `benchmarks/unseen/ledger.py`: append-only `usage-pN.jsonl`, one row per use (slice, decision or experiment id, purpose `baseline | exploratory | informed | qualifies`, date, result digest, freeze digest). It refuses a `qualifies` row for a change whose slice already has an `informed` row for it, and refuses to rewrite rows. - `tests/test_unseen_ledger.py`: append, the refusal, a rewritten row detected by a digest chain. - _Requirements: 5.2_ -- [ ] 1.5 Used-set measurement and record 01 **(coordinator, network)** +- [x] 1.5 Used-set measurement and record 01 **(coordinator, network)** - Run `eligibility.py` against the real sources, the S3 memory store included (read only). - Instrument: each source reports files read and bytes; a source with zero rows fails the run unless it is known empty. @@ -77,7 +77,7 @@ Execution rules: - _Requirements: 1.1, 1.3_ - [ ] 2. Sourcing and extraction (design 1.1, 1.3, 1.4, 2, 3; increment 2) -- [ ] 2.1 Advisory candidates and repository filters +- [x] 2.1 Advisory candidates and repository filters - `benchmarks/unseen/source.py`: from GHSA advisory JSON (pip, npm, maven, composer; published 2019 or later) keep those whose CWE maps to a family through `src/openultrasast/model/taxonomy.py` and that link one fix commit with one parent in one repository; attach the OSV first affected version; flag `post_cutoff` when @@ -87,7 +87,7 @@ Execution rules: multi-repository fixes dropped, cutoff flag, licence classing (GPL, AGPL, LGPL, none and source-available go private). - _Requirements: 1.4, 2.1_ -- [ ] 2.2 Vulnerability-introducing range +- [x] 2.2 Vulnerability-introducing range - `benchmarks/unseen/extract.py`: the `introducing` method of design 2.1 (blame of the sink line, or the declaration for an absence bug; absence at V's first parent; no merge or root; at most 50 files and 2,000 lines; OSV version bracket when present; walk-back of at most five steps) and the `last_touch` fallback. Each @@ -96,7 +96,7 @@ Execution rules: - `tests/test_unseen_extract.py` on a synthetic repository built in a temp dir: accepted V; a reformat walked back; a merge rejected; size cap; bracket rejection; fallback chosen when no V passes. - _Requirements: 2.1_ -- [ ] 2.3 Ordinary-commit draw +- [x] 2.3 Ordinary-commit draw - `extract.py`: first-parent, non-merge commits touching an analysed language; the exclusions of design 2.2 (security words of the narrow pattern, reverts, bot and bump commits, manifest- or lock-only, generated or vendored, advisory fix commits, commits touching a known vulnerable function, more than 1,000 source lines, @@ -105,7 +105,7 @@ Execution rules: - Tests: each exclusion; the narrow pattern keeps "query", "token", "session"; allocation and the seed reproduce the same draw; the 40-commit floor. - _Requirements: 2.2, 2.3_ -- [ ] 2.4 Draft manifest and slices +- [x] 2.4 Draft manifest and slices - `benchmarks/unseen/draft.py`: writes `benchmarks/unseen/draft-p1.toml` (permissive entries by name) and `benchmarks/unseen/private/draft-p1.toml` (the rest); the public file holds an opaque id and sha256(URL, commits) per private entry. Seeded shuffle into three slices of 100 repositories, stratified by @@ -158,7 +158,7 @@ Execution rules: - _Requirements: 1.1, 1.2, 1.3_ - [ ] 4. Dispatcher (design 5.3; increment 4) -- [ ] 4.1 Items, workloads and the AX lane +- [x] 4.1 Items, workloads and the AX lane - `benchmarks/ax/batch.py`: `Item` (id, input objects, result object name, command, extra env, input digest), `Workload` (image digest, allowed URL env names, deadline, result validator) and the `ax` lane, moved from `benchmarks/learn/engine_trace_ax.py`: manifest render, egress readiness, presigned GET and PUT as `.dat`, @@ -167,7 +167,7 @@ Execution rules: - `tests/test_ax_batch.py` with a fake `ax` CLI and a fake store: manifest holds only the allowed URL env names and no credential; names carry the `ousast-engine-` prefix; cleanup on success and failure. - _Requirements: 4.1, 4.2, 4.5_ -- [ ] 4.2 Scheduler, fallback and resume +- [x] 4.2 Scheduler, fallback and resume - `batch.py`: two AX lanes and one VM `docker` lane by default (one container at a time); sandbox death retried once under a new name, a second death or a JVM OOM goes to the VM fallback queue ahead of overflow; `progress.json` written by the scheduler only; a `STOP` file halts launches; resume skips items whose stored @@ -175,7 +175,7 @@ Execution rules: - Tests with fake lanes: retry then fallback; OOM straight to fallback; resume skips; `STOP`; cleanup failure halts; never more than two AX tasks in flight. - _Requirements: 4.3_ -- [ ] 4.3 Engine pass on the dispatcher +- [x] 4.3 Engine pass on the dispatcher - `benchmarks/learn/engine_trace_ax.py` and `engine_trace.run_plan` become an engine `Workload` over `batch.py`; no behaviour change. - Observable: `tests/test_ax_executor.py` and `tests/test_engine_trace.py` pass with their assertions unchanged @@ -188,14 +188,14 @@ Execution rules: - _Requirements: 4.3_ - [ ] 5. Replay task (design 5.1, 5.2; increment 5) -- [ ] 5.1 Image stage +- [x] 5.1 Image stage - `plane/Dockerfile.engine-task`: second stage `FROM engine AS replay` adding `git`, the package with its runtime dependencies and `benchmarks/unseen/replay_entry.py`; smoke `ousast pre-push --help`. The root image stays `ghcr.io/norandom/ax-task-runner:v0.3.1`; the engine stage is unchanged. - `tests/test_unseen_image.py`: parses the Dockerfile; the `replay` stage builds from `engine`; the root image pin; the engine stage's text is byte-identical to before. - _Requirements: 4.1_ -- [ ] 5.2 Task entry +- [x] 5.2 Task entry - `benchmarks/unseen/replay_entry.py`, standard library only, reusing `engine_task_entry.transfer`: GET the spec; `git init` and depth-1 fetches of head and base over HTTPS (180 s bound), verified with `git cat-file -e`; run `ousast pre-push` with the spec's hook flags (defaults, or `--deadline 300` for the long-deadline arm), heap @@ -206,7 +206,7 @@ Execution rules: fields; a missing commit, a failed fetch and a hook crash each give a failure record; zero bytes read and a replay faster than JVM start are flagged `suspect_fast`. - _Requirements: 3.1, 4.1, 4.2_ -- [ ] 5.3 Replay workload +- [x] 5.3 Replay workload - `benchmarks/unseen/replay.py`: builds `Item`s from a frozen slice and an arm (`default` or `long_deadline`); the spec object carries the repository URL so names stay out of AX status and telemetry; task names `ousast-engine-replay--` with an opaque slug; the validator rejects a record without @@ -223,28 +223,28 @@ Execution rules: exit codes, the quick-tier comparison. - _Requirements: 4.1, 4.2, 4.5_ -- [ ] 6. Scorer (design 3, 4, 7.1, 7.3; increment 6) -- [ ] 6.1 Coverage states +- [x] 6. Scorer (design 3, 4, 7.1, 7.3; increment 6) +- [x] 6.1 Coverage states - `benchmarks/unseen/score.py`: each record gets one state (`instrument_failure`, `unanalysable`, `quick_only`, `engine_covered`) from `instrument.json` and the artifact; `suspect_fast` goes to failure; a slice above 5% `instrument_failure` after re-runs is refused. - `tests/test_unseen_score.py` on synthetic records: every state; the refusal; a failure never counts as quiet or missed. - _Requirements: 3.3_ -- [ ] 6.2 Catch and false-alarm rules +- [x] 6.2 Catch and false-alarm rules - `score.py`: family match (engine family, quick-rule CWE through `taxonomy.py`), the location table of design 4.1, all three streams with the stream named per catch; false alarm on any finding of any stream; the alerts-only rate. - Tests: right family wrong place misses; `access_control` on the handler; `config_secrets` file scope; a quick-rule-only catch; one advisory finding on an ordinary change is a false alarm. - _Requirements: 3.2_ -- [ ] 6.3 Rates, intervals and breakdowns +- [x] 6.3 Rates, intervals and breakdowns - `score.py`: Wilson 95% intervals; repository-cluster bootstrap for false alarms; ICC; breakdowns per family, method, `post_cutoff`, stream, coverage state and the confirmed subset. - Tests: the n values of design section 3 (315, 483, 756, 62, 88, 97) reproduced; bootstrap seeded; ICC on a synthetic clustered set. - _Requirements: 2.3, 3.3_ -- [ ] 6.4 Budgets, paired tests, the gate and record 06 +- [x] 6.4 Budgets, paired tests, the gate and record 06 - `score.py`: per-budget thresholds (1%, 5%, 10%), labelled `exploratory` in-slice and `fixed` when supplied; a binary arm's single operating point with the budgets it meets; exact two-sided McNemar for catch and for false alarm; `gate(A, B)` returning `pass | regression | not_run` per design 7.3. @@ -319,3 +319,66 @@ Execution rules: false-alarm rates, coverage per state, spend; `exploratory` rows in `usage-p1.jsonl`. - Observable: one committed table of arms on slice 1; G1's table exists. - _Requirements: 3.4, 5.2, 5.3_ + +## Implementation Notes + +- 2026-10-04: Tasks 1.1–1.4 implemented offline and independently reviewed. Task 1.5 remains for the + coordinator; no networked measurement or commit was performed. Command and API contracts: + `benchmarks/unseen/README.md`. Offline evidence: + `benchmarks/measurements/2026-10-04-unseen-01-offline/record.json`. +- Population guard scope is preserved under the maintainer's hard rule: benchmarks/src/tests for + populations, plus plane for pools. Each directory exemption is reservation-specific, including private + pool manifests. Extending population scope to plane exposed 30 existing references; no historical + records were changed or exempted inside the existing population scope. +- S3 inventory opens through `open_store(read_only=True)` because ordinary store construction writes a + Select capability probe. Default store behavior is unchanged; inventory uses LIST/GET only. +- The freeze writer must match the canonical TOML digest encoding documented in the pool README and + exercised by `test_independent_population.py`; final pool discovery does not trust mutable status. + +### Group 2 implementation evidence (2026-10-04) + +Tasks 2.1–2.4 are implemented offline and independently reviewed. Task 2.5 is +`python -m benchmarks.unseen.draft --root "$PWD" --s3-store s3:// --seed 20260601` +with the documented PATH/PYTHONPATH and GH_TOKEN environment. No networked draw, +label check, group-2 measurement record, guard bisection, or commit was performed. +The group header remains pending its coordinator exits 2.5/2.6. + +The command rebuilds the group-1 used set, resolves renamed/fork repositories, +uses disposable external blobless clones, re-reads licences at extracted heads, +and emits counts/digests only. The README documents injected clients, method bias, +the hard 300-repository floor, and an explicitly unmeasured resource estimate. +Independent review caught and fixed ambiguous parent-function absence and stale +candidate-404 handling; regression tests cover both. + +Validation: requested pytest selection 103 passed, 1 skipped, exit 0 (population-test +output discarded); `ruff check .` exit 0; `ruff format --check .` exit 0 (411 files); +`mypy src/openultrasast` exit 0 (160 source files). The CLI `--help` smoke exited 0. +Source-read instrumentation printed 20,101 bytes for the extraction implementation. + +### Groups 4–6 offline implementation (2026-10-04) + +Tasks 4.1–4.3, 5.1–5.3 and 6.1–6.4 are implemented and independently reviewed. +Tasks 4.4 and 5.4 remain coordinator-owned network proofs; group headers 4 and 5 stay +pending those exits. No commit, network request, Docker, AX or kubectl command was run. +The sourcing job's draft.py, source.py, extract.py and journal.py are unchanged. + +The shared dispatcher retains the engine adapter's historical questions.json object +name to preserve its unchanged assertions; replay uses spec.dat and result.dat. +The engine Dockerfile body is unchanged except its required AS engine alias; CI builds +engine and replay separately. S3Store's read_only class default preserves existing +constructor-bypassing transport fixtures; initialized read-only stores still refuse writes. + +Exact coordinator commands and input contracts are in benchmarks/unseen/README.md, +including frozen TOML/private pointer verification, scoring joins, own-history proof +preparation and anonymous measurement-record generation. JVM wall time is the hook's +aggregate scan build/query timing and is labelled as such. Unsupported-only partitions +with explicit language_not_covered evidence do not pretend to have launched a JVM. + +Record 06 contains the committed synthetic fixture and scorer-output digests. The +reserved-repository guard was run as a black box with both output streams discarded. +Independent review: APPROVED; all reported integration and coverage defects resolved. + +Final verification: requested pytest command 279 passed, 8 skipped, exit 0; separate +`tests/test_ax_batch.py` 5 passed, exit 0; `ruff check .`, `ruff format --check .`, +`mypy src/openultrasast`, and the output-discarded reserved-repository guard all exit 0. +The own-history proof preparer and replay dry run read five items and exit 0. diff --git a/benchmarks/ax/batch.py b/benchmarks/ax/batch.py new file mode 100644 index 0000000..039b584 --- /dev/null +++ b/benchmarks/ax/batch.py @@ -0,0 +1,434 @@ +"""Host-only AX Task transport; credentials never enter task manifests.""" + +from __future__ import annotations + +import hashlib +import ipaddress +import json +import os +import queue +import re +import shutil +import subprocess +import threading +import time +import uuid +from collections import Counter, deque +from concurrent.futures import FIRST_COMPLETED, ThreadPoolExecutor, wait +from dataclasses import dataclass, field +from datetime import timedelta +from pathlib import Path +from urllib.parse import urlsplit + +import yaml + +from openultrasast.config import load_dotenv +from openultrasast.plane.memory import PRESIGNED_ARCHIVE_SUFFIX, S3Store, open_store + + +class EgressError(ValueError): + """Safe diagnostic without presigned URL contents.""" + + +class SandboxFailure(Exception): + """A terminal AX task without an authoritative result object.""" + + +def public_endpoint(url): + """Reject endpoints the actor's hostname-based HTTPS egress policy cannot reach.""" + parts = urlsplit(url) + host = parts.hostname or "" + try: + ipaddress.ip_address(host) + is_ip = True + except ValueError: + is_ip = False + if ( + parts.scheme != "https" + or not host + or is_ip + or "." not in host + or host.endswith((".localhost", ".local", ".internal")) + or parts.port not in (None, 443) + or parts.username + or parts.password + ): + raise EgressError( + "AX requires a public HTTPS DNS hostname in S3_ENDPOINT and presigned URLs: " + "the egress policy allows the store hostname on port 443 only (no IP or plain HTTP)" + ) + return host.lower() + + +def task_state(text, name): + """Accept AX YAML/JSON output and its human-readable tasks table.""" + try: + doc = yaml.safe_load(text) + except yaml.YAMLError: + doc = None + rows = doc.get("items", doc.get("tasks", [doc])) if isinstance(doc, dict) else doc if isinstance(doc, list) else [] + for row in rows: + if not isinstance(row, dict) or row.get("metadata", {}).get("name", row.get("name")) != name: + continue + status = row.get("status", {}) + ip = next( + (status.get(key) or row.get(key) for key in ("workerIP", "workerIp", "worker_ip", "ip") if status.get(key) or row.get(key)), + None, + ) + worker = status.get("worker") + if not ip and isinstance(worker, dict): + ip = worker.get("ip") or worker.get("address") + return str(status.get("phase", row.get("phase", ""))), ip + lines = [line.split() for line in text.splitlines() if line.strip()] + if lines: + headers = [re.sub(r"[^a-z]", "", cell.lower()) for cell in lines[0]] + for cells in lines[1:]: + if name not in cells: + continue + row = dict(zip(headers, cells, strict=False)) + phase = row.get("phase", row.get("status", "")) + ip = row.get("workerip") or row.get("ip") + if not ip: + for cell in cells: + try: + ipaddress.ip_address(cell) + ip = cell + break + except ValueError: + pass + return phase, None if ip in {"", "-", ""} else ip + listed = isinstance(doc, (dict, list)) or bool(lines and "name" in headers and {"phase", "status"}.intersection(headers)) + return "Missing" if listed else "", None + + +def bounded_read(read, timeout): + """Do not let SDK retries postpone task teardown beyond the pin deadline. + + Only the read runs in a daemon thread; it cannot publish a late result to disk. + No executor shutdown waits for an SDK call whose socket has stopped responding. + """ + if timeout <= 0: + raise TimeoutError("AX result deadline") + replies = queue.Queue(maxsize=1) + + def fetch(): + try: + replies.put((True, read())) + except Exception as exc: + replies.put((False, exc)) + + threading.Thread(target=fetch, daemon=True, name="engine-ax-result-read").start() + try: + ok, result = replies.get(timeout=timeout) + except queue.Empty: + raise TimeoutError("AX result deadline") from None + if not ok: + raise result + return result + + +class AXLane: + def __init__(self, args, workload, store=None, pause=time.sleep, clock=time.monotonic): + self.workload = workload + self.args = args + self.pause, self.clock = pause, clock + self.run = uuid.uuid4().hex + load_dotenv() + self.host = public_endpoint(os.environ.get("S3_ENDPOINT", "")) + # Validate the image before opening a store, whose constructor probes the bucket. + validate_image(workload.image) + self.store = store if store is not None else open_store() + if store is None and not isinstance(self.store, S3Store): + raise ValueError("AX requires OUSAST_MEMORY=s3://bucket[/prefix]") + + def __call__(self, item, output, attempt): + attempts = [] + try: + done, worker = self._execute_attempt(item, output, attempt, attempts) + except SandboxFailure: + return {"status": "sandbox_failure", "ax_attempts": attempts} + if done is None or done.returncode: + return {"status": "instrument_failure", "ax_attempts": attempts} + record = json.loads((output / "result.json").read_text()) + return {**record, "worker_ip": worker, "ax_attempts": attempts} + + def ax(self, *args, manifest=None, timeout=30): + env = dict(os.environ) + if self.args.kubeconfig: + env["KUBECONFIG"] = str(Path(self.args.kubeconfig).expanduser()) + done = subprocess.run( + [str(Path(self.args.ax_bin).expanduser()), *args, *(["--atespace", self.args.atespace] if args[0] != "apply" else [])], + input=json.dumps(manifest) if manifest is not None else None, + env=env, + capture_output=True, + text=True, + timeout=max(0.01, timeout), + ) + if done.returncode: + # AX errors may echo manifest URLs; report only verb and exit code. + raise ValueError(f"ax {args[0]} failed (exit {done.returncode})") + return done + + def _execute_attempt(self, item, output, attempt, attempts): + pin_id = item.id + digest = hashlib.sha256(f"{self.run}/{pin_id}/attempt-{attempt}".encode()).hexdigest()[:12] + slug = re.sub(r"[^a-z0-9]+", "-", str(pin_id).lower()).strip("-")[:22] + name = f"ousast-engine-{slug or 'pin'}-{digest}" + key_id = hashlib.sha256(str(pin_id).encode()).hexdigest() + prefix = f"engine-queue/{self.run}/{key_id}/attempt-{attempt}/" + keys = [prefix + item.object_names.get(key, key.lower() + PRESIGNED_ARCHIVE_SUFFIX) for key in item.inputs] + keys.append(prefix + item.result_object) + expires = timedelta(seconds=self.workload.deadline + 600) + worker_ip = None + submitted = False + started = self.clock() + phase = "" + try: + for key, data in zip(keys, item.inputs.values(), strict=False): + self.store._put(key, data) + urls = {env: self.store.presign_get(key, expires) for env, key in zip(item.inputs, keys, strict=False)} + urls["RESULT_URL"] = self.store.presign_put(keys[-1], expires) + if any(public_endpoint(url) != self.host for url in urls.values()): + raise EgressError("AX presign hostname must match S3_ENDPOINT for the store egress policy") + document = manifest(name, self.workload, item, urls, self.args.atespace) + end = self.clock() + self.workload.deadline + # An apply timeout can still have created the task; cleanup must cover it. + submitted = True + self.ax("apply", "-f", "-", manifest=document, timeout=min(30, self.workload.deadline)) + while self.clock() < end: + result = bounded_read(lambda: self.store._get(keys[-1]), end - self.clock()) + if self.clock() >= end: + return None, worker_ip + phase = "" + try: + state = self.ax("get", "tasks", timeout=min(30, end - self.clock())).stdout + phase, ip = task_state(state, name) + worker_ip = ip or worker_ip + except (ValueError, OSError, subprocess.SubprocessError): + # The object is authoritative even when AX status is temporarily unavailable. + pass + if self.clock() >= end: + return None, worker_ip + if result is not None: + self.workload.validate(result[0], output) + return subprocess.CompletedProcess([], 0, "", ""), worker_ip + if phase.lower() in {"failed", "missing"}: + # The worker may have uploaded while we were reading AX status. + result = bounded_read(lambda: self.store._get(keys[-1]), end - self.clock()) + if self.clock() >= end: + return None, worker_ip + if result is not None: + self.workload.validate(result[0], output) + return subprocess.CompletedProcess([], 0, "", ""), worker_ip + raise SandboxFailure + self.pause(min(2, max(0, end - self.clock()))) + return None, worker_ip + except (subprocess.TimeoutExpired, TimeoutError): + return None, worker_ip + except SandboxFailure: + raise + except Exception as exc: + # Never expose exception strings which may contain object-scoped bearer URLs. + detail = str(exc) if isinstance(exc, EgressError) else type(exc).__name__ + return subprocess.CompletedProcess([], 1, "", "AX transport failed: " + detail), worker_ip + finally: + attempts.append({"name": name, "phase": phase, "seconds": self.clock() - started}) + errors = [] + if submitted: + try: + self.ax("delete", "task", name) + except Exception as exc: + # A successful list already confirmed a disappeared task is gone. + if phase != "Missing": + errors.append("task: " + type(exc).__name__) + for key in keys: + try: + self.store._delete(key) + except Exception as exc: + errors.append("object: " + type(exc).__name__) + if errors: + raise RuntimeError("AX cleanup failed; stop dispatching: " + ", ".join(errors)) + + +@dataclass +class Item: + id: str + inputs: dict[str, bytes] + result_object: str + command: tuple[str, ...] + extra_env: dict[str, str] + input_digest: str + object_names: dict[str, str] = field(default_factory=dict) + + +@dataclass(frozen=True) +class Workload: + image: str + url_env: frozenset[str] + deadline: float + validate: object + + +def validate_image(image): + if not re.fullmatch(r"[^\s@]+@sha256:[0-9a-f]{64}", image): + raise ValueError("AX --image must be pinned by sha256 digest") + + +def manifest(name, workload, item, urls, atespace="default", *, validate_name=True): + validate_image(workload.image) + if validate_name and (not re.fullmatch(r"ousast-engine-[a-z0-9-]+", name) or len(name.encode()) > 49): + raise ValueError("invalid AX task name") + if set(urls) != workload.url_env: + raise ValueError("unexpected URL environment names") + # Extra env is deliberately a small numeric execution contract, never inherited credentials. + if set(item.extra_env) - {"DEADLINE", "QUESTION_DEADLINE"} or any( + not re.fullmatch(r"[0-9]+(?:\.[0-9]+)?", str(v)) for v in item.extra_env.values() + ): + raise ValueError("unexpected task environment") + for url in urls.values(): + public_endpoint(url) + env = [{"name": key, "value": str(value)} for key, value in {**urls, **item.extra_env}.items()] + if len(json.dumps(env).encode()) >= 32768: + raise ValueError("AX Task env exceeds the 32 KB limit") + return { + "apiVersion": "ax.io/v1alpha1", + "kind": "Task", + "metadata": {"name": name, "atespace": atespace}, + "spec": {"image": workload.image, "env": env, "command": list(item.command)}, + } + + +def write_json(path, record): + path.parent.mkdir(parents=True, exist_ok=True) + temporary = path.with_suffix(".tmp") + temporary.write_text(json.dumps(record, sort_keys=True) + "\n") + temporary.replace(path) + + +def schedule(pending, lanes, out, execute, complete, *, fallbacks=(), overflow=True, progress=None): + """The sole progress writer; callbacks carry workload-specific records, never transport.""" + if sum(lane in {"ax", "kind"} for lane in lanes) > 2 or lanes.count("docker") > 1: + raise ValueError("at most two AX lanes and one VM lane") + pending, fallbacks = deque(pending), deque(fallbacks) + progress = progress if progress is not None else {"pins_done": 0, "pins_total": len(pending) + len(fallbacks)} + progress["lanes"] = dict(Counter(lanes)) + active = {} + failures = False + + def persist(): + progress.update(pins_in_flight=len(active), active_lanes=dict(Counter(lane for _, lane in active.values()))) + if any(lane in {"ax", "kind"} for lane in lanes): + progress["fallbacks_pending"] = len(fallbacks) + write_json(out / "progress.json", progress) + + with ThreadPoolExecutor(max_workers=len(lanes)) as pool: + + def launch(lane): + if (out / "STOP").exists(): + return + if lane == "docker" and fallbacks: + item, previous = fallbacks.popleft() + elif pending and (lane != "docker" or overflow): + item, previous = pending.popleft(), None + else: + return + active[pool.submit(execute, item, lane, previous)] = item, lane + + for lane in lanes: + launch(lane) + persist() + try: + while active: + finished, _ = wait(active, return_when=FIRST_COMPLETED) + # Resolve ALL completed futures before launching; cleanup exceptions stop launches. + results = [(active.pop(future), future.result()) for future in finished] + for (item, lane), result in results: + failed, fallback = complete(item, lane, result, progress) + if fallback is not None: + fallbacks.append((item, fallback)) + else: + failures |= failed + progress["pins_done"] += 1 + for lane, count in (Counter(lanes) - Counter(lane for _, lane in active.values())).items(): + for _ in range(count): + launch(lane) + persist() + finally: + persist() + return failures + + +def dispatch(items, workload, out, executors, *, lanes=("ax", "ax", "docker"), overflow=True): + """Digest checkpoints, one instrument rerun, sandbox retry, priority VM fallback.""" + out.mkdir(parents=True, exist_ok=True) + pending, fallbacks = [], [] + + def path(item): + return out / (hashlib.sha256(item.id.encode()).hexdigest() + ".json") + + for item in items: + previous = json.loads(path(item).read_text()) if path(item).exists() else {} + if previous.get("input_digest") == item.input_digest: + if previous.get("fallback_pending"): + fallbacks.append((item, previous)) + continue + if previous.get("done"): + continue + pending.append(item) + + def execute(item, lane, previous): + output = out / hashlib.sha256(item.id.encode()).hexdigest() + output.mkdir(exist_ok=True) + attempts = [] + for attempt in range(2): + attempt_output = output / ("attempt-" + uuid.uuid4().hex) + attempt_output.mkdir() + record = executors[lane](item, attempt_output, attempt) + for artifact in attempt_output.iterdir(): + if artifact.is_file(): + shutil.copyfile(artifact, output / artifact.name) + attempts.append({"lane": lane, "status": record.get("status"), "attempt": attempt, "transport": record.get("ax_attempts", [])}) + if record.get("status") not in {"sandbox_failure", "instrument_failure"}: + break + return {**record, "attempts": (previous or {}).get("attempts", []) + attempts, "input_digest": item.input_digest, "done": True} + + def complete(item, lane, record, progress): + fallback = lane in {"ax", "kind"} and record.get("status") in {"sandbox_failure", "oom"} and "docker" in lanes + record["fallback_pending"] = fallback + write_json(path(item), record) + progress["last_item"] = hashlib.sha256(item.id.encode()).hexdigest() + return record.get("status") not in {"ok", "unanalysable", "quick_only", "engine_covered"}, record if fallback else None + + return schedule(pending, lanes, out, execute, complete, fallbacks=fallbacks, overflow=overflow) + + +class DockerLane(AXLane): + """VM lane uses the same presigned-object protocol and pinned replay image.""" + + def _execute_attempt(self, item, output, attempt, attempts): + # Reuse all object/validation/cleanup logic, replacing only the task runner. + import copy + + lane = copy.copy(self) + lane.ax = lane._docker_command + lane._docker_name = None + return AXLane._execute_attempt(lane, item, output, attempt, attempts) + + def _docker_command(self, *args, manifest=None, timeout=30): + if args[0] == "apply": + self._docker_name = manifest["metadata"]["name"] + command = ["docker", "run", "-d", "--name", self._docker_name, "--entrypoint", manifest["spec"]["command"][0]] + for env in manifest["spec"]["env"]: + command += ["-e", env["name"] + "=" + env["value"]] + command += [manifest["spec"]["image"], *manifest["spec"]["command"][1:], "--heap-profile", "vm"] + elif args[0] == "get": + command = ["docker", "inspect", "--format", "{{.State.Running}} {{.State.ExitCode}}", self._docker_name] + else: + command = ["docker", "rm", "-f", self._docker_name or args[2]] + done = subprocess.run(command, capture_output=True, text=True, timeout=timeout) + if done.returncode: + raise ValueError("docker command failed") + if args[0] == "get": + phase = "Running" if done.stdout.startswith("true") else "Failed" + done.stdout = json.dumps({"metadata": {"name": self._docker_name}, "status": {"phase": phase}}) + return done diff --git a/benchmarks/learn/engine_trace.py b/benchmarks/learn/engine_trace.py index 15d56a0..ebb3eaa 100644 --- a/benchmarks/learn/engine_trace.py +++ b/benchmarks/learn/engine_trace.py @@ -16,13 +16,13 @@ import os import re import shutil import subprocess +import sys import tempfile import textwrap import threading import time import tomllib -from collections import Counter, defaultdict, deque -from concurrent.futures import FIRST_COMPLETED, ThreadPoolExecutor, wait +from collections import Counter, defaultdict from pathlib import Path from engine_trace_k8s import IMAGE, Kubernetes @@ -39,6 +39,9 @@ from openultrasast.plane.memory import FileStore from openultrasast.plane.tasks.alerts import engine_languages from openultrasast.preprocess import detect_language +sys.path.insert(0, str(Path(__file__).resolve().parents[2])) +from benchmarks.ax.batch import schedule + ROOT = Path(__file__).resolve().parents[2] DEFAULT_OUT = Path.home() / "ousast-results/plane/engine-trace" SNAPSHOT_FILES = ( @@ -47,6 +50,7 @@ SNAPSHOT_FILES = ( "benchmarks/learn/engine_trace_k8s.py", "benchmarks/learn/engine_trace_queue.py", "benchmarks/learn/engine_trace_ax.py", + "benchmarks/ax/batch.py", "benchmarks/learn/engine_task_entry.py", "benchmarks/learn/engine_worker_service.py", "benchmarks/learn/engine_trace_parse.py", @@ -791,8 +795,8 @@ def run_plan(plan, source, provenance, inputs, args): "active_lanes": {}, } counts = Counter() - pending = deque(plan) - fallbacks = deque() + pending = list(plan) + fallbacks = [] if args.executor in {"ax", "mixed-ax"}: pending.clear() for pin in plan: @@ -803,12 +807,6 @@ def run_plan(plan, source, provenance, inputs, args): fallbacks.append((pin, old)) else: pending.append(pin) - active = {} - failures = False - - def persist(): - progress.update(pins_in_flight=len(active), active_lanes=dict(Counter(lane for _, lane in active.values()))) - write_json(args.out / "progress.json", progress) def execute(pin, lane, attempt=None): if not any(u["supported"] for u in pin["units"]): @@ -825,52 +823,20 @@ def run_plan(plan, source, provenance, inputs, args): options.image = getattr(args, "docker_image", args.image) return run_pin(pin, source, provenance, inputs, args.out, options) - with ThreadPoolExecutor(max_workers=len(lanes)) as pool: + def complete(pin, lane, failed, progress): + result = json.loads((args.out / pin_name(pin)).read_text()) + selected = {u["unit"] for u in pin["units"]} + reasons = [u.get("reason", "") for u in result["units"] if u["unit"] in selected] + sandbox_failed = args.executor == "mixed-ax" and result.get("container_exit") == 1 and "AX sandbox failed twice" in reasons + if lane == "ax" and (sandbox_failed or any("JVM out of memory" in reason for reason in reasons)): + result.update(fallback="vm-sandbox" if sandbox_failed else "vm-oom", fallback_pending=True) + write_json(args.out / pin_name(pin), result) + return False, result + counts.update(u["status"] for u in result["units"] if u["unit"] in selected) + progress.update(last_pin={"repo": pin["repo"], "pin": pin["pin"]}, statuses=dict(counts)) + return failed, None - def launch(lane): - if (args.out / "STOP").exists(): - return - attempt = None - if lane == "docker" and fallbacks: - pin, attempt = fallbacks.popleft() - elif pending and not (args.executor == "ax" and lane == "docker"): - pin = pending.popleft() - else: - return - active[pool.submit(execute, pin, lane, attempt)] = (pin, lane) - - for lane in lanes: - launch(lane) - persist() - while active: - completed, _ = wait(active, return_when=FIRST_COMPLETED) - for future in completed: - pin, lane = active.pop(future) - failed = future.result() - result = json.loads((args.out / pin_name(pin)).read_text()) - selected = {u["unit"] for u in pin["units"]} - reasons = [u.get("reason", "") for u in result["units"] if u["unit"] in selected] - sandbox_failed = args.executor == "mixed-ax" and result.get("container_exit") == 1 and "AX sandbox failed twice" in reasons - if lane == "ax" and (sandbox_failed or any("JVM out of memory" in reason for reason in reasons)): - result.update(fallback="vm-sandbox" if sandbox_failed else "vm-oom", fallback_pending=True) - write_json(args.out / pin_name(pin), result) - fallbacks.append((pin, result)) - continue - failures |= failed - selected = {u["unit"] for u in pin["units"]} - counts.update(u["status"] for u in result["units"] if u["unit"] in selected) - progress.update( - pins_done=progress["pins_done"] + 1, last_pin={"repo": pin["repo"], "pin": pin["pin"]}, statuses=dict(counts) - ) - idle = Counter(lanes) - Counter(lane for _, lane in active.values()) - # Give a waiting fallback priority over overflow on the single VM lane. - for lane, count in idle.items(): - for _ in range(count): - launch(lane) - if args.executor in {"ax", "mixed-ax"}: - progress["fallbacks_pending"] = len(fallbacks) - persist() - return failures + return schedule(pending, lanes, args.out, execute, complete, fallbacks=fallbacks, overflow=args.executor != "ax", progress=progress) def main(): diff --git a/benchmarks/learn/engine_trace_ax.py b/benchmarks/learn/engine_trace_ax.py index d6317ce..6046f8b 100644 --- a/benchmarks/learn/engine_trace_ax.py +++ b/benchmarks/learn/engine_trace_ax.py @@ -1,184 +1,50 @@ -"""Host-only AX Task transport; credentials never enter task manifests.""" +"""Engine workload adapter for the shared AX transport.""" from __future__ import annotations -import hashlib import io -import ipaddress import json -import os -import queue -import re import shutil import subprocess +import sys import tarfile import tempfile -import threading import time -import uuid -from datetime import timedelta from pathlib import Path -from urllib.parse import urlsplit -import yaml +sys.path.insert(0, str(Path(__file__).resolve().parents[2])) from engine_trace_k8s import pack -from openultrasast.config import load_dotenv -from openultrasast.plane.memory import PRESIGNED_ARCHIVE_SUFFIX, S3Store, open_store - - -class EgressError(ValueError): - """Safe diagnostic without presigned URL contents.""" - - -class SandboxFailure(Exception): - """A terminal AX task without an authoritative result object.""" - - -def public_endpoint(url): - """Reject endpoints the actor's hostname-based HTTPS egress policy cannot reach.""" - parts = urlsplit(url) - host = parts.hostname or "" - try: - ipaddress.ip_address(host) - is_ip = True - except ValueError: - is_ip = False - if ( - parts.scheme != "https" - or not host - or is_ip - or "." not in host - or host.endswith((".localhost", ".local", ".internal")) - or parts.port not in (None, 443) - or parts.username - or parts.password - ): - raise EgressError( - "AX requires a public HTTPS DNS hostname in S3_ENDPOINT and presigned URLs: " - "the egress policy allows the store hostname on port 443 only (no IP or plain HTTP)" - ) - return host.lower() +from benchmarks.ax.batch import AXLane, Item, SandboxFailure, Workload, manifest +from benchmarks.ax.batch import task_state as task_state +from openultrasast.plane.memory import open_store as open_store def task_manifest(name, args, urls): - if not re.fullmatch(r"[^\s@]+@sha256:[0-9a-f]{64}", args.image): - raise ValueError("AX --image must be pinned by sha256 digest") - if set(urls) != {"SOURCE_URL", "QUESTIONS_URL", "RESULT_URL"}: - raise ValueError("AX Task accepts only source, questions and result presigned URLs") - for url in urls.values(): - public_endpoint(url) - env = [{"name": key, "value": value} for key, value in urls.items()] - env.extend( - {"name": key, "value": str(value)} - for key, value in {"DEADLINE": args.deadline, "QUESTION_DEADLINE": args.question_deadline}.items() + return manifest( + name, engine_workload(args, lambda *_: None), engine_item({}, "pin", b"", args), urls, args.atespace, validate_name=False ) - if len(json.dumps(env).encode()) >= 32768: - raise ValueError("AX Task env exceeds the 32 KB limit") - return { - "apiVersion": "ax.io/v1alpha1", - "kind": "Task", - "metadata": {"name": name, "atespace": args.atespace}, - "spec": {"image": args.image, "env": env, "command": ["python3", "/app/benchmarks/learn/engine_task_entry.py"]}, - } -def task_state(text, name): - """Accept AX YAML/JSON output and its human-readable tasks table.""" - try: - doc = yaml.safe_load(text) - except yaml.YAMLError: - doc = None - rows = doc.get("items", doc.get("tasks", [doc])) if isinstance(doc, dict) else doc if isinstance(doc, list) else [] - for row in rows: - if not isinstance(row, dict) or row.get("metadata", {}).get("name", row.get("name")) != name: - continue - status = row.get("status", {}) - ip = next( - (status.get(key) or row.get(key) for key in ("workerIP", "workerIp", "worker_ip", "ip") if status.get(key) or row.get(key)), - None, - ) - worker = status.get("worker") - if not ip and isinstance(worker, dict): - ip = worker.get("ip") or worker.get("address") - return str(status.get("phase", row.get("phase", ""))), ip - lines = [line.split() for line in text.splitlines() if line.strip()] - if lines: - headers = [re.sub(r"[^a-z]", "", cell.lower()) for cell in lines[0]] - for cells in lines[1:]: - if name not in cells: - continue - row = dict(zip(headers, cells, strict=False)) - phase = row.get("phase", row.get("status", "")) - ip = row.get("workerip") or row.get("ip") - if not ip: - for cell in cells: - try: - ipaddress.ip_address(cell) - ip = cell - break - except ValueError: - pass - return phase, None if ip in {"", "-", ""} else ip - listed = isinstance(doc, (dict, list)) or bool(lines and "name" in headers and {"phase", "status"}.intersection(headers)) - return "Missing" if listed else "", None +def engine_workload(args, validate): + return Workload(args.image, frozenset({"SOURCE_URL", "QUESTIONS_URL", "RESULT_URL"}), args.deadline, validate) -def bounded_read(read, timeout): - """Do not let SDK retries postpone task teardown beyond the pin deadline. - - Only the read runs in a daemon thread; it cannot publish a late result to disk. - No executor shutdown waits for an SDK call whose socket has stopped responding. - """ - if timeout <= 0: - raise TimeoutError("AX result deadline") - replies = queue.Queue(maxsize=1) - - def fetch(): - try: - replies.put((True, read())) - except Exception as exc: - replies.put((False, exc)) - - threading.Thread(target=fetch, daemon=True, name="engine-ax-result-read").start() - try: - ok, result = replies.get(timeout=timeout) - except queue.Empty: - raise TimeoutError("AX result deadline") from None - if not ok: - raise result - return result +def engine_item(pin, pin_id, source, args): + return Item( + pin_id, + {"SOURCE_URL": source, "QUESTIONS_URL": json.dumps(pin).encode()}, + "result.dat", + ("python3", "/app/benchmarks/learn/engine_task_entry.py"), + {"DEADLINE": str(args.deadline), "QUESTION_DEADLINE": str(args.question_deadline)}, + "", + {"SOURCE_URL": "source.dat", "QUESTIONS_URL": "questions.json"}, + ) -class AX: +class AX(AXLane): def __init__(self, args, store=None, pause=time.sleep, clock=time.monotonic): - self.args = args - self.pause, self.clock = pause, clock - self.run = uuid.uuid4().hex - load_dotenv() - self.host = public_endpoint(os.environ.get("S3_ENDPOINT", "")) - # Validate the image before opening a store, whose constructor probes the bucket. - task_manifest("validate", args, {key: f"https://{self.host}/" for key in ("SOURCE_URL", "QUESTIONS_URL", "RESULT_URL")}) - self.store = store if store is not None else open_store() - if store is None and not isinstance(self.store, S3Store): - raise ValueError("AX requires OUSAST_MEMORY=s3://bucket[/prefix]") - - def ax(self, *args, manifest=None, timeout=30): - env = dict(os.environ) - if self.args.kubeconfig: - env["KUBECONFIG"] = str(Path(self.args.kubeconfig).expanduser()) - done = subprocess.run( - [str(Path(self.args.ax_bin).expanduser()), *args, *(["--atespace", self.args.atespace] if args[0] != "apply" else [])], - input=json.dumps(manifest) if manifest is not None else None, - env=env, - capture_output=True, - text=True, - timeout=max(0.01, timeout), - ) - if done.returncode: - # AX errors may echo manifest URLs; report only verb and exit code. - raise ValueError(f"ax {args[0]} failed (exit {done.returncode})") - return done + super().__init__(args, engine_workload(args, lambda *_: None), store, pause, clock) def receive(self, data, output, pin): # Validate in isolation: malformed results must never become usable checkpoints. @@ -220,11 +86,15 @@ class AX: shutil.copytree(root, output, dirs_exist_ok=True) def execute(self, checkout, output, pin, pin_id): + import copy + + self = copy.copy(self) + self.workload = engine_workload(self.args, lambda data, target: self.receive(data, target, pin)) attempts = [] try: for attempt in range(2): try: - return self._execute_attempt(checkout, output, pin, pin_id, attempt, attempts) + return self._execute_attempt(engine_item(pin, pin_id, pack(checkout), self.args), output, attempt, attempts) except SandboxFailure: if attempt == 0: self.pause(10) @@ -236,84 +106,3 @@ class AX: record = json.loads(result_path.read_text()) record["ax_attempts"] = attempts result_path.write_text(json.dumps(record)) - - def _execute_attempt(self, checkout, output, pin, pin_id, attempt, attempts): - digest = hashlib.sha256(f"{self.run}/{pin_id}/attempt-{attempt}".encode()).hexdigest()[:12] - slug = re.sub(r"[^a-z0-9]+", "-", str(pin_id).lower()).strip("-")[:22] - name = f"ousast-engine-{slug or 'pin'}-{digest}" - key_id = hashlib.sha256(str(pin_id).encode()).hexdigest() - prefix = f"engine-queue/{self.run}/{key_id}/attempt-{attempt}/" - keys = [prefix + suffix for suffix in (f"source{PRESIGNED_ARCHIVE_SUFFIX}", "questions.json", f"result{PRESIGNED_ARCHIVE_SUFFIX}")] - expires = timedelta(seconds=self.args.deadline + 600) - worker_ip = None - submitted = False - started = self.clock() - phase = "" - try: - self.store._put(keys[0], pack(checkout)) - self.store._put(keys[1], json.dumps(pin).encode()) - urls = { - "SOURCE_URL": self.store.presign_get(keys[0], expires), - "QUESTIONS_URL": self.store.presign_get(keys[1], expires), - "RESULT_URL": self.store.presign_put(keys[2], expires), - } - if any(public_endpoint(url) != self.host for url in urls.values()): - raise EgressError("AX presign hostname must match S3_ENDPOINT for the store egress policy") - manifest = task_manifest(name, self.args, urls) - end = self.clock() + self.args.deadline - # An apply timeout can still have created the task; cleanup must cover it. - submitted = True - self.ax("apply", "-f", "-", manifest=manifest, timeout=min(30, self.args.deadline)) - while self.clock() < end: - result = bounded_read(lambda: self.store._get(keys[2]), end - self.clock()) - if self.clock() >= end: - return None, worker_ip - phase = "" - try: - state = self.ax("get", "tasks", timeout=min(30, end - self.clock())).stdout - phase, ip = task_state(state, name) - worker_ip = ip or worker_ip - except (ValueError, OSError, subprocess.SubprocessError): - # The object is authoritative even when AX status is temporarily unavailable. - pass - if self.clock() >= end: - return None, worker_ip - if result is not None: - self.receive(result[0], output, pin) - return subprocess.CompletedProcess([], 0, "", ""), worker_ip - if phase.lower() in {"failed", "missing"}: - # The worker may have uploaded while we were reading AX status. - result = bounded_read(lambda: self.store._get(keys[2]), end - self.clock()) - if self.clock() >= end: - return None, worker_ip - if result is not None: - self.receive(result[0], output, pin) - return subprocess.CompletedProcess([], 0, "", ""), worker_ip - raise SandboxFailure - self.pause(min(2, max(0, end - self.clock()))) - return None, worker_ip - except (subprocess.TimeoutExpired, TimeoutError): - return None, worker_ip - except SandboxFailure: - raise - except Exception as exc: - # Never expose exception strings which may contain object-scoped bearer URLs. - detail = str(exc) if isinstance(exc, EgressError) else type(exc).__name__ - return subprocess.CompletedProcess([], 1, "", "AX transport failed: " + detail), worker_ip - finally: - attempts.append({"name": name, "phase": phase, "seconds": self.clock() - started}) - errors = [] - if submitted: - try: - self.ax("delete", "task", name) - except Exception as exc: - # A successful list already confirmed a disappeared task is gone. - if phase != "Missing": - errors.append("task: " + type(exc).__name__) - for key in keys: - try: - self.store._delete(key) - except Exception as exc: - errors.append("object: " + type(exc).__name__) - if errors: - raise RuntimeError("AX cleanup failed; stop dispatching: " + ", ".join(errors)) diff --git a/benchmarks/measurements/2026-10-04-unseen-01-used-set/record.json b/benchmarks/measurements/2026-10-04-unseen-01-used-set/record.json new file mode 100644 index 0000000..2554207 --- /dev/null +++ b/benchmarks/measurements/2026-10-04-unseen-01-used-set/record.json @@ -0,0 +1,86 @@ +{ + "count": 1029, + "guard": "green", + "sha256": "5a644a1e93c68a0a35c794e653286d0532b7eb8d757123ccce5f08b11bf9a4a2", + "sources": { + "harvest": { + "bytes": 501031, + "count": 467, + "files": 3, + "known_empty": false, + "rows": 446, + "skipped_values": 446 + }, + "measurements": { + "bytes": 21169903, + "count": 278, + "files": 221, + "known_empty": false, + "rows": 376, + "skipped_values": 681 + }, + "memory_local": { + "bytes": 8084523, + "count": 192, + "files": 776, + "known_empty": false, + "rows": 759, + "skipped_values": 2 + }, + "memory_s3": { + "bytes": 15928676, + "count": 572, + "files": 1499, + "known_empty": false, + "rows": 1490, + "skipped_values": 2 + }, + "pairs": { + "bytes": 1680480, + "count": 728, + "files": 28, + "known_empty": false, + "rows": 900, + "skipped_values": 1260 + }, + "pointers": { + "bytes": 892448, + "count": 537, + "files": 7, + "known_empty": false, + "rows": 586, + "skipped_values": 666 + }, + "populations": { + "bytes": 32431, + "count": 28, + "files": 2, + "known_empty": false, + "rows": 28, + "skipped_values": 0 + }, + "sweep": { + "bytes": 34006439, + "count": 1029, + "files": 1619, + "known_empty": false, + "rows": 2155, + "skipped_values": 0 + }, + "units": { + "bytes": 319641, + "count": 229, + "files": 9, + "known_empty": false, + "rows": 642, + "skipped_values": 0 + } + }, + "test_exit_code": 0, + "tests": { + "errors": 0, + "failures": 0, + "skipped": 1, + "tests": 41 + } +} diff --git a/benchmarks/measurements/2026-10-04-unseen-06-scorer/record.json b/benchmarks/measurements/2026-10-04-unseen-06-scorer/record.json new file mode 100644 index 0000000..d17f337 --- /dev/null +++ b/benchmarks/measurements/2026-10-04-unseen-06-scorer/record.json @@ -0,0 +1,41 @@ +{ + "commit": "48ae918dec25119142ae781607ba99306ae5ec17", + "fixture_bytes": 15374, + "fixture_sha256": "dd3c774a47d553105ed85b9cea8810736c233f870e5a015559a1524d5db5a14d", + "implementation_sha256": { + "benchmarks/ax/batch.py": "37aecac93beb25c02826e81eacaf064fbebb447a998fe139f6d94a07035b8685", + "benchmarks/unseen/replay.py": "32efb3e0b9e9d247f2780b5f3ceeb01b4c2cbda3656c75a741f9925621cfdfbc", + "benchmarks/unseen/replay_entry.py": "994d8e239eb505dc2c5bad4cdb8b5d45967a8a2aa9ef48fb4c2f20ba45736280", + "benchmarks/unseen/score.py": "56e0a7b34482b5d84df39824b8ee447c283e35bd932f38cfe692cf84cb52cef7" + }, + "network_proofs": { + "4.4": "coordinator_pending", + "5.4": "coordinator_pending" + }, + "scope": "offline synthetic scorer only", + "score_command_exit": 0, + "scorer_output_sha256": "77f154eedb6a4acd098ab63a0b17434291f0d226dabe8ac66f2e0b362154d002", + "scorer_test_exit": 0, + "scorer_tests": 14, + "synthetic_records": 20, + "uncommitted": true, + "validation": { + "generic_dispatcher_tests": { + "exit_code": 0, + "passed": 5 + }, + "mypy": 0, + "proof_and_scorer_tests": { + "exit_code": 0, + "passed": 16 + }, + "requested_pytest": { + "exit_code": 0, + "passed": 279, + "skipped": 8 + }, + "reserved_repository_guard": 0, + "ruff_check": 0, + "ruff_format_check": 0 + } +} diff --git a/benchmarks/unseen/README.md b/benchmarks/unseen/README.md new file mode 100644 index 0000000..9af5c38 --- /dev/null +++ b/benchmarks/unseen/README.md @@ -0,0 +1,291 @@ +# Unseen evaluation, group 1 + +The coordinator runs task 1.5 from the repository root: + +```sh +export PATH=/home/mc/Source/OpenUltraSAST/.venv/bin:$PATH +export PYTHONPATH=$PWD/src +python -m benchmarks.unseen.eligibility --s3-store s3:// --verify-tests --output benchmarks/measurements/2026-10-04-unseen-01-used-set/record.json +``` + +This is the networked command; implementation tests use fakes. It loads `.env` with +`openultrasast.config.load_dotenv` (exported values win), opens the local FileStore and +S3 MemoryStore through `openultrasast.plane.memory.open_store(..., read_only=True)`, +and uses `GH_TOKEN` for GitHub GET resolution. `s3://` selects `S3_BUCKET`; pass +`--s3-store s3://BUCKET/PREFIX` for a prefixed store and `--local-store file:///PATH` +for a different local store. Read-only S3 opening skips the normal write/Select probe; +collection uses LIST/GET only. Existing normal store opening is unchanged. + +Pair catalogs/recipes include local pointer rows (`vendored = false`, including +`fix_repo`). Additional external manifests can be supplied with repeatable +`--pointer-manifest PATH`. Registered units in memory and experiment YAML are followed; +missing units with a registered digest fail. Plans without frozen units are not used data. +Experiment units identify repositories in `group`. Binary measurement artifacts are +read and counted; structured JSON, TOML and multi-document YAML are parsed strictly. + +Each source records files/objects read, bytes read, repository-bearing rows (distinct +names per input), and distinct repository counts. The digest is SHA-256 of sorted, +normalized names joined with newlines. Source counts overlap. An empty source fails; +only after establishing it is empty may the coordinator pass `--known-empty LABEL`. +The output records that declaration. Missing/unreadable inputs still fail. + +GitHub redirects add the requested and canonical names, plus parent/source fork +identities. Used names are also resolved so an old used name excludes a renamed +candidate. GitHub cannot enumerate all historical names: injected clients may supply +`former_names`; actual historical aliases come from used references and redirects. +A 404 for a historical/synthetic used identity retains its literal identity; other +API failures stop collection. Candidate resolution always fails closed on a 404. +`resolve(candidate, api)` and `rejection(resolution, used, selected_networks)` are +available to the later sourcing task. Names stay in memory; output contains only +counts, labels and SHA-256 digests. Output files are created exclusively, never overwritten. + +`--verify-tests` runs the three group-1 test modules with both output streams discarded, +including the existing population guard. Only aggregate test counts and the exit code +enter the record; a failed test prevents the record. V3 paths are excluded before +eligibility reads; only the existing guard handles v3 inside its black-box test process. + +Population reservations retain their existing sweep over `benchmarks/`, `src/`, and +`tests/`. Pool reservations also cover `plane/`. Each reservation exempts only its own +directory; population checks include all of `benchmarks/unseen/private/`. This preserves +the existing population behavior while adding pools (the hard rule for this increment). + +All `pool-pN.toml` files require a matching `freeze-pN.json`; drafts use `draft-pN.toml`. +The freeze digest is SHA-256 of parsed public TOML encoded as UTF-8 JSON with sorted +keys, compact separators and `ensure_ascii=False`. Private entries must be represented +by opaque digests in that public manifest, as task 3.2 requires. + +The ledger exposes `append(path, row)` and `read(path)`. Rows contain `slice`, `decision` +(an opaque decision or experiment label), `purpose`, ISO `date`, `result_digest`, and +`freeze_digest`. Appends lock and validate the existing chain, reject duplicates or a +changed freeze, and reject qualification of an informed slice for the same decision. +Keep the published final digest to authenticate history: the chain detects row edits, +but cannot independently prove that an entire history was not replaced or truncated. + +## Sourcing and extraction (tasks 2.1–2.4) + +The coordinator's networked task **2.5** is now: + +```sh +export PATH=/home/mc/Source/OpenUltraSAST/.venv/bin:$PATH +export PYTHONPATH=$PWD/src +python -m benchmarks.unseen.draft --root "$PWD" --s3-store s3:// --seed 20260601 \ + --cache /home/mc/ousast-results/plane/unseen-draft --max-hours 8 +``` + +Export `GH_TOKEN` first, or supply it through the root `.env` (exported values win). +The same read-only local/S3 stores, `--pointer-manifest`, and explicit `--known-empty` +options as task 1.5 apply. Missing or unreadable sources fail. The used set is rebuilt +and canonicalized before candidate selection, including redirects and fork networks. +No population v3 file is opened and no guard/bisection is run by this command; that is +reserved for group 3. An existing draft is never overwritten. + +The command writes `--cache/journal/draft-p1.toml` and +`--cache/journal/private-draft-p1.toml` only after all sourcing stages complete. +Both stay outside the repository, with mode 0600. The public-format file contains permissive entries +and, for every private entry, only an opaque ID, slice, private marker, and SHA-256 +of canonical JSON containing its URL and ordered `[base, head]` pairs. Private entries +carry the complete records, including licences, in the private-format file. +Licensing is re-read through GitHub's licence endpoint **at the actual extracted +vulnerability replay head**; the candidate's earlier fix-parent classification is +provisional. An explicit SPDX declaration in the pinned file takes precedence over +the API's classification. Missing/unknown/compound licences remain private. + +Three seeded slices each contain 100 repositories, one vulnerability range and twelve +ordinary changes per repository. Each ecosystem/post-cutoff stratum differs from its +proportional per-slice target by less than one repository. Post-cutoff candidates are +attempted first within each ecosystem. Below 300 eligible repositories the command +exits 1 with counts and does not write a partial draft or relax any selection bar. +The group-2 measured exit and label confirmation remain coordinator tasks 2.5/2.6; +passing implementation tests does not assert that 300 eligible repositories exist. + +Clones and checkpoints default to `results_root()/unseen-draft` (normally +`~/ousast-results/plane/unseen-draft`), overridable by `--cache /outside/path`. +The cache cannot lie inside `--root`. One opaque, disposable clone exists at a time; +it is removed on success, rejection, or failure. `--filter=blob:none --no-checkout` +keeps the commit graph needed for temporal thirds, blame, and release-tag ancestry, +and fetches source blobs on demand. This is a **partial clone**, not a shallow history. +The 500 MiB disk cap is checked after Git commands, including lazy blob fetches; +one in-flight command can transiently exceed it before rejection and cleanup. +Git output is captured, never forwarded. Each repository must read a nonempty source +blob before extraction. Summary instrumentation includes source read operations and +actual bytes, clone count, cumulative/peak disk bytes, wall time, and logical API calls +and HTTP attempts. Zero source bytes or a failed Git read is an instrument failure, +not an eligibility drop. Stdout is a JSON summary of counts/digests only; no names or +URLs enter stdout or a measurement record. Raw per-change data stays in the external journal and manifests. + +Extraction uses Git and syntax boundaries, with no scanner, engine, or model. Python +uses stdlib AST; Java/PHP/JS/TS use conservative named-function brace matching. Ambiguous +or unparseable functions can reject a candidate. The first removed code line per +fix-touched function is the SZZ sink proxy; pure-addition fixes use its declaration. +Normalized text absence, non-root/non-merge ancestry, file/line limits, and the OSV +release bracket are checked before accepting `introducing`. Matching tags are `VERSION` +or `vVERSION`; an absent/ambiguous tag or missing prior release cannot certify the +bracket. After at most five walk-backs, `last_touch` uses `git log -L` with the resolved +numeric function extent (also works where Git lacks a language-specific function driver). +Method, bias, fallback reasons, head function/range, files/hunks, family, advisory, and +cutoff flag are recorded. Task 2.6 must review every third `introducing` label. + +All reviewed advisories are enumerated to exclude **all** known repository fixes from +ordinary changes, including old advisories and advisories outside the four candidate +ecosystems. Fix URLs are canonicalized before indexing so renames cannot hide them. +The ordinary filter uses only the design's narrowed security pattern. It counts +rejections (including oversized commits and repositories failing the 40-commit floor), +checks both sides for touches to known vulnerable functions, and inspects generated +headers and inherited `.gitattributes`. Sampling allocates twelve draws proportionally +across three history-time thirds and four source-size buckets, with one per third. + +For offline tests, inject GitHub/OSV clients, the clone context, and `UsedSet` into +`draft.run`; HTTP transports and Git subprocess runners also accept fakes. Tests use +sanitized REST-shaped data and a temporary synthetic local Git history. No implementation +test contacts GitHub, OSV, a model, or a cluster. HTTP retries are bounded to five attempts; +rate-limit reset/Retry-After waits and transient 5xx backoff are tested with fake clocks +and sleepers. A stale candidate 404 is counted; other service failures fail the run. + +### Checkpoint and resume + +The command above uses journal directory +`/home/mc/ousast-results/plane/unseen-draft/journal`. Run the same command to resume. +`--max-hours` is a per-invocation budget; changing it does not invalidate checkpoints. +Create `journal/STOP` to stop between units, then remove it before resuming. A page, +repository resolution, candidate decision, or repository extraction already running +finishes first, so the time budget is not a hard timeout on a large repository. +Clean stops exit 0 with `status: stopped`; failures and shortfalls exit 1. Completion +exits 0 with `status: draft`. No partial manifest is published. + +`inputs.json` binds the seed, source-labelled used-set digest, known-empty declarations, +clone cap, and design fingerprints (sourcing/extraction/eligibility/publication code and +family taxonomy). A mismatch refuses resume and prints only the changed parameter +labels. Use a new external cache for a deliberately changed design or population. +The used set is rebuilt each invocation; repository resolutions, including 404s, are +reused from the journal. Live GitHub metadata for unfinished units can still change; +completed pages and decisions are the durable sourcing snapshot. + +Each completed unit is appended to its stage JSONL and fsynced before work advances: +`advisories.jsonl` (pages), `resolution.jsonl` (alias/fork metadata), +`eligibility.jsonl` (candidate decisions), `extraction.jsonl` (repository/advisory +results, vulnerable range, ordinary draws, pinned licence, method and counts), and +`completion.jsonl` (final manifest digest). A lock permits only one writer. +`progress.json` is atomically rewritten after every unit with stage counts, last unit, +last observed GitHub rate-limit headers, and cumulative active elapsed seconds. +These files may contain identities and must stay outside version control. Stdout and +repository records contain only counts, digests and fixed labels. + +Replaying completed units makes no GitHub/OSV or clone calls. Final manifests are +reconstructed deterministically from journal results; an interrupted publication can +be retried, but a different existing manifest is refused. Completed JSONL rows are +never rewritten. Malformed or torn records fail closed instead of being interpreted +as completed work. Tests interrupt each stage with fakes, then compare resumed and +uninterrupted manifests byte for byte. + +### Planning estimate for task 2.5 (not measured) + +Let A be all reviewed advisories, R distinct fix-linked repository identities, U used +identities not already cached, C candidate advisories passing early filters, and K +repositories surviving extraction. GitHub requests are approximately +`ceil(A/100) + R + U + 2*C + K` (enumeration, canonical repositories, fix metadata, +provisional licence, final licence), plus retries. OSV requests are at most C. Git +fetches and lazy blob transfers are additional network operations, not GitHub API calls. + +For a planning scenario of A=40,000, R=15,000, U≈1,000, C=600–1,500, K=300: +about **18,000–20,000 GitHub calls** and **600–1,500 OSV calls**. The all-advisory index +makes this substantially more than 300 repository lookups. At an assumed 5,000-request +hourly allowance, allow several rate-limit windows. Actual token limits can differ. +For 300–1,000 attempted clones averaging 20–100 MB after source hydration, plan roughly +**6–100 GB transferred**, with one clone occupying about 500 MiB at most between commands +plus the retained journal and manifests after completion. Transfer volume is an estimate; disk bytes +are only a proxy. A working planning range is **8–48 hours**, potentially longer on +large histories: every first-parent commit is inspected and lazy blobs can dominate. +These are assumptions, not evidence from a network run. The command reports actual +counts and wall time; task 2.5 establishes the measured estimate. + +## Dispatcher, replay and scoring (groups 4–6) + +`benchmarks.ax.batch` owns Task transport and the shared scheduler. New workloads supply +`Item` input bytes/commands and a `Workload` image, URL allowlist, deadline and result +validator. Two AX slots and one VM slot are the default; sandbox death gets one fresh +Task before VM fallback, OOM goes straight to VM, and instrument failures get one rerun. +Cleanup failure prevents further launches. Create `STOP` in the output directory to stop +launching; remove it and repeat the command to resume matching input digests. The legacy +engine adapter retains its `questions.json` object name for compatibility; replay inputs +and outputs use `.dat`. Both use the same object polling and cleanup implementation. + +Replay reads `pool-pN.toml`, verifies its adjacent `freeze-pN.json` canonical digest, +selects `--slice`, and resolves private pointers through `private/pool-pN.toml` (or +`--private-manifest`). It never reads a draft. `--dry-run` prints only counts and digests. +The five-change proof uses a separate frozen JSON manifest from this project's own +source-changing history. Raw output must be outside the repository. + +Each output root contains `inputs.json` and digest-keyed records and artifact directories. +The scorer accepts that root directly, joins change labels, and includes missing or stale +results as instrument failures. It refuses a slice above 5% failure after reruns. Example: + +```sh +python -m benchmarks.unseen.score "$HOME/ousast-results/unseen/p1/1/default" \ + --out "$HOME/ousast-results/unseen/p1/1/default-score.json" +``` + +For paired binary arms add `--against BASELINE_ROOT --slice-id p1/2`; `--informed` forces +`unseen.gate=not_run`. The later experiment integration owns ledger verification; callers +of `gate` must pass that ledger decision. Score-bearing arms use `budgets` with fixed +thresholds and `gate(..., binary=False, threshold=..., threshold_fixed=True)`. In-slice +thresholds are exploratory and cannot qualify adoption. JVM wall time is explicitly the +sum of the hook artifact's scan build/query times, not a separately observed process timer. + +### Coordinator task 4.4 + +Run from the merged repository root. Set `ENGINE_IMAGE` to the **engine** digest printed +by Engine image CI, `ENGINE_UNITS` and `ENGINE_EXAMPLES` to the frozen input files for the +two stored pins, `ENGINE_BASELINE` to their stored record directory, and `AX_KUBECONFIG` +to the operator's kube-ax kubeconfig. Host store credentials stay on the VM; use the public +HTTPS `S3_ENDPOINT=https://files.because-security.com` and the operator's `OUSAST_MEMORY`. +Use a fresh output directory, so the smoke executes instead of resuming prior records. + +```sh +export PATH=/home/mc/Source/OpenUltraSAST/.venv/bin:$PATH +export PYTHONPATH=$PWD/src +export S3_ENDPOINT=https://files.because-security.com +python benchmarks/learn/engine_trace.py \ + --units "$ENGINE_UNITS" --examples "$ENGINE_EXAMPLES" \ + --executor ax --parallel 2 --limit 2 --image "$ENGINE_IMAGE" \ + --kubeconfig "$AX_KUBECONFIG" --out "$HOME/ousast-results/unseen/proof-04" +python -m benchmarks.unseen.proof engine --image "$ENGINE_IMAGE" \ + --records "$HOME/ousast-results/unseen/proof-04" --baseline "$ENGINE_BASELINE" \ + --out "benchmarks/measurements/$(date +%F)-unseen-04-dispatcher/record.json" +``` + +### Coordinator task 5.4 + +Merge before running this proof. `.github/workflows/engine-image.yml` builds the `engine` +and `replay` targets on push to main; it publishes `ousast-engine-task` and +`ousast-replay-task` separately and prints their digests. Set `REPLAY_IMAGE` to the latter +full `ghcr.io/...@sha256:...` reference. No manual image build is needed. Set +`KIND_KUBECONFIG` and `AX_KUBECONFIG` to the corresponding AX kubeconfigs; the kind lane +uses the identical Task API/manifest, never a Kubernetes Job. + +```sh +export PATH=/home/mc/Source/OpenUltraSAST/.venv/bin:$PATH +export PYTHONPATH=$PWD/src +export S3_ENDPOINT=https://files.because-security.com +python -m benchmarks.unseen.replay --proof-history \ + --manifest "$HOME/ousast-results/unseen/proof-05-inputs.json" +python -m benchmarks.unseen.replay --manifest "$HOME/ousast-results/unseen/proof-05-inputs.json" \ + --image "$REPLAY_IMAGE" --lane kind --kubeconfig "$KIND_KUBECONFIG" \ + --out "$HOME/ousast-results/unseen/proof-05-kind" +python -m benchmarks.unseen.replay --manifest "$HOME/ousast-results/unseen/proof-05-inputs.json" \ + --image "$REPLAY_IMAGE" --lane ax --kubeconfig "$AX_KUBECONFIG" \ + --out "$HOME/ousast-results/unseen/proof-05-ax" +python -m benchmarks.unseen.replay --manifest "$HOME/ousast-results/unseen/proof-05-inputs.json" \ + --image "$REPLAY_IMAGE" --lane docker \ + --out "$HOME/ousast-results/unseen/proof-05-vm" +python -m benchmarks.unseen.proof replay --image "$REPLAY_IMAGE" \ + --kind-records "$HOME/ousast-results/unseen/proof-05-kind" \ + --ax-records "$HOME/ousast-results/unseen/proof-05-ax" \ + --vm-records "$HOME/ousast-results/unseen/proof-05-vm" \ + --out "benchmarks/measurements/$(date +%F)-unseen-05-replay-task/record.json" +``` + +The record command fails if fewer than five results exist, identities differ, bytes/JVM +proof is missing, or quick-tier findings differ. It does not convert a quick-only run +without a JVM into an engine proof. Pool replays use `--lane mixed` (default) for the two +AX slots and single VM overflow/fallback lane; `--arm long_deadline` changes only the +hook deadline to 300 seconds and is a separate input digest. diff --git a/benchmarks/unseen/draft.py b/benchmarks/unseen/draft.py new file mode 100644 index 0000000..6597699 --- /dev/null +++ b/benchmarks/unseen/draft.py @@ -0,0 +1,472 @@ +"""Coordinator-only sourcing command: python -m benchmarks.unseen.draft --help. + +All network and git operations are injected at the run() boundary. Only counts, +digests and fixed failure codes go to stdout; identity-bearing output stays in the +private journal outside the repository. No population guard is run here. +""" + +from __future__ import annotations + +import argparse +import copy +import hashlib +import json +import os +import random +import subprocess +import tempfile +import time +from collections import Counter, defaultdict +from contextlib import contextmanager +from dataclasses import asdict +from pathlib import Path + +from . import eligibility, extract, source +from .extract import Git, InstrumentFailure +from .journal import Journal, ResolvedGitHub, ResumeMismatch, Stopped, atomic, digest, external + + +def canonical(value) -> bytes: + return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode() + + +def private_digest(row: dict) -> str: + return hashlib.sha256(canonical({"url": row["url"], "commits": [[c["base"], c["head"]] for c in row["changes"]]})).hexdigest() + + +def assign_slices(rows: list[dict], *, seed: int) -> list[dict]: + if len(rows) != 300 or len({r["repository"] for r in rows}) != 300: + raise ValueError("pool_size") + for row in rows: + kinds = Counter(c["kind"] for c in row["changes"]) + if kinds != {"vulnerability": 1, "ordinary": 12}: + raise ValueError("change_count") + if row["license_class"] not in {"permissive", "private"}: + raise ValueError("license_class") + if row["license_class"] == "permissive" and row["license_spdx"] not in source.PERMISSIVE: + raise ValueError("license_class") + rng = random.Random(seed) + groups = defaultdict(list) + for row in sorted(rows, key=lambda r: r["repository"]): + groups[row["ecosystem"], row["post_cutoff"]].append(row) + sizes = Counter({1: 0, 2: 0, 3: 0}) + assigned = [] + for key in sorted(groups): + group = groups[key] + rng.shuffle(group) + quotient, remainder = divmod(len(group), 3) + # Give each stratum its floor on all slices, then fill globally least-full slices. + quotas = {i: quotient for i in sizes} + order = list(sizes) + rng.shuffle(order) + order.sort(key=lambda i: sizes[i]) + for i in order[:remainder]: + quotas[i] += 1 + offset = 0 + for slice_id in sorted(quotas): + for row in group[offset : offset + quotas[slice_id]]: + digest = private_digest(row) + assigned.append({**row, "id": "r-" + digest[:24], "slice": slice_id}) + sizes[slice_id] += quotas[slice_id] + offset += quotas[slice_id] + if set(sizes.values()) != {100}: + raise ValueError("slice_size") + return sorted(assigned, key=lambda r: (r["slice"], r["id"])) + + +def toml(value) -> str: + """Inline TOML values keep nested change records structured without a new dependency.""" + if isinstance(value, bool): + return "true" if value else "false" + if isinstance(value, str): + return json.dumps(value, ensure_ascii=False) + if isinstance(value, int): + return str(value) + if isinstance(value, list): + return "[" + ", ".join(toml(v) for v in value) + "]" + if isinstance(value, dict): + return "{" + ", ".join(json.dumps(k) + " = " + toml(v) for k, v in sorted(value.items()) if v is not None) + "}" + raise ValueError("unsupported_manifest_type") + + +def manifest(rows: list[dict], seed: int) -> str: + lines = ['status = "draft"', 'pool = "p1"', f"seed = {seed}"] + for row in rows: + lines.append("\n[[repository]]") + lines.extend(f"{key} = {toml(value)}" for key, value in sorted(row.items()) if value is not None) + return "\n".join(lines) + "\n" + + +def write_draft(root: Path, assigned: list[dict], *, seed: int) -> tuple[Path, Path]: + public_path = root / "benchmarks/unseen/draft-p1.toml" + private_path = root / "benchmarks/unseen/private/draft-p1.toml" + if public_path.exists() or private_path.exists(): + raise FileExistsError("draft_exists") + public, private = [], [] + for row in assigned: + if row["license_class"] == "permissive": + public.append(row) + else: + private.append(row) + public.append({"id": row["id"], "sha256": private_digest(row), "slice": row["slice"], "private": True}) + private_path.parent.mkdir(parents=True, exist_ok=True, mode=0o700) + # Exclusive creates protect drafts awaiting human label review. Private first: never + # publish a pointer whose private manifest failed to write. + with private_path.open("x") as handle: + os.chmod(private_path, 0o600) + handle.write(manifest(private, seed)) + with public_path.open("x") as handle: + handle.write(manifest(public, seed)) + return public_path, private_path + + +class Clones: + """One disposable blobless clone at a time; no name in cache paths or subprocess output.""" + + def __init__(self, root: Path, cache: Path, *, runner=None, limit_bytes=500 * 1024 * 1024): + self.cache = external(root, cache) + self.runner = runner or subprocess.run + self.limit_bytes = limit_bytes + self.counts = Counter() + + @contextmanager + def open(self, url: str): + self.cache.mkdir(parents=True, exist_ok=True) + started = time.monotonic() + with tempfile.TemporaryDirectory(prefix="clone-", dir=self.cache) as directory: + path = Path(directory) + git = Git(path, runner=self.runner, limit_bytes=self.limit_bytes) + try: + # Full commit graph is needed for temporal thirds and OSV brackets; blobs + # are fetched lazily. This is a partial clone, not a truncated history. + done = self.runner( + ["git", "-c", "credential.helper=", "clone", "--quiet", "--filter=blob:none", "--no-checkout", url, str(path)], + stdin=subprocess.DEVNULL, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + check=False, + timeout=180, + env={**os.environ, "GIT_TERMINAL_PROMPT": "0"}, + ) + if done.returncode: + raise InstrumentFailure("clone_failed") + self.counts["clones"] += 1 + if git.disk_bytes() > self.limit_bytes: + raise InstrumentFailure("clone_size_cap") + yield git + finally: + size = git.disk_bytes() + self.counts["clone_bytes"] += size + self.counts["peak_clone_bytes"] = max(self.counts["peak_clone_bytes"], size) + self.counts["source_bytes_read"] += git.bytes_read + self.counts["source_files_read"] += git.files_read + self.counts["clone_wall_seconds"] += round(time.monotonic() - started, 3) + + +def prove_source(git, revision: str) -> dict: + files = git.run("ls-tree", "-r", "--name-only", revision).splitlines() + for path in files: + if Path(path).suffix in extract.LANGUAGES and not extract.generated_path(path): + text = git.blob(revision, path) + if text: + return {"files": 1, "bytes": getattr(git, "last_blob_bytes", len(text.encode()))} + raise InstrumentFailure("zero_source_bytes") + + +def advisory_index(rows: list[dict], github) -> dict[str, set[str]]: + """Include ALL enumerated fixes, even old, unmapped, multi-fix and other ecosystems. + + Canonicalize link names as well, so an old advisory URL survives repository renames. + A 404 is not treated as proof that an advisory is unrelated to a selected repository. + """ + index = defaultdict(set) + for row in rows: + for name, sha in source.fix_links(row): + try: + resolved = eligibility.resolve(name, github) + except eligibility.RepositoryNotFound: + index[name].add(sha) + continue + index[resolved.full_name].add(sha) + return dict(index) + + +def repository_changes(git, candidate, fixes: set[str], *, seed: int): + proof = prove_source(git, candidate.parent) + change = extract.introducing(git, candidate) + known = set() + expanded = set() + for fix in sorted(fixes | {candidate.fix}): + # Explicitly fetch detached advisory commits unavailable on default-branch history. + # All git including this possible network call still goes through the injected client. + git.run("fetch", "--quiet", "--filter=blob:none", "origin", fix) + full = git.run("rev-parse", fix).strip() + expanded.add(full) + parents = git.parents(full) + for parent in parents: + sites = extract.fix_sites(git, full, parent, global_ok=True) + known.update((site["path"], site["function"]) for site in sites) + rows = extract.ordinary_history(git, "HEAD") + ordinary, exclusions = extract.draw_ordinary(rows, expanded, known, seed=seed) + return [change, *ordinary], exclusions, proof + + +def inputs(used, seed, known_empty, clones): + from openultrasast.model.taxonomy import DEFAULT_FAMILIES_PATH + + parameters = { + "seed": seed, + "used_set_digest": digest({label: sorted(names) for label, names in used.sources.items()}), + "known_empty": sorted(known_empty), + "clone_limit_bytes": getattr(clones, "limit_bytes", 500 * 1024 * 1024), + "journal_version": 1, + } + # Hardcoded design thresholds and algorithms are covered as well as taxonomy data. + for name in ("draft", "source", "extract", "eligibility", "journal"): + parameters["design_" + name] = hashlib.sha256(Path(__file__).with_name(name + ".py").read_bytes()).hexdigest() + parameters["design_taxonomy"] = hashlib.sha256(DEFAULT_FAMILIES_PATH.read_bytes()).hexdigest() + return parameters + + +def enumerate_advisories(github, journal): + page, records = 1, {} + while True: + key = str(page) + if key not in journal.rows["advisories"]: + journal.boundary() + result = github.advisory_page(page) + journal.put("advisories", key, result) + result = journal.rows["advisories"][key] + records.update((row["ghsa_id"], row) for row in result["rows"]) + if result["done"]: + break + page += 1 + if not records: + raise InstrumentFailure("empty_advisories") + return list(records.values()) + + +def decide(row, github, osv, used, selected_networks): + try: + links = source.fix_links(row) + if len(links) != 1: + raise source.Rejected("fix_links") + name, _ = next(iter(links)) + resolved = eligibility.resolve(name, github) + rejected_by = eligibility.rejection(resolved, used, set(selected_networks)) + if rejected_by: + raise source.Rejected("used_" + rejected_by) + candidate = source.candidate(row, github, osv) + return {"candidate": asdict(candidate), "network": resolved.network} + except source.Rejected as exc: + return {"rejected": str(exc)} + except eligibility.RepositoryNotFound: + return {"rejected": "repository_or_commit_not_found"} + + +def extract_repository(candidate, github, clones, fixes, seed): + before = dict(clones.counts) + try: + with clones.open(candidate.url) as git: + git.run("fetch", "--quiet", "--filter=blob:none", "origin", candidate.fix) + changes, exclusions, proof = repository_changes(git, candidate, fixes, seed=seed) + pinned = changes[0]["head"] + license_record = github.get_license(candidate.repository, pinned) + spdx, license_class = source.license_info(license_record) + entry = { + **asdict(candidate), + "license_spdx": spdx, + "license_class": license_class, + "license_ref": pinned, + "license_path": license_record.get("path", ""), + "changes": changes, + "instrument": proof, + } + result = {"entry": entry, "ordinary_exclusions": exclusions} + except source.Rejected as exc: + result = {"rejected": str(exc), "ordinary_exclusions": getattr(exc, "counts", {})} + except eligibility.RepositoryNotFound: + result = {"rejected": "repository_or_commit_not_found", "ordinary_exclusions": {}} + result["counts"] = {key: value - before.get(key, 0) for key, value in clones.counts.items()} + return result + + +def publish(journal, assigned, seed): + """Idempotent publication from completed journal data, never inside the checkout.""" + public, private = [], [] + for row in assigned: + if row["license_class"] == "permissive": + public.append(row) + else: + private.append(row) + public.append({"id": row["id"], "sha256": private_digest(row), "slice": row["slice"], "private": True}) + for name, rows in (("private-draft-p1.toml", private), ("draft-p1.toml", public)): + path = journal.path / name + data = manifest(rows, seed).encode() + if path.exists(): + if path.read_bytes() != data: + raise ValueError("draft_exists") + else: + atomic(path, data) + + +def run( + root: Path, + *, + github, + osv, + clones, + used: eligibility.UsedSet, + seed: int, + known_empty=None, + journal_path=None, + max_hours=None, + clock=None, +) -> dict: + started = time.monotonic() + report = {"status": "sourcing", "seed": seed, "candidates": {}, "rejections": {}, "ordinary_exclusions": {}} + counts, rejected, ordinary_counts = Counter(), Counter(), Counter() + checkpoint = None + try: + known_empty = set(known_empty or ()) + used.report(known_empty) + cache = getattr(clones, "cache", root.parent / (root.name + "-cache")) + path = external(root, journal_path or cache / "journal") + checkpoint = Journal( + path, max_hours=max_hours, clock=clock, rate_limit=lambda: getattr(getattr(github, "transport", None), "rate_limit", {}) + ) + with checkpoint.open(inputs(used, seed, known_empty, clones)): + checkpoint.boundary() + api = ResolvedGitHub(github, checkpoint) + used = copy.deepcopy(used) # canonicalization must not mutate the resume input + eligibility.canonicalize_used(used, api) + report["used_set"] = used.report(known_empty) + rows = enumerate_advisories(api, checkpoint) + report["advisories"] = len(rows) + fixes_by_repo = advisory_index(rows, api) + selected_networks, entries = set(), [] + for row in source.ordered_advisories(rows, seed): + checkpoint.boundary() + key = row["ghsa_id"] + counts[source.ecosystem(row)] += 1 + if key not in checkpoint.rows["eligibility"]: + decision = decide(row, api, osv, used, selected_networks) + checkpoint.put("eligibility", key, decision) + decision = checkpoint.rows["eligibility"][key] + if "rejected" in decision: + rejected[decision["rejected"]] += 1 + continue + candidate = source.Candidate(**decision["candidate"]) + # Multiple advisories for one repository can have different fix/range inputs. + extraction_key = digest({"repository": candidate.repository, "advisory": key, "fix": candidate.fix}) + if extraction_key not in checkpoint.rows["extraction"]: + checkpoint.boundary() + result = extract_repository(candidate, api, clones, fixes_by_repo.get(candidate.repository, set()), seed) + checkpoint.put("extraction", extraction_key, result) + result = checkpoint.rows["extraction"][extraction_key] + ordinary_counts.update(result["ordinary_exclusions"]) + if "rejected" in result: + rejected[result["rejected"]] += 1 + continue + selected_networks.add(decision["network"]) + entries.append(result["entry"]) + if len(entries) == 300: + break + report.update( + { + "repositories": len(entries), + "candidates": dict(counts), + "rejections": dict(rejected), + "ordinary_exclusions": dict(ordinary_counts), + } + ) + checkpoint.boundary() + if len(entries) != 300: + report["status"] = "insufficient_repositories" + checkpoint.progress(report["status"]) + return report + assigned = assign_slices(entries, seed=seed) + completion = {"manifest_sha256": digest(assigned), "repositories": len(entries)} + if "all" not in checkpoint.rows["completion"]: + checkpoint.put("completion", "all", completion) + elif checkpoint.rows["completion"]["all"] != completion: + raise ValueError("journal_completion_mismatch") + checkpoint.boundary() + publish(checkpoint, assigned, seed) + report.update( + { + "status": "draft", + "changes": 3900, + "slices": 3, + "license_split": dict(Counter(e["license_class"] for e in entries)), + "methods": dict(Counter(e["changes"][0]["method"] for e in entries)), + "post_cutoff": sum(e["post_cutoff"] for e in entries), + **completion, + } + ) + checkpoint.progress("draft") + return report + except ResumeMismatch as exc: + report.update({"status": "resume_refused", "changed_parameters": exc.parameters}) + return report + except Stopped as exc: + report.update({"status": "stopped", "reason": str(exc)}) + return report + except Exception as exc: + report.update({"status": "instrument_failure", "error_type": type(exc).__name__}) + return report + finally: + report.update({"candidates": dict(counts), "rejections": dict(rejected), "ordinary_exclusions": dict(ordinary_counts)}) + report["instrument"] = dict(clones.counts) + report["wall_seconds"] = round(time.monotonic() - started, 3) + report["github_calls"] = github.calls + report["osv_calls"] = osv.calls + report["github_http_attempts"] = getattr(getattr(github, "transport", None), "attempts", github.calls) + report["osv_http_attempts"] = getattr(getattr(osv, "transport", None), "attempts", osv.calls) + + +def main(argv: list[str] | None = None) -> int: + from openultrasast import config + from openultrasast.plane import memory + from openultrasast.plane.reconciler import results_root + + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--root", type=Path, default=Path.cwd()) + parser.add_argument("--cache", type=Path) + parser.add_argument("--max-hours", type=float, help="Stop between units after this many hours per invocation") + parser.add_argument("--seed", type=int, default=20260601) + parser.add_argument("--local-store") + parser.add_argument("--s3-store", default="s3://") + parser.add_argument("--pointer-manifest", type=Path, action="append", default=[]) + parser.add_argument("--known-empty", choices=(*eligibility.SOURCES, "memory_local", "memory_s3"), action="append", default=[]) + args = parser.parse_args(argv) + try: + config.load_dotenv(args.root / ".env") + # Refuse an existing draft before doing any network work. + if any((args.root / p).exists() for p in ("benchmarks/unseen/draft-p1.toml", "benchmarks/unseen/private/draft-p1.toml")): + raise FileExistsError("draft_exists") + clones = Clones(args.root, args.cache or results_root() / "unseen-draft") + stores = { + "local": memory.open_store(args.local_store or "file://" + str(results_root() / "memory"), read_only=True), + "s3": memory.open_store(args.s3_store, read_only=True), + } + used = eligibility.build_used(args.root, stores=stores, pointer_manifests=args.pointer_manifest) + report = run( + args.root, + github=source.GitHub(os.environ.get("GH_TOKEN", "")), + osv=source.OSV(), + clones=clones, + used=used, + seed=args.seed, + known_empty=set(args.known_empty), + journal_path=clones.cache / "journal", + max_hours=args.max_hours, + ) + except Exception as exc: + report = {"status": "instrument_failure", "error_type": type(exc).__name__} + print(json.dumps(report, sort_keys=True)) + return 0 if report["status"] in {"draft", "stopped"} else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/benchmarks/unseen/eligibility.py b/benchmarks/unseen/eligibility.py new file mode 100644 index 0000000..ec1186f --- /dev/null +++ b/benchmarks/unseen/eligibility.py @@ -0,0 +1,379 @@ +"""Read-only used-set collection. Only counts, source labels and digests leave this module's CLI. + +Run from the repository root: python -m benchmarks.unseen.eligibility --help. +Population v3 is exclusively the existing guard's responsibility. +""" + +from __future__ import annotations + +import argparse +import hashlib +import json +import re +import tomllib +from collections.abc import Iterable, Mapping +from dataclasses import dataclass, field +from pathlib import Path +from urllib.parse import urlsplit + +import yaml + +from openultrasast.plane.memory import MemoryStore + +URL = re.compile(r"github\.com/([\w.-]+/[\w.-]+)", re.I) +NAME = re.compile(r"[\w.-]+/[\w.-]+") +SOURCES = ("pairs", "pointers", "populations", "harvest", "units", "measurements", "sweep") + + +def normalize(value: str) -> str: + text = value.strip().lower().rstrip("/") + if NAME.fullmatch(text): + name = text.removesuffix(".git") + else: + # Structured fields must be identities, not prose containing a URL. + if any(char.isspace() for char in text): + raise ValueError("invalid_repository") + url = urlsplit(text if "://" in text else "https://" + text) + if url.scheme not in {"http", "https", "ssh", "git"} or not url.hostname or "." not in url.hostname: + raise ValueError("invalid_repository") + parts = url.path.strip("/").split("/") + if len(parts) < 2: + raise ValueError("invalid_repository") + name = "/".join(parts[:2]).removesuffix(".git") + if not NAME.fullmatch(name): + raise ValueError("invalid_repository") + if url.hostname != "github.com": + return url.hostname + "/" + name + if not NAME.fullmatch(name) or any(part in {".", ".."} for part in name.split("/")): + raise ValueError("invalid_repository") + return name + + +def excluded(path: Path) -> bool: + return any("v3" in part.lower() or part in {".git", "__pycache__", ".venv"} for part in path.parts) + + +def repositories(value: object, counts: dict[str, int] | None = None) -> set[str]: + """Structured repo fields include bare owner/name pointers; URLs also occur in prose.""" + found = set() + if isinstance(value, dict): + for key, item in value.items(): + if key in {"repo", "fix_repo", "repository", "full_name", "origin"} or (key == "group" and "unit" in value): + try: + if not isinstance(item, str): + raise ValueError("invalid_repository") + found.add(normalize(item)) + except ValueError: + if counts is not None: + counts["skipped_values"] = counts.get("skipped_values", 0) + 1 + else: + found.update(repositories(item, counts)) + elif isinstance(value, list): + for item in value: + found.update(repositories(item, counts)) + elif isinstance(value, str): + found.update(normalize(m[1]) for m in URL.finditer(value)) + return found + + +def pointer_repositories(value: object, counts: dict[str, int] | None = None) -> set[str]: + if isinstance(value, dict): + if value.get("vendored") is False: + return repositories(value, counts) + return set().union(*(pointer_repositories(v, counts) for v in value.values())) + if isinstance(value, list): + return set().union(*(pointer_repositories(v, counts) for v in value)) + return set() + + +def parse(path: Path, raw: bytes) -> object: + structured = path.suffix in {".toml", ".json", ".jsonl", ".yaml", ".yml"} + text = raw.decode("utf-8", errors="strict" if structured else "ignore") + if path.suffix == ".toml": + return tomllib.loads(text) + if path.suffix == ".json": + return json.loads(text) + if path.suffix == ".jsonl": + return [json.loads(line) for line in text.splitlines() if line.strip()] + if path.suffix in {".yaml", ".yml"}: + documents = list(yaml.safe_load_all(text)) + return documents[0] if len(documents) == 1 else documents + return text + + +@dataclass +class UsedSet: + sources: dict[str, set[str]] = field(default_factory=lambda: {label: set() for label in SOURCES}) + instrument: dict[str, dict[str, int]] = field(default_factory=dict) + + def add(self, label: str, names: set[str], size: int, files: int = 1, skipped_values: int = 0) -> None: + self.sources.setdefault(label, set()).update(names) + counts = self.instrument.setdefault(label, {"files": 0, "bytes": 0, "rows": 0, "skipped_values": 0}) + counts["files"] += files + counts["bytes"] += size + counts["rows"] += len(names) + counts["skipped_values"] += skipped_values + + def report(self, known_empty: set[str] | None = None) -> dict: + known_empty = known_empty or set() + if known_empty - self.sources.keys(): + raise ValueError("unknown_source_label") + records = {} + for label, names in self.sources.items(): + counts = self.instrument.get(label, {"files": 0, "bytes": 0, "rows": 0, "skipped_values": 0}) + if not counts["rows"] and label not in known_empty: + raise ValueError("empty_source:" + label) + records[label] = {**counts, "count": len(names), "known_empty": label in known_empty} + names = set().union(*self.sources.values()) + return {"sources": records, "count": len(names), "sha256": hashlib.sha256("\n".join(sorted(names)).encode()).hexdigest()} + + +def build_used(root: Path, *, stores: Mapping[str, MemoryStore] | None = None, pointer_manifests: Iterable[Path] = ()) -> UsedSet: + used = UsedSet() + + def read(label: str, paths: Iterable[Path], structured: bool = True) -> list[object]: + values = [] + for path in sorted(set(paths)): + if excluded(path): + continue + raw = path.read_bytes() # failure is an instrument failure, never a quiet empty source + value = parse(path, raw) if structured else raw.decode("utf-8", errors="ignore") + counts: dict[str, int] = {} + names = repositories(value, counts) + used.add(label, names, len(raw), skipped_values=counts.get("skipped_values", 0)) + if label == "pairs": + pointer_counts: dict[str, int] = {} + pointers = pointer_repositories(value, pointer_counts) + if pointers or pointer_counts: + used.add("pointers", pointers, len(raw), skipped_values=pointer_counts.get("skipped_values", 0)) + values.append(value) + return values + + pairs = root / "benchmarks/pairs" + read("pairs", (p for p in pairs.rglob("*") if p.is_file() and p.suffix in {".toml", ".json", ".jsonl", ".csv", ".yaml", ".yml"})) + read("pointers", pointer_manifests) + read("populations", (p for p in (root / "benchmarks/independent").glob("population-v[12].toml"))) + read("harvest", (p for directory in pairs.glob("advisory-fixes*") for p in directory.rglob("*.toml"))) + units = set((root / "plane/experiments").glob("*.units.jsonl")) + for path in (root / "plane/experiments").glob("*.y*ml"): + if excluded(path): + continue + raw = path.read_bytes() + manifest = parse(path, raw) + counts = {} + names = repositories(manifest, counts) + used.add("units", names, len(raw), skipped_values=counts.get("skipped_values", 0)) + if isinstance(manifest, dict) and manifest.get("units_file"): + declared = root / Path(manifest["units_file"]).expanduser() + if declared.exists(): # an unregistered plan may not have frozen units yet + units.add(declared) + for label, store in (stores or {}).items(): + if label not in {"local", "s3"}: + raise ValueError("unknown_store_label") + source = "memory_" + label + used.add(source, set(), 0, 0) + # Read actual objects, not a serialized row size masquerading as bytes read. + for key in store._keys("repos/"): + if not key.endswith(".jsonl") or excluded(Path(key)): + continue + result = store._get(key) + if result is None: + raise ValueError("missing_memory_object") + raw = result[0] + rows = parse(Path("rows.jsonl"), raw) + names = set() + counts = {} + for row in rows: + if row.get("kind") == "example" and "repo" in row: + names.update(repositories({"repo": row["repo"]}, counts)) + if row.get("kind") == "experiment" and row.get("units_file"): + registered = root / Path(row["units_file"]).expanduser() + if row.get("units_digest") or registered.exists(): + units.add(registered) + used.add(source, names, len(raw), skipped_values=counts.get("skipped_values", 0)) + read("units", units) + read("measurements", (p for top in ("measurements", "experiments") for p in (root / "benchmarks" / top).rglob("*") if p.is_file())) + read( + "sweep", + ( + p + for top in ("benchmarks", "src", "tests", "plane", ".kiro") + for p in (root / top).rglob("*") + if p.is_file() and not p.is_relative_to(root / "benchmarks/unseen") and not excluded(p) + ), + structured=False, + ) + return used + + +@dataclass(frozen=True) +class Resolution: + full_name: str + names: frozenset[str] + network: str + + +def resolve(candidate: str, api) -> Resolution: + name = normalize(candidate) + if name.count("/") > 1: + return Resolution(name, frozenset({name}), name) + data = api.get_repo(name) + canonical = normalize(data["full_name"]) + names = {name, canonical, *(normalize(n) for n in data.get("former_names", []))} + for key in ("parent", "source"): + if data.get(key): + names.add(normalize(data[key]["full_name"])) + network = normalize((data.get("source") or data.get("parent") or data)["full_name"]) + return Resolution(canonical, frozenset(names), network) + + +def canonicalize_used(used: UsedSet, api) -> None: + """Resolve old used names too: GitHub exposes redirects, not a complete rename history.""" + cache = {} + for names in used.sources.values(): + for name in sorted(names): + if name not in cache: + try: + cache[name] = resolve(name, api) + except RepositoryNotFound: + cache[name] = Resolution(name, frozenset({name}), name) + names.update(cache[name].names) + + +def rejection(candidate: Resolution, used: UsedSet, selected: set[str]) -> str | None: + for label, names in used.sources.items(): + if candidate.names & names: + return label + if candidate.network in selected: + return "fork_network" + selected.add(candidate.network) + return None + + +class RepositoryNotFound(ValueError): + """A historical or synthetic used identity no longer resolves; retain it verbatim.""" + + +class GitHub: + """Coordinator-only GET client. Exceptions never disclose request URLs or tokens.""" + + def __init__(self, token: str): + if not token: + raise ValueError("missing_gh_token") + self.token = token + + def get_repo(self, name: str) -> dict: + from urllib.error import HTTPError + from urllib.request import Request, urlopen + + request = Request( + "https://api.github.com/repos/" + normalize(name), + headers={ + "Authorization": "Bearer " + self.token, + "Accept": "application/vnd.github+json", + }, + ) + try: + with urlopen(request, timeout=30) as response: + return json.load(response) + except HTTPError as exc: + if exc.code == 404: + raise RepositoryNotFound("github_not_found") from None + raise ValueError("github_resolution_failed") from None + except Exception: + raise ValueError("github_resolution_failed") from None + + +def verify_tests(root: Path) -> dict: + """The coordinator runs the guard as a black box; only exit codes and counts escape.""" + import subprocess + import sys + import tempfile + from xml.etree import ElementTree + + with tempfile.TemporaryDirectory() as directory: + report = Path(directory) / "tests.xml" + done = subprocess.run( + [ + sys.executable, + "-m", + "pytest", + "-q", + "-p", + "no:cacheprovider", + "tests/test_unseen_eligibility.py", + "tests/test_unseen_ledger.py", + "tests/test_independent_population.py", + "--junitxml=" + str(report), + ], + cwd=root, + stdin=subprocess.DEVNULL, + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + check=False, + ) + if done.returncode != 0: + raise ValueError("guard_or_tests_failed") + suites = ElementTree.parse(report).getroot().iter("testsuite") + counts = {key: 0 for key in ("tests", "failures", "errors", "skipped")} + for suite in suites: + for key in counts: + counts[key] += int(suite.attrib[key]) + if not counts["tests"]: + raise ValueError("empty_test_run") + return {"guard": "green", "test_exit_code": done.returncode, "tests": counts} + + +def main(argv: list[str] | None = None) -> int: + import os + + from openultrasast import config + from openultrasast.plane import memory + from openultrasast.plane.reconciler import results_root + + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--root", type=Path, default=Path.cwd()) + parser.add_argument("--local-store", default=None) + parser.add_argument("--s3-store", default="s3://") + parser.add_argument("--pointer-manifest", type=Path, action="append", default=[]) + parser.add_argument("--known-empty", choices=(*SOURCES, "memory_local", "memory_s3"), action="append", default=[]) + parser.add_argument("--output", type=Path) + parser.add_argument("--verify-tests", action="store_true") + args = parser.parse_args(argv) + step = "open_stores" + try: + config.load_dotenv(args.root / ".env") + stores = { + "local": memory.open_store(args.local_store or "file://" + str(results_root() / "memory"), read_only=True), + "s3": memory.open_store(args.s3_store, read_only=True), + } + step = "build_used" + used = build_used(args.root, stores=stores, pointer_manifests=args.pointer_manifest) + step = "report" + used.report(set(args.known_empty)) # prove inputs before API work + step = "canonicalize" + canonicalize_used(used, GitHub(os.environ.get("GH_TOKEN", ""))) + step = "report" + report = used.report(set(args.known_empty)) + if args.verify_tests: + step = "verify_tests" + report.update(verify_tests(args.root)) + step = "report" + text = json.dumps(report, sort_keys=True, indent=2) + "\n" + if args.output: + args.output.parent.mkdir(parents=True, exist_ok=True) + with args.output.open("x") as handle: + handle.write(text) + print(text, end="") + return 0 + except Exception as exc: + # Source paths, store exceptions and JSON parser errors can contain repository names. + report = {"status": "instrument_failure", "step": step, "type": type(exc).__name__} + if str(exc) in {"empty_source:" + label for label in (*SOURCES, "memory_local", "memory_s3")}: + report["source"] = str(exc).partition(":")[2] + print(json.dumps(report)) + return 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/benchmarks/unseen/extract.py b/benchmarks/unseen/extract.py new file mode 100644 index 0000000..0e7a5a5 --- /dev/null +++ b/benchmarks/unseen/extract.py @@ -0,0 +1,474 @@ +"""Git-only change extraction; clients and subprocess execution are injectable. + +No scanner, engine or model runs here. Function boundaries are syntax, not a security +judgement: the first removed code line of a fix is the SZZ sink proxy. Ambiguous sites +are dropped, and the recorded SZZ bias remains subject to task 2.6's label check. +""" + +from __future__ import annotations + +import ast +import json +import re +import subprocess +from pathlib import Path + +from .source import Candidate, Rejected + +LANGUAGES = {".py", ".js", ".jsx", ".mjs", ".cjs", ".ts", ".tsx", ".php", ".java"} +HUNK = re.compile(r"^@@ -(\d+)(?:,(\d+))? \+(\d+)(?:,(\d+))? @@", re.M) +SHA = re.compile(r"[a-f0-9]{40}") +BIAS = { + "introducing": "SZZ last-writer proxy: refactors, moves and reformats can misidentify the introduction in either direction.", + "last_touch": "Pessimistic lower bound: the flaw may already exist in base, so a delta hook can correctly stay quiet.", +} + + +class InstrumentFailure(ValueError): + """A failed read or command must never masquerade as an empty history.""" + + +class Git: + def __init__(self, path: Path, *, runner=None, limit_bytes: int | None = None): + self.path = path + self.runner = runner or subprocess.run + self.bytes_read = 0 + self.files_read = 0 + self.last_blob_bytes = 0 + self.limit_bytes = limit_bytes + + def run(self, *args: str, allowed=(0,)) -> str: + try: + done = self.runner( + ["git", "-c", "core.quotePath=false", "-C", str(self.path), *args], + stdin=subprocess.DEVNULL, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + timeout=180, + check=False, + ) + except Exception: + raise InstrumentFailure("git_launch_failed") from None + if done.returncode not in allowed: + raise InstrumentFailure("git_exit_" + str(done.returncode)) + if self.limit_bytes is not None and self.disk_bytes() > self.limit_bytes: + raise InstrumentFailure("clone_size_cap") + self.last_blob_bytes = len(done.stdout) + return done.stdout.decode("utf-8", errors="replace") + + def disk_bytes(self) -> int: + return sum(p.stat().st_size for p in self.path.rglob("*") if p.is_file() and not p.is_symlink()) + + def parents(self, sha: str) -> list[str]: + return self.run("rev-list", "--parents", "-n", "1", sha).split()[1:] + + def blob(self, sha: str, path: str) -> str | None: + # Missing is established by a successful tree read, never by swallowing a show error. + if not self.run("ls-tree", sha, "--", path).strip(): + return None + text = self.run("show", f"{sha}:{path}") + if Path(path).suffix in LANGUAGES: + self.files_read += 1 + self.bytes_read += self.last_blob_bytes + return text + + def stats(self, base: str, head: str) -> list[tuple[str, int, int]]: + rows = [] + for line in self.run("diff", "--no-renames", "--numstat", base, head, "--").splitlines(): + added, removed, path = line.split("\t", 2) + rows.append((path, int(added) if added.isdigit() else 0, int(removed) if removed.isdigit() else 0)) + return rows + + def hunks(self, base: str, head: str, path: str) -> list[tuple[int, int, int, int]]: + diff = self.run("diff", "--no-ext-diff", "--no-renames", "-U0", base, head, "--", path) + return [(int(m[1]), int(m[2] or 1), int(m[3]), int(m[4] or 1)) for m in HUNK.finditer(diff)] + + def ancestor(self, older: str, newer: str) -> bool: + # rev-list avoids interpreting a command failure as a negative ancestry result. + return not self.run("rev-list", "--max-count=1", older, "--not", newer).strip() + + def blame(self, revision: str, path: str, line: int) -> tuple[str, str, int]: + text = self.run("blame", "-w", "-M", "-C", "--porcelain", f"-L{line},{line}", revision, "--", path) + header = text.splitlines()[0].split() + if len(header) < 3 or not SHA.fullmatch(header[0]): + raise InstrumentFailure("invalid_blame") + filename = next((row[9:] for row in text.splitlines() if row.startswith("filename ")), path) + if filename.startswith('"'): + filename = json.loads(filename) + return header[0], filename, int(header[1]) + + +def functions(path: str, text: str) -> list[tuple[str, int, int]]: + """Conservative syntax boundaries. No scanner import; unresolved sites are rejected.""" + if Path(path).suffix == ".py": + try: + tree = ast.parse(text) + except SyntaxError: + raise Rejected("function_parse") from None + return [ + (n.name, n.lineno, n.end_lineno or n.lineno) for n in ast.walk(tree) if isinstance(n, ast.FunctionDef | ast.AsyncFunctionDef) + ] + # Blank comments and quoted strings before brace matching. Keep newlines/offsets. + masked = re.sub( + r'/\*.*?\*/|//[^\n]*|\#[^\n]*|"(?:\\.|[^"\\])*"|\'(?:\\.|[^\'\\])*\'|`(?:\\.|[^`\\])*`', + lambda m: "".join("\n" if c == "\n" else " " for c in m[0]), + text, + flags=re.S, + ) + declaration = re.compile( + r"\bfunction\s*&?\s*(\w+)\s*\([^)]*\)\s*(?::[^\n{]+)?\s*\{|" + r"\b(?:const|let|var)\s+(\w+)\s*=\s*(?:async\s*)?(?:\([^)]*\)|\w+)\s*=>\s*\{|" + r"\b([\w$]+)\s*\([^;{}]*\)\s*(?:throws\s+[\w., ]+)?\s*\{" + ) + found = [] + for match in declaration.finditer(masked): + name = next(g for g in match.groups() if g) + if name in {"if", "while", "for", "switch", "catch", "with", "synchronized"}: + continue + depth = 1 + end = match.end() + while end < len(masked) and depth: + depth += (masked[end] == "{") - (masked[end] == "}") + end += 1 + if depth == 0: + found.append((name, masked.count("\n", 0, match.start()) + 1, masked.count("\n", 0, end) + 1)) + return found + + +def enclosing(path: str, text: str, line: int, *, global_ok=False) -> tuple[str, int, int] | None: + spans = [s for s in functions(path, text) if s[1] <= line <= s[2]] + if spans: + return min(spans, key=lambda s: s[2] - s[1]) + return ("", 1, max(1, len(text.splitlines()))) if global_ok else None + + +def normalized(text: str) -> str: + return re.sub(r"\s+", "", text) + + +def site_at(git, revision: str, path: str, name: str): + text = git.blob(revision, path) + if text is None: + return None + spans = [("", 1, max(1, len(text.splitlines())))] if name == "" else [s for s in functions(path, text) if s[0] == name] + if len(spans) != 1: + return None + return text, spans[0] + + +def fix_sites(git, fix: str, parent: str, *, global_ok=False) -> list[dict]: + sites = [] + for path, _, _ in git.stats(parent, fix): + if Path(path).suffix not in LANGUAGES: + continue + text = git.blob(parent, path) + if text is None: + continue + lines = text.splitlines() + for start, count, _, _ in git.hunks(parent, fix, path): + for line in range(max(1, start), max(1, start) + max(1, count)): + if line > len(lines): + continue + span = enclosing(path, text, line, global_ok=global_ok) + if span is None: + continue + sink_line = line if count else span[1] + sink = lines[sink_line - 1] + if not sink.strip() or sink.lstrip().startswith(("#", "//", "/*", "*")): + continue + key = (path, span[0]) + if any((s["path"], s["function"]) == key for s in sites): + continue + sites.append({"path": path, "function": span[0], "line": sink_line, "text": sink, "absence": count == 0, "span": span}) + return sites + + +def version_bracket(git, candidate: Candidate, revision: str) -> bool: + if not candidate.first_affected: + return True + tags = git.run("tag", "--sort=version:refname").splitlines() + matching = [t for t in tags if t.removeprefix("v") == candidate.first_affected] + if len(matching) != 1: + return False + affected = matching[0] + earlier = [t for t in tags[: tags.index(affected)] if git.ancestor(t, affected)] + return bool(earlier) and git.ancestor(revision, affected) and not git.ancestor(revision, earlier[-1]) + + +def change_files(git, base: str, head: str) -> list[dict]: + return [ + {"path": path, "head_lines": [[max(1, new), max(1, new + max(count, 1) - 1)] for _, _, new, count in git.hunks(base, head, path)]} + for path, _, _ in git.stats(base, head) + ] + + +def introducing(git, candidate: Candidate) -> dict: + from collections import Counter + + if git.parents(candidate.fix) != [candidate.parent]: + raise Rejected("fix_parents") + sites = fix_sites(git, candidate.fix, candidate.parent, global_ok=candidate.family == "config_secrets") + reasons = Counter() + + def record(method, base, head, path, name): + resolved = site_at(git, head, path, name) + if resolved is None: + return None + return { + "kind": "vulnerability", + "base": base, + "head": head, + "method": method, + "family": candidate.family, + "function": name, + "path": path, + "function_lines": list(resolved[1][1:]), + "files": change_files(git, base, head), + "post_cutoff": candidate.post_cutoff, + "advisory": candidate.advisory, + "bias": BIAS[method], + "fallback_reasons": dict(reasons), + } + + for site in sites: + revision, path, line = candidate.parent, site["path"], site["line"] + for _ in range(6): # initial blame plus at most five walk-back steps + blamed, blamed_path, blamed_line = git.blame(revision, path, line) + parents = git.parents(blamed) + if len(parents) != 1: + reasons["merge_or_root"] += 1 + break + base = parents[0] + head_site = site_at(git, blamed, blamed_path, site["function"]) + if head_site is None: + reasons["unresolved_function"] += 1 + break + parent_text = git.blob(base, blamed_path) + if parent_text is not None and site["function"] != "": + matching = [span for span in functions(blamed_path, parent_text) if span[0] == site["function"]] + if len(matching) > 1: + reasons["ambiguous_parent"] += 1 + break + old_site = site_at(git, base, blamed_path, site["function"]) + occurrences = [] + if old_site: + text, (_, start, end) = old_site + occurrences = [ + i for i, value in enumerate(text.splitlines(), 1) if start <= i <= end and normalized(value) == normalized(site["text"]) + ] + if occurrences: + reasons["text_present"] += 1 + revision, path, line = base, blamed_path, occurrences[0] + continue + stats = git.stats(base, blamed) + if len(stats) > 50 or sum(a + d for _, a, d in stats) > 2000: + reasons["size_cap"] += 1 + break + if not version_bracket(git, candidate, blamed): + reasons["version_bracket"] += 1 + break + result = record("introducing", base, blamed, blamed_path, site["function"]) + if result: + return result + else: + reasons["walkback_limit"] += 1 + for site in sites: + # Numeric function extent implements log -L for all five languages, including + # languages for which Git has no built-in :function: driver. + _, start, end = site["span"] + history = git.run("log", "--format=%H", "--no-patch", f"-L{start},{end}:{site['path']}", candidate.parent).splitlines() + if not history: + continue + latest = history[0] + parents = git.parents(latest) + if len(parents) != 1: + continue + result = record("last_touch", parents[0], candidate.parent, site["path"], site["function"]) + if result: + return result + raise Rejected("no_range") + + +SECURITY = re.compile(r"secur|cve|ghsa|vuln|inject|xss|csrf|ssrf|exploit|sanitiz|escap", re.I) +DEPENDENCY = re.compile(r"\bbump\b|\bupgrade .* from\b", re.I) +MANIFESTS = { + "package.json", + "package-lock.json", + "yarn.lock", + "pnpm-lock.yaml", + "composer.json", + "composer.lock", + "pom.xml", + "build.gradle", + "build.gradle.kts", + "gradle.lockfile", + "pyproject.toml", + "setup.py", + "setup.cfg", + "Pipfile", + "Pipfile.lock", + "poetry.lock", + "uv.lock", +} + + +def generated_path(path: str) -> bool: + return bool({"vendor", "node_modules", "dist"} & set(Path(path).parts)) or path.endswith(".min.js") + + +def manifest(path: str) -> bool: + name = Path(path).name + return name in MANIFESTS or name.endswith(".lock") or bool(re.fullmatch(r"requirements[^/]*\.txt", name)) + + +def ordinary_rejection(row: dict, fixes: set[str], known_functions: set[tuple[str, str]]) -> str | None: + if len(row["parents"]) != 1: + return "merge_or_root" + if row["head"] in fixes: + return "advisory_fix" + if SECURITY.search(row["message"]): + return "security" + if re.search(r"^Revert|This reverts commit", row["message"], re.I | re.M): + return "revert" + if re.search(r"dependabot|renovate|\[bot\]", row["author"], re.I) or DEPENDENCY.search(row["message"]): + return "dependency" + files = row["files"] + if files and all(manifest(f["path"]) for f in files): + return "manifest_only" + source = [f for f in files if Path(f["path"]).suffix in LANGUAGES and not manifest(f["path"])] + if not source: + return "no_source" + if all(f["generated"] or generated_path(f["path"]) for f in source): + return "generated" + if any((f["path"], name) in known_functions for f in files for name in f["functions"]): + return "vulnerable_function" + if any((f["path"], "") in known_functions for f in files): + return "vulnerable_function" + if sum(f["added"] + f["removed"] for f in source) > 1000: + return "size_cap" + return None + + +def generated(git, revision: str, path: str, text: str) -> bool: + import fnmatch + + if generated_path(path): + return True + header = "\n".join(text.splitlines()[:20]) + if re.search(r"(?im)^\s*(?:#|//|/\*|\*|