AIOS Governed Continuation
91f104ef62
autonomy-smoke-tests focused receipt/replay and Task lifecycle smoke: PASS
autonomy-unit-tests full Linux regression: 834 passed, 3 skipped, 154 subtests passed
campaign-approval-gate standing Owner V15 exact Repair Task authorization and formal Task verification: PASS
git-diff-check exact two-file governed diff and whitespace check: PASS
py-compile active-runtime and tests compileall: PASS
reconciliation-contract WriterLease completed replay and governed branch convergence: PASS
release-state-machine full Linux regression includes release state-machine contract: PASS; no release mutation
secret-scan exact changed-file high-signal credential/private-key scan: PASS
signer-validator-contract receipt framing/replay/adoption contract: 18 focused tests PASS; private key non-exportable
292 lines
14 KiB
Python
292 lines
14 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import sys
|
|
|
|
import pytest
|
|
|
|
APP = Path(__file__).resolve().parents[1] / "active-runtime" / "app"
|
|
sys.path.insert(0, str(APP))
|
|
import trusted_builder_signer_provisioning as provisioning # noqa: E402
|
|
|
|
|
|
def _source(tmp_path: Path) -> tuple[Path, list[str]]:
|
|
root = tmp_path / "source"
|
|
for relative in provisioning.SOURCE_EVIDENCE_PATHS:
|
|
path = root / relative
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
if relative != provisioning.PACKAGE_MANIFEST_PATH:
|
|
path.write_text("source:" + relative + "\n", encoding="utf-8")
|
|
os.chmod(path, 0o644)
|
|
manifest = root / provisioning.PACKAGE_MANIFEST_PATH
|
|
manifest.write_bytes(provisioning.pretty_json(provisioning.package_contract_document()))
|
|
os.chmod(manifest, 0o644)
|
|
return root, list(provisioning.SOURCE_EVIDENCE_PATHS)
|
|
|
|
|
|
def _payload(tmp_path: Path):
|
|
source, paths = _source(tmp_path)
|
|
output = tmp_path / "payload.json"
|
|
value = provisioning.build_payload(
|
|
source, output, main_oid="c" * 40, tree_oid="d" * 40,
|
|
pr_number=70, pr_head_oid="e" * 40,
|
|
merge_parent_oids=["a" * 40, "e" * 40], source_files=paths,
|
|
nonce="test-provisioning-nonce-001",
|
|
)
|
|
return source, output, value
|
|
|
|
|
|
def test_fresh_payload_is_bound_to_final_common_main_and_ordered_pr_parent(tmp_path: Path):
|
|
_source_root, output, payload = _payload(tmp_path)
|
|
assert output.read_bytes().startswith(b"{")
|
|
provisioning._validate_payload(payload)
|
|
assert payload["source"]["gitea_main_oid"] == payload["source"]["governed_main_oid"]
|
|
assert payload["source"]["merge_parent_oids"][1] == payload["source"]["pr_head_oid"]
|
|
assert set(payload["source"]["source_evidence_sha256"]) == set(provisioning.SOURCE_EVIDENCE_PATHS)
|
|
assert set(payload["allowed_capabilities"]) == provisioning.ALLOWED_CAPABILITIES
|
|
assert set(payload["denied_capabilities"]) == provisioning.DENIED_CAPABILITIES
|
|
|
|
|
|
@pytest.mark.parametrize("field", ["allowed_capabilities", "denied_capabilities"])
|
|
def test_payload_rejects_capability_scope_tampering(tmp_path: Path, field: str):
|
|
_source_root, _output, payload = _payload(tmp_path)
|
|
payload[field] = ["owner_release_approval"]
|
|
with pytest.raises(provisioning.ProvisioningError, match="CAPABILITY_SCOPE_INVALID"):
|
|
provisioning._validate_payload(payload)
|
|
|
|
|
|
def test_payload_rejects_main_divergence_and_wrong_second_parent(tmp_path: Path):
|
|
_source_root, _output, payload = _payload(tmp_path)
|
|
payload["source"]["governed_main_oid"] = "f" * 40
|
|
with pytest.raises(provisioning.ProvisioningError, match="SOURCE_MAIN_NOT_CONVERGED"):
|
|
provisioning._validate_payload(payload)
|
|
_source_root, _output, payload = _payload(tmp_path / "second")
|
|
payload["source"]["merge_parent_oids"][1] = "a" * 40
|
|
with pytest.raises(provisioning.ProvisioningError, match="SOURCE_PR_HEAD_SECOND_PARENT_INVALID"):
|
|
provisioning._validate_payload(payload)
|
|
|
|
|
|
def test_payload_rejects_missing_or_extra_source_evidence(tmp_path: Path):
|
|
_source_root, _output, payload = _payload(tmp_path)
|
|
payload["source"]["source_evidence_sha256"].pop(next(iter(payload["source"]["source_evidence_sha256"])))
|
|
with pytest.raises(provisioning.ProvisioningError, match="SOURCE_EVIDENCE_SET_INVALID"):
|
|
provisioning._validate_payload(payload)
|
|
_source_root, _output, payload = _payload(tmp_path / "second")
|
|
payload["source"]["source_evidence_sha256"]["extra.txt"] = "0" * 64
|
|
with pytest.raises(provisioning.ProvisioningError, match="SOURCE_EVIDENCE_SET_INVALID"):
|
|
provisioning._validate_payload(payload)
|
|
|
|
|
|
def test_package_contract_rejects_incompatible_source_mode_and_tampering(tmp_path: Path):
|
|
source, _paths = _source(tmp_path)
|
|
service = source / "deploy/systemd/aios-trusted-candidate-builder.service"
|
|
os.chmod(service, 0o601)
|
|
with pytest.raises(provisioning.ProvisioningError, match="SOURCE_EVIDENCE_MODE_INVALID"):
|
|
provisioning.verify_package_contract(source)
|
|
os.chmod(service, 0o644)
|
|
manifest = source / provisioning.PACKAGE_MANIFEST_PATH
|
|
value = json.loads(manifest.read_text(encoding="utf-8"))
|
|
value["activation"]["credential_installation"] = True
|
|
manifest.write_text(json.dumps(value), encoding="utf-8")
|
|
with pytest.raises(provisioning.ProvisioningError, match="PACKAGE_CONTRACT_INVALID"):
|
|
provisioning.verify_package_contract(source)
|
|
|
|
|
|
def test_package_contract_declares_socket_activation_no_credentials_and_0755_launcher(tmp_path: Path):
|
|
source, _paths = _source(tmp_path)
|
|
provisioning.verify_package_contract(source)
|
|
value = json.loads((source / provisioning.PACKAGE_MANIFEST_PATH).read_text(encoding="utf-8"))
|
|
launcher = next(item for item in value["files"] if item["source_path"] == "deploy/libexec/aios-trusted-builder-signer-provisioner")
|
|
assert launcher["install_mode"] == "0755"
|
|
assert value["activation"] == {
|
|
"daemon_reload": True,
|
|
"enable_units": ["aios-trusted-candidate-builder.socket"],
|
|
"service_start": "socket-activation-only",
|
|
"credential_installation": False,
|
|
"signer_invocation": False,
|
|
}
|
|
|
|
|
|
def test_provisioner_self_test_has_no_mutation(capsys):
|
|
assert provisioning.main(["--self-test"]) == 0
|
|
assert capsys.readouterr().out.strip() == "PASS"
|
|
|
|
|
|
def _replay_fixture(tmp_path: Path):
|
|
_source_root, payload_path, payload = _payload(tmp_path)
|
|
raw = payload_path.read_bytes()
|
|
signature = tmp_path / "owner.sig"
|
|
signature.write_bytes(b"owner-signature-test")
|
|
os.chmod(signature, 0o600)
|
|
|
|
config = provisioning.ProvisionerConfig()
|
|
config.credential_store = tmp_path / "credstore"
|
|
config.public_keyring = tmp_path / "state" / "public-keyring.json"
|
|
config.binding_file = tmp_path / "state" / "service-principal-binding.json"
|
|
config.receipt_log = tmp_path / "state" / "provisioning-receipts.jsonl"
|
|
config.credential_store.mkdir(mode=0o700)
|
|
os.chmod(config.credential_store, 0o700)
|
|
config.public_keyring.parent.mkdir(mode=0o700)
|
|
|
|
public = b"p" * 32
|
|
now = provisioning.utc_now()
|
|
keyring = provisioning._keyring(payload["signing_identity"], public, now)
|
|
config.public_keyring.write_bytes(provisioning.pretty_json(keyring))
|
|
os.chmod(config.public_keyring, 0o600)
|
|
binding = {
|
|
"schema": "aios.trusted-builder.service-principal-binding.v1",
|
|
"purpose": provisioning.PURPOSE,
|
|
"service_principal": payload["signing_identity"]["service_principal"],
|
|
"key_id": payload["signing_identity"]["key_id"],
|
|
"generation": payload["signing_identity"]["key_generation"],
|
|
"algorithm": "Ed25519",
|
|
"private_key_export": False,
|
|
"public_key_sha256": provisioning.sha256_bytes(public),
|
|
"source_commit": payload["source"]["gitea_main_oid"],
|
|
"source_tree": payload["source"]["gitea_main_tree_oid"],
|
|
"payload_sha256": provisioning.sha256_bytes(raw),
|
|
}
|
|
config.binding_file.write_bytes(provisioning.pretty_json(binding))
|
|
os.chmod(config.binding_file, 0o600)
|
|
for name in ("signing_private_key", "builder_hmac"):
|
|
target = config.credential_store / name
|
|
target.write_bytes(("encrypted-" + name).encode())
|
|
os.chmod(target, 0o600)
|
|
signature_bytes = signature.read_bytes()
|
|
receipt = {
|
|
"schema": "aios.trusted-builder-signer-provisioning-receipt.v1",
|
|
"status": "COMPLETED",
|
|
"created_at": provisioning.iso(now),
|
|
"operation_key": "payload:" + provisioning.sha256_bytes(raw) + ":signature:" + provisioning.sha256_bytes(signature_bytes),
|
|
"payload_sha256": provisioning.sha256_bytes(raw),
|
|
"signature_sha256": provisioning.sha256_bytes(signature_bytes),
|
|
"source_commit": payload["source"]["gitea_main_oid"],
|
|
"source_tree": payload["source"]["gitea_main_tree_oid"],
|
|
"key_id": payload["signing_identity"]["key_id"],
|
|
"generation": payload["signing_identity"]["key_generation"],
|
|
"algorithm": "Ed25519",
|
|
"purpose": provisioning.PURPOSE,
|
|
"service_principal": payload["signing_identity"]["service_principal"],
|
|
"service_principal_bound": True,
|
|
"public_key_sha256": provisioning.sha256_bytes(public),
|
|
"private_key_exportable": False,
|
|
"credential_projection": "systemd-encrypted-host-credential",
|
|
"negative_capability_tests": {name: True for name in sorted(provisioning.DENIED_CAPABILITIES)},
|
|
"production_mutation": False,
|
|
"protected_ref_mutation": False,
|
|
"tag_mutation": False,
|
|
"baseline_mutation": False,
|
|
}
|
|
return config, payload_path, payload, signature, receipt
|
|
|
|
|
|
def test_legacy_multiline_receipt_parse(tmp_path: Path):
|
|
path = tmp_path / "receipts.jsonl"
|
|
receipt = {"schema": "legacy", "status": "COMPLETED", "nested": {"value": True}}
|
|
path.write_bytes(provisioning.pretty_json(receipt))
|
|
assert provisioning._receipts(path) == [receipt]
|
|
|
|
|
|
def test_new_canonical_jsonl_receipt_parse(tmp_path: Path):
|
|
path = tmp_path / "receipts.jsonl"
|
|
rows = [{"id": 1}, {"id": 2}]
|
|
for row in rows:
|
|
provisioning._append(path, row)
|
|
assert path.read_bytes() == b"".join(provisioning.canonical_json(row) + b"\n" for row in rows)
|
|
assert provisioning._receipts(path) == rows
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"content",
|
|
[b'{"status":"COMPLETED"}\nnot-json\n', b'{"status":"COMPLETED"'],
|
|
ids=["malformed", "truncated"],
|
|
)
|
|
def test_malformed_or_truncated_receipt_fails_closed(tmp_path: Path, content: bytes):
|
|
path = tmp_path / "receipts.jsonl"
|
|
path.write_bytes(content)
|
|
with pytest.raises(provisioning.ProvisioningError, match="RECEIPT_LOG_INVALID"):
|
|
provisioning._receipts(path)
|
|
|
|
|
|
def test_preexisting_valid_state_is_adopted_without_mutation(tmp_path: Path, monkeypatch):
|
|
config, payload_path, _payload_value, signature, receipt = _replay_fixture(tmp_path)
|
|
config.receipt_log.write_bytes(provisioning.pretty_json(receipt))
|
|
os.chmod(config.receipt_log, 0o600)
|
|
before = {
|
|
path: path.read_bytes()
|
|
for path in (
|
|
config.receipt_log, config.public_keyring, config.binding_file,
|
|
config.credential_store / "signing_private_key",
|
|
config.credential_store / "builder_hmac",
|
|
)
|
|
}
|
|
monkeypatch.setattr(provisioning, "_run_signature_verify", lambda *_args: None)
|
|
monkeypatch.setattr(provisioning, "_verify_replay_source", lambda *_args: None)
|
|
monkeypatch.setattr(provisioning.os, "geteuid", lambda: 0)
|
|
result = provisioning.provision(config, payload_path, signature)
|
|
assert result["status"] == "IDEMPOTENT_REPLAY_PASS"
|
|
assert result["mutation_executed"] is False
|
|
assert result["existing_signer_state_adopted"] is True
|
|
assert result["private_key_exportable"] is False
|
|
assert before == {path: path.read_bytes() for path in before}
|
|
|
|
|
|
def test_duplicate_replay_is_idempotent(tmp_path: Path, monkeypatch):
|
|
config, payload_path, _payload_value, signature, receipt = _replay_fixture(tmp_path)
|
|
config.receipt_log.write_bytes(provisioning.pretty_json(receipt))
|
|
os.chmod(config.receipt_log, 0o600)
|
|
monkeypatch.setattr(provisioning, "_run_signature_verify", lambda *_args: None)
|
|
monkeypatch.setattr(provisioning, "_verify_replay_source", lambda *_args: None)
|
|
monkeypatch.setattr(provisioning.os, "geteuid", lambda: 0)
|
|
first = provisioning.provision(config, payload_path, signature)
|
|
receipt_bytes = config.receipt_log.read_bytes()
|
|
second = provisioning.provision(config, payload_path, signature)
|
|
assert first == second
|
|
assert config.receipt_log.read_bytes() == receipt_bytes
|
|
|
|
|
|
def test_wrong_payload_is_rejected_against_preexisting_state(tmp_path: Path, monkeypatch):
|
|
config, payload_path, payload, signature, receipt = _replay_fixture(tmp_path)
|
|
config.receipt_log.write_bytes(provisioning.pretty_json(receipt))
|
|
os.chmod(config.receipt_log, 0o600)
|
|
payload["nonce"] = "wrong-payload-nonce-0002"
|
|
payload_path.write_bytes(provisioning.pretty_json(payload))
|
|
monkeypatch.setattr(provisioning, "_run_signature_verify", lambda *_args: None)
|
|
monkeypatch.setattr(provisioning, "_verify_source", lambda *_args: None)
|
|
monkeypatch.setattr(provisioning.os, "geteuid", lambda: 0)
|
|
with pytest.raises(provisioning.ProvisioningError, match="PREEXISTING_STATE_CONFLICT"):
|
|
provisioning.provision(config, payload_path, signature)
|
|
|
|
|
|
def test_wrong_signature_is_rejected(tmp_path: Path, monkeypatch):
|
|
config, payload_path, _payload_value, signature, receipt = _replay_fixture(tmp_path)
|
|
config.receipt_log.write_bytes(provisioning.pretty_json(receipt))
|
|
monkeypatch.setattr(
|
|
provisioning,
|
|
"_run_signature_verify",
|
|
lambda *_args: (_ for _ in ()).throw(provisioning.ProvisioningError("OWNER_SIGNATURE_VERIFY_FAILED")),
|
|
)
|
|
monkeypatch.setattr(provisioning.os, "geteuid", lambda: 0)
|
|
with pytest.raises(provisioning.ProvisioningError, match="OWNER_SIGNATURE_VERIFY_FAILED"):
|
|
provisioning.provision(config, payload_path, signature)
|
|
|
|
|
|
def test_wrong_purpose_is_rejected(tmp_path: Path):
|
|
_source_root, _payload_path, payload = _payload(tmp_path)
|
|
payload["signing_identity"]["purpose"] = "WRONG_PURPOSE"
|
|
with pytest.raises(provisioning.ProvisioningError, match="SIGNING_IDENTITY_INVALID"):
|
|
provisioning._validate_payload(payload)
|
|
|
|
|
|
def test_replay_rejects_state_binding_tampering(tmp_path: Path, monkeypatch):
|
|
config, payload_path, _payload_value, signature, receipt = _replay_fixture(tmp_path)
|
|
receipt["service_principal"] = "wrong.service"
|
|
config.receipt_log.write_bytes(provisioning.pretty_json(receipt))
|
|
monkeypatch.setattr(provisioning, "_run_signature_verify", lambda *_args: None)
|
|
monkeypatch.setattr(provisioning, "_verify_replay_source", lambda *_args: None)
|
|
monkeypatch.setattr(provisioning.os, "geteuid", lambda: 0)
|
|
with pytest.raises(provisioning.ProvisioningError, match="RECEIPT_CONFLICT"):
|
|
provisioning.provision(config, payload_path, signature)
|