from __future__ import annotations
from datetime import datetime, timezone
import hashlib
import importlib.util
import json
import os
from pathlib import Path
import shlex
import shutil
import subprocess
import sys
import tempfile
from typing import Any
ROOT = Path(__file__).resolve().parents[1]
VERIFIER = ROOT / "scripts" / "verify-readiness-bundle.py"
SWITCH = ROOT / "scripts" / "switch-resolved-transactionally-v2.sh"
BUILDER = ROOT / "scripts" / "build-reproducible-release.sh"
CERTIFIER = ROOT / "scripts" / "certify-replacement-v2.sh"
PROOF_FIXTURE_PATH = ROOT / "tests" / "test-replacement-proof-validator.py"
CONTRACT = json.loads(
(ROOT / "scripts" / "replacement-certificate-contract.json").read_text(
encoding="utf-8"
)
)
PROOF_FIXTURE_SPEC = importlib.util.spec_from_file_location(
"replacement_proof_fixtures", PROOF_FIXTURE_PATH
)
assert PROOF_FIXTURE_SPEC is not None and PROOF_FIXTURE_SPEC.loader is not None
PROOF_FIXTURES = importlib.util.module_from_spec(PROOF_FIXTURE_SPEC)
PROOF_FIXTURE_SPEC.loader.exec_module(PROOF_FIXTURES)
CC_RS_ENV_BASES = {
"CC",
"CXX",
"AR",
"CFLAGS",
"CXXFLAGS",
"ARFLAGS",
"RANLIB",
"RANLIBFLAGS",
}
def is_cc_rs_override(name: str) -> bool:
if name == "CRATE_CC_NO_DEFAULTS":
return True
if name.startswith(("HOST_", "TARGET_")):
return name.split("_", 1)[1] in CC_RS_ENV_BASES
return any(name.startswith(f"{base}_") for base in CC_RS_ENV_BASES)
def digest(path: Path) -> str:
return hashlib.sha256(path.read_bytes()).hexdigest()
def git(directory: Path, *arguments: str) -> str:
return subprocess.check_output(
["git", "-C", str(directory), *arguments], text=True
).strip()
def write_json(path: Path, payload: dict[str, Any]) -> None:
path.write_text(json.dumps(payload, sort_keys=True) + "\n", encoding="utf-8")
def run(
certificate: Path,
source_root: Path,
*arguments: str,
environment: dict[str, str] | None = None,
) -> subprocess.CompletedProcess[str]:
if environment is None:
environment = os.environ.copy()
for name in list(environment):
if name.startswith("GIT_"):
environment.pop(name)
return subprocess.run(
[
"/usr/bin/python3",
"-I",
str(VERIFIER),
"--certificate",
str(certificate),
"--source-root",
str(source_root),
*arguments,
],
check=False,
capture_output=True,
text=True,
env=environment,
)
def initialize_source(directory: Path, contract: dict[str, Any]) -> tuple[Path, str, str]:
source = directory / "source"
scripts = source / "scripts"
scripts.mkdir(parents=True)
write_json(scripts / "replacement-certificate-contract.json", contract)
(scripts / "validate-replacement-proof.py").write_bytes(
(ROOT / "scripts" / "validate-replacement-proof.py").read_bytes()
)
baseline = source / "compat" / "upstream-systemd"
baseline.mkdir(parents=True)
(baseline / "commit").write_text("3" * 40 + "\n", encoding="ascii")
subprocess.run(["git", "init", "--quiet", str(source)], check=True)
subprocess.run(
["git", "-C", str(source), "config", "user.email", "test@example.invalid"],
check=True,
)
subprocess.run(
["git", "-C", str(source), "config", "user.name", "Readiness Test"],
check=True,
)
subprocess.run(["git", "-C", str(source), "add", "."], check=True)
subprocess.run(
["git", "-C", str(source), "commit", "--quiet", "-m", "fixture"],
check=True,
)
return source, git(source, "rev-parse", "HEAD"), git(source, "rev-parse", "HEAD^{tree}")
def portable_entry(bundle: Path, path: Path) -> dict[str, str]:
return {
"path": f"/stale/runner/{path.name}",
"artifact_path": str(path.relative_to(bundle)),
"sha256": digest(path),
}
def portable_external_proof(
bundle: Path, proof_root: Path, proof: Path
) -> dict[str, object]:
payload = json.loads(proof.read_text(encoding="utf-8"))
gate = payload["gate"]
entries = []
for raw in payload["artifacts"]:
name = raw["name"]
path = proof_root / "artifacts" / gate / name
entries.append(
{
"name": name,
"size": path.stat().st_size,
"artifact_path": str(path.relative_to(bundle)),
"sha256": digest(path),
}
)
return {"proof": portable_entry(bundle, proof), "artifacts": entries}
def fixture(
directory: Path,
contract: dict[str, Any] | None = None,
) -> tuple[Path, dict[str, Any], Path, Path, Path]:
active_contract = CONTRACT if contract is None else contract
source, source_commit, source_tree = initialize_source(directory, active_contract)
bundle = directory / "bundle"
artifacts = bundle / "certificate.d" / "artifacts"
reproducible = artifacts / "reproducible-release"
reproducible.mkdir(parents=True)
final_paths: dict[str, Path] = {}
for key, name, content in (
("binary", "systemd-resolved", b"daemon\n"),
("client", "resolvectl", b"client\n"),
("nss", "libnss_resolve.so.2", b"nss\n"),
):
path = artifacts / name
path.write_bytes(content)
final_paths[key] = path
(reproducible / name).write_bytes(content)
(reproducible / "rustd-resolved.tar.gz").write_bytes(b"package\n")
(reproducible / "files.sha256").write_bytes(b"files\n")
(reproducible / "rust-toolchain.txt").write_text(
"rustc 1.74.0\n"
"cargo 1.74.0\n"
f"rustc_binary_sha256 {'8' * 64}\n"
f"cargo_binary_sha256 {'9' * 64}\n",
encoding="utf-8",
)
manifest_artifacts = []
for name in (
"systemd-resolved",
"resolvectl",
"libnss_resolve.so.2",
"rustd-resolved.tar.gz",
"files.sha256",
"rust-toolchain.txt",
):
path = reproducible / name
manifest_artifacts.append(
{"name": name, "size": path.stat().st_size, "sha256": digest(path)}
)
manifest = reproducible / "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": "3" * 40,
"source_date_epoch": 1,
"generated_at": datetime.fromtimestamp(1, tz=timezone.utc).isoformat(),
"rustc_release": "1.74.0",
"cargo_release": "1.74.0",
"rustc_sha256": "8" * 64,
"cargo_sha256": "9" * 64,
"artifacts": manifest_artifacts,
},
)
proof_root = artifacts / "external-proofs"
PROOF_FIXTURES.SOURCE_COMMIT = source_commit
PROOF_FIXTURES.SOURCE_TREE = source_tree
PROOF_FIXTURES.UPSTREAM_COMMIT = "3" * 40
PROOF_FIXTURES.DAEMON_HASH = digest(final_paths["binary"])
PROOF_FIXTURES.CLIENT_HASH = digest(final_paths["client"])
PROOF_FIXTURES.NSS_HASH = digest(final_paths["nss"])
proofs = {
"reproducible-release": PROOF_FIXTURES.reproducible_fixture(proof_root),
"upstream-test-75": PROOF_FIXTURES.upstream_fixture(
proof_root, PROOF_FIXTURES.NSS_HASH
),
"upstream-test-89-mdns": PROOF_FIXTURES.upstream_mdns_fixture(proof_root),
"security-suite": PROOF_FIXTURES.security_fixture(proof_root, source_commit),
"boot-replacement": PROOF_FIXTURES.boot_fixture(proof_root, source_tree),
}
required_gates = active_contract.get("required_gates", [])
payload: dict[str, Any] = {
"schema": 3,
"certified": True,
"contract_errors": [],
"generated_at": datetime.now(timezone.utc).isoformat(),
"source_commit": source_commit,
"source_tree": source_tree,
"upstream_commit": "3" * 40,
"toolchain": {
"rustc_release": "1.74.0",
"cargo_release": "1.74.0",
"rustc_sha256": "8" * 64,
"cargo_sha256": "9" * 64,
},
"gates": [{"name": name, "status": "pass"} for name in required_gates],
"binary": portable_entry(bundle, final_paths["binary"]),
"client": portable_entry(bundle, final_paths["client"]),
"nss": portable_entry(bundle, final_paths["nss"]),
"reproducible_release": {
"manifest": portable_entry(bundle, manifest),
"package": portable_entry(bundle, reproducible / "rustd-resolved.tar.gz"),
"files": portable_entry(bundle, reproducible / "files.sha256"),
"toolchain": portable_entry(bundle, reproducible / "rust-toolchain.txt"),
},
"external_proofs": {
gate: portable_external_proof(bundle, proof_root, proof)
for gate, proof in proofs.items()
},
}
certificate = bundle / "certificate.json"
write_json(certificate, payload)
return certificate, payload, final_paths["nss"], source, manifest
def assert_rejected(result: subprocess.CompletedProcess[str], detail: str) -> None:
assert result.returncode != 0, result.stdout
assert detail in result.stderr, result.stderr
def switch_fixture(directory: Path) -> dict[str, Any]:
repository = directory / "repository"
scripts = repository / "scripts"
scripts.mkdir(parents=True)
state = directory / "state"
system_copy = directory / "system-copy"
guard = directory / "guard.service"
dropin = directory / "dropin"
install_base = directory / "installed"
dropin.mkdir()
dropin_path = dropin / "90-rustd-resolved.conf"
system_copy.write_text("previous system copy\n", encoding="utf-8")
system_copy.chmod(0o751)
guard.write_text("previous guard\n", encoding="utf-8")
dropin_path.write_text("previous dropin\n", encoding="utf-8")
switch_text = SWITCH.read_text(encoding="utf-8")
replacements = {
"STATE_ROOT=/var/lib/rustd-resolved": f"STATE_ROOT={shlex.quote(str(state))}",
"SYSTEM_COPY=/usr/lib/systemd/rustd-resolved-switch": (
f"SYSTEM_COPY={shlex.quote(str(system_copy))}"
),
"INSTALL_BASE=/usr/lib/systemd/rustd-resolved": (
f"INSTALL_BASE={shlex.quote(str(install_base))}"
),
"GUARD_UNIT=/etc/systemd/system/rustd-resolved-guard.service": (
f"GUARD_UNIT={shlex.quote(str(guard))}"
),
"DROPIN_DIR=/etc/systemd/system/systemd-resolved.service.d": (
f"DROPIN_DIR={shlex.quote(str(dropin))}"
),
"if [[ $MODE == install ]]; then": (
"if [[ $MODE == install && false == true ]]; then"
),
'[[ ${EUID:-$(id -u)} -eq 0 ]] || {': "[[ 0 -eq 0 ]] || {",
}
for original, replacement in replacements.items():
assert original in switch_text
switch_text = switch_text.replace(original, replacement, 1)
test_switch = scripts / SWITCH.name
test_switch.write_text(switch_text, encoding="utf-8")
test_switch.chmod(0o755)
write_json(scripts / "replacement-certificate-contract.json", {"schema": 3})
subprocess.run(["git", "init", "--quiet", str(repository)], check=True)
subprocess.run(
["git", "-C", str(repository), "config", "user.email", "test@example.invalid"],
check=True,
)
subprocess.run(
["git", "-C", str(repository), "config", "user.name", "Switch Test"],
check=True,
)
subprocess.run(["git", "-C", str(repository), "add", "scripts"], check=True)
subprocess.run(
["git", "-C", str(repository), "commit", "--quiet", "-m", "fixture"],
check=True,
)
bundle = directory / "bundle"
bundle.mkdir()
certificate = bundle / "certificate.json"
certificate.write_text("{}\n", encoding="utf-8")
daemon = bundle / "systemd-resolved"
client = bundle / "resolvectl"
nss = bundle / "libnss_resolve.so.2"
daemon.write_text("#!/bin/sh\nexit 0\n", encoding="utf-8")
client.write_text("#!/bin/sh\nexit 0\n", encoding="utf-8")
nss.write_bytes(b"nss\n")
daemon.chmod(0o755)
client.chmod(0o755)
nss.chmod(0o755)
source_commit = git(repository, "rev-parse", "HEAD")
source_tree = git(repository, "rev-parse", "HEAD^{tree}")
upstream_commit = "c" * 40
verifier = scripts / "verify-readiness-bundle.py"
verifier.write_text(
"\n".join(
(
f"print('{source_commit}')",
f"print('{source_tree}')",
f"print('{daemon}')",
f"print('{digest(daemon)}')",
f"print('{client}')",
f"print('{digest(client)}')",
f"print('{nss}')",
f"print('{digest(nss)}')",
f"print('{upstream_commit}')",
)
)
+ "\n",
encoding="utf-8",
)
shim = directory / "shim"
shim.mkdir()
systemctl_marker = directory / "systemctl.log"
systemctl = shim / "systemctl"
systemctl.write_text(
"#!/bin/sh\n"
+ f"printf '%s\\n' \"$*\" >>{shlex.quote(str(systemctl_marker))}\n"
+ "if [ \"${RUSTD_RESOLVED_SWITCH_TEST_SYSTEMCTL_FAIL:-}\" = \"$*\" ]; then\n"
+ " exit 97\n"
+ "fi\n"
+ "case \"$1\" in\n"
+ " is-enabled) printf '%s\\n' disabled ;;\n"
+ " is-active) printf '%s\\n' inactive ;;\n"
+ "esac\n"
+ "exit 0\n",
encoding="utf-8",
)
systemctl.chmod(0o755)
environment = os.environ.copy()
for name in list(environment):
if name.startswith("GIT_"):
environment.pop(name)
environment["PATH"] = str(shim) + os.pathsep + environment["PATH"]
return {
"switch": test_switch,
"certificate": certificate,
"state": state,
"system_copy": system_copy,
"guard": guard,
"dropin": dropin_path,
"install_base": install_base,
"systemctl_marker": systemctl_marker,
"environment": environment,
"source_commit": source_commit,
}
def prepare_rolled_back_transaction(directory: Path) -> tuple[dict[str, Any], Path]:
switch = switch_fixture(directory)
environment = switch["environment"].copy()
environment["RUSTD_RESOLVED_SWITCH_TEST_FAIL_AT"] = "install-root"
result = subprocess.run(
[
"bash",
str(switch["switch"]),
"--certificate",
str(switch["certificate"]),
],
check=False,
text=True,
capture_output=True,
env=environment,
timeout=10,
)
assert result.returncode == 99, result.stderr
transactions = list((switch["state"] / "transactions").iterdir())
assert len(transactions) == 1, transactions
switch["systemctl_marker"].unlink(missing_ok=True)
return switch, transactions[0]
def main() -> None:
with tempfile.TemporaryDirectory(prefix="readiness-bundle-test-") as name:
certificate, payload, nss, source, _ = fixture(Path(name))
result = run(certificate, source, "--shell-values")
assert result.returncode == 0, result.stderr
assert len(result.stdout.splitlines()) == 9, result.stdout
with tempfile.TemporaryDirectory(prefix="readiness-foreign-git-") as foreign_name:
foreign = Path(foreign_name) / "source"
subprocess.run(
["git", "clone", "--quiet", "--no-local", str(source), str(foreign)],
check=True,
)
hidden_dirt = source / "foreign-git-hidden-dirt"
hidden_dirt.write_text("dirty\n", encoding="utf-8")
try:
git_environment = os.environ.copy()
for env_name in list(git_environment):
if env_name.startswith("GIT_"):
git_environment.pop(env_name)
git_environment["GIT_DIR"] = str(foreign / ".git")
git_environment["GIT_WORK_TREE"] = str(foreign)
redirected = run(
certificate,
source,
environment=git_environment,
)
finally:
hidden_dirt.unlink(missing_ok=True)
assert_rejected(redirected, "ambient Git controls are not permitted")
python_environment = os.environ.copy()
for env_name in list(python_environment):
if env_name.startswith("GIT_"):
python_environment.pop(env_name)
python_environment["PYTHONPATH"] = str(Path(name) / "untrusted-python")
assert_rejected(
run(
certificate,
source,
environment=python_environment,
),
"ambient Python controls are not permitted: PYTHONPATH",
)
startup_directory = Path(name) / "python-startup"
startup_directory.mkdir()
startup_marker = Path(name) / "sitecustomize-ran"
(startup_directory / "sitecustomize.py").write_text(
"from pathlib import Path\n"
f"Path({str(startup_marker)!r}).write_text('ran\\n')\n",
encoding="utf-8",
)
startup_environment = os.environ.copy()
for env_name in list(startup_environment):
if env_name.startswith("GIT_"):
startup_environment.pop(env_name)
startup_environment["PYTHONPATH"] = str(startup_directory)
assert_rejected(
run(certificate, source, environment=startup_environment),
"ambient Python controls are not permitted: PYTHONPATH",
)
assert not startup_marker.exists()
contract_path = source / "scripts" / "replacement-certificate-contract.json"
contract_original = contract_path.read_bytes()
try:
subprocess.run(
[
"git",
"-C",
str(source),
"update-index",
"--assume-unchanged",
"scripts/replacement-certificate-contract.json",
],
check=True,
)
contract_path.write_bytes(contract_original + b"\n")
hidden_contract = run(certificate, source)
finally:
contract_path.write_bytes(contract_original)
subprocess.run(
[
"git",
"-C",
str(source),
"update-index",
"--no-assume-unchanged",
"scripts/replacement-certificate-contract.json",
],
check=True,
)
assert_rejected(hidden_contract, "tracked path has unsafe index flags")
verifier_text = VERIFIER.read_text(encoding="utf-8")
assert "safe.directory={source_root}" in verifier_text
assert '"/usr/bin/git"' in verifier_text
assert '"core.fsmonitor=false"' in verifier_text
assert '"/usr/bin/python3",\n "-I"' in verifier_text
switch_text = SWITCH.read_text(encoding="utf-8")
assert '/usr/bin/python3 -I "$VERIFIER"' in switch_text
assert switch_text.index('certificate_output=$(/usr/bin/python3 -I "$VERIFIER"') < switch_text.index(
'install -d -m 0700 "$STATE_ROOT"'
)
assert "Install mode is blocked until authenticated artifact provenance" in switch_text
assert 'install -m 0755 "$0" "$SYSTEM_COPY"' not in switch_text
assert 'install -m 0755 "$SYSTEM_COPY_CANDIDATE" "$SYSTEM_COPY"' in switch_text
assert '"$SOURCE_COMMIT:scripts/switch-resolved-transactionally-v2.sh"' in switch_text
assert "git --no-replace-objects" in switch_text
assert 'certified_git hash-object "$SYSTEM_COPY_CANDIDATE"' in switch_text
assert 'require_active_transaction "$TRANSACTION"' in switch_text
assert "unsupported certificate schema" not in switch_text
assert switch_text.index("trap 'rollback_on_failure $?' EXIT") < switch_text.index(
"mutation_checkpoint install-root"
)
assert "system-copy.previous" in switch_text
assert "validated_transaction_directory" in switch_text
assert "preflight_restore_transaction" in switch_text
assert "write_rollback_phase" in switch_text
assert "Automatic resolver rollback failed with status" in switch_text
blocked_install = subprocess.run(
["bash", str(SWITCH), "--certificate", str(certificate)],
check=False,
text=True,
capture_output=True,
)
assert blocked_install.returncode == 2
assert "authenticated artifact provenance" in blocked_install.stderr
for invalid_transaction in (
"../escape",
"/tmp/escape",
"20260814T010203-1-deadbeefcafe\nescape",
):
rejected_transaction = subprocess.run(
["bash", str(SWITCH), "--confirm", invalid_transaction],
check=False,
text=True,
capture_output=True,
)
assert rejected_transaction.returncode == 2
assert "Invalid transaction id" in rejected_transaction.stderr
for checkpoint in (
"install-root",
"system-copy",
"dropin",
"guard",
"active",
"service",
):
with tempfile.TemporaryDirectory(prefix="switch-fault-test-") as fault_name:
switch = switch_fixture(Path(fault_name))
environment = switch["environment"].copy()
environment["RUSTD_RESOLVED_SWITCH_TEST_FAIL_AT"] = checkpoint
result = subprocess.run(
[
"bash",
str(switch["switch"]),
"--certificate",
str(switch["certificate"]),
],
check=False,
text=True,
capture_output=True,
env=environment,
timeout=10,
)
assert result.returncode == 99, result.stderr
assert switch["system_copy"].read_text(encoding="utf-8") == (
"previous system copy\n"
)
assert switch["system_copy"].stat().st_mode & 0o777 == 0o751
assert switch["guard"].read_text(encoding="utf-8") == "previous guard\n"
assert switch["dropin"].read_text(encoding="utf-8") == "previous dropin\n"
assert not (switch["state"] / "active").exists()
assert not (
switch["install_base"] / switch["source_commit"]
).exists()
with tempfile.TemporaryDirectory(
prefix="switch-explicit-restore-failure-test-"
) as switch_name:
switch, transaction = prepare_rolled_back_transaction(Path(switch_name))
(switch["state"] / "active").symlink_to(transaction)
environment = switch["environment"].copy()
environment["RUSTD_RESOLVED_SWITCH_TEST_SYSTEMCTL_FAIL"] = "daemon-reload"
failed_rollback = subprocess.run(
[
"bash",
str(switch["switch"]),
"--rollback",
transaction.name,
],
check=False,
text=True,
capture_output=True,
env=environment,
timeout=5,
)
assert failed_rollback.returncode != 0
assert "Rolled back resolver transaction" not in failed_rollback.stdout
assert (switch["state"] / "active").resolve() == transaction.resolve()
assert (transaction / "rollback-phase").read_text(encoding="utf-8").strip() == (
"restoring-files"
)
with tempfile.TemporaryDirectory(
prefix="switch-automatic-restore-failure-test-"
) as switch_name:
switch = switch_fixture(Path(switch_name))
environment = switch["environment"].copy()
environment["RUSTD_RESOLVED_SWITCH_TEST_FAIL_AT"] = "service"
environment["RUSTD_RESOLVED_SWITCH_TEST_SYSTEMCTL_FAIL"] = "daemon-reload"
failed_automatic_rollback = subprocess.run(
[
"bash",
str(switch["switch"]),
"--certificate",
str(switch["certificate"]),
],
check=False,
text=True,
capture_output=True,
env=environment,
timeout=10,
)
assert failed_automatic_rollback.returncode == 125
assert "Automatic resolver rollback failed with status" in (
failed_automatic_rollback.stderr
)
assert "Rolled back resolver transaction" not in (
failed_automatic_rollback.stdout
)
transactions = list((switch["state"] / "transactions").iterdir())
assert len(transactions) == 1
transaction = transactions[0]
assert (switch["state"] / "active").resolve() == transaction.resolve()
assert (transaction / "pending").is_file()
for corruption in (
"dropin.previous",
"guard.previous",
"system-copy.previous",
"active.previous-without-marker",
"resolved-enabled.before",
):
with tempfile.TemporaryDirectory(
prefix="switch-preflight-test-"
) as switch_name:
switch, transaction = prepare_rolled_back_transaction(
Path(switch_name)
)
(switch["state"] / "active").symlink_to(transaction)
if corruption == "active.previous-without-marker":
(transaction / "active.previous").symlink_to(transaction)
elif corruption == "resolved-enabled.before":
(transaction / corruption).write_text(
"attacker-controlled\n", encoding="utf-8"
)
else:
(transaction / corruption).unlink()
result = subprocess.run(
[
"bash",
str(switch["switch"]),
"--rollback",
transaction.name,
],
check=False,
text=True,
capture_output=True,
env=switch["environment"],
timeout=5,
)
assert result.returncode != 0, result.stdout
assert not switch["systemctl_marker"].exists()
assert switch["system_copy"].read_text(encoding="utf-8") == (
"previous system copy\n"
)
assert switch["system_copy"].stat().st_mode & 0o777 == 0o751
assert switch["guard"].read_text(encoding="utf-8") == (
"previous guard\n"
)
assert switch["dropin"].read_text(encoding="utf-8") == (
"previous dropin\n"
)
assert (switch["state"] / "active").resolve() == transaction.resolve()
with tempfile.TemporaryDirectory(
prefix="switch-stale-ancestor-test-"
) as switch_name:
switch, ancestor = prepare_rolled_back_transaction(Path(switch_name))
child_id = "20260814T010204-124-feedfacecafe"
child = ancestor.parent / child_id
shutil.copytree(ancestor, child, symlinks=True)
metadata_path = child / "transaction.json"
metadata = json.loads(metadata_path.read_text(encoding="utf-8"))
metadata["transaction"] = child_id
metadata["parent_transaction"] = ancestor.name
write_json(metadata_path, metadata)
(child / "active.existed").write_bytes(b"")
(child / "active.previous").symlink_to(ancestor)
(switch["state"] / "active").symlink_to(child)
result = subprocess.run(
[
"bash",
str(switch["switch"]),
"--rollback",
ancestor.name,
],
check=False,
text=True,
capture_output=True,
env=switch["environment"],
timeout=5,
)
assert result.returncode == 2, result.stdout
assert "Selected transaction is not active" in result.stderr
assert not switch["systemctl_marker"].exists()
assert (switch["state"] / "active").resolve() == child.resolve()
assert switch["system_copy"].read_text(encoding="utf-8") == (
"previous system copy\n"
)
with tempfile.TemporaryDirectory(prefix="switch-git-env-test-") as switch_name:
switch = switch_fixture(Path(switch_name))
environment = switch["environment"].copy()
environment["GIT_DIR"] = str(Path(switch_name) / "foreign.git")
result = subprocess.run(
[
"bash",
str(switch["switch"]),
"--certificate",
str(switch["certificate"]),
],
check=False,
text=True,
capture_output=True,
env=environment,
timeout=5,
)
assert result.returncode == 2
assert "Ambient Git control is not permitted" in result.stderr
assert not switch["state"].exists()
assert not switch["systemctl_marker"].exists()
valid_transaction = "20260814T010203-123-deadbeefcafe"
with tempfile.TemporaryDirectory(prefix="switch-symlink-test-") as switch_name:
switch = switch_fixture(Path(switch_name))
transactions = switch["state"] / "transactions"
transactions.mkdir(parents=True)
outside = Path(switch_name) / "outside" / valid_transaction
outside.mkdir(parents=True)
(transactions / valid_transaction).symlink_to(
outside, target_is_directory=True
)
(switch["state"] / "active").symlink_to(
transactions / valid_transaction, target_is_directory=True
)
result = subprocess.run(
["bash", str(switch["switch"]), "--confirm", valid_transaction],
check=False,
text=True,
capture_output=True,
env=switch["environment"],
timeout=5,
)
assert result.returncode != 0
assert "Unknown or unsafe transaction" in result.stderr
assert not switch["systemctl_marker"].exists()
with tempfile.TemporaryDirectory(prefix="switch-active-escape-test-") as switch_name:
switch = switch_fixture(Path(switch_name))
transactions = switch["state"] / "transactions"
transactions.mkdir(parents=True)
outside = Path(switch_name) / "outside" / valid_transaction
outside.mkdir(parents=True)
(switch["state"] / "active").symlink_to(
outside, target_is_directory=True
)
result = subprocess.run(
["bash", str(switch["switch"]), "--rollback"],
check=False,
text=True,
capture_output=True,
env=switch["environment"],
timeout=5,
)
assert result.returncode == 2
assert "Unknown or unsafe transaction" in result.stderr
assert not switch["systemctl_marker"].exists()
with tempfile.TemporaryDirectory(prefix="switch-parent-link-test-") as switch_name:
switch = switch_fixture(Path(switch_name))
transactions = switch["state"] / "transactions"
transaction = transactions / valid_transaction
transaction.mkdir(parents=True)
(switch["state"] / "active").symlink_to(transaction)
parent = "20260814T010204-124-feedfacecafe"
outside_parent = Path(switch_name) / "outside" / parent
outside_parent.mkdir(parents=True)
(transactions / parent).symlink_to(
outside_parent, target_is_directory=True
)
install_root = switch["install_base"] / switch["source_commit"]
write_json(
transaction / "transaction.json",
{
"schema": 2,
"transaction": valid_transaction,
"parent_transaction": parent,
"source_commit": switch["source_commit"],
"source_tree": "b" * 40,
"upstream_commit": "c" * 40,
"installed_binary": str(install_root / "systemd-resolved"),
"installed_client": str(install_root / "resolvectl"),
"installed_nss": str(install_root / "libnss_resolve.so.2"),
"daemon_sha256": "d" * 64,
"client_sha256": "e" * 64,
"nss_sha256": "f" * 64,
"external_name": None,
},
)
result = subprocess.run(
["bash", str(switch["switch"]), "--rollback", valid_transaction],
check=False,
text=True,
capture_output=True,
env=switch["environment"],
timeout=5,
)
assert result.returncode != 0
assert "Unknown or unsafe transaction" in result.stderr
assert not switch["systemctl_marker"].exists()
with tempfile.TemporaryDirectory(prefix="switch-metadata-test-") as switch_name:
switch = switch_fixture(Path(switch_name))
transaction = switch["state"] / "transactions" / valid_transaction
transaction.mkdir(parents=True)
(switch["state"] / "active").symlink_to(transaction)
write_json(
transaction / "transaction.json",
{
"schema": 2,
"transaction": valid_transaction,
"parent_transaction": "../escape",
"source_commit": switch["source_commit"],
"source_tree": "b" * 40,
"upstream_commit": "c" * 40,
"installed_binary": "/bin/sh",
"installed_client": "/bin/sh",
"installed_nss": "/tmp/nss",
"daemon_sha256": "d" * 64,
"client_sha256": "e" * 64,
"nss_sha256": "f" * 64,
"external_name": None,
},
)
result = subprocess.run(
["bash", str(switch["switch"]), "--rollback", valid_transaction],
check=False,
text=True,
capture_output=True,
env=switch["environment"],
timeout=5,
)
assert result.returncode != 0
assert "parent id is invalid" in result.stderr
assert not switch["systemctl_marker"].exists()
with tempfile.TemporaryDirectory(prefix="switch-path-escape-test-") as switch_name:
switch = switch_fixture(Path(switch_name))
transaction = switch["state"] / "transactions" / valid_transaction
transaction.mkdir(parents=True)
(switch["state"] / "active").symlink_to(transaction)
write_json(
transaction / "transaction.json",
{
"schema": 2,
"transaction": valid_transaction,
"parent_transaction": None,
"source_commit": switch["source_commit"],
"source_tree": "b" * 40,
"upstream_commit": "c" * 40,
"installed_binary": "/bin/sh",
"installed_client": "/bin/sh",
"installed_nss": "/tmp/nss",
"daemon_sha256": "d" * 64,
"client_sha256": "e" * 64,
"nss_sha256": "f" * 64,
"external_name": None,
},
)
result = subprocess.run(
["bash", str(switch["switch"]), "--rollback", valid_transaction],
check=False,
text=True,
capture_output=True,
env=switch["environment"],
timeout=5,
)
assert result.returncode != 0
assert "installed_binary escapes its install root" in result.stderr
assert not switch["systemctl_marker"].exists()
with tempfile.TemporaryDirectory(prefix="switch-metadata-link-test-") as switch_name:
switch = switch_fixture(Path(switch_name))
transaction = switch["state"] / "transactions" / valid_transaction
transaction.mkdir(parents=True)
(switch["state"] / "active").symlink_to(transaction)
outside_metadata = Path(switch_name) / "outside.json"
outside_metadata.write_text("{}\n", encoding="utf-8")
(transaction / "transaction.json").symlink_to(outside_metadata)
result = subprocess.run(
["bash", str(switch["switch"]), "--confirm", valid_transaction],
check=False,
text=True,
capture_output=True,
env=switch["environment"],
timeout=5,
)
assert result.returncode == 2
assert "metadata is missing or unsafe" in result.stderr
assert not switch["systemctl_marker"].exists()
builder_text = BUILDER.read_text(encoding="utf-8")
assert '${RUSTFLAGS:-}' not in builder_text
assert 'export RUSTFLAGS="--remap-path-prefix=' in builder_text
assert "-O2 -g -fPIC -std=c17 -Wall -Wextra -Werror -Wpedantic" in builder_text
assert "-fstack-protector-strong -U_FORTIFY_SOURCE -D_FORTIFY_SOURCE=3" in builder_text
assert "verify_nss_hardening" in builder_text
assert builder_text.count("require_clean_source") == 3
assert builder_text.index("require_clean_source\ncd \"$ROOT\"") < builder_text.index(
'SOURCE_COMMIT="$(source_git rev-parse HEAD)"'
)
assert builder_text.index('cd "$ROOT"') < builder_text.index(
'"$RUSTUP_BIN" run 1.74.0 cargo build'
)
assert builder_text.index("build_once 2\n\nrequire_clean_source") < builder_text.index(
'cp "$WORK/rustd-resolved-1.tar.gz"'
)
assert '"$RUSTUP_BIN" run 1.74.0 cargo build' in builder_text
assert 'case "$OUTPUT_PARENT" in' in builder_text
assert '"$TARGET_ROOT"|"$TMP_ROOT"' in builder_text
assert 'rm -rf -- "$OUTPUT"' not in builder_text
assert "Output must not already exist" in builder_text
assert "mv --no-clobber --no-target-directory" in builder_text
assert "BASE_ENV=(\n env -i" in builder_text
assert "CANONICAL_ENV=(" in builder_text
assert "git --no-replace-objects" in builder_text
assert "core.fsmonitor=false" in builder_text
assert '"ls-files", "-v", "-z"' in builder_text
assert '"PATH=$SYSTEM_PATH"' in builder_text
escaped = Path(name) / "escaped"
escaped.mkdir()
victim = escaped / "victim"
victim.mkdir()
sentinel = victim / "sentinel"
sentinel.write_text("preserve\n", encoding="utf-8")
ancestor = Path(name) / "ancestor"
ancestor.symlink_to(escaped, target_is_directory=True)
unsafe_build = subprocess.run(
["bash", str(BUILDER), "--output", str(ancestor / "victim")],
check=False,
text=True,
capture_output=True,
)
assert unsafe_build.returncode == 2, unsafe_build.stderr
assert "Output parent must resolve exactly to" in unsafe_build.stderr
assert sentinel.read_text(encoding="utf-8") == "preserve\n"
escaped_report = escaped / "certificate.d"
escaped_report.mkdir()
escaped_certificate_sentinel = escaped_report / "sentinel"
escaped_certificate_sentinel.write_text("preserve\n", encoding="utf-8")
escaped_certification = subprocess.run(
["bash", str(CERTIFIER), "--output", str(ancestor / "certificate.json")],
check=False,
text=True,
capture_output=True,
timeout=5,
)
assert escaped_certification.returncode == 2, escaped_certification.stderr
assert "Certificate output parent must resolve exactly to" in escaped_certification.stderr
assert escaped_certificate_sentinel.read_text(encoding="utf-8") == "preserve\n"
unsafe_certificate = Path(name) / "certificate"
unsafe_report = Path(f"{unsafe_certificate}.d")
unsafe_report.mkdir()
certificate_sentinel = unsafe_report / "sentinel"
certificate_sentinel.write_text("preserve\n", encoding="utf-8")
unsafe_certification = subprocess.run(
["bash", str(CERTIFIER), "--output", str(unsafe_certificate)],
check=False,
text=True,
capture_output=True,
timeout=5,
)
assert unsafe_certification.returncode == 2, unsafe_certification.stderr
assert "Certificate output must be a nonempty .json leaf" in unsafe_certification.stderr
assert certificate_sentinel.read_text(encoding="utf-8") == "preserve\n"
with tempfile.TemporaryDirectory(
prefix="certifier-report-sentinel-", suffix=".d"
) as report_name:
protected_report = Path(report_name)
protected_output = Path(str(protected_report)[:-2] + ".json")
protected_sentinel = protected_report / "sentinel"
protected_sentinel.write_text("preserve\n", encoding="utf-8")
rejected_report = subprocess.run(
[
"bash",
str(CERTIFIER),
"--output",
str(protected_output),
],
check=False,
text=True,
capture_output=True,
timeout=5,
)
assert rejected_report.returncode == 2, rejected_report.stderr
assert "Certificate report path must not already exist" in rejected_report.stderr
assert protected_sentinel.read_text(encoding="utf-8") == "preserve\n"
assert not protected_output.exists()
certifier_text = CERTIFIER.read_text(encoding="utf-8")
assert '--local-reproducible-directory "$REPRODUCIBLE_DIR"' in certifier_text
assert 'cp -a "$REPRODUCIBLE_DIR/$artifact"' in certifier_text
assert 'cp -a "$PROOF_DIR/artifacts/$gate/."' not in certifier_text
assert '"$ARTIFACT_DIR/external-proofs" "$gate"' in certifier_text
assert 'source.is_symlink()' in certifier_text
assert 'source.resolve().parent != source_directory' in certifier_text
assert 'rm -rf -- "$REPORT_DIR"' not in certifier_text
assert ".rustd-resolved-certification-stage" in certifier_text
assert 'mv --no-clobber --no-target-directory "$REPORT_DIR"' in certifier_text
rejected_environment = os.environ.copy()
for variable in list(rejected_environment):
if variable.startswith("GIT_") or variable in {
"RUSTC",
"RUSTDOC",
"RUSTC_WRAPPER",
"RUSTC_WORKSPACE_WRAPPER",
"RUSTUP_HOME",
"RUSTUP_TOOLCHAIN",
"CARGO_HOME",
"CARGO_BUILD_RUSTC",
"CARGO_BUILD_RUSTDOC",
"CARGO_BUILD_RUSTC_WRAPPER",
"CARGO_BUILD_RUSTC_WORKSPACE_WRAPPER",
"CARGO_BUILD_RUSTFLAGS",
"CARGO_BUILD_RUSTDOCFLAGS",
"CARGO_ENCODED_RUSTFLAGS",
"CARGO_TARGET_DIR",
"CC",
"CXX",
"FC",
"AR",
"ARFLAGS",
"LD",
"NM",
"RANLIB",
"RANLIBFLAGS",
"STRIP",
"GCC_EXEC_PREFIX",
"COMPILER_PATH",
"LIBRARY_PATH",
"CPATH",
"C_INCLUDE_PATH",
"CPLUS_INCLUDE_PATH",
"OBJC_INCLUDE_PATH",
"LDFLAGS_SO",
"MAKEFILES",
"MAKEFLAGS",
"GNUMAKEFLAGS",
"MFLAGS",
"MAKEOVERRIDES",
} or is_cc_rs_override(variable) or (
variable.startswith("CARGO_BUILD_")
and variable != "CARGO_BUILD_JOBS"
) or variable.startswith("CARGO_PROFILE_RELEASE_") or (
variable.startswith("CARGO_TARGET_")
and variable.endswith(
("_RUSTFLAGS", "_RUSTDOCFLAGS", "_LINKER", "_RUNNER", "_AR")
)
):
rejected_environment.pop(variable)
with tempfile.TemporaryDirectory(prefix="builder-existing-reject-") as output_name:
output = Path(output_name)
output_sentinel = output / "sentinel"
output_sentinel.write_text("preserve\n", encoding="utf-8")
rejected_existing = subprocess.run(
["bash", str(BUILDER), "--output", str(output)],
check=False,
text=True,
capture_output=True,
env=rejected_environment,
timeout=5,
)
assert rejected_existing.returncode == 2, rejected_existing.stderr
assert "Output must not already exist" in rejected_existing.stderr
assert output_sentinel.read_text(encoding="utf-8") == "preserve\n"
with tempfile.TemporaryDirectory(prefix="builder-wrapper-reject-") as output_name:
sentinel_root = Path(output_name)
output = Path(f"{output_name}.output")
output_sentinel = sentinel_root / "sentinel"
output_sentinel.write_text("preserve\n", encoding="utf-8")
wrapper_marker = sentinel_root / "wrapper-ran"
wrapper = sentinel_root / "wrapper"
wrapper.write_text(
f"#!/bin/sh\n: >'{wrapper_marker}'\nexit 99\n",
encoding="utf-8",
)
wrapper.chmod(0o755)
wrapper_environment = rejected_environment.copy()
wrapper_environment["RUSTC_WRAPPER"] = str(wrapper)
rejected_wrapper = subprocess.run(
["bash", str(BUILDER), "--output", str(output)],
check=False,
text=True,
capture_output=True,
env=wrapper_environment,
timeout=5,
)
assert rejected_wrapper.returncode == 2, rejected_wrapper.stderr
assert "Ambient Rust/Cargo override is not permitted" in rejected_wrapper.stderr
assert output_sentinel.read_text(encoding="utf-8") == "preserve\n"
assert not wrapper_marker.exists()
assert not output.exists()
with tempfile.TemporaryDirectory(prefix="builder-target-reject-") as output_name:
sentinel_root = Path(output_name)
output = Path(f"{output_name}.output")
output_sentinel = sentinel_root / "sentinel"
output_sentinel.write_text("preserve\n", encoding="utf-8")
target_environment = rejected_environment.copy()
target_environment["CARGO_BUILD_TARGET"] = "x86_64-unknown-linux-musl"
rejected_target = subprocess.run(
["bash", str(BUILDER), "--output", str(output)],
check=False,
text=True,
capture_output=True,
env=target_environment,
timeout=5,
)
assert rejected_target.returncode == 2, rejected_target.stderr
assert "Ambient Rust/Cargo override is not permitted" in rejected_target.stderr
assert "CARGO_BUILD_TARGET" in rejected_target.stderr
assert output_sentinel.read_text(encoding="utf-8") == "preserve\n"
assert not output.exists()
for variable in (
"CC",
"CARGO_TARGET_X86_64_UNKNOWN_LINUX_GNU_AR",
"CC_x86_64_unknown_linux_gnu",
"CC_x86_64-unknown-linux-gnu",
"HOST_CC",
"TARGET_AR",
"CFLAGS_x86_64_unknown_linux_gnu",
"CRATE_CC_NO_DEFAULTS",
"LDFLAGS_SO",
"MAKEFILES",
"MAKEFLAGS",
"GNUMAKEFLAGS",
"RUSTUP_HOME",
"GCC_EXEC_PREFIX",
"COMPILER_PATH",
"LIBRARY_PATH",
"CPATH",
"C_INCLUDE_PATH",
"CPLUS_INCLUDE_PATH",
"OBJC_INCLUDE_PATH",
):
with tempfile.TemporaryDirectory(prefix="builder-native-reject-") as output_name:
sentinel_root = Path(output_name)
output = Path(f"{output_name}.output")
output_sentinel = sentinel_root / "sentinel"
output_sentinel.write_text("preserve\n", encoding="utf-8")
tool_marker = sentinel_root / "tool-ran"
tool = sentinel_root / "tool"
tool.write_text(
f"#!/bin/sh\n: >'{tool_marker}'\nexit 99\n",
encoding="utf-8",
)
tool.chmod(0o755)
native_environment = rejected_environment.copy()
native_environment[variable] = str(tool)
rejected_native = subprocess.run(
["bash", str(BUILDER), "--output", str(output)],
check=False,
text=True,
capture_output=True,
env=native_environment,
timeout=5,
)
assert rejected_native.returncode == 2, rejected_native.stderr
assert "override is not permitted" in rejected_native.stderr
assert variable in rejected_native.stderr
assert output_sentinel.read_text(encoding="utf-8") == "preserve\n"
assert not tool_marker.exists()
assert not output.exists()
with tempfile.TemporaryDirectory(prefix="builder-git-reject-") as output_name:
sentinel_root = Path(output_name)
foreign = sentinel_root / "foreign"
subprocess.run(
["git", "clone", "--quiet", "--no-local", str(ROOT), str(foreign)],
check=True,
)
output = Path(f"{output_name}.output")
output_sentinel = sentinel_root / "sentinel"
output_sentinel.write_text("preserve\n", encoding="utf-8")
git_environment = rejected_environment.copy()
git_environment["GIT_DIR"] = str(foreign / ".git")
git_environment["GIT_WORK_TREE"] = str(foreign)
rejected_git = subprocess.run(
["bash", str(BUILDER), "--output", str(output)],
check=False,
text=True,
capture_output=True,
env=git_environment,
timeout=5,
)
assert rejected_git.returncode == 2, rejected_git.stderr
assert "Ambient Git control is not permitted: GIT_DIR" in rejected_git.stderr
assert output_sentinel.read_text(encoding="utf-8") == "preserve\n"
assert not output.exists()
for flag in ("--assume-unchanged", "--skip-worktree"):
with tempfile.TemporaryDirectory(
prefix="builder-index-flag-reject-"
) as fixture_name:
fixture_root = Path(fixture_name) / "source"
fixture_scripts = fixture_root / "scripts"
fixture_scripts.mkdir(parents=True)
fixture_builder = fixture_scripts / BUILDER.name
fixture_builder.write_bytes(BUILDER.read_bytes())
fixture_builder.chmod(0o755)
(fixture_root / "Cargo.toml").write_text(
"[package]\nname='fixture'\nversion='0.0.0'\n",
encoding="utf-8",
)
(fixture_root / ".gitignore").write_text(
"/target/\n", encoding="utf-8"
)
subprocess.run(["git", "init", "--quiet", str(fixture_root)], check=True)
subprocess.run(
["git", "-C", str(fixture_root), "config", "user.email", "test@example.invalid"],
check=True,
)
subprocess.run(
["git", "-C", str(fixture_root), "config", "user.name", "Builder Test"],
check=True,
)
subprocess.run(["git", "-C", str(fixture_root), "add", "."], check=True)
subprocess.run(
["git", "-C", str(fixture_root), "commit", "--quiet", "-m", "fixture"],
check=True,
)
subprocess.run(
["git", "-C", str(fixture_root), "update-index", flag, "Cargo.toml"],
check=True,
)
flagged_output = Path(f"{fixture_name}.output")
rejected_flag = subprocess.run(
["bash", str(fixture_builder), "--output", str(flagged_output)],
check=False,
text=True,
capture_output=True,
env=rejected_environment,
timeout=5,
)
assert rejected_flag.returncode == 2, rejected_flag.stderr
assert "tracked path has unsafe index flags" in rejected_flag.stderr
assert not flagged_output.exists()
with tempfile.TemporaryDirectory(prefix="builder-dirty-reject-") as output_name:
sentinel_root = Path(output_name)
output = Path(f"{output_name}.output")
output_sentinel = sentinel_root / "sentinel"
output_sentinel.write_text("preserve\n", encoding="utf-8")
untracked = ROOT / ".builder-untracked-cleanliness-test"
assert not untracked.exists()
try:
untracked.write_text("dirty\n", encoding="utf-8")
rejected_dirty = subprocess.run(
["bash", str(BUILDER), "--output", str(output)],
check=False,
text=True,
capture_output=True,
env=rejected_environment,
timeout=5,
)
finally:
untracked.unlink(missing_ok=True)
assert rejected_dirty.returncode == 2, rejected_dirty.stderr
assert "requires a clean source checkout" in rejected_dirty.stderr
assert "?? .builder-untracked-cleanliness-test" in rejected_dirty.stderr
assert output_sentinel.read_text(encoding="utf-8") == "preserve\n"
assert not output.exists()
with tempfile.TemporaryDirectory(prefix="builder-tracked-reject-") as output_name:
sentinel_root = Path(output_name)
output = Path(f"{output_name}.output")
output_sentinel = sentinel_root / "sentinel"
output_sentinel.write_text("preserve\n", encoding="utf-8")
tracked = ROOT / "Cargo.toml"
tracked_original = tracked.read_bytes()
try:
tracked.write_bytes(
tracked_original + b"\n# builder tracked cleanliness test\n"
)
rejected_tracked = subprocess.run(
["bash", str(BUILDER), "--output", str(output)],
check=False,
text=True,
capture_output=True,
env=rejected_environment,
timeout=5,
)
finally:
tracked.write_bytes(tracked_original)
assert rejected_tracked.returncode == 2, rejected_tracked.stderr
assert "requires a clean source checkout" in rejected_tracked.stderr
assert "Cargo.toml" in rejected_tracked.stderr
assert output_sentinel.read_text(encoding="utf-8") == "preserve\n"
assert not output.exists()
cargo_config_directory = ROOT / ".cargo"
cargo_config = cargo_config_directory / "config.toml"
existing_cargo_config = None
if cargo_config.exists():
existing_cargo_config = cargo_config.read_bytes()
try:
cargo_config_directory.mkdir(exist_ok=True)
cargo_config.write_text("[build]\nrustc-wrapper = 'false'\n", encoding="utf-8")
with tempfile.TemporaryDirectory(prefix="builder-config-reject-") as output_name:
sentinel_root = Path(output_name)
output = Path(f"{output_name}.output")
output_sentinel = sentinel_root / "sentinel"
output_sentinel.write_text("preserve\n", encoding="utf-8")
rejected_config = subprocess.run(
["bash", str(BUILDER), "--output", str(output)],
check=False,
text=True,
capture_output=True,
env=rejected_environment,
timeout=5,
)
assert rejected_config.returncode == 2, rejected_config.stderr
assert "Ambient Cargo configuration is not permitted" in rejected_config.stderr
assert output_sentinel.read_text(encoding="utf-8") == "preserve\n"
assert not output.exists()
finally:
if existing_cargo_config is not None:
cargo_config.write_bytes(existing_cargo_config)
else:
cargo_config.unlink(missing_ok=True)
try:
cargo_config_directory.rmdir()
except OSError:
pass
nss.write_bytes(b"tampered\n")
assert_rejected(run(certificate, source), "NSS module artifact hash mismatch")
nss.write_bytes(b"nss\n")
payload["nss"]["artifact_path"] = "../libnss_resolve.so.2"
write_json(certificate, payload)
assert_rejected(run(certificate, source), "NSS module artifact path is unsafe")
payload["nss"]["artifact_path"] = "certificate.d/artifacts/libnss_resolve.so.2"
del payload["nss"]["artifact_path"]
write_json(certificate, payload)
assert_rejected(run(certificate, source), "NSS module artifact path is missing")
payload["nss"]["artifact_path"] = "certificate.d/artifacts/libnss_resolve.so.2"
payload["nss"]["sha256"] = "z" * 64
write_json(certificate, payload)
assert_rejected(run(certificate, source), "NSS module hash is missing or invalid")
with tempfile.TemporaryDirectory(prefix="readiness-bundle-test-") as name:
certificate, payload, _, source, _ = fixture(Path(name))
payload["gates"].pop()
write_json(certificate, payload)
assert_rejected(run(certificate, source), "certificate gate set differs")
with tempfile.TemporaryDirectory(prefix="readiness-bundle-test-") as name:
certificate, payload, _, source, _ = fixture(Path(name))
payload["gates"].append(dict(payload["gates"][0]))
write_json(certificate, payload)
assert_rejected(run(certificate, source), "certificate contains duplicate gates")
with tempfile.TemporaryDirectory(prefix="readiness-bundle-test-") as name:
certificate, payload, _, source, _ = fixture(Path(name))
payload["gates"].append({"name": "unexpected", "status": "pass"})
write_json(certificate, payload)
assert_rejected(run(certificate, source), "certificate gate set differs")
with tempfile.TemporaryDirectory(prefix="readiness-bundle-test-") as name:
certificate, payload, _, source, _ = fixture(Path(name))
payload["gates"][0] = {"name": payload["gates"][0]["name"]}
write_json(certificate, payload)
assert_rejected(run(certificate, source), "certificate contains a nonpassing gate")
with tempfile.TemporaryDirectory(prefix="readiness-bundle-test-") as name:
certificate, payload, _, source, _ = fixture(Path(name))
payload["gates"][0]["status"] = "fail"
write_json(certificate, payload)
assert_rejected(run(certificate, source), "certificate contains a nonpassing gate")
with tempfile.TemporaryDirectory(prefix="readiness-bundle-test-") as name:
certificate, payload, _, source, _ = fixture(Path(name))
payload["schema"] = 2
write_json(certificate, payload)
assert_rejected(run(certificate, source), "unsupported certificate schema")
with tempfile.TemporaryDirectory(prefix="readiness-bundle-test-") as name:
certificate, payload, _, source, _ = fixture(Path(name))
payload["generated_at"] = "2000-01-01T00:00:00+00:00"
write_json(certificate, payload)
assert_rejected(run(certificate, source), "outside the allowed window")
with tempfile.TemporaryDirectory(prefix="readiness-bundle-test-") as name:
certificate, _, _, source, _ = fixture(Path(name))
daemon = certificate.parent / "certificate.d" / "artifacts" / "systemd-resolved"
daemon.unlink()
assert_rejected(run(certificate, source), "daemon artifact is missing")
with tempfile.TemporaryDirectory(prefix="readiness-bundle-test-") as name:
certificate, _, _, source, _ = fixture(Path(name))
daemon = certificate.parent / "certificate.d" / "artifacts" / "systemd-resolved"
daemon.write_bytes(b"tampered daemon\n")
assert_rejected(run(certificate, source), "daemon artifact hash mismatch")
with tempfile.TemporaryDirectory(prefix="readiness-bundle-test-") as name:
certificate, _, _, source, _ = fixture(Path(name))
package = (
certificate.parent
/ "certificate.d"
/ "artifacts"
/ "reproducible-release"
/ "rustd-resolved.tar.gz"
)
package.write_bytes(b"tampered\n")
assert_rejected(run(certificate, source), "reproducible package artifact hash mismatch")
with tempfile.TemporaryDirectory(prefix="readiness-bundle-test-") as name:
certificate, _, _, source, _ = fixture(Path(name))
package = (
certificate.parent
/ "certificate.d"
/ "artifacts"
/ "reproducible-release"
/ "rustd-resolved.tar.gz"
)
package.unlink()
assert_rejected(run(certificate, source), "reproducible package artifact is missing")
with tempfile.TemporaryDirectory(prefix="readiness-bundle-test-") as name:
certificate, payload, _, source, _ = fixture(Path(name))
del payload["external_proofs"]
write_json(certificate, payload)
assert_rejected(
run(certificate, source),
"external proof set differs from the certificate contract",
)
with tempfile.TemporaryDirectory(prefix="readiness-bundle-test-") as name:
certificate, payload, _, source, _ = fixture(Path(name))
proof_path = (
certificate.parent
/ payload["external_proofs"]["security-suite"]["proof"]["artifact_path"]
)
proof_path.unlink()
assert_rejected(run(certificate, source), "external security-suite proof artifact is missing")
with tempfile.TemporaryDirectory(prefix="readiness-bundle-test-") as name:
certificate, payload, _, source, _ = fixture(Path(name))
evidence_entry = payload["external_proofs"]["security-suite"]["artifacts"][0]
evidence = certificate.parent / evidence_entry["artifact_path"]
evidence.unlink()
assert_rejected(
run(certificate, source),
"external security-suite artifact security-evidence.json artifact is missing",
)
with tempfile.TemporaryDirectory(prefix="readiness-bundle-test-") as name:
certificate, payload, _, source, _ = fixture(Path(name))
evidence_entry = payload["external_proofs"]["security-suite"]["artifacts"][0]
evidence = certificate.parent / evidence_entry["artifact_path"]
evidence.write_bytes(b"tampered evidence\n")
assert_rejected(
run(certificate, source),
"external security-suite artifact security-evidence.json artifact hash mismatch",
)
with tempfile.TemporaryDirectory(prefix="readiness-bundle-test-") as name:
certificate, _, _, source, _ = fixture(Path(name))
subprocess.run(
["git", "-C", str(source), "commit", "--quiet", "--allow-empty", "-m", "later"],
check=True,
)
assert_rejected(run(certificate, source), "current checkout commit differs")
with tempfile.TemporaryDirectory(prefix="readiness-bundle-test-") as name:
certificate, payload, _, source, _ = fixture(Path(name))
payload["source_tree"] = "f" * 40
write_json(certificate, payload)
assert_rejected(run(certificate, source), "current checkout tree differs")
with tempfile.TemporaryDirectory(prefix="readiness-bundle-test-") as name:
certificate, payload, _, source, _ = fixture(Path(name))
payload["upstream_commit"] = "4" * 40
write_json(certificate, payload)
assert_rejected(
run(certificate, source),
"certificate upstream commit differs from the tracked baseline",
)
with tempfile.TemporaryDirectory(prefix="readiness-bundle-test-") as name:
certificate, _, _, source, _ = fixture(Path(name))
contract_path = source / "scripts" / "replacement-certificate-contract.json"
contract_path.write_text(contract_path.read_text(encoding="utf-8") + "\n", encoding="utf-8")
assert_rejected(run(certificate, source), "current checkout is not clean")
with tempfile.TemporaryDirectory(prefix="readiness-bundle-test-") as name:
certificate, _, _, source, _ = fixture(Path(name))
(source / "untracked").write_text("dirty\n", encoding="utf-8")
assert_rejected(run(certificate, source), "current checkout is not clean")
with tempfile.TemporaryDirectory(prefix="readiness-bundle-test-") as name:
certificate, payload, _, source, manifest = fixture(Path(name))
manifest_payload = json.loads(manifest.read_text(encoding="utf-8"))
manifest_payload["byte_identical"].append("systemd-resolved")
write_json(manifest, manifest_payload)
payload["reproducible_release"]["manifest"]["sha256"] = digest(manifest)
write_json(certificate, payload)
assert_rejected(run(certificate, source), "byte-identical set differs")
with tempfile.TemporaryDirectory(prefix="readiness-bundle-test-") as name:
certificate, payload, _, source, manifest = fixture(Path(name))
reproduced_daemon = manifest.parent / "systemd-resolved"
reproduced_daemon.write_bytes(b"different daemon\n")
manifest_payload = json.loads(manifest.read_text(encoding="utf-8"))
daemon_entry = next(
item for item in manifest_payload["artifacts"] if item["name"] == "systemd-resolved"
)
daemon_entry["size"] = reproduced_daemon.stat().st_size
daemon_entry["sha256"] = digest(reproduced_daemon)
write_json(manifest, manifest_payload)
payload["reproducible_release"]["manifest"]["sha256"] = digest(manifest)
write_json(certificate, payload)
assert_rejected(run(certificate, source), "reproducible daemon hash differs")
with tempfile.TemporaryDirectory(prefix="readiness-bundle-test-") as name:
certificate, payload, _, source, manifest = fixture(Path(name))
manifest_payload = json.loads(manifest.read_text(encoding="utf-8"))
manifest_payload["rustc_release"] = "1.75.0"
write_json(manifest, manifest_payload)
payload["reproducible_release"]["manifest"]["sha256"] = digest(manifest)
write_json(certificate, payload)
assert_rejected(run(certificate, source), "reproducible manifest toolchain differs")
with tempfile.TemporaryDirectory(prefix="readiness-bundle-test-") as name:
certificate, payload, _, source, _ = fixture(Path(name))
payload["toolchain"]["rustc_sha256"] = "7" * 64
write_json(certificate, payload)
assert_rejected(
run(certificate, source),
"certificate toolchain hashes differ from the manifest",
)
with tempfile.TemporaryDirectory(prefix="readiness-bundle-test-") as name:
certificate, payload, _, source, manifest = fixture(Path(name))
manifest_payload = json.loads(manifest.read_text(encoding="utf-8"))
manifest_payload["rustc_sha256"] = "7" * 64
write_json(manifest, manifest_payload)
payload["reproducible_release"]["manifest"]["sha256"] = digest(manifest)
payload["toolchain"]["rustc_sha256"] = "7" * 64
write_json(certificate, payload)
assert_rejected(
run(certificate, source),
"reproducible toolchain executable hash evidence differs",
)
with tempfile.TemporaryDirectory(prefix="readiness-bundle-test-") as name:
certificate, payload, _, source, manifest = fixture(Path(name))
manifest_payload = json.loads(manifest.read_text(encoding="utf-8"))
manifest_payload["generated_at"] = "2026-01-01T00:00:00+00:00"
write_json(manifest, manifest_payload)
payload["reproducible_release"]["manifest"]["sha256"] = digest(manifest)
write_json(certificate, payload)
assert_rejected(
run(certificate, source),
"reproducible manifest generated timestamp differs",
)
invalid_contract = {
"schema": 3,
"required_gates": list(CONTRACT["required_gates"]),
"unexpected": True,
}
with tempfile.TemporaryDirectory(prefix="readiness-bundle-test-") as name:
certificate, _, _, source, _ = fixture(Path(name), invalid_contract)
assert_rejected(run(certificate, source), "replacement certificate contract is invalid")
duplicate_contract = {
"schema": 3,
"required_gates": [*CONTRACT["required_gates"], CONTRACT["required_gates"][0]],
}
with tempfile.TemporaryDirectory(prefix="readiness-bundle-test-") as name:
certificate, _, _, source, _ = fixture(Path(name), duplicate_contract)
assert_rejected(
run(certificate, source),
"replacement certificate contract contains duplicate gates",
)
empty_contract = {
"schema": 3,
"required_gates": [*CONTRACT["required_gates"], ""],
}
with tempfile.TemporaryDirectory(prefix="readiness-bundle-test-") as name:
certificate, _, _, source, _ = fixture(Path(name), empty_contract)
assert_rejected(run(certificate, source), "replacement certificate contract is invalid")
print("readiness bundle regression tests passed")
if __name__ == "__main__":
main()