from __future__ import annotations
import argparse
from datetime import datetime, timezone
import hashlib
import json
import os
from pathlib import Path
import stat
import subprocess
import sys
from typing import Any
class BundleError(RuntimeError):
pass
EXTERNAL_PROOF_GATES = (
"reproducible-release",
"upstream-test-75",
"upstream-test-89-mdns",
"security-suite",
"boot-replacement",
)
SYSTEM_PATH = "/usr/bin:/bin"
BASE_ENV = {
"PATH": SYSTEM_PATH,
"LANG": "C",
"LC_ALL": "C",
"TZ": "UTC",
}
GIT_ENV = {
**BASE_ENV,
"GIT_CONFIG_NOSYSTEM": "1",
"GIT_CONFIG_GLOBAL": "/dev/null",
}
def reject_ambient_controls() -> None:
git_controls = sorted(name for name in os.environ if name.startswith("GIT_"))
if git_controls:
raise BundleError(
"ambient Git controls are not permitted: " + ", ".join(git_controls)
)
python_controls = sorted(
name
for name in os.environ
if name
in {
"PYTHONHOME",
"PYTHONINSPECT",
"PYTHONPATH",
"PYTHONSTARTUP",
"PYTHONWARNINGS",
}
)
if python_controls:
raise BundleError(
"ambient Python controls are not permitted: "
+ ", ".join(python_controls)
)
def is_sha256(value: Any) -> bool:
return (
isinstance(value, str)
and len(value) == 64
and all(character in "0123456789abcdef" for character in value)
)
def sha256(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 resolve_artifact(
certificate: Path, entry: dict[str, Any], label: str
) -> tuple[Path, str]:
expected = entry.get("sha256")
if not is_sha256(expected):
raise BundleError(f"{label} hash is missing or invalid")
relative = entry.get("artifact_path")
if not isinstance(relative, str) or not relative:
raise BundleError(f"{label} artifact path is missing")
relative_path = Path(relative)
if relative_path.is_absolute() or ".." in relative_path.parts:
raise BundleError(f"{label} artifact path is unsafe")
candidate = (certificate.parent / relative_path).resolve()
parent = certificate.parent.resolve()
if candidate != parent and parent not in candidate.parents:
raise BundleError(f"{label} artifact escapes the readiness bundle")
if not candidate.is_file():
raise BundleError(f"{label} artifact is missing: {candidate}")
actual = sha256(candidate)
if actual != expected:
raise BundleError(
f"{label} artifact hash mismatch: expected {expected}, got {actual}"
)
return candidate, expected
def arguments() -> argparse.Namespace:
parser = argparse.ArgumentParser()
parser.add_argument("--certificate", required=True, type=Path)
parser.add_argument("--maximum-age", type=int, default=86400)
parser.add_argument("--shell-values", action="store_true")
parser.add_argument(
"--source-root",
type=Path,
default=Path(__file__).resolve().parents[1],
)
return parser.parse_args()
def git_bytes(source_root: Path, *arguments: str) -> bytes:
try:
return subprocess.check_output(
[
"/usr/bin/git",
"--no-replace-objects",
"-c",
f"safe.directory={source_root}",
"-c",
"core.fsmonitor=false",
"-c",
"core.untrackedCache=false",
"-c",
"core.ignorestat=false",
"-C",
str(source_root),
*arguments,
],
stderr=subprocess.PIPE,
env=GIT_ENV,
)
except subprocess.CalledProcessError as error:
detail = (
error.stderr.decode("utf-8", "replace").strip()
if error.stderr
else str(error)
)
raise BundleError(f"git {' '.join(arguments)} failed: {detail}") from error
def git(source_root: Path, *arguments: str) -> str:
return git_bytes(source_root, *arguments).decode("utf-8", "strict").strip()
def verify_index_worktree(source_root: Path) -> None:
flags = git_bytes(source_root, "ls-files", "-v", "-z")
for entry in flags.split(b"\0"):
if entry and not entry.startswith(b"H "):
path = os.fsdecode(entry[2:] if len(entry) > 2 else entry)
raise BundleError(f"tracked path has unsafe index flags: {path}")
index = git_bytes(source_root, "ls-files", "--stage", "-z")
for entry in index.split(b"\0"):
if not entry:
continue
metadata, separator, relative = entry.partition(b"\t")
fields = metadata.split()
if not separator or len(fields) != 3 or fields[2] != b"0":
raise BundleError("tracked index entry is malformed or unmerged")
mode, oid = fields[:2]
path = source_root / os.fsdecode(relative)
try:
status = path.lstat()
except OSError as error:
raise BundleError(
f"tracked path is missing or unsafe: {path}: {error}"
) from error
expected = git_bytes(source_root, "cat-file", "blob", oid.decode("ascii"))
if mode in {b"100644", b"100755"}:
if not stat.S_ISREG(status.st_mode) or path.is_symlink():
raise BundleError(
f"tracked regular-file type differs from the index: {path}"
)
executable = bool(status.st_mode & stat.S_IXUSR)
if executable != (mode == b"100755") or path.read_bytes() != expected:
raise BundleError(f"tracked file differs from the index: {path}")
elif mode == b"120000":
if (
not stat.S_ISLNK(status.st_mode)
or os.fsencode(os.readlink(path)) != expected
):
raise BundleError(
f"tracked symbolic link differs from the index: {path}"
)
else:
raise BundleError(
f"unsupported tracked index mode {mode.decode('ascii', 'replace')}: {path}"
)
def verify_clean_source(source_root: Path) -> None:
try:
verify_index_worktree(source_root)
except BundleError as error:
raise BundleError(f"current checkout is not clean: {error}") from error
if git(source_root, "status", "--porcelain=v1", "--untracked-files=all"):
raise BundleError("current checkout is not clean")
def validate_reproducible_release(
certificate: Path,
data: dict[str, Any],
source_commit: str,
source_tree: str,
upstream_commit: str,
binary_hash: str,
client_hash: str,
nss_hash: str,
) -> dict[str, str]:
release = data.get("reproducible_release")
if not isinstance(release, dict) or set(release) != {
"manifest",
"package",
"files",
"toolchain",
}:
raise BundleError("reproducible release metadata is incomplete")
resolved: dict[str, tuple[Path, str]] = {}
for key in ("manifest", "package", "files", "toolchain"):
entry = release.get(key)
if not isinstance(entry, dict):
raise BundleError(f"reproducible {key} metadata is incomplete")
resolved[key] = resolve_artifact(certificate, entry, f"reproducible {key}")
manifest_path = resolved["manifest"][0]
try:
manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
except (OSError, json.JSONDecodeError) as error:
raise BundleError(f"cannot read reproducible manifest: {error}") from error
if not isinstance(manifest, dict):
raise BundleError("reproducible manifest root is not an object")
if manifest.get("schema") != 2 or manifest.get("reproducible") is not True:
raise BundleError("reproducible manifest schema or result is invalid")
if manifest.get("build_count") != 2:
raise BundleError("reproducible manifest build count differs")
if manifest.get("source_commit") != source_commit:
raise BundleError("reproducible manifest source commit differs")
if manifest.get("source_tree") != source_tree:
raise BundleError("reproducible manifest source tree differs")
if manifest.get("upstream_commit") != upstream_commit:
raise BundleError("reproducible manifest upstream commit differs")
source_date_epoch = manifest.get("source_date_epoch")
if (
not isinstance(source_date_epoch, int)
or source_date_epoch < 0
or source_date_epoch > 253402300799
):
raise BundleError("reproducible manifest source date epoch is invalid")
expected_generated_at = datetime.fromtimestamp(
source_date_epoch, tz=timezone.utc
).isoformat()
if manifest.get("generated_at") != expected_generated_at:
raise BundleError("reproducible manifest generated timestamp differs")
if manifest.get("rustc_release") != "1.74.0" or manifest.get("cargo_release") != "1.74.0":
raise BundleError("reproducible manifest toolchain differs")
for label, value in (
("rustc", manifest.get("rustc_sha256")),
("cargo", manifest.get("cargo_sha256")),
):
if not is_sha256(value):
raise BundleError(
f"reproducible manifest {label} executable hash is invalid"
)
expected_identical = {
"systemd-resolved",
"resolvectl",
"libnss_resolve.so.2",
"files.sha256",
"rustd-resolved.tar.gz",
}
raw_identical = manifest.get("byte_identical")
if (
not isinstance(raw_identical, list)
or len(raw_identical) != len(expected_identical)
or set(raw_identical) != expected_identical
):
raise BundleError("reproducible manifest byte-identical set differs")
raw_artifacts = manifest.get("artifacts")
expected_names = expected_identical | {"rust-toolchain.txt"}
if not isinstance(raw_artifacts, list):
raise BundleError("reproducible manifest artifact list is missing")
entries: dict[str, dict[str, Any]] = {}
for item in raw_artifacts:
if not isinstance(item, dict):
raise BundleError("reproducible manifest artifact entry is malformed")
name = item.get("name")
if (
not isinstance(name, str)
or not name
or Path(name).name != name
or name in entries
):
raise BundleError("reproducible manifest artifact name is invalid")
entries[name] = item
if set(entries) != expected_names:
raise BundleError("reproducible manifest artifact set differs")
bundle_parent = certificate.parent.resolve()
for name, entry in entries.items():
path = (manifest_path.parent / name).resolve()
if bundle_parent not in path.parents:
raise BundleError(f"reproducible artifact escapes the readiness bundle: {name}")
if not path.is_file():
raise BundleError(f"reproducible artifact is missing: {name}")
if (
not isinstance(entry.get("size"), int)
or entry.get("size") < 0
or not is_sha256(entry.get("sha256"))
or entry.get("size") != path.stat().st_size
or entry.get("sha256") != sha256(path)
):
raise BundleError(f"reproducible artifact binding differs: {name}")
for key, name in (
("package", "rustd-resolved.tar.gz"),
("files", "files.sha256"),
("toolchain", "rust-toolchain.txt"),
):
if resolved[key][0] != (manifest_path.parent / name).resolve():
raise BundleError(f"reproducible {key} path differs from the manifest")
if entries["systemd-resolved"].get("sha256") != binary_hash:
raise BundleError("reproducible daemon hash differs from the certificate")
if entries["resolvectl"].get("sha256") != client_hash:
raise BundleError("reproducible client hash differs from the certificate")
if entries["libnss_resolve.so.2"].get("sha256") != nss_hash:
raise BundleError("reproducible NSS hash differs from the certificate")
toolchain_text = resolved["toolchain"][0].read_text(encoding="utf-8")
if "rustc 1.74.0" not in toolchain_text or "cargo 1.74.0" not in toolchain_text:
raise BundleError("reproducible toolchain evidence differs")
if (
f"rustc_binary_sha256 {manifest['rustc_sha256']}" not in toolchain_text
or f"cargo_binary_sha256 {manifest['cargo_sha256']}" not in toolchain_text
):
raise BundleError("reproducible toolchain executable hash evidence differs")
return {key: str(value[0]) for key, value in resolved.items()}
def validate_external_proofs(
certificate: Path,
data: dict[str, Any],
source_root: Path,
source_commit: str,
source_tree: str,
upstream_commit: str,
binary_hash: str,
client_hash: str,
nss_hash: str,
local_reproducible_directory: Path,
) -> dict[str, dict[str, object]]:
external = data.get("external_proofs")
expected_gates = set(EXTERNAL_PROOF_GATES)
if not isinstance(external, dict) or set(external) != expected_gates:
raise BundleError("external proof set differs from the certificate contract")
validator = source_root / "scripts" / "validate-replacement-proof.py"
if validator.is_symlink() or not validator.is_file():
raise BundleError("checked-in replacement proof validator is missing or unsafe")
proof_root: Path | None = None
verified: dict[str, dict[str, object]] = {}
for gate in EXTERNAL_PROOF_GATES:
raw_gate = external.get(gate)
if not isinstance(raw_gate, dict) or set(raw_gate) != {"proof", "artifacts"}:
raise BundleError(f"external proof metadata is malformed: {gate}")
proof_entry = raw_gate.get("proof")
if not isinstance(proof_entry, dict):
raise BundleError(f"external proof file metadata is malformed: {gate}")
proof, proof_hash = resolve_artifact(
certificate, proof_entry, f"external {gate} proof"
)
if proof.name != f"{gate}.json":
raise BundleError(f"external proof filename differs: {gate}")
if proof_root is None:
proof_root = proof.parent
elif proof.parent != proof_root:
raise BundleError("external proof files do not share one bundle root")
try:
proof_payload = json.loads(proof.read_text(encoding="utf-8"))
except (OSError, json.JSONDecodeError) as error:
raise BundleError(f"cannot read external proof {gate}: {error}") from error
raw_proof_artifacts = (
proof_payload.get("artifacts")
if isinstance(proof_payload, dict)
else None
)
raw_certificate_artifacts = raw_gate.get("artifacts")
if (
not isinstance(raw_proof_artifacts, list)
or not raw_proof_artifacts
or not isinstance(raw_certificate_artifacts, list)
or len(raw_certificate_artifacts) != len(raw_proof_artifacts)
):
raise BundleError(f"external proof artifact set is incomplete: {gate}")
proof_entries: dict[str, tuple[int, str]] = {}
for raw in raw_proof_artifacts:
if not isinstance(raw, dict):
raise BundleError(f"external proof artifact is malformed: {gate}")
name = raw.get("name")
size = raw.get("size")
expected_hash = raw.get("sha256")
if (
not isinstance(name, str)
or not name
or name in {".", ".."}
or Path(name).name != name
or name in proof_entries
or not isinstance(size, int)
or size < 0
or not is_sha256(expected_hash)
):
raise BundleError(f"external proof artifact entry is invalid: {gate}")
proof_entries[name] = (size, expected_hash)
certificate_entries: dict[str, tuple[int, str, Path]] = {}
for raw in raw_certificate_artifacts:
if not isinstance(raw, dict) or set(raw) != {
"name",
"size",
"artifact_path",
"sha256",
}:
raise BundleError(
f"external certificate artifact entry is malformed: {gate}"
)
name = raw.get("name")
size = raw.get("size")
if (
not isinstance(name, str)
or not name
or name in {".", ".."}
or Path(name).name != name
or name in certificate_entries
or not isinstance(size, int)
or size < 0
):
raise BundleError(
f"external certificate artifact entry is invalid: {gate}"
)
path, expected_hash = resolve_artifact(
certificate, raw, f"external {gate} artifact {name}"
)
expected_path = (proof.parent / "artifacts" / gate / name).resolve()
if path != expected_path or path.stat().st_size != size:
raise BundleError(f"external proof artifact path or size differs: {gate}/{name}")
certificate_entries[name] = (size, expected_hash, path)
certificate_bindings = {
name: (size, expected_hash)
for name, (size, expected_hash, _) in certificate_entries.items()
}
if certificate_bindings != proof_entries:
raise BundleError(f"external proof artifact bindings differ: {gate}")
assert proof_root is not None
command = [
"/usr/bin/python3",
"-I",
str(validator),
"--proof",
str(proof),
"--gate",
gate,
"--source-commit",
source_commit,
"--source-tree",
source_tree,
"--upstream-commit",
upstream_commit,
"--expected-daemon-sha256",
binary_hash,
"--expected-client-sha256",
client_hash,
"--expected-nss-sha256",
nss_hash,
"--proof-directory",
str(proof_root),
]
if gate == "reproducible-release":
command.extend(
[
"--local-reproducible-directory",
str(local_reproducible_directory),
]
)
completed = subprocess.run(
command,
check=False,
capture_output=True,
text=True,
env=BASE_ENV,
)
if completed.returncode != 0:
detail = completed.stderr.strip() or completed.stdout.strip()
raise BundleError(f"external proof validation failed for {gate}: {detail}")
verified[gate] = {
"proof": str(proof),
"sha256": proof_hash,
"artifacts": {
name: str(entry[2])
for name, entry in sorted(certificate_entries.items())
},
}
return verified
def main() -> int:
os.environ["PATH"] = SYSTEM_PATH
options = arguments()
reject_ambient_controls()
if options.maximum_age <= 0:
raise BundleError("maximum certificate age must be positive")
source_root = options.source_root.resolve()
verify_clean_source(source_root)
contract_path = source_root / "scripts" / "replacement-certificate-contract.json"
try:
contract = json.loads(contract_path.read_text(encoding="utf-8"))
except (OSError, json.JSONDecodeError) as error:
raise BundleError(f"cannot read replacement certificate contract: {error}") from error
if not isinstance(contract, dict) or set(contract) != {"schema", "required_gates"}:
raise BundleError("replacement certificate contract is invalid")
expected_schema = contract.get("schema")
expected_gate_order = contract.get("required_gates")
if (
expected_schema != 3
or not isinstance(expected_gate_order, list)
or not all(isinstance(name, str) and name for name in expected_gate_order)
):
raise BundleError("replacement certificate contract is invalid")
expected_gates = set(expected_gate_order)
if len(expected_gates) != len(expected_gate_order):
raise BundleError("replacement certificate contract contains duplicate gates")
certificate = options.certificate.resolve()
try:
data = json.loads(certificate.read_text(encoding="utf-8"))
except (OSError, json.JSONDecodeError) as error:
raise BundleError(f"cannot read certificate: {error}") from error
if not isinstance(data, dict) or data.get("schema") != expected_schema:
raise BundleError("unsupported certificate schema")
if data.get("certified") is not True:
raise BundleError("certificate is not certified")
if data.get("contract_errors") != []:
raise BundleError("certificate reports contract errors")
gates = data.get("gates")
if not isinstance(gates, list) or not gates:
raise BundleError("certificate contains no gates")
if any(
not isinstance(gate, dict)
or not isinstance(gate.get("name"), str)
or not gate.get("name")
or gate.get("status") != "pass"
for gate in gates
):
raise BundleError("certificate contains a nonpassing gate")
gate_names = [str(gate["name"]) for gate in gates]
if len(gate_names) != len(set(gate_names)):
raise BundleError("certificate contains duplicate gates")
if set(gate_names) != expected_gates:
raise BundleError("certificate gate set differs from the contract")
try:
generated = datetime.fromisoformat(str(data["generated_at"]))
except (KeyError, ValueError) as error:
raise BundleError("certificate timestamp is invalid") from error
if generated.tzinfo is None:
raise BundleError("certificate timestamp has no timezone")
age = (datetime.now(timezone.utc) - generated.astimezone(timezone.utc)).total_seconds()
if age < -300 or age > options.maximum_age:
raise BundleError(
f"certificate age {age:.0f}s is outside the allowed window"
)
source_commit = data.get("source_commit")
source_tree = data.get("source_tree")
upstream_commit = data.get("upstream_commit")
for label, value in (
("source commit", source_commit),
("source tree", source_tree),
("upstream commit", upstream_commit),
):
if (
not isinstance(value, str)
or len(value) != 40
or any(character not in "0123456789abcdef" for character in value)
):
raise BundleError(f"{label} is invalid")
if git(source_root, "rev-parse", "HEAD") != source_commit:
raise BundleError("current checkout commit differs from the certificate")
if git(source_root, "rev-parse", "HEAD^{tree}") != source_tree:
raise BundleError("current checkout tree differs from the certificate")
baseline_path = source_root / "compat" / "upstream-systemd" / "commit"
try:
tracked_upstream_commit = baseline_path.read_text(encoding="ascii").strip()
except OSError as error:
raise BundleError(f"cannot read tracked upstream baseline: {error}") from error
if (
len(tracked_upstream_commit) != 40
or any(
character not in "0123456789abcdef"
for character in tracked_upstream_commit
)
or tracked_upstream_commit != upstream_commit
):
raise BundleError("certificate upstream commit differs from the tracked baseline")
toolchain = data.get("toolchain")
if (
not isinstance(toolchain, dict)
or set(toolchain)
!= {"rustc_release", "cargo_release", "rustc_sha256", "cargo_sha256"}
or toolchain.get("rustc_release") != "1.74.0"
or toolchain.get("cargo_release") != "1.74.0"
or not is_sha256(toolchain.get("rustc_sha256"))
or not is_sha256(toolchain.get("cargo_sha256"))
):
raise BundleError("certificate toolchain differs")
binary_entry = data.get("binary")
client_entry = data.get("client")
nss_entry = data.get("nss")
if (
not isinstance(binary_entry, dict)
or not isinstance(client_entry, dict)
or not isinstance(nss_entry, dict)
):
raise BundleError("certificate release artifact metadata is incomplete")
binary, binary_hash = resolve_artifact(certificate, binary_entry, "daemon")
client, client_hash = resolve_artifact(certificate, client_entry, "client")
nss, nss_hash = resolve_artifact(certificate, nss_entry, "NSS module")
reproducible = validate_reproducible_release(
certificate,
data,
source_commit,
source_tree,
upstream_commit,
binary_hash,
client_hash,
nss_hash,
)
reproducible_manifest = json.loads(
Path(reproducible["manifest"]).read_text(encoding="utf-8")
)
if (
toolchain["rustc_sha256"] != reproducible_manifest.get("rustc_sha256")
or toolchain["cargo_sha256"] != reproducible_manifest.get("cargo_sha256")
):
raise BundleError("certificate toolchain hashes differ from the manifest")
external_proofs = validate_external_proofs(
certificate,
data,
source_root,
source_commit,
source_tree,
upstream_commit,
binary_hash,
client_hash,
nss_hash,
Path(reproducible["manifest"]).parent,
)
verify_clean_source(source_root)
if options.shell_values:
for value in (
source_commit,
source_tree,
str(binary),
binary_hash,
str(client),
client_hash,
str(nss),
nss_hash,
upstream_commit,
):
print(value)
else:
print(
json.dumps(
{
"certified": True,
"certificate": str(certificate),
"age_seconds": int(age),
"source_commit": source_commit,
"source_tree": source_tree,
"upstream_commit": upstream_commit,
"binary": {"path": str(binary), "sha256": binary_hash},
"client": {"path": str(client), "sha256": client_hash},
"nss": {"path": str(nss), "sha256": nss_hash},
"reproducible_release": reproducible,
"external_proofs": external_proofs,
},
indent=2,
sort_keys=True,
)
)
return 0
if __name__ == "__main__":
try:
raise SystemExit(main())
except (OSError, BundleError) as error:
print(f"verify-readiness-bundle: {error}", file=sys.stderr)
raise SystemExit(1) from error