1028 lines
32 KiB
Python
1028 lines
32 KiB
Python
from __future__ import annotations
|
|
|
|
import base64
|
|
import dataclasses
|
|
import json
|
|
import os
|
|
import sys
|
|
import tempfile
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from cryptography.hazmat.primitives import serialization
|
|
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(ROOT / "active-runtime" / "app"))
|
|
|
|
from silent_release_controller import ( # noqa: E402
|
|
ApprovalError,
|
|
ApprovalLedger,
|
|
DeploymentController,
|
|
EnvironmentAllowList,
|
|
ReleaseCandidate,
|
|
SilentReleasePolicy,
|
|
SystemdDeploymentAdapter,
|
|
SystemdDeploymentConfig,
|
|
)
|
|
from silent_release_hardening import ( # noqa: E402
|
|
ApprovalKeyRing,
|
|
ApprovalRevocationStore,
|
|
BaselineController,
|
|
CapabilityAuthorizationError,
|
|
CapabilityController,
|
|
CapabilityLedger,
|
|
ClaimProbeResult,
|
|
DeploymentPlan,
|
|
PreSwitchValidator,
|
|
HardenedTagController,
|
|
ReleaseSagaStore,
|
|
RegistryController,
|
|
SagaConflict,
|
|
SagaState,
|
|
SignedApprovalVerifier,
|
|
SignedReleasePlanApproval,
|
|
SilentReleaseActionService,
|
|
StaleApprovalError,
|
|
)
|
|
|
|
|
|
COMMIT = "1" * 40
|
|
TREE = "2" * 40
|
|
ARTIFACT = "3" * 64
|
|
MANIFEST = "4" * 64
|
|
DEPLOYMENT = "5" * 64
|
|
POLICY = "6" * 64
|
|
EVIDENCE = "7" * 40
|
|
OWNER_RECORD = "8" * 40
|
|
|
|
|
|
def candidate() -> ReleaseCandidate:
|
|
return ReleaseCandidate(
|
|
candidate_id="rc-hardening-001",
|
|
source_commit=COMMIT,
|
|
tree_oid=TREE,
|
|
artifact_sha256=ARTIFACT,
|
|
manifest_sha256=MANIFEST,
|
|
)
|
|
|
|
|
|
def deployment_plan() -> DeploymentPlan:
|
|
return DeploymentPlan(
|
|
service="aios-gitops-model-api",
|
|
release_path="/data/aios-gitops/runtime/releases/" + COMMIT,
|
|
release_root="/data/aios-gitops/runtime/releases",
|
|
dropin_path=(
|
|
"/data/aios-gitops/runtime/releases/"
|
|
+ COMMIT
|
|
+ "/deployment/production.conf"
|
|
),
|
|
dropin_sha256="9" * 64,
|
|
rollback_path="/usr/libexec/aios-gitops-runtime-anchor",
|
|
rollback_sha256="a" * 64,
|
|
source_commit=COMMIT,
|
|
artifact_sha256=ARTIFACT,
|
|
manifest_sha256=MANIFEST,
|
|
policy_sha256=POLICY,
|
|
health_contract={
|
|
"runtime_commit": COMMIT,
|
|
"writer": False,
|
|
"repository_identity": "aios/aios",
|
|
},
|
|
)
|
|
|
|
|
|
def unsigned_approval(
|
|
plan: DeploymentPlan | None = None,
|
|
) -> SignedReleasePlanApproval:
|
|
plan = plan or deployment_plan()
|
|
return SignedReleasePlanApproval(
|
|
release_plan_id="release-plan-001",
|
|
candidate_id="rc-hardening-001",
|
|
source_commit=COMMIT,
|
|
tree_oid=TREE,
|
|
artifact_sha256=ARTIFACT,
|
|
manifest_sha256=MANIFEST,
|
|
deployment_plan_sha256=plan.sha256,
|
|
policy_sha256=POLICY,
|
|
approval_plan_id="owner-plan-001",
|
|
evidence_closure_oid=EVIDENCE,
|
|
owner_approval_record_oid=OWNER_RECORD,
|
|
capabilities=(
|
|
"runtime_switch",
|
|
"tag_publish",
|
|
"baseline_registry_finalize",
|
|
),
|
|
expires_at="2099-01-01T00:00:00Z",
|
|
nonce="nonce-0000000000000001",
|
|
key_id="owner-key-1",
|
|
signature="",
|
|
)
|
|
|
|
|
|
def signing_material():
|
|
private = Ed25519PrivateKey.generate()
|
|
public = private.public_key().public_bytes(
|
|
serialization.Encoding.Raw,
|
|
serialization.PublicFormat.Raw,
|
|
)
|
|
return private, public
|
|
|
|
|
|
def sign(
|
|
value: SignedReleasePlanApproval,
|
|
private: Ed25519PrivateKey,
|
|
) -> SignedReleasePlanApproval:
|
|
signature = base64.b64encode(
|
|
private.sign(value.canonical_bytes())
|
|
).decode("ascii")
|
|
return dataclasses.replace(value, signature=signature)
|
|
|
|
|
|
def verifier(
|
|
root: Path,
|
|
approval: SignedReleasePlanApproval | None = None,
|
|
):
|
|
private, public = signing_material()
|
|
value = sign(approval or unsigned_approval(), private)
|
|
keyring = ApprovalKeyRing({"owner-key-1": public})
|
|
revocations = ApprovalRevocationStore(root / "revocations")
|
|
return value, SignedApprovalVerifier(keyring, revocations), keyring, revocations
|
|
|
|
|
|
def snapshot(plan: DeploymentPlan | None = None):
|
|
value = plan or deployment_plan()
|
|
return {
|
|
"source_commit": COMMIT,
|
|
"tree_oid": TREE,
|
|
"artifact_sha256": ARTIFACT,
|
|
"manifest_sha256": MANIFEST,
|
|
"deployment_plan_sha256": value.sha256,
|
|
"policy_sha256": POLICY,
|
|
"rollback_ready": True,
|
|
"git_refs_consistent": True,
|
|
"repository_identity": "aios/aios",
|
|
"writer_disabled": True,
|
|
}
|
|
|
|
|
|
class FakePreSwitchProbe:
|
|
def __init__(self, result=None):
|
|
self.result = result or snapshot()
|
|
self.calls = 0
|
|
|
|
def collect(self, release_candidate, plan):
|
|
self.calls += 1
|
|
return dict(self.result)
|
|
|
|
|
|
def test_valid_signed_release_plan_is_verified():
|
|
with tempfile.TemporaryDirectory() as td:
|
|
value, approval_verifier, _, _ = verifier(Path(td))
|
|
verified = approval_verifier.verify(
|
|
value, candidate(), deployment_plan(), POLICY
|
|
)
|
|
assert verified.fingerprint == value.fingerprint
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"field,value",
|
|
[
|
|
("candidate_id", "rc-forged"),
|
|
("source_commit", "a" * 40),
|
|
("tree_oid", "b" * 40),
|
|
("artifact_sha256", "c" * 64),
|
|
("manifest_sha256", "d" * 64),
|
|
("owner_approval_record_oid", "e" * 40),
|
|
],
|
|
)
|
|
def test_signed_plan_tampering_is_rejected(field, value):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
signed, approval_verifier, _, _ = verifier(Path(td))
|
|
forged = dataclasses.replace(signed, **{field: value})
|
|
with pytest.raises(CapabilityAuthorizationError):
|
|
approval_verifier.verify(
|
|
forged, candidate(), deployment_plan(), POLICY
|
|
)
|
|
|
|
|
|
def test_unknown_signing_key_is_rejected():
|
|
with tempfile.TemporaryDirectory() as td:
|
|
private, _ = signing_material()
|
|
signed = sign(unsigned_approval(), private)
|
|
approval_verifier = SignedApprovalVerifier(
|
|
ApprovalKeyRing({}),
|
|
ApprovalRevocationStore(Path(td) / "revocations"),
|
|
)
|
|
with pytest.raises(CapabilityAuthorizationError):
|
|
approval_verifier.verify(
|
|
signed, candidate(), deployment_plan(), POLICY
|
|
)
|
|
|
|
|
|
def test_key_rotation_accepts_new_key_and_revokes_old_key():
|
|
with tempfile.TemporaryDirectory() as td:
|
|
old_private, old_public = signing_material()
|
|
new_private, new_public = signing_material()
|
|
keys = ApprovalKeyRing(
|
|
{"owner-key-1": old_public, "owner-key-2": new_public}
|
|
)
|
|
revocations = ApprovalRevocationStore(Path(td) / "revocations")
|
|
approval_verifier = SignedApprovalVerifier(keys, revocations)
|
|
old = sign(unsigned_approval(), old_private)
|
|
approval_verifier.verify(
|
|
old, candidate(), deployment_plan(), POLICY
|
|
)
|
|
keys.revoke("owner-key-1")
|
|
with pytest.raises(CapabilityAuthorizationError):
|
|
approval_verifier.verify(
|
|
old, candidate(), deployment_plan(), POLICY
|
|
)
|
|
new = sign(
|
|
dataclasses.replace(
|
|
unsigned_approval(), key_id="owner-key-2"
|
|
),
|
|
new_private,
|
|
)
|
|
assert approval_verifier.verify(
|
|
new, candidate(), deployment_plan(), POLICY
|
|
)
|
|
|
|
|
|
def test_approval_revocation_is_fail_closed():
|
|
with tempfile.TemporaryDirectory() as td:
|
|
signed, approval_verifier, _, revocations = verifier(Path(td))
|
|
revocations.revoke(
|
|
signed.approval_plan_id, "owner revoked release plan"
|
|
)
|
|
with pytest.raises(CapabilityAuthorizationError):
|
|
approval_verifier.verify(
|
|
signed, candidate(), deployment_plan(), POLICY
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"capability",
|
|
[
|
|
"runtime_switch",
|
|
"tag_publish",
|
|
"baseline_registry_finalize",
|
|
],
|
|
)
|
|
def test_one_signed_plan_authorizes_each_declared_capability(capability):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
signed, approval_verifier, _, _ = verifier(Path(td))
|
|
ledger = CapabilityLedger(Path(td) / "capabilities")
|
|
result = ledger.consume(
|
|
approval_verifier.verify(
|
|
signed, candidate(), deployment_plan(), POLICY
|
|
),
|
|
capability,
|
|
operation=capability,
|
|
)
|
|
assert result["status"] == "CONSUMED"
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"capability",
|
|
[
|
|
"force_push",
|
|
"tag_rewrite",
|
|
"direct_main_push",
|
|
"baseline_mutation",
|
|
"registry_mutation",
|
|
],
|
|
)
|
|
def test_undeclared_or_forbidden_capability_is_rejected(capability):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
signed, approval_verifier, _, _ = verifier(Path(td))
|
|
ledger = CapabilityLedger(Path(td) / "capabilities")
|
|
with pytest.raises(CapabilityAuthorizationError):
|
|
ledger.consume(
|
|
approval_verifier.verify(
|
|
signed, candidate(), deployment_plan(), POLICY
|
|
),
|
|
capability,
|
|
operation=capability,
|
|
)
|
|
|
|
|
|
def test_capability_replay_is_idempotent_for_same_fingerprint():
|
|
with tempfile.TemporaryDirectory() as td:
|
|
signed, approval_verifier, _, _ = verifier(Path(td))
|
|
verified = approval_verifier.verify(
|
|
signed, candidate(), deployment_plan(), POLICY
|
|
)
|
|
ledger = CapabilityLedger(Path(td) / "capabilities")
|
|
first = ledger.consume(
|
|
verified, "runtime_switch", operation="runtime"
|
|
)
|
|
second = ledger.consume(
|
|
verified, "runtime_switch", operation="runtime"
|
|
)
|
|
assert first["status"] == "CONSUMED"
|
|
assert second["status"] == "IDEMPOTENT"
|
|
|
|
|
|
def test_capability_nonce_replay_with_different_fingerprint_is_rejected():
|
|
with tempfile.TemporaryDirectory() as td:
|
|
signed, approval_verifier, _, _ = verifier(Path(td))
|
|
verified = approval_verifier.verify(
|
|
signed, candidate(), deployment_plan(), POLICY
|
|
)
|
|
ledger = CapabilityLedger(Path(td) / "capabilities")
|
|
ledger.consume(verified, "runtime_switch", operation="runtime")
|
|
with pytest.raises(CapabilityAuthorizationError):
|
|
ledger.consume(
|
|
dataclasses.replace(
|
|
verified, release_plan_id="release-plan-forged"
|
|
),
|
|
"runtime_switch",
|
|
operation="runtime",
|
|
)
|
|
|
|
|
|
def test_capability_failure_stays_reserved_and_retry_can_complete():
|
|
with tempfile.TemporaryDirectory() as td:
|
|
signed, approval_verifier, _, _ = verifier(Path(td))
|
|
verified = approval_verifier.verify(
|
|
signed, candidate(), deployment_plan(), POLICY
|
|
)
|
|
ledger = CapabilityLedger(Path(td) / "capabilities")
|
|
failing = CapabilityController(
|
|
capability="runtime_switch",
|
|
ledger=ledger,
|
|
adapter=lambda: {"verified": False},
|
|
)
|
|
with pytest.raises(CapabilityAuthorizationError):
|
|
failing.execute(
|
|
verified, operation="runtime-switch-001"
|
|
)
|
|
reservation = ledger.reserve(
|
|
verified,
|
|
"runtime_switch",
|
|
operation="runtime-switch-001",
|
|
)
|
|
assert reservation["status"] == "RESUME"
|
|
recovered = CapabilityController(
|
|
capability="runtime_switch",
|
|
ledger=ledger,
|
|
adapter=lambda: {"verified": True},
|
|
recovery_probe=lambda: ClaimProbeResult.NOT_APPLIED,
|
|
).execute(verified, operation="runtime-switch-001")
|
|
assert recovered["status"] == "EXECUTED"
|
|
assert ledger.reserve(
|
|
verified,
|
|
"runtime_switch",
|
|
operation="runtime-switch-001",
|
|
)["status"] == "IDEMPOTENT"
|
|
|
|
|
|
def test_tag_proof_failure_does_not_complete_capability():
|
|
with tempfile.TemporaryDirectory() as td:
|
|
signed, approval_verifier, _, _ = verifier(Path(td))
|
|
verified = approval_verifier.verify(
|
|
signed, candidate(), deployment_plan(), POLICY
|
|
)
|
|
ledger = CapabilityLedger(Path(td) / "capabilities")
|
|
controller = HardenedTagController(
|
|
ledger, lambda: {"gitea_listing_visible": False}
|
|
)
|
|
with pytest.raises(CapabilityAuthorizationError):
|
|
controller.execute(
|
|
verified, operation="tag-V2R1C3"
|
|
)
|
|
assert ledger.reserve(
|
|
verified,
|
|
"tag_publish",
|
|
operation="tag-V2R1C3",
|
|
)["status"] == "RESUME"
|
|
|
|
|
|
def test_tag_controller_requires_matching_annotated_target():
|
|
with tempfile.TemporaryDirectory() as td:
|
|
signed, approval_verifier, _, _ = verifier(Path(td))
|
|
verified = approval_verifier.verify(
|
|
signed, candidate(), deployment_plan(), POLICY
|
|
)
|
|
proof = {
|
|
"evidence_closure_oid": verified.evidence_closure_oid,
|
|
"approval_plan_id": verified.approval_plan_id,
|
|
"repository_identity": "aios/aios",
|
|
"remote_tag_oid": "9" * 40,
|
|
"remote_peeled_target_oid": "a" * 40,
|
|
"gitea_api_tag_oid": "9" * 40,
|
|
"gitea_api_peeled_target_oid": "a" * 40,
|
|
"gitea_listing_visible": True,
|
|
}
|
|
controller = HardenedTagController(
|
|
CapabilityLedger(Path(td) / "capabilities"),
|
|
lambda: proof,
|
|
expected_repository="aios/aios",
|
|
)
|
|
with pytest.raises(CapabilityAuthorizationError):
|
|
controller.execute(
|
|
verified, operation="tag-V2R1C3"
|
|
)
|
|
|
|
|
|
def test_baseline_and_registry_remain_separate_controllers():
|
|
assert BaselineController is not RegistryController
|
|
assert (
|
|
BaselineController.validate_result
|
|
is not RegistryController.validate_result
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"controller_type,target_field,operation",
|
|
[
|
|
(
|
|
BaselineController,
|
|
"baseline_target_oid",
|
|
"baseline-V2R1C3",
|
|
),
|
|
(
|
|
RegistryController,
|
|
"registry_target_oid",
|
|
"registry-V2R1C3",
|
|
),
|
|
],
|
|
)
|
|
def test_finalize_controllers_require_governance_lineage(
|
|
controller_type, target_field, operation
|
|
):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
signed, approval_verifier, _, _ = verifier(Path(td))
|
|
verified = approval_verifier.verify(
|
|
signed, candidate(), deployment_plan(), POLICY
|
|
)
|
|
result = {
|
|
"verified": True,
|
|
target_field: verified.source_commit,
|
|
"evidence_closure_oid": verified.evidence_closure_oid,
|
|
"approval_plan_id": verified.approval_plan_id,
|
|
"tag_verified": True,
|
|
"remote_peeled_target_oid": verified.source_commit,
|
|
"gitea_api_peeled_target_oid": verified.source_commit,
|
|
"gitea_listing_visible": True,
|
|
}
|
|
controller = controller_type(
|
|
CapabilityLedger(Path(td) / "capabilities"),
|
|
lambda: result,
|
|
)
|
|
assert controller.execute(
|
|
verified, operation=operation
|
|
)["status"] == "EXECUTED"
|
|
|
|
|
|
@pytest.mark.parametrize("state", list(SagaState))
|
|
def test_saga_state_round_trips_for_every_state(state):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
store = ReleaseSagaStore(Path(td) / "sagas")
|
|
saga = store.create(
|
|
"rc-hardening-001", "release-plan-001"
|
|
)
|
|
loaded = store.force_state_for_recovery_test(
|
|
saga.candidate_id, state
|
|
)
|
|
assert store.load(saga.candidate_id).state == state
|
|
assert loaded.state == state
|
|
|
|
|
|
def test_saga_cas_rejects_stale_revision():
|
|
with tempfile.TemporaryDirectory() as td:
|
|
store = ReleaseSagaStore(Path(td) / "sagas")
|
|
saga = store.create(
|
|
"rc-hardening-001", "release-plan-001"
|
|
)
|
|
store.compare_and_set(
|
|
saga.candidate_id,
|
|
expected_revision=0,
|
|
expected_states={SagaState.PREPARING},
|
|
target=SagaState.VERIFYING,
|
|
event="verification_started",
|
|
)
|
|
with pytest.raises(SagaConflict):
|
|
store.compare_and_set(
|
|
saga.candidate_id,
|
|
expected_revision=0,
|
|
expected_states={SagaState.PREPARING},
|
|
target=SagaState.VERIFYING,
|
|
event="stale_writer",
|
|
)
|
|
|
|
|
|
def test_saga_event_journal_survives_new_store_instance():
|
|
with tempfile.TemporaryDirectory() as td:
|
|
root = Path(td) / "sagas"
|
|
store = ReleaseSagaStore(root)
|
|
saga = store.create(
|
|
"rc-hardening-001", "release-plan-001"
|
|
)
|
|
store.compare_and_set(
|
|
saga.candidate_id,
|
|
expected_revision=0,
|
|
expected_states={SagaState.PREPARING},
|
|
target=SagaState.VERIFYING,
|
|
event="verification_started",
|
|
)
|
|
recovered = ReleaseSagaStore(root).load(saga.candidate_id)
|
|
assert recovered.revision == 1
|
|
assert recovered.events[-1]["event"] == "verification_started"
|
|
assert (root / "rc-hardening-001.events.jsonl").is_file()
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"state,action",
|
|
[
|
|
(SagaState.SWITCH_PENDING, "resume_runtime_switch"),
|
|
(SagaState.SWITCHED, "verify_runtime"),
|
|
(SagaState.RUNTIME_VERIFYING, "verify_runtime"),
|
|
(SagaState.TAG_PENDING, "reconcile_tag"),
|
|
(
|
|
SagaState.BASELINE_REGISTRY_PENDING,
|
|
"reconcile_baseline_registry",
|
|
),
|
|
(SagaState.ROLLBACK_PENDING, "resume_rollback"),
|
|
(
|
|
SagaState.MANUAL_OWNER_REVIEW_REQUIRED,
|
|
"manual_owner_review",
|
|
),
|
|
],
|
|
)
|
|
def test_saga_recovery_action_is_deterministic(state, action):
|
|
assert ReleaseSagaStore.recovery_action(state) == action
|
|
|
|
|
|
def test_two_controller_race_allows_only_one_cas_winner():
|
|
with tempfile.TemporaryDirectory() as td:
|
|
store_a = ReleaseSagaStore(Path(td) / "sagas")
|
|
store_b = ReleaseSagaStore(Path(td) / "sagas")
|
|
saga = store_a.create(
|
|
"rc-hardening-001", "release-plan-001"
|
|
)
|
|
first = store_a.compare_and_set(
|
|
saga.candidate_id,
|
|
expected_revision=0,
|
|
expected_states={SagaState.PREPARING},
|
|
target=SagaState.VERIFYING,
|
|
event="writer_a",
|
|
)
|
|
assert first.revision == 1
|
|
with pytest.raises(SagaConflict):
|
|
store_b.compare_and_set(
|
|
saga.candidate_id,
|
|
expected_revision=0,
|
|
expected_states={SagaState.PREPARING},
|
|
target=SagaState.VERIFYING,
|
|
event="writer_b",
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"name",
|
|
[
|
|
"TOKEN",
|
|
"GITEA_TOKEN",
|
|
"API_KEY",
|
|
"PASSWORD",
|
|
"CREDENTIAL",
|
|
"AUTHORIZATION",
|
|
"AWS_SECRET_ACCESS_KEY",
|
|
"DATABASE_URL",
|
|
"SSH_AUTH_SOCK",
|
|
"GIT_ASKPASS",
|
|
"PYTHONPATH",
|
|
"LD_PRELOAD",
|
|
],
|
|
)
|
|
def test_environment_allowlist_rejects_sensitive_or_internal_names(name):
|
|
environment = EnvironmentAllowList({"PATH", "LANG"})
|
|
with pytest.raises(ValueError):
|
|
environment.build({name: "must-not-survive"})
|
|
|
|
|
|
def test_environment_allowlist_does_not_inherit_process_environment(
|
|
monkeypatch,
|
|
):
|
|
monkeypatch.setenv("GITEA_TOKEN", "live-secret")
|
|
result = EnvironmentAllowList({"PATH", "LANG"}).build(
|
|
{"LANG": "C.UTF-8"}
|
|
)
|
|
assert "GITEA_TOKEN" not in result
|
|
assert result["LANG"] == "C.UTF-8"
|
|
|
|
|
|
def test_environment_allowlist_rejects_secret_values_in_allowed_name():
|
|
environment = EnvironmentAllowList({"PATH", "LANG"})
|
|
with pytest.raises(ValueError):
|
|
environment.build(
|
|
{"LANG": "Authorization: Bearer hidden-value"}
|
|
)
|
|
|
|
|
|
def test_environment_allowlist_returns_only_explicit_safe_values():
|
|
result = EnvironmentAllowList({"PATH", "LANG"}).build(
|
|
{"PATH": "/usr/bin:/bin", "LANG": "C.UTF-8"}
|
|
)
|
|
assert result == {"PATH": "/usr/bin:/bin", "LANG": "C.UTF-8"}
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"field,value",
|
|
[
|
|
("service", "unapproved.service"),
|
|
("release_path", "/tmp/escape"),
|
|
("dropin_path", "/etc/passwd"),
|
|
("dropin_sha256", "b" * 64),
|
|
("rollback_sha256", "c" * 64),
|
|
("policy_sha256", "d" * 64),
|
|
],
|
|
)
|
|
def test_deployment_plan_mutation_invalidates_pre_switch(field, value):
|
|
plan = deployment_plan()
|
|
signed = unsigned_approval(plan)
|
|
mutated = dataclasses.replace(plan, **{field: value})
|
|
validator = PreSwitchValidator(FakePreSwitchProbe(snapshot(plan)))
|
|
with pytest.raises(StaleApprovalError):
|
|
validator.validate(
|
|
candidate(),
|
|
signed,
|
|
mutated,
|
|
expected_repository="aios/aios",
|
|
)
|
|
|
|
|
|
def test_pre_switch_validator_accepts_exact_fresh_plan():
|
|
plan = deployment_plan()
|
|
result = PreSwitchValidator(
|
|
FakePreSwitchProbe(snapshot(plan))
|
|
).validate(
|
|
candidate(),
|
|
unsigned_approval(plan),
|
|
plan,
|
|
expected_repository="aios/aios",
|
|
)
|
|
assert result["valid"] is True
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"field,value",
|
|
[
|
|
("artifact_sha256", "f" * 64),
|
|
("policy_sha256", "e" * 64),
|
|
("writer_disabled", False),
|
|
("repository_identity", "other/repo"),
|
|
],
|
|
)
|
|
def test_pre_switch_live_mutation_requires_reapproval(field, value):
|
|
changed = snapshot()
|
|
changed[field] = value
|
|
validator = PreSwitchValidator(FakePreSwitchProbe(changed))
|
|
with pytest.raises(
|
|
StaleApprovalError,
|
|
match="APPROVAL_STALE_REAPPROVAL_REQUIRED",
|
|
):
|
|
validator.validate(
|
|
candidate(),
|
|
unsigned_approval(),
|
|
deployment_plan(),
|
|
expected_repository="aios/aios",
|
|
)
|
|
|
|
|
|
def test_shadow_action_service_creates_candidate_without_execution():
|
|
with tempfile.TemporaryDirectory() as td:
|
|
service = SilentReleaseActionService.shadow(Path(td))
|
|
created = service.create_release_candidate(
|
|
candidate().to_dict()
|
|
)
|
|
assert created["candidate_id"] == "rc-hardening-001"
|
|
result = service.start_deployment("rc-hardening-001")
|
|
assert result["execution_enabled"] is False
|
|
assert result["side_effects"] == []
|
|
|
|
|
|
def test_shadow_mode_never_invokes_runtime_tag_or_finalize_adapters():
|
|
class Bomb:
|
|
def __getattr__(self, name):
|
|
raise AssertionError(name)
|
|
|
|
with tempfile.TemporaryDirectory() as td:
|
|
service = SilentReleaseActionService.shadow(
|
|
Path(td), runtime_executor=Bomb()
|
|
)
|
|
service.create_release_candidate(candidate().to_dict())
|
|
result = service.start_deployment("rc-hardening-001")
|
|
assert result["status"] == "SHADOW_COMPLETED"
|
|
assert result["side_effects_executed"] is False
|
|
|
|
|
|
def test_action_service_candidate_and_saga_are_durable():
|
|
with tempfile.TemporaryDirectory() as td:
|
|
root = Path(td)
|
|
first = SilentReleaseActionService.shadow(root)
|
|
first.create_release_candidate(candidate().to_dict())
|
|
second = SilentReleaseActionService.shadow(root)
|
|
assert second.get_release_candidate(
|
|
"rc-hardening-001"
|
|
)["candidate_id"] == "rc-hardening-001"
|
|
assert second.get_release_saga(
|
|
"rc-hardening-001"
|
|
)["state"] == SagaState.PREPARING.value
|
|
|
|
|
|
def test_action_service_rejects_execution_without_signed_approval():
|
|
with tempfile.TemporaryDirectory() as td:
|
|
with pytest.raises(CapabilityAuthorizationError):
|
|
SilentReleaseActionService(
|
|
Path(td), execution_enabled=True
|
|
)
|
|
|
|
|
|
def test_deployment_plan_rejects_release_root_escape():
|
|
with pytest.raises(ValueError):
|
|
dataclasses.replace(
|
|
deployment_plan(), release_path="/tmp/escape"
|
|
).validate(
|
|
service_allowlist={"aios-gitops-model-api"},
|
|
dropin_allowlist={
|
|
(
|
|
"/data/aios-gitops/runtime/releases/"
|
|
+ COMMIT
|
|
+ "/deployment/production.conf"
|
|
)
|
|
},
|
|
)
|
|
|
|
|
|
def test_deployment_plan_rejects_unapproved_service():
|
|
with pytest.raises(ValueError):
|
|
deployment_plan().validate(
|
|
service_allowlist={"other-service"},
|
|
dropin_allowlist={deployment_plan().dropin_path},
|
|
)
|
|
|
|
|
|
def test_deployment_plan_accepts_contained_allowlisted_paths():
|
|
deployment_plan().validate(
|
|
service_allowlist={"aios-gitops-model-api"},
|
|
dropin_allowlist={deployment_plan().dropin_path},
|
|
)
|
|
|
|
|
|
def test_no_private_key_is_persisted_in_approval_or_keyring(tmp_path):
|
|
private, public = signing_material()
|
|
signed = sign(unsigned_approval(), private)
|
|
keyring = ApprovalKeyRing({"owner-key-1": public})
|
|
target = tmp_path / "public-keys.json"
|
|
keyring.save_public(target)
|
|
combined = (
|
|
json.dumps(signed.to_dict(), sort_keys=True)
|
|
+ target.read_text(encoding="utf-8")
|
|
)
|
|
assert "PRIVATE KEY" not in combined
|
|
assert base64.b64encode(public).decode("ascii") in combined
|
|
|
|
|
|
def test_expired_signed_approval_is_rejected(tmp_path):
|
|
expired = dataclasses.replace(
|
|
unsigned_approval(), expires_at="2020-01-01T00:00:00Z"
|
|
)
|
|
signed, approval_verifier, _, _ = verifier(tmp_path, expired)
|
|
with pytest.raises(CapabilityAuthorizationError):
|
|
approval_verifier.verify(
|
|
signed, candidate(), deployment_plan(), POLICY
|
|
)
|
|
|
|
|
|
def test_invalid_base64_signature_is_rejected(tmp_path):
|
|
_, approval_verifier, _, _ = verifier(tmp_path)
|
|
invalid = dataclasses.replace(
|
|
unsigned_approval(), signature="not:base64"
|
|
)
|
|
with pytest.raises(CapabilityAuthorizationError):
|
|
approval_verifier.verify(
|
|
invalid, candidate(), deployment_plan(), POLICY
|
|
)
|
|
|
|
|
|
def test_action_service_verifies_and_stores_owner_approval(tmp_path):
|
|
signed, approval_verifier, _, _ = verifier(tmp_path)
|
|
service = SilentReleaseActionService(
|
|
tmp_path / "control",
|
|
approval_verifier=approval_verifier,
|
|
policy_sha256=POLICY,
|
|
allow_legacy_candidate_input=True,
|
|
)
|
|
service.create_release_candidate(
|
|
{
|
|
**candidate().to_dict(),
|
|
"deployment_plan": deployment_plan().to_dict(),
|
|
}
|
|
)
|
|
result = service.submit_owner_approval(signed.to_dict())
|
|
assert result["status"] == "VERIFIED_AND_STORED"
|
|
assert service.get_release_saga(
|
|
candidate().candidate_id
|
|
)["release_plan_id"] == signed.release_plan_id
|
|
|
|
|
|
def test_action_service_owner_approval_submission_is_idempotent(tmp_path):
|
|
signed, approval_verifier, _, _ = verifier(tmp_path)
|
|
service = SilentReleaseActionService(
|
|
tmp_path / "control",
|
|
approval_verifier=approval_verifier,
|
|
policy_sha256=POLICY,
|
|
allow_legacy_candidate_input=True,
|
|
)
|
|
service.create_release_candidate(
|
|
{
|
|
**candidate().to_dict(),
|
|
"deployment_plan": deployment_plan().to_dict(),
|
|
}
|
|
)
|
|
service.submit_owner_approval(signed.to_dict())
|
|
assert (
|
|
service.submit_owner_approval(signed.to_dict())["status"]
|
|
== "IDEMPOTENT"
|
|
)
|
|
|
|
|
|
def test_action_service_rejects_owner_approval_without_public_key(tmp_path):
|
|
service = SilentReleaseActionService(
|
|
tmp_path / "control",
|
|
policy_sha256=POLICY,
|
|
allow_legacy_candidate_input=True,
|
|
)
|
|
service.create_release_candidate(
|
|
{
|
|
**candidate().to_dict(),
|
|
"deployment_plan": deployment_plan().to_dict(),
|
|
}
|
|
)
|
|
with pytest.raises(CapabilityAuthorizationError):
|
|
service.submit_owner_approval(unsigned_approval().to_dict())
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"endpoint",
|
|
[
|
|
"/createReleaseCandidate",
|
|
"/getReleaseCandidate",
|
|
"/getApprovalEnvelope",
|
|
"/submitOwnerApproval",
|
|
"/startDeployment",
|
|
"/getReleaseSaga",
|
|
],
|
|
)
|
|
def test_openapi_exposes_silent_release_control_actions(endpoint):
|
|
schema = (
|
|
ROOT / "docs" / "gitops" / "openapi-chatgpt-actions.yaml"
|
|
).read_text(encoding="utf-8")
|
|
assert f" {endpoint}:" in schema
|
|
|
|
|
|
def test_policy_keeps_silent_release_execution_disabled_by_default():
|
|
policy = json.loads(
|
|
(
|
|
ROOT / "config" / "gitops" / "silent-release-policy.json"
|
|
).read_text(encoding="utf-8")
|
|
)
|
|
assert policy["execution"]["default_enabled"] is False
|
|
assert policy["execution"]["shadow_side_effects_allowed"] is False
|
|
|
|
|
|
def test_policy_never_grants_forbidden_git_capabilities():
|
|
policy = json.loads(
|
|
(
|
|
ROOT / "config" / "gitops" / "silent-release-policy.json"
|
|
).read_text(encoding="utf-8")
|
|
)
|
|
assert set(policy["approval"]["capabilities"]) == {
|
|
"runtime_switch",
|
|
"tag_publish",
|
|
"baseline_registry_finalize",
|
|
}
|
|
assert {
|
|
"force_push",
|
|
"tag_rewrite",
|
|
"direct_main_push",
|
|
}.issubset(policy["forbidden"])
|
|
|
|
|
|
def test_action_api_execution_flag_is_fail_closed_by_default():
|
|
source = (
|
|
ROOT
|
|
/ "active-runtime"
|
|
/ "app"
|
|
/ "aios_chatgpt_actions_api.py"
|
|
).read_text(encoding="utf-8")
|
|
assert (
|
|
"'AIOS_SILENT_RELEASE_EXECUTION_ENABLED', 'false'" in source
|
|
)
|
|
assert "SILENT_RELEASE_EXECUTION_ENABLED" in source
|
|
|
|
|
|
def test_release_subprocesses_never_expand_entire_process_environment():
|
|
source = (
|
|
ROOT
|
|
/ "active-runtime"
|
|
/ "app"
|
|
/ "silent_release_controller.py"
|
|
).read_text(encoding="utf-8")
|
|
assert "{**os.environ" not in source
|
|
assert "env=dict(SAFE_PROCESS_ENVIRONMENT)" in source
|
|
|
|
|
|
def test_legacy_unsigned_runtime_approval_is_disabled_by_default(
|
|
tmp_path,
|
|
):
|
|
controller = DeploymentController(
|
|
object(),
|
|
ApprovalLedger(tmp_path / "legacy-approvals"),
|
|
SilentReleasePolicy.default(),
|
|
)
|
|
with pytest.raises(ApprovalError, match="legacy unsigned"):
|
|
controller.deploy(candidate(), object())
|
|
|
|
|
|
def systemd_config(root: Path, **overrides):
|
|
values = {
|
|
"release_root": root / "releases",
|
|
"cutover_link": root / "systemd" / "production.conf",
|
|
"service_name": "aios-gitops-model-api",
|
|
"health_url": "http://127.0.0.1:18890/api/run",
|
|
"deployment_lock": root / "locks" / "deployment.lock",
|
|
}
|
|
values.update(overrides)
|
|
return SystemdDeploymentConfig(**values)
|
|
|
|
|
|
def test_systemd_adapter_rejects_service_outside_allowlist(tmp_path):
|
|
with pytest.raises(ValueError):
|
|
SystemdDeploymentAdapter(
|
|
systemd_config(
|
|
tmp_path,
|
|
service_name="unapproved-service",
|
|
)
|
|
)
|
|
|
|
|
|
def test_systemd_adapter_rejects_dropin_outside_allowlist(tmp_path):
|
|
with pytest.raises(ValueError):
|
|
SystemdDeploymentAdapter(
|
|
systemd_config(
|
|
tmp_path,
|
|
dropin_name="unapproved.conf",
|
|
)
|
|
)
|
|
|
|
|
|
def test_systemd_adapter_rejects_release_dropin_symlink(tmp_path):
|
|
config = systemd_config(tmp_path)
|
|
release = (
|
|
config.release_root
|
|
/ COMMIT
|
|
/ "deployment"
|
|
)
|
|
release.mkdir(parents=True)
|
|
outside = tmp_path / "outside.conf"
|
|
outside.write_text("outside", encoding="utf-8")
|
|
(release / config.dropin_name).symlink_to(outside)
|
|
adapter = SystemdDeploymentAdapter(config)
|
|
with pytest.raises(RuntimeError, match="must not be a symlink"):
|
|
adapter.atomic_switch(candidate())
|
|
assert adapter._lock_handle is None
|
|
|
|
|
|
def test_systemd_adapter_rejects_existing_cutover_escape(tmp_path):
|
|
config = systemd_config(tmp_path)
|
|
target = (
|
|
config.release_root
|
|
/ COMMIT
|
|
/ "deployment"
|
|
/ config.dropin_name
|
|
)
|
|
target.parent.mkdir(parents=True)
|
|
target.write_text("approved", encoding="utf-8")
|
|
config.cutover_link.parent.mkdir(parents=True)
|
|
config.cutover_link.symlink_to(tmp_path / "outside.conf")
|
|
adapter = SystemdDeploymentAdapter(config)
|
|
with pytest.raises(RuntimeError, match="escapes release root"):
|
|
adapter.atomic_switch(candidate())
|
|
assert adapter._lock_handle is None
|
|
|
|
|
|
def test_systemd_adapter_tracks_pid_and_start_timestamp_in_source():
|
|
source = (
|
|
ROOT
|
|
/ "active-runtime"
|
|
/ "app"
|
|
/ "silent_release_controller.py"
|
|
).read_text(encoding="utf-8")
|
|
assert "--property=MainPID" in source
|
|
assert "--property=ActiveEnterTimestampMonotonic" in source
|
|
assert "deployment lock is not held" in source
|