marius/Open_Ultra_SAST
1
0
Fork 0
mirror of https://github.com/norandom/OpenUltraSAST.git synced 2026-10-16 03:49:26 +00:00
Code Issues Packages Projects Releases Wiki Activity

ci: green again -- format and lint the whole tree; push the scan and runner images to GHCR on tags

Browse Source 
CI and the release job check `ruff format --check .` and `ruff check .` over the
whole repository, while the local gate checked only src and tests, so eight
benchmark scripts drifted and the v2.0.0 release job stopped before building.
The fixes are formatting, two noqa comments and one renamed loop variable; no
script computes anything differently. The local gate now checks the whole tree.

The release workflow gains an images job: Dockerfile -> ghcr.io/<owner>/openultrasast,
plane/Dockerfile.runner -> ghcr.io/<owner>/openultrasast-runner, tagged with the
version, and the digests written to the job summary.

Co-Authored-By: Claude Opus 5.5 <[email protected]>
Claude-Session: https://claude.ai/code/session_01EmM1j46qgKMh3zqD1QMwEo
This commit is contained in:
Marius Ciepluch
2026-10-02 20:09:28 +02:00
co-authored by Claude Opus 5.5
parent c978de9ff8
commit d1aabf0f66
12 changed files with 256 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.github/workflows/release.yml
+42
View File
@@ -34,3 +34,45 @@ jobs:
generate_release_notes: true
prerelease: ${{ contains(github.ref_name, '-') }}
make_latest: ${{ !contains(github.ref_name, '-') }}
images:
# The scan image (Joern + php-cli, `docker run ... scan`) and the plane's task runner, pushed to GHCR under the
# tag's version. Pin them by the digests this job prints, not by tag: Agent Substrate rejects tags.
runs-on: ubuntu-latest
permissions:
contents: read
packages: write
strategy:
fail-fast: false
matrix:
include:
- name: openultrasast
dockerfile: Dockerfile
- name: openultrasast-runner
dockerfile: plane/Dockerfile.runner
steps:
- uses: actions/checkout@v5
- uses: docker/setup-buildx-action@v3
- uses: docker/login-action@v3
with:
registry: ghcr.io
username: ${{ github.actor }}
password: ${{ secrets.GITHUB_TOKEN }}
- id: meta
uses: docker/metadata-action@v5
with:
images: ghcr.io/${{ github.repository_owner }}/${{ matrix.name }}
tags: |
type=semver,pattern={{version}}
type=semver,pattern={{major}}.{{minor}},enable=${{ !contains(github.ref_name, '-') }}
type=raw,value=latest,enable=${{ !contains(github.ref_name, '-') }}
- id: build
uses: docker/build-push-action@v6
with:
context: .
file: ${{ matrix.dockerfile }}
push: true
tags: ${{ steps.meta.outputs.tags }}
labels: ${{ steps.meta.outputs.labels }}
- run: |
echo "ghcr.io/${{ github.repository_owner }}/${{ matrix.name }}@${{ steps.build.outputs.digest }}" >> "$GITHUB_STEP_SUMMARY"
benchmarks/archive/hunt.py
+38 -3
View File
@@ -40,8 +40,34 @@ from openultrasast.preprocess import build_file_target, enumerate_source_files
# The model names a family in its own words; the protocol scores three. Fixed before any v2 pin is hunted.
FAMILY_WORDS = {
"untrusted_destination": ("ssrf", "server-side request", "open redirect", "redirect", "url fetch", "request forgery"),
"config_secrets": ("cors", "debug", "secret", "hardcoded", "hard-coded", "credential", "misconfig", "configuration", "tls", "certificate"),
"injection": ("inject", "sql", "sqli", "nosql", "command", "rce", "code execution", "eval", "template", "ssti", "xss", "deserializ", "ldap", "xpath"),
"config_secrets": (
"cors",
"debug",
"secret",
"hardcoded",
"hard-coded",
"credential",
"misconfig",
"configuration",
"tls",
"certificate",
),
"injection": (
"inject",
"sql",
"sqli",
"nosql",
"command",
"rce",
"code execution",
"eval",
"template",
"ssti",
"xss",
"deserializ",
"ldap",
"xpath",
),
}
@@ -182,7 +208,16 @@ def main() -> int:
result["root"] = "/case"
out.write_text(json.dumps(result, indent=2) + "\n")
print(
json.dumps({"case": case["id"], "pin": label, "findings": len(result["findings"]), "hunts": f"{result['completed']}/{result['questions']}", "usd": result["cost_usd"], "total_usd": round(sum(spent), 4)}),
json.dumps(
{
"case": case["id"],
"pin": label,
"findings": len(result["findings"]),
"hunts": f"{result['completed']}/{result['questions']}",
"usd": result["cost_usd"],
"total_usd": round(sum(spent), 4),
}
),
flush=True,
)
return 0
benchmarks/archive/model_sinks.py
+13 -5
View File
@@ -37,7 +37,9 @@ from openultrasast.config import load_config # noqa: E402
from openultrasast.model.endpoint import DEFAULT_DETECTOR_MODEL # noqa: E402
from openultrasast.preprocess import detect_language, enumerate_source_files # noqa: E402
NOT_PRODUCT = re.compile(r"(^|/)(tests?|__tests__|spec|specs|docs?|examples?|fixtures?|e2e|cypress|storybook)/|\.(test|spec|stories)\.[jt]sx?$|_test\.py$|(^|/)test_[^/]*\.py$|\.min\.js$|(^|/)dist/|(^|/)build/")
NOT_PRODUCT = re.compile(
r"(^|/)(tests?|__tests__|spec|specs|docs?|examples?|fixtures?|e2e|cypress|storybook)/|\.(test|spec|stories)\.[jt]sx?$|_test\.py$|(^|/)test_[^/]*\.py$|\.min\.js$|(^|/)dist/|(^|/)build/"
)
OPERATIONS = {
"injection": (
"executes a database query (SQL, NoSQL, JCR/graph query), an operating-system command or shell, code or "
@@ -69,7 +71,9 @@ def product_files(root: Path, languages: set[str]) -> list[Path]:
def chunks(text: str, size: int) -> list[str]:
lines = text.splitlines()
return ["\n".join(f"{n + 1:5d}| {line}" for n, line in enumerate(lines[i : i + size], start=i)) for i in range(0, max(len(lines), 1), size)]
return [
"\n".join(f"{n + 1:5d}| {line}" for n, line in enumerate(lines[i : i + size], start=i)) for i in range(0, max(len(lines), 1), size)
]
class Meter:
@@ -85,7 +89,9 @@ class Meter:
return sum(float(c.cost_usd()) for c in self.clients)
def classify(root: Path, path: Path, family: str, args: argparse.Namespace, meter: Meter, local: threading.local) -> tuple[list[tuple[str, str, int]], int, int]:
def classify(
root: Path, path: Path, family: str, args: argparse.Namespace, meter: Meter, local: threading.local
) -> tuple[list[tuple[str, str, int]], int, int]:
from openultrasast import tool_hunter
if not hasattr(local, "client"):
@@ -99,7 +105,9 @@ def classify(root: Path, path: Path, family: str, args: argparse.Namespace, mete
asked += 1
prompt = PROMPT.format(operation=OPERATIONS[family], path=relative, code=code)
try:
response = local.client.complete(model=args.model, messages=[{"role": "user", "content": prompt}], tools=[], timeout_seconds=90, json_object=True)
response = local.client.complete(
model=args.model, messages=[{"role": "user", "content": prompt}], tools=[], timeout_seconds=90, json_object=True
)
items = json.loads(response.content or "{}").get("functions", [])
except Exception: # noqa: BLE001 -- counted as unclassified, never as "no sinks"
failed += 1
@@ -138,7 +146,7 @@ def main() -> int:
files = product_files(checkout, {case["language"], *case.get("also", [])})
found, failed, asked = [], 0, 0
with concurrent.futures.ThreadPoolExecutor(max_workers=args.workers) as pool:
for f, fail, ask in pool.map(lambda p: classify(checkout, p, case["family"], args, meter, local), files):
for f, fail, ask in pool.map(lambda p, c=checkout, fam=case["family"]: classify(c, p, fam, args, meter, local), files):
found += f
failed += fail
asked += ask
benchmarks/archive/score_batched.py
+2 -1
View File
@@ -66,7 +66,8 @@ def main() -> int:
"agreed_findings": len(vulnerable["findings"]), "disputed_findings": len(vulnerable.get("disputed", [])),
"triaged_out": vulnerable.get("triaged_out"), "candidates": vulnerable.get("questions"),
"fixed_side_alerts": fixed_alerts, "fixed_agreed_total": len(fixed["findings"]) if fixed else None,
"removed_by_fix": fixed.get("removed_by_fix") if fixed else None, "usd": vulnerable.get("usd"), "usage": vulnerable.get("usage"),
"removed_by_fix": fixed.get("removed_by_fix") if fixed else None,
"usd": vulnerable.get("usd"), "usage": vulnerable.get("usage"),
}
) # fmt: skip
pool += [{"case": case["id"], **f} for f in vulnerable["findings"]]
benchmarks/archive/sink_candidates.py
+3 -1
View File
@@ -36,7 +36,9 @@ from openultrasast.preprocess import detect_language, enumerate_source_files #
DECLARATION = {
"python": re.compile(r"^\s*(?:async\s+)?def\s+(\w+)"),
"php": re.compile(r"^\s*(?:(?:public|private|protected|static|final|abstract)\s+)*function\s+&?(\w+)"),
"javascript": re.compile(r"^\s*(?:export\s+)?(?:async\s+)?function\s*\*?\s*(\w+)|^\s*(?:(?:public|private|protected|static|async)\s+)*(\w+)\s*\([^;]*\)\s*(?::[^{=]+)?\{\s*$|^\s*(?:export\s+)?(?:const|let|var)\s+(\w+)\s*=\s*(?:async\s*)?(?:function|\()"),
"javascript": re.compile(
r"^\s*(?:export\s+)?(?:async\s+)?function\s*\*?\s*(\w+)|^\s*(?:(?:public|private|protected|static|async)\s+)*(\w+)\s*\([^;]*\)\s*(?::[^{=]+)?\{\s*$|^\s*(?:export\s+)?(?:const|let|var)\s+(\w+)\s*=\s*(?:async\s*)?(?:function|\()"
),
}
DECLARATION["typescript"] = DECLARATION["javascript"]
KEYWORDS = {"if", "for", "while", "switch", "catch", "return", "function", "else"}
benchmarks/archive/verify_batched.py
+87 -14
View File
@@ -101,13 +101,22 @@ def triage(root: Path, path: str, candidates: list[list], family: str, model: st
except OSError:
return {}
names = sorted({fn for _, fn, _ in candidates})
prompt = TRIAGE.format(operation=OPERATIONS[family], functions=json.dumps(names), path=path, code=_relevant_code(text, [ln for _, _, ln in candidates], chunk_lines))
prompt = TRIAGE.format(
operation=OPERATIONS[family],
functions=json.dumps(names),
path=path,
code=_relevant_code(text, [ln for _, _, ln in candidates], chunk_lines),
)
response = client.complete(model=model, messages=[{"role": "user", "content": prompt}], tools=[], timeout_seconds=90, json_object=True)
try:
items = json.loads(response.content or "{}").get("functions", [])
except (json.JSONDecodeError, AttributeError):
return {}
return {str(i.get("function")): str(i.get("data_origin", "")).strip().lower() for i in items if isinstance(i, dict) and i.get("function") in names}
return {
str(i.get("function")): str(i.get("data_origin", "")).strip().lower()
for i in items
if isinstance(i, dict) and i.get("function") in names
}
def hunt(root: Path, path: str, group: list[list], family: str, model: str, steps: int) -> tuple[list[dict], dict]:
@@ -120,8 +129,30 @@ def hunt(root: Path, path: str, group: list[list], family: str, model: str, step
f"Operations to judge, all in `{path}`, each of which {OPERATIONS[family]}:\n{listing}\n"
"The file is above. For each, trace where the operation's data comes from and whether it is guarded."
)
spots = [Hotspot(path=path, function_name=fn, score=1.0, band="candidate", signals={}, rationale="model-classified sink", test_hint=None, inventory_finding_ids=()) for _, fn, _ in group]
found = tool_hunter.run_tool_hunter(root, spots, client=recording, model=model, max_steps=steps, system_prompt=SYSTEM, user_prompt=prompt, context_files=[path], tags=("sink-verifier",))
spots = [
Hotspot(
path=path,
function_name=fn,
score=1.0,
band="candidate",
signals={},
rationale="model-classified sink",
test_hint=None,
inventory_finding_ids=(),
)
for _, fn, _ in group
]
found = tool_hunter.run_tool_hunter(
root,
spots,
client=recording,
model=model,
max_steps=steps,
system_prompt=SYSTEM,
user_prompt=prompt,
context_files=[path],
tags=("sink-verifier",),
)
names = {fn for _, fn, _ in group}
rows = []
for f in found:
@@ -131,7 +162,16 @@ def hunt(root: Path, path: str, group: list[list], family: str, model: str, step
# The verdict is about the candidate operation, so the site is the candidate's file and function; the
# model's own location is kept beside it (it often names the file where the input enters instead).
line = f.line if f.path == path else next(c[2] for c in group if c[1] == fn)
rows.append({"candidate": f"{path}::{fn}", "site": f"{path}:{line or 0}:{fn}", "reported_at": f"{f.path}:{f.line or 0}", "family": family, "title": f.title, "witness": f.rationale[:600]})
rows.append(
{
"candidate": f"{path}::{fn}",
"site": f"{path}:{line or 0}:{fn}",
"reported_at": f"{f.path}:{f.line or 0}",
"family": family,
"title": f.title,
"witness": f.rationale[:600],
}
)
return rows, recording.usage()
@@ -143,7 +183,12 @@ def verify_file(root: str, path: str, candidates: list[list], family: str, args_
try:
if not args_["no_triage"]:
record["triage"] = triage(Path(root), path, candidates, family, args_["model"], args_["chunk_lines"])
kept = [c for c in candidates if record["triage"].get(c[1], "unclassified") in VERIFY | {"unclassified"} or record["triage"].get(c[1]) not in {"internal", "constant"}]
kept = [
c
for c in candidates
if record["triage"].get(c[1], "unclassified") in VERIFY | {"unclassified"}
or record["triage"].get(c[1]) not in {"internal", "constant"}
]
record["kept"] = [c[1] for c in kept]
usage: dict[str, int] = {}
for _pass in range(args_["passes"]):
@@ -169,14 +214,16 @@ def agreement(record: dict) -> tuple[list[dict], list[dict]]:
some = set.union(*(set(b) for b in by_pass))
first = by_pass[0]
pick = lambda c: next(b[c] for b in by_pass if c in b) # noqa: E731
return [dict(first[c], passes=len(by_pass)) for c in sorted(every)], [dict(pick(c), passes=sum(c in b for b in by_pass)) for c in sorted(some - every)]
return [dict(first[c], passes=len(by_pass)) for c in sorted(every)], [
dict(pick(c), passes=sum(c in b for b in by_pass)) for c in sorted(some - every)
]
def main() -> int:
parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
parser.add_argument("--population", type=Path, default=HERE / "population-v2.toml")
parser.add_argument("--only", default="")
parser.add_argument("--subset", type=Path, help='JSON {case_id: [[path, function], ...]}: verify only these candidates')
parser.add_argument("--subset", type=Path, help="JSON {case_id: [[path, function], ...]}: verify only these candidates")
parser.add_argument("--model", default=DEFAULT_DETECTOR_MODEL)
parser.add_argument("--max-steps", type=int, default=6)
parser.add_argument("--per-hunt", type=int, default=6)
@@ -206,7 +253,11 @@ def main() -> int:
if out.is_file():
continue
log = out_dir / f"{case['id']}--{label}.jsonl"
done = {r["path"]: r for r in (json.loads(row) for row in log.read_text().splitlines()) if "unverified" not in r} if log.is_file() else {}
done = (
{r["path"]: r for r in (json.loads(row) for row in log.read_text().splitlines()) if "unverified" not in r}
if log.is_file()
else {}
)
spent += sum(float(r.get("usd", 0.0)) for r in done.values())
record = json.loads((candidates_dir / f"{case['id']}.json").read_text())
unique = list({(p, fn): [p, fn, ln] for p, fn, ln in sorted(record["candidates"])}.values())
@@ -265,19 +316,41 @@ def main() -> int:
agreed += a
disputed += d
triaged_out = sum(len(r["candidates"]) - len(r.get("kept", [])) for r in done.values())
usage = {k: sum(int(r.get("usage", {}).get(k, 0)) for r in done.values()) for k in ("prompt_tokens", "prompt_cache_hit_tokens", "completion_tokens", "calls")}
usage = {
k: sum(int(r.get("usage", {}).get(k, 0)) for r in done.values())
for k in ("prompt_tokens", "prompt_cache_hit_tokens", "completion_tokens", "calls")
}
complete = len(done) >= len(by_file) and not account_error
result = {
"root": "/case", "families": [case["family"]], "questions": len(unique), "completed": sum(len(r["candidates"]) for r in done.values()),
"root": "/case", "families": [case["family"]],
"questions": len(unique), "completed": sum(len(r["candidates"]) for r in done.values()),
"completed_regions": sorted(f"{c[0]}:{c[1]}" for r in done.values() for c in r["candidates"]),
"triaged_out": triaged_out, "passes": args.passes, "usage": usage, "usd": round(sum(float(r.get("usd", 0.0)) for r in done.values()), 4),
"seconds": round(time.monotonic() - started, 1), "findings": sorted(agreed, key=lambda f: f["site"]), "disputed": sorted(disputed, key=lambda f: f["site"]),
"triaged_out": triaged_out, "passes": args.passes, "usage": usage,
"usd": round(sum(float(r.get("usd", 0.0)) for r in done.values()), 4),
"seconds": round(time.monotonic() - started, 1),
"findings": sorted(agreed, key=lambda f: f["site"]),
"disputed": sorted(disputed, key=lambda f: f["site"]),
} # fmt: skip
if args.fixed:
result["removed_by_fix"] = removed
if complete:
out.write_text(json.dumps(result, indent=1) + "\n")
print(json.dumps({"case": case["id"], "files": f"{len(done)}/{len(by_file)}", "candidates": len(unique), "triaged_out": triaged_out, "agreed": len(agreed), "disputed": len(disputed), "usd": result["usd"], "total_usd": round(spent, 4), "complete": complete}), flush=True)
print(
json.dumps(
{
"case": case["id"],
"files": f"{len(done)}/{len(by_file)}",
"candidates": len(unique),
"triaged_out": triaged_out,
"agreed": len(agreed),
"disputed": len(disputed),
"usd": result["usd"],
"total_usd": round(spent, 4),
"complete": complete,
}
),
flush=True,
)
if account_error:
print(json.dumps({"stopped": "account", "error": account_error}), flush=True)
return 2
benchmarks/archive/verify_sinks.py
+53 -7
View File
@@ -104,7 +104,16 @@ def verify(root: str, candidate: list, family: str, model: str, max_steps: int,
client = _client()
before = float(client.cost_usd())
path, function, line = candidate
spot = Hotspot(path=path, function_name=function, score=1.0, band="candidate", signals={}, rationale="model-classified sink", test_hint=None, inventory_finding_ids=())
spot = Hotspot(
path=path,
function_name=function,
score=1.0,
band="candidate",
signals={},
rationale="model-classified sink",
test_hint=None,
inventory_finding_ids=(),
)
prompt = (
f"Operation to judge: in `{path}`, function `{function}`, around line {line}, code that {OPERATIONS[family]}. "
"The file is above. Trace where the operation's data comes from and whether it is guarded."
@@ -120,9 +129,19 @@ def verify(root: str, candidate: list, family: str, model: str, max_steps: int,
second = recording.usage()
usage = {k: usage.get(k, 0) + second.get(k, 0) for k in set(usage) | set(second)}
except Exception as error: # noqa: BLE001 -- recorded; an unverified candidate is not a clean one
return {"candidate": candidate, "unverified": f"{type(error).__name__}: {str(error)[:120]}", "usd": float(client.cost_usd()) - before}
return {
"candidate": candidate,
"unverified": f"{type(error).__name__}: {str(error)[:120]}",
"usd": float(client.cost_usd()) - before,
}
findings = [
{"site": f"{f.path}:{f.line or line}:{f.function_name or function}", "family": family, "title": f.title, "witness": f.rationale[:600], "candidate": f"{path}::{function}"}
{
"site": f"{f.path}:{f.line or line}:{f.function_name or function}",
"family": family,
"title": f.title,
"witness": f.rationale[:600],
"candidate": f"{path}::{function}",
}
for f in found
]
return {"candidate": candidate, "findings": findings, "usd": float(client.cost_usd()) - before, "escalated": escalated, "usage": usage}
@@ -134,7 +153,12 @@ def main() -> int:
parser.add_argument("--only", default="")
parser.add_argument("--model", default=DEFAULT_DETECTOR_MODEL)
parser.add_argument("--max-steps", type=int, default=6)
parser.add_argument("--escalate-steps", type=int, default=0, help="rerun a hunt that used every step without a verdict; 0 disables (measured: 88% escalate, +45% cost)")
parser.add_argument(
"--escalate-steps",
type=int,
default=0,
help="rerun a hunt that used every step without a verdict; 0 disables (measured: 88% escalate, +45% cost)",
)
parser.add_argument("--workers", type=int, default=4)
parser.add_argument("--budget-usd", type=float, default=40.0)
parser.add_argument("--fixed", action="store_true", help="recheck detected sites at the fixed pin")
@@ -185,7 +209,9 @@ def main() -> int:
with concurrent.futures.ProcessPoolExecutor(max_workers=args.workers) as pool, log.open("a") as sink:
futures = []
for candidate in pending:
futures.append(pool.submit(verify, str(checkout), list(candidate), case["family"], args.model, args.max_steps, args.escalate_steps))
futures.append(
pool.submit(verify, str(checkout), list(candidate), case["family"], args.model, args.max_steps, args.escalate_steps)
)
for future in concurrent.futures.as_completed(futures):
if future.cancelled(): # cancelled at the ceiling: never run, nothing to record
continue
@@ -219,9 +245,29 @@ def main() -> int:
} # fmt: skip
if complete:
out.write_text(json.dumps(result, indent=1) + "\n")
tokens = {k: sum(int(d.get("usage", {}).get(k, 0)) for d in done) for k in ("prompt_tokens", "prompt_cache_hit_tokens", "completion_tokens")}
tokens = {
k: sum(int(d.get("usage", {}).get(k, 0)) for d in done)
for k in ("prompt_tokens", "prompt_cache_hit_tokens", "completion_tokens")
}
result["usage"] = tokens | {"escalated": sum(1 for d in done if d.get("escalated"))}
print(json.dumps({"case": case["id"], "pin": label, "candidates": len(todo), "verified": len(verified), "findings": len(findings), "unverified": len(unverified), "escalated": result["usage"]["escalated"], "cache_share": round(tokens["prompt_cache_hit_tokens"] / max(tokens["prompt_tokens"], 1), 2), "usd": result["usd"], "total_usd": round(spent, 4), "complete": complete}), flush=True)
print(
json.dumps(
{
"case": case["id"],
"pin": label,
"candidates": len(todo),
"verified": len(verified),
"findings": len(findings),
"unverified": len(unverified),
"escalated": result["usage"]["escalated"],
"cache_share": round(tokens["prompt_cache_hit_tokens"] / max(tokens["prompt_tokens"], 1), 2),
"usd": result["usd"],
"total_usd": round(spent, 4),
"complete": complete,
}
),
flush=True,
)
if account_error:
print(json.dumps({"stopped": "account", "error": account_error, "spent_usd": round(spent, 4)}), flush=True)
return 2
benchmarks/dev/bulk_read.py
+7 -2
View File
@@ -36,9 +36,9 @@ def main() -> int:
parser.add_argument("files", nargs="+", type=Path)
parser.add_argument("--max-usd", type=float, default=0.05)
args = parser.parse_args()
from openultrasast import tool_hunter
from openultrasast.config import load_config
from openultrasast.model.endpoint import DEFAULT_DETECTOR_MODEL
from openultrasast import tool_hunter
load_config()
client = tool_hunter.resolve_hunter_client()
@@ -51,7 +51,12 @@ def main() -> int:
if float(client.cost_usd()) >= args.max_usd:
answers.append(f"[stopped at the ${args.max_usd} ceiling before: {chunk.splitlines()[0]}]")
break
response = client.complete(model=DEFAULT_DETECTOR_MODEL, messages=[{"role": "user", "content": PROMPT.format(question=args.question, files=chunk)}], tools=[], timeout_seconds=90)
response = client.complete(
model=DEFAULT_DETECTOR_MODEL,
messages=[{"role": "user", "content": PROMPT.format(question=args.question, files=chunk)}],
tools=[],
timeout_seconds=90,
)
answers.append((response.content or "").strip())
print("\n\n".join(a for a in answers if a))
print(f"\n[bulk_read: {len(chunks)} chunk(s), ${float(client.cost_usd()):.4f}]", file=sys.stderr)
benchmarks/dev/read_guard.py
+5 -1
View File
@@ -37,7 +37,11 @@ def main() -> int:
f"{path} has {lines} lines (limit {LIMIT} for a whole-file read). Read a range with offset/limit, or ask "
f"the bulk reader for what you need: .venv/bin/python benchmarks/dev/bulk_read.py '<question>' {path}"
)
print(json.dumps({"hookSpecificOutput": {"hookEventName": "PreToolUse", "permissionDecision": "deny", "permissionDecisionReason": reason}}))
print(
json.dumps(
{"hookSpecificOutput": {"hookEventName": "PreToolUse", "permissionDecision": "deny", "permissionDecisionReason": reason}}
)
)
return 0
benchmarks/independent/evaluate.py
+4 -2
View File
@@ -258,7 +258,7 @@ def score_v2() -> int:
per_case, pool, unstable_pool = [], [], 0
changing, total_questions, total_completed = [0, 0], 0, 0
for case in cases():
path = lambda label: RESULTS / "scans" / f"{case['id']}--{label}.json" # noqa: E731
path = lambda label: RESULTS / "scans" / f"{case['id']}--{label}.json" # noqa: E731, B023 -- called only inside this iteration
scans = {label: json.loads(path(label).read_text()) for label in pins(case) if path(label).is_file()}
if len(scans) < 6:
per_case.append({"id": case["id"], "status": "incomplete", "scans": sorted(scans)})
@@ -279,7 +279,9 @@ def score_v2() -> int:
]
site_files = {site.partition("::")[0] for site in case.get("sites", [])}
covered = set(old) | site_files
answered = any(r.split(":")[0] in covered for label in ("vulnerable_a", "vulnerable_b") for r in scans[label].get("completed_regions", []))
answered = any(
r.split(":")[0] in covered for label in ("vulnerable_a", "vulnerable_b") for r in scans[label].get("completed_regions", [])
)
diff_v, union_v = stability(va, vb)
diff_f, union_f = stability(fa, fb)
changing[0] += diff_v + diff_f
benchmarks/independent/freeze.py
+1 -1
View File
@@ -44,7 +44,7 @@ EXTENSIONS = {
# scopes to API tokens" -- access control and input validation, which is exactly what a benign control must not be.
SECURITY_WORDS = re.compile(
r"secur|cve|ghsa|vuln|inject|xss|csrf|ssrf|sanitiz|escap|exploit|attack|auth|permission|cors|access|restrict|harden"
r"|token|scope|valid|filter|privileg|role|owner|password|secret|redirect|sql|query|login|log in|ldap|session|ssl|tls|verif|cert|crypt|hash|cookie",
r"|token|scope|valid|filter|privileg|role|owner|password|secret|redirect|sql|query|login|log in|ldap|session|ssl|tls|verif|cert|crypt|hash|cookie", # noqa: E501
re.I,
)
MAX_BENIGN_LINES = 200
benchmarks/measurements/2026-09-30-php-quick-rules/measure_php_quick.py
+1 -1
View File
@@ -162,7 +162,7 @@ def per_rule(result: dict) -> dict:
rows[fid.split(":", 1)[0]][label] += 1
for fid in pin["at_known_site"]:
rows[fid.split(":", 1)[0]][label + "_at_site"] += 1
for name, fixture in result["fixtures"].items():
for _name, fixture in result["fixtures"].items():
for fid in fixture["true_positives"]:
rows[fid.split(":", 1)[0]]["fixture_tp"] += 1
for fid in fixture["false_positives"]: