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

test(m3): Python transfer detects all three declared targets (M3 13.1)

Browse Source 
Generalized the replay instrument so the case family is an argument rather than a
hard-coded target, since a harness that only knows one language cannot say whether
the design transfers. Targets, pins and expectations come from the committed inputs,
so Python is asked exactly the way Node was, through the same core, ranker,
comparison and admission.

  VAmPI-SQLI        models/user_model.py:73:get_user        detected, answered
  VAmPI-BOLA        api_views/users.py:187:update_password  detected, answered
  VAmPI-BOLA-BOOKS  api_views/books.py:51:get_by_title      detected, answered

35 of 35 questions complete on every case, nothing reported new, zero admitted
defects.

Two results stand out. The SQL injection is the evidence the milestone spec names
as previously unresolved. And two of the three targets are access control, the
family that could not complete a question on any repository before group 12, so
this is the first time broken object-level authorization has been asked on Python
at all, and it answered on both endpoints.

VAmPI has the same revision on both sides, so the regression mode establishes
detection and completion, never novelty. It asserts that directly: with nothing
changed, a target reported new would be a comparison defect, and none was. That
tests the novelty logic from the opposite side to the NodeGoat case.

Identical revisions cannot show novelty, and VAmPI is inspected teaching data that
cannot qualify production Python support. No capability enabled, no latency claim.

1,407 tests passed, nine skipped.

