From 86a4efc85ecd0ceca8b29114b35154d216178668 Mon Sep 17 00:00:00 2001 From: BaaRai <130491490+BaaRai@users.noreply.github.com> Date: Mon, 12 Oct 2026 07:01:36 +0200 Subject: [PATCH] feat: L3 benchmark harness (whole-app corpus, independent judge) Model benchmarks measure recognition on snippets. Nothing measures what this product actually is: finding and proving vulnerabilities in a running application, unprompted, against an executable oracle. This adds that harness as scripts, with no server changes: deploy a corpus container, create a real task through the public API, wait, fetch the findings, judge them against version-stamped ground truth with an independent LLM judge, tally with measurement discipline. scripts/benchmark/run_bench.py plus ground_truth/juice-shop.json (8 stable entries: SQLi login bypass, admin section access, basket IDOR, /ftp exposure, forged feedback, error disclosure, token tampering, XSS) and a README documenting the standing rules: - multi-gate scoring (component 1 / class 2 / keywords 1 / proof 1), partial credit, gate pattern over aggregate - findings beyond ground truth are discovery candidates; precision is reported as bounds until verified - the judge must not be the model under test (refuses to start, override knowingly), and must pass a falsification battery before grading - invalid findings (errored or empty) are excluded from every rate and named, never counted as misses - every rate carries its denominator, the aggregate a Wilson 95% interval; results are stamped with method + ground-truth versions The task goal bakes in the standing quality gates, so the benchmark measures the product as it behaves, prompt rules included. Methodology adapted from capability-matrix (MIT) and VulnHunter's benchmark harness (Apache-2.0). --- .gitignore | 3 + scripts/benchmark/README.md | 94 ++++ .../benchmark/ground_truth/juice-shop.json | 91 ++++ scripts/benchmark/run_bench.py | 422 ++++++++++++++++++ 4 files changed, 610 insertions(+) create mode 100755 scripts/benchmark/README.md create mode 100755 scripts/benchmark/ground_truth/juice-shop.json create mode 100755 scripts/benchmark/run_bench.py diff --git a/.gitignore b/.gitignore index 3689e17..88a29c8 100644 --- a/.gitignore +++ b/.gitignore @@ -58,3 +58,6 @@ Thumbs.db /bench/ /Dockerfile.bench /Dockerfile.bench.v2backups/ + +# benchmark runner output (one dir per run) +benchmark-results/ diff --git a/scripts/benchmark/README.md b/scripts/benchmark/README.md new file mode 100755 index 0000000..30f78de --- /dev/null +++ b/scripts/benchmark/README.md @@ -0,0 +1,94 @@ +# ARTEX L3 benchmark + +Measures whether the product can find and prove real vulnerabilities in a real +application, unprompted, against an executable oracle. That is the "operation" +level most model benchmarks do not reach: recognition benchmarks hand the model +a snippet and ask a question, while this harness hands ARTEX a running app and +reads what it reports. + +Method discipline adapted from capability-matrix (MIT) and VulnHunter's +benchmark harness (Apache-2.0). + +## Layout + +| | | +|---|---| +| `run_bench.py` | The runner: deploy corpus, create task, wait, fetch findings, judge, tally | +| `ground_truth/.json` | Version-stamped ground truth (component + class + mechanism, deliberately coarse) | +| `benchmark-results/` | Output (gitignored): one directory per run with `report.md` + `results.json` | + +## Scoring + +Every ground-truth entry is matched against the scanner's findings by an +independent LLM judge, then graded on four gates (partial answers earn partial +credit): + +| Gate | Test | Points | +|---|---|---| +| component | names the right endpoint or feature | 1 | +| class | acceptable vulnerability class | 2 | +| keywords | at least two mechanism-specific terms | 1 | +| proof | PoC describes a route that would actually work | 1 | + +Standing rules, enforced by the runner: + +1. **Multi-gate scoring, never pass/fail.** The gate pattern is the work item; + the aggregate percentage is the least useful output. +2. **Discovery candidates.** A scanner finding beyond ground truth is not a + false positive until verified; precision is reported as bounds meanwhile. +3. **The judge must not be the model under test.** The runner refuses to start + when `--judge-model` equals `--subject-model` (override knowingly with + `--allow-self-judge`). +4. **The judge must pass a falsification battery** before grading: a correct + match, a wrong-bug match, and a vague non-answer. A grader that cannot fail + the wrong ones does not run. +5. **Invalid never counts.** Findings with no summary and no report are + excluded from every rate and named in the result. +6. **Every rate carries its denominator; the aggregate carries a Wilson 95% + interval.** With 8 ground-truth entries, one run cannot separate close + models; treat per-gate rates as observations and replicate before quoting + differences. +7. **Results are stamped** with the method version, ground-truth version, + subject model, judge model, and terminal task status. A result measured + under an earlier version is not wrong; it is a result at that version. + +## Usage + +```bash +cd scripts/benchmark + +# against a locally deployed ARTEX, with an independent judge endpoint +ARTEX_TOKEN= \ +JUDGE_BASE_URL=https://api.example.com JUDGE_MODEL= JUDGE_API_KEY=... \ +python3 run_bench.py --corpus juice-shop --subject-model glm-5.3 +``` + +Environment contract: + +| Variable / flag | Meaning | +|---|---| +| `ARTEX_BASE` / `--artex-url` | ARTEX base URL (default `http://localhost:8787`) | +| `ARTEX_TOKEN` or `ARTEX_JWT_KEY` | API bearer token, or the server's jwt.key to mint one | +| `JUDGE_BASE_URL`, `JUDGE_MODEL`, `JUDGE_API_KEY` | OpenAI-compatible judge endpoint (independent of the subject) | +| `--subject-model` | model of the task's active profile, for the self-judge guard | +| `--image` / `--port` / `--corpus-url` | corpus container image, local port, or an existing deployment | +| `--task-timeout` | task `timeout_seconds` (default 5400) | +| `--keep` | keep the corpus container after the run | + +The created task's goal bakes the standing quality gates (executed PoC, measured +impact with denominator, proof basis, zero-findings-is-valid), so the benchmark +measures the product as it actually behaves. + +## Adding a corpus + +Add `ground_truth/.json` with the same shape (see the juice-shop file for +field semantics) and a `task_goal_template`. Pin the image tag in the file's +`version` note. Entries stay coarse on purpose: component + class + mechanism. +Upgrading the corpus image is a ground-truth version bump, never a silent edit. + +## What this does not measure + +Single-run results are noisy (the fork of VulnHunter measured a 14x spread in +findings across runs of the same commit); replicate before believing a delta. +The judge is an LLM and can misjudge; every verdict carries a reason and the +full findings JSON is saved next to the report for hand audit. diff --git a/scripts/benchmark/ground_truth/juice-shop.json b/scripts/benchmark/ground_truth/juice-shop.json new file mode 100755 index 0000000..7dff422 --- /dev/null +++ b/scripts/benchmark/ground_truth/juice-shop.json @@ -0,0 +1,91 @@ +{ + "corpus": "owasp-juice-shop", + "version": "v1 (targets the standard juice-shop image; entries are deliberately coarse: component + class + mechanism, no line numbers)", + "task_goal_template": "Map and hunt the highest-signal security weaknesses of the deliberately vulnerable training application at {url}: authentication bypass, injection, broken access control (IDOR, admin surfaces), sensitive data exposure, XSS. Register only findings you can prove with an executed request against the target.", + "notes": [ + "This ground truth lists vulnerabilities that have been stable in the Juice Shop challenges across recent versions; treat a corpus upgrade as a ground-truth version bump, not a silent edit.", + "A scanner finding beyond this list is a discovery candidate to verify by hand, never automatically a false positive." + ], + "findings": [ + { + "id": "GT-001", + "title": "SQL injection authentication bypass on the login form", + "cwe": "CWE-89", + "classes": ["SQL injection", "authentication bypass", "injection"], + "components": ["login form", "login page", "/rest/user/login", "sign in"], + "mechanism_keywords": ["' OR 1=1", "SQL", "quote", "authentication bypass", "admin session", "email field"], + "proof_hint": "submitting a tautology payload such as ' OR 1=1-- in the email field authenticates as an administrator", + "severity": "critical" + }, + { + "id": "GT-002", + "title": "Administration section reachable without administrator rights", + "cwe": "CWE-284", + "classes": ["broken access control", "improper authorization", "privilege escalation"], + "components": ["administration page", "/administration", "admin section", "navbar admin link"], + "mechanism_keywords": ["administration", "admin", "unauthorized access", "role", "privilege"], + "proof_hint": "the /administration route is served to a non-administrator session (or becomes reachable after a role/privilege manipulation)", + "severity": "high" + }, + { + "id": "GT-003", + "title": "IDOR: reading another user's shopping basket", + "cwe": "CWE-639", + "classes": ["IDOR", "broken access control", "insecure direct object reference"], + "components": ["basket", "cart", "/rest/basket"], + "mechanism_keywords": ["basket id", "another user", "object reference", "unauthorized read"], + "proof_hint": "requesting /rest/basket/ of another user as a different (non-admin) account returns that basket's contents", + "severity": "high" + }, + { + "id": "GT-004", + "title": "Sensitive files exposed under the /ftp directory", + "cwe": "CWE-538", + "classes": ["sensitive data exposure", "information disclosure", "misconfiguration"], + "components": ["/ftp", "ftp directory", "directory listing"], + "mechanism_keywords": ["ftp", "file disclosure", "acquisitions", "directory traversal of static files", "sensitive document"], + "proof_hint": "the /ftp path serves documents (for example acquisitions or incident files) to unauthenticated users", + "severity": "medium" + }, + { + "id": "GT-005", + "title": "Feedback posted as another user (forged author identity)", + "cwe": "CWE-284", + "classes": ["broken access control", "identity spoofing", "mass assignment"], + "components": ["feedback", "contact form", "/api/Feedbacks"], + "mechanism_keywords": ["feedback", "author", "user id", "another user", "forged"], + "proof_hint": "the feedback endpoint accepts a caller-chosen user identity so one account can leave feedback attributed to another", + "severity": "medium" + }, + { + "id": "GT-006", + "title": "Unhandled error responses disclosing stack traces", + "cwe": "CWE-209", + "classes": ["information exposure through error message", "information disclosure", "error handling"], + "components": ["error response", "server error", "stack trace", "checkout", "product search", "tracking"], + "mechanism_keywords": ["stack trace", "error message", "internal path", "exception", "500"], + "proof_hint": "a malformed input (for example at checkout, order tracking, or search) produces a server error body containing a stack trace or internal details", + "severity": "low" + }, + { + "id": "GT-007", + "title": "Client-side privilege escalation via token tampering (premium content)", + "cwe": "CWE-345", + "classes": ["broken authentication", "JWT tampering", "client-side enforcement"], + "components": ["premium", "paywall", "token", "JWT", "account page"], + "mechanism_keywords": ["token", "role", "premium", "tamper", "client side"], + "proof_hint": "modifying the session token (for example its role claim) unlocks premium content without paying", + "severity": "medium" + }, + { + "id": "GT-008", + "title": "Cross-site scripting through user-controlled content", + "cwe": "CWE-79", + "classes": ["XSS", "cross-site scripting", "injection"], + "components": ["search", "feedback", "product reviews", "address book", "profile"], + "mechanism_keywords": ["script", "XSS", "payload", "renders", "unescaped", "HTML injection"], + "proof_hint": "user-controlled text (search, review, or profile fields) executes as script in the browser context", + "severity": "high" + } + ] +} diff --git a/scripts/benchmark/run_bench.py b/scripts/benchmark/run_bench.py new file mode 100755 index 0000000..ea37a2e --- /dev/null +++ b/scripts/benchmark/run_bench.py @@ -0,0 +1,422 @@ +#!/usr/bin/env python3 +"""ARTEX L3 benchmark runner. + +Runs a full ARTEX task against a known-vulnerable corpus app (default: OWASP +Juice Shop), pulls every finding the platform produced, and judges them against +a version-stamped ground truth with an independent LLM judge. + +Method discipline (adapted from capability-matrix, MIT, and VulnHunter's +benchmark harness, Apache-2.0): +- multi-gate scoring, never pass/fail: component (1), class (2), mechanism + keywords (1), proof (1) -> partial answers earn partial credit +- findings beyond ground truth are DISCOVERY CANDIDATES, not false positives; + precision is reported as bounds until candidates are verified +- the judge must not be the model under test; the judge must pass a + falsification battery before it is allowed to grade +- invalid (errored/empty) findings never count as misses or passes +- every rate carries its denominator; the aggregate detection rate carries a + Wilson 95% interval; per-gate rates are observations +- results are stamped with METHOD_VERSION and the ground-truth version + +Usage: + ARTEX_TOKEN=... JUDGE_API_KEY=... python3 run_bench.py \ + --corpus juice-shop --subject-model glm-5.3 --judge-model deepseek-chat +See README.md for the full contract. +""" +from __future__ import annotations + +import argparse +import base64 +import hashlib +import hmac +import json +import math +import os +import pathlib +import subprocess +import sys +import time +import urllib.error +import urllib.request + +METHOD_VERSION = "bench-v1" +TERMINAL_STATUSES = {"done", "timeout", "failed", "stopped", "archived", "canceled"} + + +# ---------------------------------------------------------------- api helpers + +class Api: + def __init__(self, base: str, token: str): + self.base = base.rstrip("/") + self.token = token + + def req(self, method: str, path: str, body: dict | None = None) -> dict: + data = json.dumps(body).encode() if body is not None else None + r = urllib.request.Request(self.base + path, data=data, method=method) + r.add_header("Authorization", "Bearer " + self.token) + if data: + r.add_header("Content-Type", "application/json") + try: + with urllib.request.urlopen(r, timeout=60) as resp: + return json.loads(resp.read().decode() or "{}") + except urllib.error.HTTPError as exc: + raise SystemExit(f"{method} {path}: HTTP {exc.code}: {exc.read().decode(errors='replace')[:300]}") + + def get(self, path: str) -> dict: + return self.req("GET", path) + + def post(self, path: str, body: dict) -> dict: + return self.req("POST", path, body) + + +def mint_jwt(key_path: str) -> str: + """Mint a 7-day HS256 token from the server's jwt.key (sub=ARTEX).""" + key = pathlib.Path(key_path).read_bytes().strip() + + def b64(b: bytes) -> str: + return base64.urlsafe_b64encode(b).decode().rstrip("=") + + header = b64(json.dumps({"alg": "HS256", "typ": "JWT"}).encode()) + now = int(time.time()) + payload = b64(json.dumps({"sub": "ARTEX", "iat": now, "exp": now + 7 * 86400}).encode()) + sig = b64(hmac.new(key, f"{header}.{payload}".encode(), hashlib.sha256).digest()) + return f"{header}.{payload}.{sig}" + + +def llm(base_url: str, api_key: str, model: str, system: str, user: str, max_tokens: int, timeout: int) -> str: + payload = json.dumps({ + "model": model, + "messages": [{"role": "system", "content": system}, {"role": "user", "content": user}], + "temperature": 0, + "max_tokens": max_tokens, + }).encode() + r = urllib.request.Request(base_url.rstrip("/") + "/chat/completions", data=payload, method="POST") + r.add_header("Content-Type", "application/json") + r.add_header("Authorization", "Bearer " + api_key) + try: + with urllib.request.urlopen(r, timeout=timeout) as resp: + body = json.loads(resp.read().decode()) + return body["choices"][0]["message"]["content"] or "" + except urllib.error.HTTPError as exc: + raise SystemExit(f"judge call failed: HTTP {exc.code}: {exc.read().decode(errors='replace')[:200]}") + except (KeyError, IndexError): + return "" # empty response -> invalid, never a verdict + + +# ---------------------------------------------------------------- phases + +def deploy_corpus(args) -> str: + """Start the corpus container; returns its base URL.""" + if args.corpus_url: + print(f"[corpus] using existing deployment {args.corpus_url}") + return args.corpus_url + name = f"bench-{args.corpus}-{args.run_id}" + subprocess.run(["docker", "rm", "-f", name], capture_output=True) + port = str(args.port) + print(f"[corpus] docker run {args.image} -> 127.0.0.1:{port}") + subprocess.run(["docker", "run", "-d", "--rm", "--name", name, "-p", f"127.0.0.1:{port}:{args.container_port}", args.image], check=True, capture_output=True) + url = f"http://127.0.0.1:{port}" + for _ in range(90): + try: + with urllib.request.urlopen(url, timeout=5) as resp: + if resp.status == 200: + print(f"[corpus] up at {url}") + return url + except Exception: + time.sleep(2) + raise SystemExit("corpus did not become reachable; run with --keep and inspect the container") + + +def create_task(api: Api, args, corpus_url: str, gt: dict) -> dict: + host = corpus_url.split("//", 1)[1] + company = api.post("/api/companies", {"name": f"bench-{args.run_id}", "scope": [{"kind": "domain", "value": host.split(":")[0]}]}) + cid = company.get("id") or company.get("company", {}).get("id") + if not cid: + raise SystemExit(f"company create returned no id: {json.dumps(company)[:200]}") + goal = gt.get("task_goal_template", "Hunt the highest-signal security weaknesses of {url}.") \ + .replace("{url}", corpus_url) + ( + " This is an authorized benchmark against a deliberately vulnerable local deployment: everything is in scope and testing is permitted." + " Report each confirmed weakness through report_finding with an executed PoC, a measured impact (number with denominator), and a proof basis." + " Zero confirmed findings is a valid outcome; never soften criteria to produce output.") + task = api.post("/api/tasks", { + "name": f"bench-{args.corpus}-{args.run_id}", + "company_ids": [cid], + "goal": goal, + "timeout_seconds": args.task_timeout, + }) + tid = task.get("id") + if not tid: + raise SystemExit(f"task create returned no id: {json.dumps(task)[:200]}") + print(f"[task] created #{tid}") + return {"id": tid, "url": corpus_url} + + +def wait_task(api: Api, tid, budget_s: int): + deadline = time.time() + budget_s + last = "" + while time.time() < deadline: + t = api.get(f"/api/tasks/{tid}") + status = str(t.get("status", "")).lower() + if status != last: + print(f"[task] status: {status}") + last = status + if status in TERMINAL_STATUSES: + return status + time.sleep(20) + raise SystemExit(f"task {tid} did not reach a terminal state within {budget_s}s") + + +def fetch_findings(api: Api, tid) -> list[dict]: + res = api.get(f"/api/exploration/findings?task={tid}") + items = res.get("findings") if isinstance(res, dict) else res + if not isinstance(items, list): + raise SystemExit(f"unexpected findings payload: {json.dumps(res)[:200]}") + out = [] + for f in items: + fid = f.get("id") or f.get("node_id") + detail = {} + try: + detail = api.get(f"/api/exploration/findings/{fid}") + except SystemExit: + raise + except Exception: + pass + report = detail.get("report") or f.get("report") or "" + summary = f.get("summary") or detail.get("summary") or "" + name = f.get("name") or f.get("title") or summary[:80] + out.append({ + "id": fid, + "name": str(name)[:200], + "severity": str(f.get("severity") or detail.get("severity") or "?"), + "summary": str(summary)[:600], + "report": str(report)[:1600], + }) + print(f"[findings] fetched {len(out)}") + return out + + +JUDGE_SYSTEM = """You are a security finding comparator. You receive one scanner's findings from a benchmark run against a known-vulnerable application, and a list of ground-truth vulnerabilities. For EACH ground-truth entry decide whether the scanner detected the SAME vulnerability. + +Matching criteria (ALL must be true for detected): +- Same vulnerability CLASS, or the scanner found a vulnerability earlier in the same data flow whose fix would also prevent the ground truth +- Same COMPONENT (same endpoint, page, or feature, or a closely related one) +- Same MECHANISM (same missing check, same injection type, same bypass) + +It is NOT required that severities, finding IDs, or wording match. + +For each ground-truth entry also grade four gates against the matched scanner finding (all false if not detected): +- component: the scanner names the right endpoint/feature (1 point) +- class: the scanner's classification is an acceptable label for the ground-truth CWE (2 points) +- keywords: the scanner's text uses at least two terms describing the specific mechanism (1 point) +- proof: the scanner's PoC/steps describe a route that would actually work for this ground truth (1 point) + +Output ONLY a valid JSON array, one object per ground-truth entry: +[{"gt_id": "...", "detected": true|false, "confidence": "high"|"medium"|"low", "matched_finding_id": , "gates": {"component": bool, "class": bool, "keywords": bool, "proof": bool}, "reason": "1-2 sentences"}]""" + + +def judge_battery(args) -> None: + """The judge must pass a falsification battery before grading (capability-matrix rule: + assume the grader is the broken part).""" + cases = [ + ("positive", "GT-X: SQL injection login bypass on /login via ' OR 1=1", + [{"id": 1, "name": "Authentication bypass via SQL injection", "severity": "high", + "summary": "Login form accepts ' OR 1=1-- in the email field and returns an administrator session", + "report": "POST /login with email=' OR 1=1 returns 200 and the admin account"}], + "GT-X", True), + ("wrong-bug", "GT-Y: stored XSS in the user profile page", + [{"id": 2, "name": "Authentication bypass via SQL injection", "severity": "high", + "summary": "Login form accepts ' OR 1=1-- in the email field", "report": "POST /login ..."}], + "GT-Y", False), + ("vague", "GT-Z: IDOR allowing access to another user's basket", + [{"id": 3, "name": "Possible issue", "severity": "low", + "summary": "The application might have some problems in some areas", "report": "unclear"}], + "GT-Z", False), + ] + for label, gt_text, findings, gt_id, expect in cases: + user = f"## Scanner findings\n{json.dumps(findings, indent=1)}\n\n## Ground truth\n### {gt_text}\n\nRespond with the JSON array." + raw = llm(args.judge_url, args.judge_key, args.judge_model, JUDGE_SYSTEM, user, 2000, args.judge_timeout) + try: + verdict = json.loads(raw[raw.find("["): raw.rfind("]") + 1])[0] + got = bool(verdict.get("detected")) + except Exception: + raise SystemExit(f"judge battery case '{label}': unparseable output; grader not trusted:\n{raw[:300]}") + if got != expect: + raise SystemExit(f"judge battery FAILED on '{label}' (expected detected={expect}, got {got}); a grader that cannot fail the wrong answers does not run") + + +def run_judge(args, gt_findings: list[dict], scanner_findings: list[dict]) -> list[dict]: + gt_block = "\n\n".join( + f"### Ground truth {g['id']}: {g['title']}\nCWE: {g['cwe']}. Acceptable classes: {', '.join(g['classes'])}.\n" + f"Component(s): {', '.join(g['components'])}. Mechanism keywords: {', '.join(g['mechanism_keywords'])}.\n" + f"Proof hint: {g['proof_hint']}" + for g in gt_findings) + user = (f"## Scanner findings\n{json.dumps(scanner_findings, indent=1)}\n\n## Ground truth entries\n{gt_block}\n\n" + "For EACH ground-truth entry, respond in the JSON array.") + for attempt in range(3): + raw = llm(args.judge_url, args.judge_key, args.judge_model, JUDGE_SYSTEM, user, 4000, args.judge_timeout) + start, end = raw.find("["), raw.rfind("]") + if start != -1 and end > start: + try: + verdicts = json.loads(raw[start:end + 1]) + if isinstance(verdicts, list) and len(verdicts) == len(gt_findings): + return verdicts + except json.JSONDecodeError: + pass + print(f"[judge] unparseable output (attempt {attempt + 1}/3), retrying") + raise SystemExit("judge output unparseable after 3 attempts; grader not trusted") + + +# ---------------------------------------------------------------- scoring + +def wilson(p_hat: float, n: int, z: float = 1.96) -> tuple[float, float]: + if n == 0: + return (0.0, 0.0) + denom = 1 + z * z / n + center = (p_hat + z * z / (2 * n)) / denom + half = z * math.sqrt(p_hat * (1 - p_hat) / n + z * z / (4 * n * n)) / denom + return (max(0.0, center - half), min(1.0, center + half)) + + +def tally(gt_findings, verdicts, scanner_findings, invalid_ids) -> dict: + detected = [v for v in verdicts if v.get("detected")] + matched_ids = {v.get("matched_finding_id") for v in detected} + candidates = [f for f in scanner_findings if f["id"] not in matched_ids and f["id"] not in invalid_ids] + gates = {"component": 0, "class": 0, "keywords": 0, "proof": 0} + gate_scores = [] + for v in detected: + g = v.get("gates") or {} + score = (1 if g.get("component") else 0) + (2 if g.get("class") else 0) \ + + (1 if g.get("keywords") else 0) + (1 if g.get("proof") else 0) + gate_scores.append(score) + for k in gates: + gates[k] += 1 if g.get(k) else 0 + n = len(gt_findings) + rate = len(detected) / n if n else 0.0 + lo, hi = wilson(rate, n) + tp = len(detected) + precision_lo = tp / (tp + len(candidates) + len(invalid_ids)) if (tp + len(candidates) + len(invalid_ids)) else 0.0 + precision_hi = tp / tp if tp else 0.0 + return { + "ground_truth_total": n, + "detected": tp, + "detection_rate": round(rate, 4), + "detection_rate_wilson95": [round(lo, 4), round(hi, 4)], + "gate_totals_over_detected": gates, + "mean_gate_score_of_6": round(sum(gate_scores) / len(gate_scores), 2) if gate_scores else None, + "discovery_candidates": [{"id": c["id"], "name": c["name"]} for c in candidates], + "precision_bounds_until_verified": [round(precision_lo, 4), round(precision_hi, 4)], + "invalid_findings_excluded": invalid_ids, + "scanner_findings_total": len(scanner_findings), + } + + +def write_report(out_dir: pathlib.Path, corpus: str, args, task_info, status, findings, verdicts, result) -> None: + out_dir.mkdir(parents=True, exist_ok=True) + (out_dir / "results.json").write_text(json.dumps({ + "method_version": METHOD_VERSION, + "corpus": corpus, + "ground_truth_version": args.gt_version, + "task": task_info, + "task_status": status, + "subject_model": args.subject_model, + "judge_model": args.judge_model, + "result": result, + "verdicts": verdicts, + "findings": findings, + }, indent=2)) + lines = [ + f"# ARTEX benchmark run: {corpus}", + f"- method `{METHOD_VERSION}`, ground truth `{args.gt_version}`, subject model `{args.subject_model}`, judge `{args.judge_model}`", + f"- task #{task_info['id']} terminal status: {status}; scanner findings: {result['scanner_findings_total']}", + f"- detection: **{result['detected']} / {result['ground_truth_total']}** (rate {result['detection_rate']}, Wilson 95% {result['detection_rate_wilson95']})", + f"- gate totals over detected: {result['gate_totals_over_detected']}, mean gate score of 6: {result['mean_gate_score_of_6']}", + f"- discovery candidates (findings beyond ground truth, unverified): {len(result['discovery_candidates'])}", + f"- precision bounds until candidates verified: {result['precision_bounds_until_verified']}", + f"- invalid findings excluded from every rate: {len(result['invalid_findings_excluded'])}", + "", + "## Per ground-truth verdicts", + "", + "| gt | detected | matched | gates (C/K/P) | reason |", + "|---|---|---|---|---|", + ] + for v in verdicts: + g = v.get("gates") or {} + lines.append(f"| {v.get('gt_id')} | {v.get('detected')} | {v.get('matched_finding_id')} | " + f"{g.get('component')}/{g.get('class')}/{g.get('keywords')}/{g.get('proof')} | {(v.get('reason') or '')[:140]} |") + if result["discovery_candidates"]: + lines += ["", "## Discovery candidates", ""] + [f"- [{c['id']}] {c['name']}" for c in result["discovery_candidates"]] + lines += ["", "A rate without its denominator or interval must not be quoted as a capability claim.", ""] + (out_dir / "report.md").write_text("\n".join(lines)) + print(f"[out] {out_dir / 'report.md'}") + + +# ---------------------------------------------------------------- main + +def main() -> None: + ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) + ap.add_argument("--corpus", default="juice-shop", help="ground-truth file stem under ground_truth/") + ap.add_argument("--run-id", default=time.strftime("%Y%m%d-%H%M%S")) + ap.add_argument("--image", default="bkimminich/juice-shop:latest") + ap.add_argument("--port", type=int, default=13100) + ap.add_argument("--container-port", type=int, default=3000) + ap.add_argument("--corpus-url", help="use an already-running deployment instead of docker") + ap.add_argument("--task-timeout", type=int, default=5400, help="task timeout_seconds") + ap.add_argument("--artex-url", default=os.environ.get("ARTEX_BASE", "http://localhost:8787")) + ap.add_argument("--gt-version", default="v1") + ap.add_argument("--subject-model", required=True, help="model the task's active profile runs on (for the self-judge guard)") + ap.add_argument("--judge-url", default=os.environ.get("JUDGE_BASE_URL", "")) + ap.add_argument("--judge-model", default=os.environ.get("JUDGE_MODEL", "")) + ap.add_argument("--judge-timeout", type=int, default=900) + ap.add_argument("--allow-self-judge", action="store_true") + ap.add_argument("--keep", action="store_true", help="keep the corpus container running after the run") + args = ap.parse_args() + + args.judge_key = os.environ.get("JUDGE_API_KEY", "") + token = os.environ.get("ARTEX_TOKEN", "") + if not token: + key_path = os.environ.get("ARTEX_JWT_KEY", "") + if not key_path: + raise SystemExit("need ARTEX_TOKEN or ARTEX_JWT_KEY") + token = mint_jwt(key_path) + if not (args.judge_url and args.judge_model and args.judge_key): + raise SystemExit("need JUDGE_BASE_URL, JUDGE_MODEL, JUDGE_API_KEY") + + # A model never judges its own output (capability-matrix standing rule 5). + if args.judge_model == args.subject_model and not args.allow_self_judge: + raise SystemExit(f"judge model equals subject model ({args.judge_model}); refusing. Pass --allow-self-judge to override knowingly.") + + gt_path = pathlib.Path(__file__).parent / "ground_truth" / f"{args.corpus}.json" + gt = json.loads(gt_path.read_text()) + gt_findings = gt["findings"] + + api = Api(args.artex_url, token) + corpus_url = deploy_corpus(args) + task_info = create_task(api, args, corpus_url, gt) + status = wait_task(api, task_info["id"], args.task_timeout + 1800) + findings = fetch_findings(api, task_info["id"]) + + invalid_ids = [f["id"] for f in findings if not (f["summary"].strip() or f["report"].strip())] + usable = [f for f in findings if f["id"] not in invalid_ids] + + if not usable: + print("[judge] no usable findings; scoring zeros") + verdicts = [{"gt_id": g["id"], "detected": False, "matched_finding_id": None, + "gates": {"component": False, "class": False, "keywords": False, "proof": False}, + "reason": "no usable scanner findings"} for g in gt_findings] + else: + print("[judge] running falsification battery") + judge_battery(args) + print("[judge] grading") + verdicts = run_judge(args, gt_findings, usable) + + result = tally(gt_findings, verdicts, findings, invalid_ids) + out_dir = pathlib.Path("benchmark-results") / args.corpus / args.run_id + write_report(out_dir, args.corpus, args, task_info, status, findings, verdicts, result) + print(json.dumps(result, indent=2)[:1500]) + + if not (args.keep or args.corpus_url): + subprocess.run(["docker", "rm", "-f", f"bench-{args.corpus}-{args.run_id}"], capture_output=True) + + +if __name__ == "__main__": + main()