Backup/ARTEX
1
0
Fork 0
mirror of https://github.com/BaaRai/ARTEX.git synced 2026-11-13 11:40:53 +08:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #59 from BaaRai/feat/l3-benchmark

Browse Source 
feat: L3 benchmark harness (whole-app corpus, independent judge)
This commit is contained in:
BaaRai
2026-10-12 08:02:00 +02:00
committed by GitHub
parent 0e53de78c2 86a4efc85e
commit ba5c31fd01
4 changed files with 610 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.gitignore
+3
View File
@@ -58,3 +58,6 @@ Thumbs.db
/bench/
/Dockerfile.bench
/Dockerfile.bench.v2backups/
# benchmark runner output (one dir per run)
benchmark-results/
scripts/benchmark/README.md
Executable
+94
View File
@@ -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/<corpus>.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=<jwt or mint via ARTEX_JWT_KEY=/app/jwt.key copy> \
JUDGE_BASE_URL=https://api.example.com JUDGE_MODEL=<other-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/<name>.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.
scripts/benchmark/ground_truth/juice-shop.json
Executable
+91
View File
@@ -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/<id> 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"
}
]
}
scripts/benchmark/run_bench.py
Executable
+422
View File
@@ -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": <scanner finding id or null>, "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()