Co-Authored-By: Claude Opus 5 (1M context) <[email protected]>
Claude-Session: https://claude.ai/code/session_01EmM1j46qgKMh3zqD1QMwEo
This commit is contained in:
Marius Ciepluch
2026-09-17 18:17:31 +02:00
co-authored by Claude Opus 5
parent fc5dc453b6
commit bb72fcd413
4 changed files with 803 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
benchmarks/push/transfer_replay.py
+309
View File
@@ -0,0 +1,309 @@
"""M3: replay a declared vulnerable/fixed/benign family in any supported language.
The NodeGoat instrument hard-codes its target and its three comparisons, which was right for M1
and is wrong for a transfer claim: a harness that only knows one language cannot say whether the
design transfers. This one takes the case family as an argument and reads its target, pins and
expectations from the committed inputs, so PHP and Python are asked exactly the way Node was.
What it records per revision is the same: the raw findings at the declared target, whether the
target question completed, the production comparison verdict, the recorded veto list, and the
census and source receipts. Expectations are written before the first scan.
Development evidence under a named budget. It is not independent qualification, it enables no
capability, and it makes no latency claim.
"""
from __future__ import annotations
import argparse
import hashlib
import json
import os
import platform
import subprocess
import sys
import time
from pathlib import Path
from typing import Any
sys.path.insert(0, str(Path(__file__).resolve().parent))
from measure import git, pin_snapshot # noqa: E402
from openultrasast.model.scan import ScanBudget # noqa: E402
from openultrasast.push.policy import CapabilityAdmission, CapabilityKey # noqa: E402
from openultrasast.push.runner import _provenance # noqa: E402
from openultrasast.push.vetoes import FLAG, LABEL # noqa: E402
from openultrasast.push_inputs import validate_manifest # noqa: E402
from openultrasast.repos import checkout_path, load_repo_recipe # noqa: E402
# What each declared comparison is expected to show, stated before anything runs. `introduce`
# reintroduces the reviewed operation, `repair` removes it, `benign` changes something else.
EXPECTATIONS = {
"introduce": "raw finding at the declared target lines; recorded novelty new",
"repair": "fixed twin analyzed and quiet at the target; base-only findings retained",
"benign": "backlog unchanged at the target; never new or worsened",
# A regression snapshot has the same revision on both sides, so there is no change and no
# novelty to claim. What it can show is whether the declared target is still detected and
# whether its question completes. Anything reported as new here would be a defect in the
# comparison, not a finding, so the verdict checks for exactly that.
"regression": "declared target still detected and its question answered; never new, because nothing changed",
}
# The manifest labels a case by what it IS (vulnerable, fixed, benign, regression); the case id
# says what the comparison DOES (introduce, repair, benign). Selecting on one and keying
# expectations by the other silently ran a single case and called the rest missing.
ROLES = {"vulnerable": "introduce", "fixed": "repair", "benign": "benign", "regression": "regression"}
def role_of(case: dict[str, Any]) -> str | None:
return ROLES.get(str(case.get("label", "")))
def write(path: Path, data: object) -> None:
path.write_text(json.dumps(data, indent=2, allow_nan=False, default=str) + "\n")
def declaration_for(semantics: str, language: str, family: str, symbols: tuple[str, ...]) -> list[dict[str, Any]]:
"""An unreviewed declaration so the evaluation explanation renders. Never enabled, never PASS."""
key = CapabilityKey(language, "unspecified", "unspecified", family, "taint", "unreviewed", semantics)
declaration = CapabilityAdmission(
key,
"unreviewed-m3-transfer-v1",
"experimental://m3-transfer",
"experimental",
"A request value reaches {operation}, which interpolates it into a statement the runtime executes ({context} capability).",
"Bind the value at {operation} through the platform's parameterised form instead of interpolating it ({context} capability).",
operation_symbols=symbols,
)
assert not declaration.enabled and declaration.verdict != "PASS", "the transfer declaration must stay unqualified"
return [declaration.to_payload()]
def summarize(case: dict[str, Any], target: dict[str, Any], artifact: dict[str, Any]) -> dict[str, Any]:
experimental = artifact.get("experimental") or {}
head = [record["scan"] for record in artifact["scans"] if record["side"] == "head"]
base = [record["scan"] for record in artifact["scans"] if record["side"] == "base"]
at_target = [
finding
for scan in head
for finding in scan["findings"]
if finding["family"] == target["family"] and finding["site"].startswith(target["path"] + ":")
]
questions = [
outcome
for scan in head
for outcome in scan["question_outcomes"]
if outcome["identity"]["path"] == target["path"] and outcome["identity"]["family"] == target["family"]
]
answered = bool(questions) and all(q["raw_rows_json"] is not None for q in questions)
findings = experimental.get("findings") if isinstance(experimental.get("findings"), list) else []
recorded = [f for f in findings if f["family"] == target["family"] and f["site"].startswith(target["path"] + ":")]
return {
"expected": EXPECTATIONS[role_of(case) or "introduce"],
"identical_revisions": case["base"] == case["tip"],
"declared_target": target,
"findings_at_target": len(at_target),
"finding_sites": [f["site"] for f in at_target][:8],
"target_lines_hit": sorted({int(f["site"].split(":")[1]) for f in at_target if f["site"].split(":")[1].isdigit()}),
"target_questions": [{"function": q["identity"]["function"], "status": q["status"], "reason": q["reason"]} for q in questions],
"target_questions_answered": answered,
"questions_completed": sum(1 for s in head for q in s["question_outcomes"] if q["status"] == "completed"),
"questions_total": sum(len(s["question_outcomes"]) for s in head),
"base_questions_completed": sum(1 for s in base for q in s["question_outcomes"] if q["status"] == "completed"),
"census": [{"language": p.get("language"), "files": len(p.get("paths", ()))} for s in head for p in s.get("partitions", ())],
"production_result": [artifact["result"]["finding_status"], artifact["result"]["coverage_status"]],
"recorded_status": experimental.get("status"),
"recorded_at_target": [
{
"site": f["site"],
"production": [f["production_novelty"], f["production_reason"]],
"recorded": [f["recorded_novelty"], f["recorded_reason"]],
"admission": f["production_admission"],
"unlifted_vetoes": [[v["stage"], v["veto"]] for v in f["vetoes"] if not v["lifted"]],
}
for f in recorded
],
"evaluation": experimental.get("evaluation", []),
"base_only_findings": len(experimental.get("base_only_findings", []) or []),
"coverage_reasons": artifact["admission"]["coverage_reasons"][:12],
"admitted_defects": len(artifact["admission"]["defects"]),
"timings": {k: v for k, v in artifact["timings"].items() if k.endswith("_seconds")},
}
def main() -> int:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--manifest", type=Path, required=True)
parser.add_argument("--cache", type=Path, required=True)
parser.add_argument("--out", type=Path, required=True)
parser.add_argument("--family", required=True, help="case id prefix, for example pmpro or nodegoat")
parser.add_argument("--language", required=True)
parser.add_argument("--symbols", default="", help="comma-separated operation symbols for the unreviewed declaration")
parser.add_argument("--deadline", type=float, default=900.0)
args = parser.parse_args()
args.out.mkdir(parents=True, exist_ok=False)
out = args.out.resolve()
manifest = json.loads(args.manifest.read_bytes())
targets = json.loads((args.manifest.parent / "diagnostic-config-v1.json").read_text())["targets"]
validation = validate_manifest(args.manifest, cache_root=args.cache, fetch=False)
write(out / "input-validation.json", validation)
ready = {s["id"] for s in validation["snapshots"] if s["status"] == "ready"}
prefix = args.family if args.family.endswith("-") else args.family + "-"
cases = [c for c in manifest["cases"] if c["id"].startswith(prefix) and role_of(c) is not None]
if not cases:
raise SystemExit(f"no vulnerable/fixed/benign cases for {args.family}")
snapshots = {s["id"]: s for s in manifest["snapshots"]}
missing = [c["id"] for c in cases if c["base"] not in ready or c["tip"] not in ready]
if missing:
raise SystemExit(f"inputs not ready: {missing}")
recipe = load_repo_recipe(args.manifest.parent / snapshots[cases[0]["base"]]["recipe"])
bare = out / "objects.git"
bare.mkdir()
git(bare, "init", "--bare")
alternates = set()
pins: dict[str, str] = {}
for case in cases:
for sid in (case["base"], case["tip"]):
if sid in pins:
continue
snapshot = snapshots[sid]
source = checkout_path(load_repo_recipe(args.manifest.parent / snapshot["recipe"]), args.cache)
objects = git(source, "rev-parse", "--path-format=absolute", "--git-path", "objects").decode().strip()
alternates.add(objects)
(bare / "objects/info/alternates").write_text("\n".join(sorted(alternates)) + "\n")
pins[sid] = pin_snapshot(bare, snapshot.get("commit", recipe.commit), snapshot)
settings = ScanBudget(max_model_calls=0, max_regions=500, order_by_evidence=True)
installed = _provenance(bare, settings)
family = next((targets[c["id"]]["family"] for c in cases if c["id"] in targets), "injection")
symbols = tuple(s.strip() for s in args.symbols.split(",") if s.strip())
declarations = out / "experimental-declarations.json"
write(declarations, declaration_for(installed["semantics"], args.language, family, symbols))
record: dict[str, Any] = {
"schema_version": 1,
"label": LABEL,
"milestone": "M3",
"family": args.family,
"language": args.language,
"flag": FLAG,
"development_budget_seconds": args.deadline,
"budget_note": "Explicitly named development budget; this run cannot satisfy or measure the hook latency gate.",
"expectations": EXPECTATIONS,
"pins": pins,
"targets": {c["id"]: targets.get(c["id"]) for c in cases},
"source_receipts": {
sid: {
"path": targets[cases[0]["id"]]["path"],
"bytes": len(git(bare, "cat-file", "blob", f"{oid}:{targets[cases[0]['id']]['path']}")),
"sha256": hashlib.sha256(git(bare, "cat-file", "blob", f"{oid}:{targets[cases[0]['id']]['path']}")).hexdigest(),
}
for sid, oid in pins.items()
},
"provenance": installed,
"hardware": {"platform": platform.platform(), "cpu_affinity": sorted(os.sched_getaffinity(0))},
"instrument_sha256": hashlib.sha256(Path(__file__).read_bytes()).hexdigest(),
"cases": [],
}
write(out / "transfer-record.json", record)
for case in cases:
target = targets.get(case["id"])
if target is None:
record["cases"].append({"case_id": case["id"], "status": "no_declared_target"})
write(out / "transfer-record.json", record)
continue
artifact = out / (case["id"] + ".json")
command = [
sys.executable,
"-m",
"openultrasast.cli",
"pre-push",
str(bare),
"--base",
pins[case["base"]],
"--head",
pins[case["tip"]],
"--artifact",
str(artifact),
"--deadline",
str(args.deadline),
"--cancellation-allowance",
"2",
"--max-regions",
"500",
"--cache-dir",
str(out / "cache"),
FLAG,
"--experimental-declarations",
str(declarations),
]
started = time.monotonic()
process = subprocess.run(command, capture_output=True, text=True, timeout=args.deadline + 60) # noqa: S603
elapsed = time.monotonic() - started
(out / (case["id"] + ".stdout")).write_text(process.stdout)
(out / (case["id"] + ".stderr")).write_text(process.stderr)
entry: dict[str, Any] = {
"case_id": case["id"],
"label": case["label"],
"role": role_of(case),
"base_oid": pins[case["base"]],
"head_oid": pins[case["tip"]],
"exit_code": process.returncode,
"elapsed_seconds": elapsed,
"terminal": process.stdout,
}
if artifact.is_file():
data = json.loads(artifact.read_text())
entry["provenance_matches"] = all(
data["provenance"].get(k) == installed[k] for k in ("engine", "facts", "queries", "policy", "core", "semantics")
)
entry["outcome"] = summarize(case, target, data)
else:
entry["outcome"] = {"instrument_error": "missing_artifact"}
record["cases"].append(entry)
write(out / "transfer-record.json", record)
print(json.dumps({"case": case["id"], "elapsed": round(elapsed, 1), "exit": process.returncode}), flush=True)
outcomes = {c["case_id"]: c.get("outcome", {}) for c in record["cases"]}
regressions = [(c["case_id"], c.get("outcome", {})) for c in record["cases"] if c.get("role") == "regression"]
if regressions:
record["verdict"] = {
"targets_detected": {cid: o.get("findings_at_target", 0) for cid, o in regressions},
"all_targets_detected": all(o.get("findings_at_target") for _, o in regressions),
"all_target_questions_answered": all(o.get("target_questions_answered") for _, o in regressions),
"nothing_reported_new": not any(
f["recorded"][0] in ("new", "worsened") for _, o in regressions for f in o.get("recorded_at_target", [])
),
"all_recorded": all(o.get("recorded_status") == "evaluated" for _, o in regressions),
"no_admitted_defects": all(o.get("admitted_defects") == 0 for _, o in regressions),
"note": (
"Identical revisions on both sides: this establishes detection and completion, not novelty. "
"A target reported new here would be a comparison defect."
),
}
write(out / "transfer-record.json", record)
print(json.dumps(record["verdict"]), flush=True)
return 0 if all(v for k, v in record["verdict"].items() if k not in ("note", "targets_detected")) else 1
by_role = {c.get("role"): c.get("outcome", {}) for c in record["cases"]}
introduce, repair, benign = by_role.get("introduce", {}), by_role.get("repair", {}), by_role.get("benign", {})
record["verdict"] = {
"raw_witness_present": bool(introduce.get("findings_at_target")),
"fixed_twin_quiet": bool(repair.get("target_questions_answered")) and repair.get("findings_at_target") == 0,
"unchanged_not_new_regression": bool(benign.get("target_questions_answered"))
and not any(f["recorded"][0] in ("new", "worsened") for f in benign.get("recorded_at_target", [])),
"all_recorded": all(o.get("recorded_status") == "evaluated" for o in outcomes.values() if o),
"no_admitted_defects": all(o.get("admitted_defects") == 0 for o in outcomes.values() if o),
"note": "Development transfer evidence. Silence from a failure is not a pass; an unanalyzed fixed twin fails.",
}
write(out / "transfer-record.json", record)
print(json.dumps(record["verdict"]), flush=True)
return 0 if all(v for k, v in record["verdict"].items() if k != "note") else 1
if __name__ == "__main__":
raise SystemExit(main())