358 lines
14 KiB
Python
358 lines
14 KiB
Python
"""Versioned benchmark-case and immutable run-artifact contracts."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import copy
|
|
import hashlib
|
|
import json
|
|
import platform
|
|
import subprocess
|
|
from collections.abc import Mapping, Sequence
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
CASE_SCHEMA_VERSION = 1
|
|
RUN_ARTIFACT_SCHEMA_VERSION = 1
|
|
|
|
CASE_TYPES = frozenset(
|
|
{
|
|
"positive-semantic",
|
|
"negative-near-miss",
|
|
"vulnerable-fixed",
|
|
"metamorphic",
|
|
"differential",
|
|
"dynamic-confirmation",
|
|
"real-project-advisory",
|
|
"finding-set",
|
|
}
|
|
)
|
|
ORIGINS = frozenset({"synthetic", "organic", "revived", "mutated", "metamorphic"})
|
|
BACKENDS = frozenset({"source", "binary", "hybrid", "rust", "swift"})
|
|
|
|
|
|
def _nonempty_string(value: Any) -> bool:
|
|
return isinstance(value, str) and bool(value.strip())
|
|
|
|
|
|
def _sha256_string(value: Any) -> bool:
|
|
return (
|
|
isinstance(value, str)
|
|
and len(value) == 64
|
|
and all(character in "0123456789abcdef" for character in value.lower())
|
|
)
|
|
|
|
|
|
def _infer_backend(document: Mapping[str, Any]) -> str:
|
|
if "binary_build_source" in document or (
|
|
"sources" in document and "binary" in document
|
|
):
|
|
return "hybrid"
|
|
if "binary" in document or "build_source" in document:
|
|
return "binary"
|
|
mode = str(document.get("mode", "source"))
|
|
return mode if mode in BACKENDS else "source"
|
|
|
|
|
|
def _slug(value: str) -> str:
|
|
normalized = "".join(ch.lower() if ch.isalnum() else "-" for ch in value)
|
|
return "-".join(part for part in normalized.split("-") if part)
|
|
|
|
|
|
def normalize_benchmark_case(
|
|
document: Mapping[str, Any], *, source_path: str = ""
|
|
) -> dict[str, Any]:
|
|
"""Normalize a canonical or legacy golden without discarding legacy fields."""
|
|
|
|
raw = copy.deepcopy(dict(document))
|
|
legacy = "schema_version" not in raw
|
|
name = str(raw.get("name", raw.get("case_id", "")))
|
|
normalized = {
|
|
"schema_version": CASE_SCHEMA_VERSION,
|
|
"case_id": str(raw.get("case_id", _slug(name))),
|
|
"name": name,
|
|
"case_type": str(raw.get("case_type", "finding-set")),
|
|
"origin": str(raw.get("origin", "synthetic")),
|
|
"backend": str(raw.get("backend", _infer_backend(raw))),
|
|
"oracle": copy.deepcopy(raw.get("oracle", raw.get("expect", {}))),
|
|
"source_path": source_path,
|
|
"legacy": legacy,
|
|
"manifest": raw,
|
|
}
|
|
for field in (
|
|
"corpus",
|
|
"lineage",
|
|
"relation",
|
|
"runtime_oracle",
|
|
"differential_oracle",
|
|
"slices",
|
|
):
|
|
if field in raw:
|
|
normalized[field] = copy.deepcopy(raw[field])
|
|
return normalized
|
|
|
|
|
|
def validate_benchmark_case(
|
|
document: Mapping[str, Any], *, allow_legacy: bool = True, source_path: str = ""
|
|
) -> dict[str, Any]:
|
|
normalized = normalize_benchmark_case(document, source_path=source_path)
|
|
errors: list[str] = []
|
|
warnings: list[str] = []
|
|
if normalized["legacy"]:
|
|
if not allow_legacy:
|
|
errors.append("schema_version is required")
|
|
else:
|
|
warnings.append("legacy manifest normalized to benchmark-case schema v1")
|
|
elif document.get("schema_version") != CASE_SCHEMA_VERSION:
|
|
errors.append(f"unsupported schema_version: {document.get('schema_version')}")
|
|
|
|
if not _nonempty_string(normalized["case_id"]):
|
|
errors.append("case_id must be a non-empty string")
|
|
if normalized["case_type"] not in CASE_TYPES:
|
|
errors.append(f"unknown case_type: {normalized['case_type']}")
|
|
if normalized["origin"] not in ORIGINS:
|
|
errors.append(f"unknown origin: {normalized['origin']}")
|
|
if normalized["backend"] not in BACKENDS:
|
|
errors.append(f"unknown backend: {normalized['backend']}")
|
|
if not isinstance(normalized["oracle"], dict):
|
|
errors.append("oracle/expect must be an object")
|
|
|
|
case_type = normalized["case_type"]
|
|
if case_type == "vulnerable-fixed":
|
|
lineage = normalized.get("lineage")
|
|
if not isinstance(lineage, dict) or not all(
|
|
_nonempty_string(lineage.get(field))
|
|
for field in ("vulnerable_revision", "fixed_revision")
|
|
):
|
|
errors.append(
|
|
"vulnerable-fixed cases require lineage.vulnerable_revision and "
|
|
"lineage.fixed_revision"
|
|
)
|
|
if normalized["origin"] == "organic":
|
|
corpus = normalized.get("corpus")
|
|
entries = corpus.get("entries", []) if isinstance(corpus, dict) else []
|
|
if not isinstance(corpus, dict) or not _nonempty_string(corpus.get("kind")):
|
|
errors.append("organic vulnerable-fixed cases require corpus.kind")
|
|
if not isinstance(entries, list) or not entries:
|
|
errors.append("organic vulnerable-fixed cases require corpus.entries")
|
|
else:
|
|
required = (
|
|
"name",
|
|
"cve",
|
|
"project",
|
|
"repository",
|
|
"reference",
|
|
"vulnerable_revision",
|
|
"fixed_revision",
|
|
"upstream_path",
|
|
"license",
|
|
"adaptation",
|
|
"vulnerable_source",
|
|
"fixed_source",
|
|
)
|
|
names: set[str] = set()
|
|
for index, entry in enumerate(entries):
|
|
if not isinstance(entry, dict):
|
|
errors.append(f"corpus.entries[{index}] must be an object")
|
|
continue
|
|
missing = [
|
|
field
|
|
for field in required
|
|
if not _nonempty_string(entry.get(field))
|
|
]
|
|
if missing:
|
|
errors.append(
|
|
f"corpus.entries[{index}] missing provenance fields: "
|
|
+ ", ".join(missing)
|
|
)
|
|
name = str(entry.get("name", ""))
|
|
if name in names:
|
|
errors.append(f"duplicate organic corpus entry name: {name}")
|
|
names.add(name)
|
|
for field in ("vulnerable_sha256", "fixed_sha256"):
|
|
if not _sha256_string(entry.get(field)):
|
|
errors.append(
|
|
f"corpus.entries[{index}].{field} must be a SHA-256"
|
|
)
|
|
expected = entry.get("expected")
|
|
if not isinstance(expected, dict) or not all(
|
|
isinstance(expected.get(field), bool)
|
|
for field in ("vulnerable_detected", "fixed_reported")
|
|
):
|
|
errors.append(
|
|
f"corpus.entries[{index}].expected must define boolean "
|
|
"vulnerable_detected and fixed_reported"
|
|
)
|
|
elif not expected["vulnerable_detected"] and not _nonempty_string(
|
|
entry.get("known_gap")
|
|
):
|
|
errors.append(
|
|
f"corpus.entries[{index}] requires known_gap when the "
|
|
"vulnerable revision is not expected to be detected"
|
|
)
|
|
if case_type == "metamorphic":
|
|
relation = normalized.get("relation")
|
|
if not isinstance(relation, dict) or not _nonempty_string(relation.get("kind")):
|
|
errors.append("metamorphic cases require relation.kind")
|
|
if case_type == "dynamic-confirmation":
|
|
oracle = normalized.get("runtime_oracle")
|
|
required = ("hazard", "input_channel", "checker", "replay_command")
|
|
if not isinstance(oracle, dict) or not all(
|
|
_nonempty_string(oracle.get(field)) for field in required
|
|
):
|
|
errors.append(
|
|
"dynamic-confirmation cases require runtime_oracle hazard, "
|
|
"input_channel, checker, and replay_command"
|
|
)
|
|
if case_type == "differential":
|
|
oracle = normalized.get("differential_oracle")
|
|
comparators = oracle.get("comparators", []) if isinstance(oracle, dict) else []
|
|
if (
|
|
not isinstance(oracle, dict)
|
|
or not _nonempty_string(oracle.get("agreement"))
|
|
or not isinstance(comparators, list)
|
|
or len(comparators) < 2
|
|
or not all(_nonempty_string(item) for item in comparators)
|
|
):
|
|
errors.append(
|
|
"differential cases require differential_oracle.agreement and "
|
|
"at least two comparator names"
|
|
)
|
|
return {
|
|
"valid": not errors,
|
|
"errors": errors,
|
|
"warnings": warnings,
|
|
"case": normalized,
|
|
}
|
|
|
|
|
|
def executable_version(path: Path) -> str:
|
|
"""Return the executable's canonical one-line version provenance."""
|
|
|
|
completed = subprocess.run(
|
|
[str(path), "--version"],
|
|
check=False,
|
|
text=True,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.STDOUT,
|
|
timeout=10,
|
|
)
|
|
version = completed.stdout.strip().splitlines()
|
|
if completed.returncode != 0 or not version:
|
|
raise ValueError(f"cannot read tool version from {path}")
|
|
return version[0]
|
|
|
|
|
|
def sha256_file(path: Path) -> str:
|
|
digest = hashlib.sha256()
|
|
with path.open("rb") as stream:
|
|
for chunk in iter(lambda: stream.read(1024 * 1024), b""):
|
|
digest.update(chunk)
|
|
return digest.hexdigest()
|
|
|
|
|
|
def artifact_record(role: str, path: Path, media_type: str) -> dict[str, str]:
|
|
record = {"role": role, "path": str(path), "media_type": media_type, "sha256": ""}
|
|
if path.is_file():
|
|
record["sha256"] = sha256_file(path)
|
|
return record
|
|
|
|
|
|
def utc_now() -> str:
|
|
return datetime.now(timezone.utc).isoformat()
|
|
|
|
|
|
def build_run_artifact(
|
|
*,
|
|
experiment_id: str,
|
|
case_id: str,
|
|
tool: Mapping[str, Any],
|
|
target: Mapping[str, Any],
|
|
command: Sequence[str],
|
|
started_at: str,
|
|
ended_at: str,
|
|
exit_status: int,
|
|
budgets: Mapping[str, Any],
|
|
artifacts: Sequence[Mapping[str, Any]],
|
|
measurements: Mapping[str, Any] | None = None,
|
|
observations: Mapping[str, Any] | None = None,
|
|
seed: str | int | None = None,
|
|
) -> dict[str, Any]:
|
|
execution: dict[str, Any] = {
|
|
"command": list(command),
|
|
"started_at": started_at,
|
|
"ended_at": ended_at,
|
|
"exit_status": exit_status,
|
|
"budgets": dict(budgets),
|
|
"machine": {
|
|
"system": platform.system(),
|
|
"release": platform.release(),
|
|
"machine": platform.machine(),
|
|
"python": platform.python_version(),
|
|
},
|
|
}
|
|
if seed is not None:
|
|
execution["seed"] = seed
|
|
return {
|
|
"schema_version": RUN_ARTIFACT_SCHEMA_VERSION,
|
|
"experiment_id": experiment_id,
|
|
"case_id": case_id,
|
|
"tool": dict(tool),
|
|
"target": dict(target),
|
|
"execution": execution,
|
|
"artifacts": [dict(artifact) for artifact in artifacts],
|
|
"measurements": dict(measurements or {}),
|
|
"observations": dict(observations or {}),
|
|
}
|
|
|
|
|
|
def validate_run_artifact(document: Mapping[str, Any]) -> list[str]:
|
|
errors: list[str] = []
|
|
if document.get("schema_version") != RUN_ARTIFACT_SCHEMA_VERSION:
|
|
errors.append(f"unsupported schema_version: {document.get('schema_version')}")
|
|
for field in ("experiment_id", "case_id"):
|
|
if not _nonempty_string(document.get(field)):
|
|
errors.append(f"{field} must be a non-empty string")
|
|
for field in ("tool", "target", "execution", "measurements", "observations"):
|
|
if not isinstance(document.get(field), dict):
|
|
errors.append(f"{field} must be an object")
|
|
# Current writers always preserve tool.version. It remains optional for
|
|
# schema-v1 readers so immutable artifacts created before that writer
|
|
# requirement are still valid under the unchanged schema version.
|
|
execution = document.get("execution", {})
|
|
if isinstance(execution, dict):
|
|
command = execution.get("command")
|
|
if (
|
|
not isinstance(command, list)
|
|
or not command
|
|
or not all(_nonempty_string(item) for item in command)
|
|
):
|
|
errors.append("execution.command must be a non-empty string array")
|
|
if isinstance(execution.get("exit_status"), bool) or not isinstance(
|
|
execution.get("exit_status"), int
|
|
):
|
|
errors.append("execution.exit_status must be an integer")
|
|
for field in ("started_at", "ended_at"):
|
|
if not _nonempty_string(execution.get(field)):
|
|
errors.append(f"execution.{field} must be a non-empty string")
|
|
for field in ("budgets", "machine"):
|
|
if not isinstance(execution.get(field), dict):
|
|
errors.append(f"execution.{field} must be an object")
|
|
artifacts = document.get("artifacts")
|
|
if not isinstance(artifacts, list):
|
|
errors.append("artifacts must be an array")
|
|
else:
|
|
for index, artifact in enumerate(artifacts):
|
|
if not isinstance(artifact, dict):
|
|
errors.append(f"artifacts[{index}] must be an object")
|
|
continue
|
|
for field in ("role", "path", "media_type", "sha256"):
|
|
if not isinstance(artifact.get(field), str):
|
|
errors.append(f"artifacts[{index}].{field} must be a string")
|
|
return errors
|
|
|
|
|
|
def write_json(path: Path, document: Mapping[str, Any]) -> None:
|
|
path.write_text(
|
|
json.dumps(document, indent=2, sort_keys=True) + "\n", encoding="utf-8"
|
|
)
|