from __future__ import annotations
import hashlib
import json
from pathlib import Path
import shutil
import subprocess
import sys
import tempfile
from typing import Any
ROOT = Path(__file__).resolve().parents[1]
VALIDATOR = ROOT / "scripts" / "validate-replacement-proof.py"
REPRODUCIBLE_VALIDATOR = ROOT / "scripts" / "validate-reproducible-release.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
RUSTC_HASH = "8" * 64
CARGO_HASH = "9" * 64
SECURITY_REPOSITORY = "example/resolver"
SECURITY_WORKFLOW_ID = 101
SECURITY_RUN_ID = 202
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 manifest_artifact(path: Path) -> dict[str, Any]:
return {
"name": path.name,
"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],
"host": {"github_repository": SECURITY_REPOSITORY},
},
)
return proof
def validate(
directory: Path,
proof: Path,
source_commit: str = SOURCE_COMMIT,
daemon_hash: str = DAEMON_HASH,
client_hash: str = CLIENT_HASH,
nss_hash: str = NSS_HASH,
local_reproducible: Path | None = None,
):
arguments = [
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),
"--expected-daemon-sha256",
daemon_hash,
"--expected-client-sha256",
client_hash,
"--expected-nss-sha256",
nss_hash,
]
if local_reproducible is not None:
arguments.extend(
["--local-reproducible-directory", str(local_reproducible)]
)
return subprocess.run(
arguments,
check=False,
capture_output=True,
text=True,
)
def validate_local_reproducible(
directory: Path, daemon_hash: str, client_hash: str, nss_hash: str
) -> subprocess.CompletedProcess[str]:
return subprocess.run(
[
sys.executable,
str(REPRODUCIBLE_VALIDATOR),
"--directory",
str(directory / "artifacts" / "reproducible-release"),
"--source-commit",
SOURCE_COMMIT,
"--source-tree",
SOURCE_TREE,
"--upstream-commit",
UPSTREAM_COMMIT,
"--expected-daemon-sha256",
daemon_hash,
"--expected-client-sha256",
client_hash,
"--expected-nss-sha256",
nss_hash,
],
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 = "RUSTD_RESOLVED_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 upstream_mdns_fixture(directory: Path) -> Path:
gate = "upstream-test-89-mdns"
artifacts = directory / "artifacts" / gate
log = artifacts / "TEST-89-RESOLVED-MDNS.log"
log.parent.mkdir(parents=True)
marker = "RUSTD_RESOLVED_TEST_89_RESOLVED_MDNS_" + SOURCE_TREE + "_" + DAEMON_HASH
log.write_text(marker + "\ncandidate suite passed\n", encoding="utf-8")
evidence = artifacts / "evidence.json"
write_json(
evidence,
{
"schema": 1,
"suite": "TEST-89-RESOLVED-MDNS",
"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),
},
},
)
return write_proof(
directory,
gate,
{
"suite": "TEST-89-RESOLVED-MDNS",
"unmodified-recorded-files": "true",
},
[evidence, log],
)
def reproducible_fixture(
directory: Path,
rustc_release: str = "1.74.0",
omit_artifact: str | None = None,
) -> Path:
gate = "reproducible-release"
artifacts = directory / "artifacts" / gate
artifacts.mkdir(parents=True)
paths: list[Path] = []
for name, content in (
("systemd-resolved", b"daemon\n"),
("resolvectl", b"client\n"),
("libnss_resolve.so.2", b"nss\n"),
("rustd-resolved.tar.gz", b"package\n"),
("files.sha256", b"files\n"),
(
"rust-toolchain.txt",
(
f"rustc {rustc_release}\ncargo 1.74.0\n"
f"rustc_binary_sha256 {RUSTC_HASH}\n"
f"cargo_binary_sha256 {CARGO_HASH}\n"
).encode(),
),
):
path = artifacts / name
path.write_bytes(content)
if name != omit_artifact:
paths.append(path)
manifest = artifacts / "manifest.json"
write_json(
manifest,
{
"schema": 2,
"reproducible": True,
"build_count": 2,
"byte_identical": [
"systemd-resolved",
"resolvectl",
"libnss_resolve.so.2",
"files.sha256",
"rustd-resolved.tar.gz",
],
"source_commit": SOURCE_COMMIT,
"source_tree": SOURCE_TREE,
"upstream_commit": UPSTREAM_COMMIT,
"source_date_epoch": 1,
"generated_at": "1970-01-01T00:00:01+00:00",
"rustc_release": rustc_release,
"cargo_release": "1.74.0",
"rustc_sha256": RUSTC_HASH,
"cargo_sha256": CARGO_HASH,
"artifacts": [manifest_artifact(path) for path in paths],
},
)
proof_artifacts = [manifest, *paths]
return write_proof(
directory,
gate,
{
"build-count": "2",
"byte-identical": "true",
"cargo-release": "1.74.0",
"rustc-release": "1.74.0",
},
proof_artifacts,
)
def rebind_reproducible_artifact(
directory: Path, proof: Path, name: str, content: bytes
) -> None:
artifact_path = directory / "artifacts" / "reproducible-release" / name
artifact_path.write_bytes(content)
manifest_path = artifact_path.parent / "manifest.json"
manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
entry = next(item for item in manifest["artifacts"] if item["name"] == name)
entry["size"] = artifact_path.stat().st_size
entry["sha256"] = digest(artifact_path)
write_json(manifest_path, manifest)
payload = json.loads(proof.read_text(encoding="utf-8"))
by_name = {
Path(str(item["path"])).name: item for item in payload["artifacts"]
}
for path in (artifact_path, manifest_path):
item = by_name[path.name]
item["size"] = path.stat().st_size
item["sha256"] = digest(path)
write_json(proof, payload)
def rebind_reproducible_manifest(proof: Path, mutate: Any) -> None:
manifest_path = proof.parent / "artifacts" / "reproducible-release" / "manifest.json"
manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
mutate(manifest)
write_json(manifest_path, manifest)
payload = json.loads(proof.read_text(encoding="utf-8"))
item = next(
item
for item in payload["artifacts"]
if item.get("name") == "manifest.json"
)
item["size"] = manifest_path.stat().st_size
item["sha256"] = digest(manifest_path)
write_json(proof, payload)
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",
}
evidence_items = []
matched = {}
for index, category in enumerate(required, start=1):
item = {
"repository": SECURITY_REPOSITORY,
"requested_source_commit": evidence_commit,
"workflow_id": SECURITY_WORKFLOW_ID,
"workflow_name": "Replacement security gates",
"workflow_path": ".github/workflows/replacement-security-gates.yml",
"run_id": SECURITY_RUN_ID,
"run_attempt": 1,
"html_url": f"https://github.example/actions/runs/{SECURITY_RUN_ID}",
"event": "workflow_dispatch",
"workflow_head_sha": SOURCE_COMMIT,
"workflow_head_branch": "main",
"workflow_head_repository": SECURITY_REPOSITORY,
"workflow_display_title": f"Replacement security gates {SOURCE_COMMIT}",
"created_at": "2026-01-01T00:00:00Z",
"updated_at": "2026-01-01T00:01:00Z",
"job_id": 300 + index,
"job_name": jobs[category],
"job_conclusion": "success",
"job_run_id": SECURITY_RUN_ID,
"job_run_attempt": 1,
"job_head_sha": SOURCE_COMMIT,
"job_head_branch": "main",
"job_workflow_name": f"Replacement security gates {SOURCE_COMMIT}",
"steps": ["Verify exact source identity", "Run security gate"],
}
evidence_items.append(item)
matched[category] = [item]
write_json(
evidence,
{
"schema": 2,
"repository": SECURITY_REPOSITORY,
"source_commit": SOURCE_COMMIT,
"workflow": {
"id": SECURITY_WORKFLOW_ID,
"name": "Replacement security gates",
"path": ".github/workflows/replacement-security-gates.yml",
},
"run": {
"id": SECURITY_RUN_ID,
"attempt": 1,
"event": "workflow_dispatch",
"conclusion": "success",
"repository": SECURITY_REPOSITORY,
"head_repository": SECURITY_REPOSITORY,
"workflow_head_sha": SOURCE_COMMIT,
"head_branch": "main",
"workflow_path": ".github/workflows/replacement-security-gates.yml",
"workflow_name": f"Replacement security gates {SOURCE_COMMIT}",
"display_title": f"Replacement security gates {SOURCE_COMMIT}",
},
"required_categories": required,
"missing": [],
"matched": matched,
"all_successful_job_evidence": evidence_items,
},
)
return write_proof(
directory,
gate,
{"profiles": "fuzz,asan,ubsan,miri,tsan,valgrind"},
[evidence],
)
def rebind_security_evidence(
directory: Path, proof: Path, mutate: Any
) -> None:
evidence = directory / "artifacts" / "security-suite" / "security-evidence.json"
payload = json.loads(evidence.read_text(encoding="utf-8"))
mutate(payload)
write_json(evidence, payload)
proof_payload = json.loads(proof.read_text(encoding="utf-8"))
item = next(
item
for item in proof_payload["artifacts"]
if Path(str(item["path"])).name == evidence.name
)
item["size"] = evidence.stat().st_size
item["sha256"] = digest(evidence)
write_json(proof, proof_payload)
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"RUSTD_RESOLVED_CANDIDATE_BOOT_1_{DAEMON_HASH}",
f"RUSTD_RESOLVED_CANDIDATE_BOOT_2_{DAEMON_HASH}",
f"RUSTD_RESOLVED_CANDIDATE_NSS_BOOT_1_{NSS_HASH}",
f"RUSTD_RESOLVED_CANDIDATE_NSS_BOOT_2_{NSS_HASH}",
f"RUSTD_RESOLVED_BOOT_ROLLBACK_PASS_{UPSTREAM_COMMIT}",
f"RUSTD_RESOLVED_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:
producer = (ROOT / "scripts" / "run-boot-replacement-vm.sh").read_text(encoding="utf-8")
assert '"rollback_verified": True,' in producer
assert '"rollback_healthy": True,' not in producer
assert '"daemon_sha256": daemon_hash,' in producer
assert '"client_sha256": client_hash,' in producer
assert '"nss_module_sha256": nss_hash,' in producer
assert '"candidate_binary_sha256": daemon_hash,' not in producer
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",
daemon_hash="7" * 64,
),
"upstream daemon hash differs",
)
assert_rejected(
validate(
directory,
directory / "upstream-test-75.json",
client_hash="7" * 64,
),
"upstream client hash differs",
)
assert_rejected(
validate(
directory,
directory / "upstream-test-75.json",
nss_hash="7" * 64,
),
"upstream NSS module hash differs",
)
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)
proof = upstream_mdns_fixture(directory)
assert_pass(validate(directory, proof))
assert_rejected(
validate(directory, proof, client_hash="7" * 64),
"upstream client hash differs",
)
assert_rejected(
validate(directory, proof, daemon_hash="7" * 64),
"upstream daemon hash differs",
)
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")
security_mutations = (
(
lambda payload: payload["matched"].update(
{"extra": [payload["matched"]["fuzz"][0]]}
),
"security evidence matched-job set differs",
),
(
lambda payload: payload["workflow"].update({"id": 999}),
"security category evidence is malformed",
),
(
lambda payload: payload["workflow"].update({"name": "Other workflow"}),
"security workflow identity differs",
),
(
lambda payload: payload["workflow"].update({"path": ".github/workflows/other.yml"}),
"security workflow identity differs",
),
(
lambda payload: payload.update({"repository": "attacker/repository"}),
"security proof repository binding differs",
),
(
lambda payload: payload["run"].update({"workflow_head_sha": "8" * 40}),
"security workflow run identity differs",
),
(
lambda payload: payload["run"].update({"display_title": "Other source"}),
"security workflow run identity differs",
),
(
lambda payload: payload["matched"]["fuzz"][0].update(
{"requested_source_commit": "8" * 40}
),
"security category evidence is malformed",
),
(
lambda payload: payload["run"].update({"id": 999}),
"security category evidence is malformed",
),
(
lambda payload: payload["run"].update({"attempt": 0}),
"security workflow run identity differs",
),
(
lambda payload: payload["run"].update({"event": "push"}),
"security workflow run identity differs",
),
(
lambda payload: payload["matched"]["fuzz"][0].update(
{"job_id": payload["matched"]["asan"][0]["job_id"]}
),
"security evidence contains duplicate job ids",
),
(
lambda payload: payload["matched"]["fuzz"][0].update(
{"steps": ["Run security gate"]}
),
"security category evidence is malformed",
),
)
for mutate, expected_error in security_mutations:
with tempfile.TemporaryDirectory(prefix="replacement-proof-test-") as name:
directory = Path(name)
proof = security_fixture(directory)
rebind_security_evidence(directory, proof, mutate)
assert_rejected(validate(directory, proof), expected_error)
with tempfile.TemporaryDirectory(prefix="replacement-proof-test-") as name:
directory = Path(name)
assert_pass(validate(directory, boot_fixture(directory)))
assert_rejected(
validate(
directory,
directory / "boot-replacement.json",
nss_hash="7" * 64,
),
"boot NSS module hash differs",
)
assert_rejected(
validate(
directory,
directory / "boot-replacement.json",
daemon_hash="7" * 64,
),
"boot daemon hash differs",
)
assert_rejected(
validate(
directory,
directory / "boot-replacement.json",
client_hash="7" * 64,
),
"boot client hash differs",
)
stale = boot_fixture(directory, "6" * 40)
assert_rejected(validate(directory, stale), "boot evidence source tree is stale")
with tempfile.TemporaryDirectory(prefix="replacement-proof-test-") as name:
directory = Path(name)
proof = reproducible_fixture(directory)
daemon = digest(directory / "artifacts" / "reproducible-release" / "systemd-resolved")
client = digest(directory / "artifacts" / "reproducible-release" / "resolvectl")
nss = digest(directory / "artifacts" / "reproducible-release" / "libnss_resolve.so.2")
local_reproducible = directory / "local-reproducible"
shutil.copytree(
directory / "artifacts" / "reproducible-release",
local_reproducible,
)
assert_pass(validate_local_reproducible(directory, daemon, client, nss))
assert_rejected(
validate_local_reproducible(directory, "7" * 64, client, nss),
"systemd-resolved hash differs",
)
assert_pass(
validate(
directory,
proof,
daemon_hash=daemon,
client_hash=client,
nss_hash=nss,
local_reproducible=local_reproducible,
)
)
assert_rejected(
validate(directory, proof, daemon_hash="7" * 64, client_hash=client, nss_hash=nss),
"reproducible daemon hash differs",
)
assert_rejected(
validate(directory, proof, daemon_hash=daemon, client_hash="7" * 64, nss_hash=nss),
"reproducible client hash differs",
)
assert_rejected(
validate(directory, proof, daemon_hash=daemon, client_hash=client, nss_hash="7" * 64),
"reproducible NSS module hash differs",
)
for artifact_name, replacement in (
("rustd-resolved.tar.gz", b"rebound external package\n"),
("files.sha256", b"rebound external file list\n"),
(
"rust-toolchain.txt",
(
"rustc 1.74.0\ncargo 1.74.0\n"
f"rustc_binary_sha256 {RUSTC_HASH}\n"
f"cargo_binary_sha256 {CARGO_HASH}\n"
"external-only marker\n"
).encode(),
),
):
with tempfile.TemporaryDirectory(prefix="replacement-proof-test-") as name:
directory = Path(name)
proof = reproducible_fixture(directory)
external = directory / "artifacts" / "reproducible-release"
local_reproducible = directory / "local-reproducible"
shutil.copytree(external, local_reproducible)
daemon = digest(external / "systemd-resolved")
client = digest(external / "resolvectl")
nss = digest(external / "libnss_resolve.so.2")
rebind_reproducible_artifact(
directory, proof, artifact_name, replacement
)
assert_rejected(
validate(
directory,
proof,
daemon_hash=daemon,
client_hash=client,
nss_hash=nss,
local_reproducible=local_reproducible,
),
"external reproducible artifacts differ from the local canonical release",
)
with tempfile.TemporaryDirectory(prefix="replacement-proof-test-") as name:
directory = Path(name)
proof = reproducible_fixture(directory, rustc_release="1.75.0")
assert_rejected(validate(directory, proof), "reproducible manifest rustc release differs")
with tempfile.TemporaryDirectory(prefix="replacement-proof-test-") as name:
directory = Path(name)
proof = reproducible_fixture(directory)
rebind_reproducible_manifest(
proof, lambda manifest: manifest.update({"rustc_sha256": "z" * 64})
)
assert_rejected(
validate(directory, proof),
"reproducible manifest rustc executable hash is invalid",
)
with tempfile.TemporaryDirectory(prefix="replacement-proof-test-") as name:
directory = Path(name)
proof = reproducible_fixture(directory)
rebind_reproducible_manifest(
proof, lambda manifest: manifest.update({"rustc_sha256": "7" * 64})
)
assert_rejected(
validate(directory, proof),
"reproducible toolchain executable hash evidence differs",
)
with tempfile.TemporaryDirectory(prefix="replacement-proof-test-") as name:
directory = Path(name)
proof = reproducible_fixture(directory, omit_artifact="files.sha256")
assert_rejected(validate(directory, proof), "reproducible proof artifact set differs")
print("replacement proof validator regression tests passed")
if __name__ == "__main__":
main()