from __future__ import annotations
import hashlib
import json
from pathlib import Path
import subprocess
import sys
import tempfile
from typing import Any
ROOT = Path(__file__).resolve().parents[1]
VALIDATOR = ROOT / "scripts" / "validate-replacement-proof.py"
SOURCE_COMMIT = "1" * 40
SOURCE_TREE = "2" * 40
UPSTREAM_COMMIT = "3" * 40
DAEMON_HASH = "4" * 64
CLIENT_HASH = "5" * 64
NSS_HASH = "6" * 64
def digest(path: Path) -> str:
return hashlib.sha256(path.read_bytes()).hexdigest()
def write_json(path: Path, payload: dict[str, Any]) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(json.dumps(payload, sort_keys=True) + "\n", encoding="utf-8")
def artifact(path: Path) -> dict[str, Any]:
return {
"name": path.name,
"path": str(path),
"size": path.stat().st_size,
"sha256": digest(path),
}
def write_proof(
directory: Path,
gate: str,
metadata: dict[str, str],
artifacts: list[Path],
) -> Path:
proof = directory / f"{gate}.json"
write_json(
proof,
{
"schema": 1,
"gate": gate,
"result": "pass",
"source_commit": SOURCE_COMMIT,
"source_tree": SOURCE_TREE,
"upstream_commit": UPSTREAM_COMMIT,
"metadata": metadata,
"artifacts": [artifact(path) for path in artifacts],
},
)
return proof
def validate(directory: Path, proof: Path, source_commit: str = SOURCE_COMMIT):
return subprocess.run(
[
sys.executable,
str(VALIDATOR),
"--proof",
str(proof),
"--gate",
proof.stem,
"--source-commit",
source_commit,
"--source-tree",
SOURCE_TREE,
"--upstream-commit",
UPSTREAM_COMMIT,
"--proof-directory",
str(directory),
],
check=False,
capture_output=True,
text=True,
)
def upstream_fixture(directory: Path, nss_hash: str | None = NSS_HASH) -> Path:
gate = "upstream-test-75"
artifacts = directory / "artifacts" / gate
log = artifacts / "TEST-75-RESOLVED.log"
log.parent.mkdir(parents=True)
marker = "RESOLVED_RS_TEST_75_" + SOURCE_TREE + "_" + DAEMON_HASH
if nss_hash is not None:
marker += "_" + nss_hash
log.write_text(marker + "\ncandidate suite passed\n", encoding="utf-8")
evidence = artifacts / "evidence.json"
payload = {
"schema": 1,
"suite": "TEST-75-RESOLVED",
"unmodified_recorded_upstream_files": True,
"upstream_commit": UPSTREAM_COMMIT,
"source_tree": SOURCE_TREE,
"daemon_sha256": DAEMON_HASH,
"client_sha256": CLIENT_HASH,
"runtime_marker": marker,
"log": {
"name": log.name,
"size": log.stat().st_size,
"sha256": digest(log),
},
}
if nss_hash is not None:
payload["nss_module_sha256"] = nss_hash
write_json(evidence, payload)
return write_proof(
directory,
gate,
{"suite": "TEST-75-RESOLVED", "unmodified-recorded-files": "true"},
[evidence, log],
)
def security_fixture(directory: Path, evidence_commit: str = SOURCE_COMMIT) -> Path:
gate = "security-suite"
artifacts = directory / "artifacts" / gate
evidence = artifacts / "security-evidence.json"
required = ["asan", "fuzz", "miri", "tsan", "ubsan", "valgrind"]
jobs = {
"fuzz": "libFuzzer corpus and smoke",
"asan": "Address Sanitizer ASan",
"ubsan": "Undefined Behavior Sanitizer UBSan",
"miri": "Miri strict provenance",
"tsan": "Thread Sanitizer TSan",
"valgrind": "Valgrind NSS Varlink and DNS fallback",
}
write_json(
evidence,
{
"schema": 1,
"source_commit": SOURCE_COMMIT,
"required_categories": required,
"missing": [],
"matched": {
category: [
{
"head_sha": evidence_commit,
"workflow_name": "Replacement security gates",
"event": "workflow_dispatch",
"job_name": jobs[category],
"run_id": 10,
"job_id": 20,
}
]
for category in required
},
},
)
return write_proof(
directory,
gate,
{"profiles": "fuzz,asan,ubsan,miri,tsan,valgrind"},
[evidence],
)
def boot_fixture(directory: Path, evidence_tree: str = SOURCE_TREE) -> Path:
gate = "boot-replacement"
artifacts = directory / "artifacts" / gate
artifacts.mkdir(parents=True, exist_ok=True)
build_log = artifacts / "mkosi-build.log"
console_log = artifacts / "qemu-console.log"
build_log.write_text("image built\n", encoding="utf-8")
console_log.write_text(
"\n".join(
(
f"RESOLVED_RS_CANDIDATE_BOOT_1_{DAEMON_HASH}",
f"RESOLVED_RS_CANDIDATE_BOOT_2_{DAEMON_HASH}",
f"RESOLVED_RS_CANDIDATE_NSS_BOOT_1_{NSS_HASH}",
f"RESOLVED_RS_CANDIDATE_NSS_BOOT_2_{NSS_HASH}",
f"RESOLVED_RS_BOOT_ROLLBACK_PASS_{UPSTREAM_COMMIT}",
f"RESOLVED_RS_BOOT_PROOF_PASS_{SOURCE_TREE}_{DAEMON_HASH}",
)
)
+ "\n",
encoding="utf-8",
)
evidence = artifacts / "evidence.json"
write_json(
evidence,
{
"schema": 1,
"environment": "qemu",
"distribution": "ubuntu",
"release": "noble",
"boot_count": 2,
"candidate_healthy_each_boot": True,
"rollback_verified": True,
"source_commit": SOURCE_COMMIT,
"source_tree": evidence_tree,
"upstream_commit": UPSTREAM_COMMIT,
"daemon_sha256": DAEMON_HASH,
"client_sha256": CLIENT_HASH,
"nss_module_sha256": NSS_HASH,
"artifacts": [artifact(build_log), artifact(console_log)],
},
)
return write_proof(
directory,
gate,
{"environment": "qemu", "boot-count": "2", "rollback-verified": "true"},
[evidence, build_log, console_log],
)
def assert_pass(result: subprocess.CompletedProcess[str]) -> None:
assert result.returncode == 0, result.stderr
def assert_rejected(result: subprocess.CompletedProcess[str], detail: str) -> None:
assert result.returncode != 0, result.stdout
assert detail in result.stderr, result.stderr
def main() -> None:
with tempfile.TemporaryDirectory(prefix="replacement-proof-test-") as name:
directory = Path(name)
assert_pass(validate(directory, upstream_fixture(directory)))
assert_rejected(
validate(directory, directory / "upstream-test-75.json", "6" * 40),
"proof source commit is stale",
)
with tempfile.TemporaryDirectory(prefix="replacement-proof-test-") as name:
directory = Path(name)
assert_rejected(
validate(directory, upstream_fixture(directory, None)),
"upstream evidence nss_module_sha256 is invalid",
)
with tempfile.TemporaryDirectory(prefix="replacement-proof-test-") as name:
directory = Path(name)
assert_pass(validate(directory, security_fixture(directory)))
stale = security_fixture(directory, "6" * 40)
assert_rejected(validate(directory, stale), "security category evidence is malformed")
with tempfile.TemporaryDirectory(prefix="replacement-proof-test-") as name:
directory = Path(name)
assert_pass(validate(directory, boot_fixture(directory)))
stale = boot_fixture(directory, "6" * 40)
assert_rejected(validate(directory, stale), "boot evidence source tree is stale")
print("replacement proof validator regression tests passed")
if __name__ == "__main__":
main()