coulomb/railiance-platform
5
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

RPF-WP-0015: T02 synthetic sender lane, T03 checklist revision, ops-warden re-acceptance

Browse source 
Addresses round-1 owner review requests from audit-core, rapp-postgres,
railiance-cluster and railiance-infra; refreshes the review contract.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

Assistant: claude-code
Assistant-Model: opus
Assistant-Process: 680842@bnt-lap001
Assistant-Session: 54eb9e8e-7d6b-4f0c-b7a6-a6db62075883
This commit is contained in:
codex 2026-10-14 12:49:40 +02:00
parent aee6fe79d7
commit ce8286dff9
13 changed files with 2032 additions and 52 deletions
Download patch file Download diff file Expand all files Collapse all files

17
tests/test_audit_core_database_lease_recovery.py
View file

@ -35,6 +35,12 @@ def approved_receipt(now: datetime) -> dict:
for owner in ("audit-core", "rapp-postgres", "railiance-platform")
},
"synthetic_load": {"contract_id": "load-1", "driver_revision": "abc123"},
"synthetic_sender": {
"identity": "t02-synthetic-sender",
"source": "t02-synthetic",
"tenant": "tenant:trial:t02-synthetic",
"expires_at": "2026-08-22T18:15:00Z",
},
}
@ -64,6 +70,17 @@ class DatabaseLeaseRecoveryTests(unittest.TestCase):
self.assertEqual("serving", module.pod_state(Remote())["uid"])
def test_approval_requires_projected_synthetic_sender(self) -> None:
now = datetime(2026, 8, 22, 12, 0, tzinfo=UTC)
with tempfile.TemporaryDirectory() as directory:
path = Path(directory) / "approval.json"
path.write_text(json.dumps(approved_receipt(now)), encoding="utf-8")
self.assertTrue(module.load_approval(path, now, require_open_window=True)["synthetic_sender"])
for sender in (None, {}, {"identity": "", "source": "s", "tenant": "t", "expires_at": "x"}):
path.write_text(json.dumps({**approved_receipt(now), "synthetic_sender": sender}), encoding="utf-8")
with self.assertRaises(module.ProcedureError):
module.load_approval(path, now, require_open_window=True)
def test_approval_requires_exact_owner_and_window_contract(self) -> None:
now = datetime(2026, 8, 22, 12, 0, tzinfo=UTC)
with tempfile.TemporaryDirectory() as directory:

249
tests/test_audit_core_recovery_preflight.py
View file

@ -19,6 +19,22 @@ SPEC.loader.exec_module(module)
class RecoveryPreflightTests(unittest.TestCase):
def test_ops_warden_receipt_reads_current_decision_of_review_history(self) -> None:
revision = "a" * 40
history = {
"workplan_id": "RPF-WP-0017",
"current_decision": "accept",
"current_reviewed_revision": revision,
"reviews": [{"decision": "reject"}, {"decision": "accept"}],
}
with tempfile.TemporaryDirectory() as tmp:
path = Path(tmp) / "receipt.json"
path.write_text(json.dumps(history), encoding="utf-8")
self.assertEqual(revision, module.validate_ops_warden_receipt(path)["accepted_revision"])
path.write_text(json.dumps({**history, "current_decision": "reject"}), encoding="utf-8")
with self.assertRaises(module.PreflightError):
module.validate_ops_warden_receipt(path)
def test_serving_pods_ignores_job_owned_attest_chain_pods(self) -> None:
def pod(name: str, kind: str) -> dict:
return {"metadata": {"name": name, "ownerReferences": [{"kind": kind}]}}
@ -92,6 +108,7 @@ class RecoveryPreflightTests(unittest.TestCase):
"external_secrets": {"a": {"ready": True}, "b": {"ready": True}},
"audit_core": {
"image_matches_reviewed_digest": True,
"running_pod_image_matches_reviewed_digest": True,
"ready_replicas": 1,
"replicas": 1,
},
@ -99,6 +116,9 @@ class RecoveryPreflightTests(unittest.TestCase):
self.assertTrue(module.automated_common_pass(state))
state["openbao"]["sealed"] = True
self.assertFalse(module.automated_common_pass(state))
state["openbao"]["sealed"] = False
state["audit_core"]["running_pod_image_matches_reviewed_digest"] = False
self.assertFalse(module.automated_common_pass(state))
def test_expected_database_projection_is_exact_key_set(self) -> None:
self.assertEqual(
@ -191,5 +211,234 @@ host: 50 bytes
)
DIGEST = module.EXPECTED_IMAGE.rsplit("@", 1)[1]
def workload(ns: str, kind: str, name: str, ready: int = 1, replicas: int = 1) -> dict:
return {
"kind": kind,
"metadata": {"namespace": ns, "name": name},
"spec": {"replicas": replicas},
"status": {"readyReplicas": ready},
}
def es_item(name: str, reason: str, status: str = "True", ns: str | None = None) -> dict:
meta = {"name": name, **({"namespace": ns} if ns else {})}
return {"metadata": meta, "status": {"conditions": [{"type": "Ready", "status": status, "reason": reason}]}}
def cnpg_cluster(primary: str = "pg-1", archiving: str | None = "True") -> dict:
cluster = {
"metadata": {"namespace": "databases", "name": "pg", "uid": "uid-1"},
"spec": {"instances": 1, **({"backup": {"barmanObjectStore": {}}} if archiving else {})},
"status": {"readyInstances": 1, "currentPrimary": primary, "healthyPVC": [primary], "phase": "Cluster in healthy state"},
}
if archiving:
cluster["status"]["conditions"] = [{"type": "ContinuousArchiving", "status": archiving}]
return cluster
class NodeRebootPreflightTests(unittest.TestCase):
def test_running_pod_image_uses_resolved_digest(self) -> None:
pod = {
"spec": {"containers": [{"name": "app", "image": module.EXPECTED_IMAGE}]},
"status": {"containerStatuses": [{"name": "app", "imageID": "forgejo.coulomb.social/coulomb/audit-core@" + DIGEST}]},
}
self.assertEqual(DIGEST, module.running_pod_image(pod)["digest"])
pod["status"]["containerStatuses"][0]["imageID"] = "x@sha256:" + "0" * 64
self.assertNotEqual(DIGEST, module.running_pod_image(pod)["digest"])
pod["status"]["containerStatuses"] = []
self.assertIsNone(module.running_pod_image(pod)["digest"])
def test_core_workloads_require_named_and_all_in_namespace(self) -> None:
items = [workload(*key) for key in module.CORE_PLATFORM_WORKLOADS]
self.assertTrue(module.core_platform_workloads({"items": items})["ready"])
result = module.core_platform_workloads({"items": items[1:]})
self.assertEqual(["kube-system/Deployment/coredns"], result["missing"])
extra = items + [workload("argocd", "Deployment", "argocd-server", ready=0)]
result = module.core_platform_workloads({"items": extra})
self.assertEqual(["argocd/Deployment/argocd-server"], result["unready"])
self.assertFalse(result["ready"])
ignored = items + [workload("coulomb", "Deployment", "probe", ready=0)]
self.assertTrue(module.core_platform_workloads({"items": ignored})["ready"])
def test_failing_pods_and_baseline_diff(self) -> None:
def pod(name: str, phase: str, ready: bool = True) -> dict:
return {"metadata": {"namespace": "ns", "name": name},
"status": {"phase": phase, "containerStatuses": [{"ready": ready}]}}
pods = {"items": [pod("ok", "Running"), pod("done", "Succeeded"), pod("bad", "Failed", False),
pod("pending", "Pending", False), pod("unready", "Running", False)]}
current = module.failing_pods(pods)
self.assertEqual(["ns/bad", "ns/pending", "ns/unready"], current)
diff = module.failing_pod_diff(["ns/bad", "ns/gone"], current)
self.assertEqual(["ns/pending", "ns/unready"], diff["new_failing"])
self.assertEqual(["ns/gone"], diff["resolved"])
self.assertFalse(diff["no_new_failures"])
self.assertTrue(module.failing_pod_diff(current, current)["no_new_failures"])
def test_cnpg_identity_and_health(self) -> None:
self.assertEqual("7672", module.system_identifier("x\nDatabase system identifier: 7672\ny"))
self.assertIsNone(module.system_identifier("nothing"))
identity = module.cnpg_identity(cnpg_cluster(), {"pg-1": "pvc-uid"})
identity["system_id"] = "7672"
self.assertEqual("pvc-uid", identity["primary_pvc_uid"])
self.assertTrue(module.cnpg_healthy(identity))
no_backup = module.cnpg_identity(cnpg_cluster(archiving=None), {"pg-1": "pvc-uid"})
no_backup["system_id"] = "7672"
self.assertIsNone(no_backup["continuous_archiving"])
self.assertTrue(module.cnpg_healthy(no_backup))
broken = module.cnpg_identity(cnpg_cluster(archiving="False"), {"pg-1": "pvc-uid"})
broken["system_id"] = "7672"
self.assertFalse(module.cnpg_healthy(broken))
missing_pvc = module.cnpg_identity(cnpg_cluster(), {})
missing_pvc["system_id"] = "7672"
self.assertFalse(module.cnpg_healthy(missing_pvc))
def test_backups_grouped_per_cluster(self) -> None:
backups = {"items": [
{"metadata": {"namespace": "databases", "name": "b1"}, "spec": {"cluster": {"name": "platform-pg"}},
"status": {"phase": "completed", "stoppedAt": "2026-10-14T00:00:00Z"}},
{"metadata": {"namespace": "databases", "name": "b2"}, "spec": {"cluster": {"name": "apps-pg"}},
"status": {"phase": "failed"}},
]}
result = module.latest_backups_by_cluster(backups, datetime(2026, 10, 14, 2, tzinfo=UTC))
self.assertEqual(2.0, result["databases/platform-pg"]["age_hours"])
self.assertIsNone(result["databases/apps-pg"])
def test_eso_inventory_requires_every_store_and_projection(self) -> None:
stores = {"items": [es_item("a", "Valid"), es_item("b", "Valid")]}
projections = {"items": [es_item("x", "SecretSynced", ns="n")]}
self.assertTrue(module.eso_inventory(stores, projections)["ready"])
stores["items"].append(es_item("c", "InvalidProviderConfig", "False"))
projections["items"].append(es_item("y", "SecretSyncedError", "False", ns="n"))
result = module.eso_inventory(stores, projections)
self.assertEqual(["c"], result["invalid_cluster_stores"])
self.assertEqual(["n/y"], result["unsynced_external_secrets"])
self.assertFalse(result["ready"])
self.assertFalse(module.eso_inventory({"items": []}, {"items": []})["ready"])
def test_ufw_matches_declaration_only(self) -> None:
ok = """Status: active
To Action From
-- ------ ----
OpenSSH ALLOW Anywhere
2224/tcp ALLOW Anywhere # nydus
OpenSSH (v6) ALLOW Anywhere (v6)
2224/tcp (v6) ALLOW Anywhere (v6) # nydus
"""
self.assertTrue(module.ufw_state(ok)["matches_declaration"])
opened = ok + "6443/tcp ALLOW Anywhere\n"
result = module.ufw_state(opened)
self.assertEqual(["6443/tcp"], result["forbidden_ports_open"])
self.assertFalse(result["matches_declaration"])
self.assertFalse(module.ufw_state("Status: inactive\n")["matches_declaration"])
def test_boot_count_and_reboot_comparison(self) -> None:
pre_time = datetime(2026, 10, 14, 8, tzinfo=UTC)
after = int((pre_time + timedelta(minutes=5)).timestamp() * 1_000_000)
before = int((pre_time - timedelta(days=1)).timestamp() * 1_000_000)
boots = json.dumps([{"boot_id": "a", "first_entry": before}, {"boot_id": "b", "first_entry": after}])
self.assertEqual(1, module.boots_since(boots, pre_time))
db = {"databases/pg": {"uid": "u", "primary_pvc_uid": "p", "system_id": "s"}}
pre = {"host": {"boot_id": "a"}, "cnpg_clusters": db, "failing_pods": ["ns/old"]}
post = {"host": {"boot_id": "b"}, "cnpg_clusters": json.loads(json.dumps(db)), "failing_pods": ["ns/old"]}
self.assertTrue(module.reboot_comparison(pre, post, 1)["passed"])
self.assertFalse(module.reboot_comparison(pre, post, 2)["exactly_one_reboot"])
same_boot = {**post, "host": {"boot_id": "a"}}
self.assertFalse(module.reboot_comparison(pre, same_boot, 1)["passed"])
post["cnpg_clusters"]["databases/pg"]["system_id"] = "rebootstrapped"
result = module.reboot_comparison(pre, post, 1)
self.assertEqual(["databases/pg:system_id"], result["database_identity_changes"])
self.assertFalse(result["passed"])
post["cnpg_clusters"] = {}
self.assertFalse(module.reboot_comparison(pre, post, 1)["passed"])
def test_consumed_decisions_and_scenario_are_not_reusable(self) -> None:
gates = module.fresh_decision_gates(
"WARDEN-WP-0027-T02-DRILL-20260822-01", "449a697a-5303-4582-aa9e-b0bc8b35ab2d"
)
self.assertFalse(any(gates.values()))
self.assertFalse(module.fresh_decision_gates("x", "85724c0c-e70f")["fresh_human_decision_recorded"])
gates = module.fresh_decision_gates("RAILIANCE-WP-0024-T03-REBOOT-20261020-01", "0123abcd-new")
self.assertTrue(all(gates.values()))
self.assertFalse(any(module.fresh_decision_gates(None, None).values()))
def _write(self, directory: str, name: str, text: str) -> Path:
path = Path(directory) / name
path.write_text(text, encoding="utf-8")
return path
def test_ops_warden_receipt_requires_acceptance_and_revision(self) -> None:
base = {"workplan_id": "RPF-WP-0017", "reviewed_revision": "a" * 40,
"secret_values_observed": False, "sensitive_material_recorded": False}
with tempfile.TemporaryDirectory() as directory:
ok = self._write(directory, "ok.json", json.dumps({**base, "decision": "accept"}))
self.assertEqual("a" * 40, module.validate_ops_warden_receipt(ok)["accepted_revision"])
for name, body in (
("reject.json", {**base, "decision": "reject"}),
("norev.json", {**base, "decision": "accept", "reviewed_revision": "HEAD"}),
("other.json", {**base, "decision": "accept", "workplan_id": "RPF-WP-0099"}),
("leak.json", {**base, "decision": "accept", "note": "hvs.abc"}),
("sensitive.json", {**base, "decision": "accept", "sensitive_material_recorded": True}),
):
with self.assertRaises(module.PreflightError, msg=name):
module.validate_ops_warden_receipt(self._write(directory, name, json.dumps(body)))
with self.assertRaises(module.PreflightError):
module.validate_ops_warden_receipt(Path(directory) / "missing.json")
def test_console_receipt_same_day_window_and_independent(self) -> None:
now = datetime(2026, 10, 15, 9, 0, tzinfo=UTC)
good = {
"receipt_id": "RAIL01-CONSOLE-20261015T080000Z", "host": "railiance01",
"attested_by_role": "host-europe-account-holder", "attested_at": "2026-10-15T08:00:00Z",
"panel_login": "ok", "mfa_used": True, "console_mechanism": "VNC console",
"console_shows_host": True, "independent_of_guest_network": True,
"rescue_mode_offered": True, "hard_reboot_offered": True, "reboot_window_id": "WIN-20261015-01",
}
name = good["receipt_id"] + ".yaml"
with tempfile.TemporaryDirectory() as directory:
ok = self._write(directory, name, json.dumps(good)) # JSON is valid YAML
result = module.validate_console_receipt(ok, window_id="WIN-20261015-01", now=now)
self.assertTrue(result["verified"])
with self.assertRaises(module.PreflightError):
module.validate_console_receipt(ok, window_id="WIN-OTHER", now=now)
with self.assertRaises(module.PreflightError):
module.validate_console_receipt(ok, window_id="WIN-20261015-01", now=now + timedelta(days=1))
for change in (
{"independent_of_guest_network": "unknown"},
{"console_shows_host": False},
{"panel_login": "failed"},
{"console_mechanism": "<as named in the panel>"},
{"note": "password=hunter2"},
):
path = self._write(directory, name, json.dumps({**good, **change}))
with self.assertRaises(module.PreflightError, msg=str(change)):
module.validate_console_receipt(path, window_id="WIN-20261015-01", now=now)
renamed = self._write(directory, "RAIL01-CONSOLE-20261015T070000Z.yaml", json.dumps(good))
with self.assertRaises(module.PreflightError):
module.validate_console_receipt(renamed, window_id="WIN-20261015-01", now=now)
def test_reboot_automated_pass_names_each_boundary(self) -> None:
extra = {
"host": {"ntp_synchronized": True, "firewall": {"matches_declaration": True}},
"k3s_api_tunnel_answers": True,
"core_platform_workloads": {"ready": True},
"missing_expected_cnpg_clusters": [],
"cnpg_clusters": {},
"cnpg_backups": {"databases/platform-pg": {"age_hours": 3}},
"eso": {"ready": True},
}
self.assertTrue(all(module.reboot_automated_pass(extra, 26).values()))
extra["host"]["ntp_synchronized"] = False
extra["missing_expected_cnpg_clusters"] = ["databases/state-hub-db"]
checks = module.reboot_automated_pass(extra, 26)
self.assertFalse(checks["ntp_synchronized"])
self.assertFalse(checks["all_cnpg_clusters_healthy"])
extra["cnpg_backups"] = {}
self.assertFalse(module.reboot_automated_pass(extra, 26)["required_backups_fresh"])
if __name__ == "__main__":
unittest.main()

242
tests/test_audit_core_synthetic_sender_lane.py Normal file
View file

@ -0,0 +1,242 @@
from __future__ import annotations
import argparse
import importlib.util
import io
import json
import os
import tempfile
import unittest
from contextlib import redirect_stderr, redirect_stdout
from datetime import UTC, datetime, timedelta
from pathlib import Path
from unittest import mock
ROOT = Path(__file__).resolve().parents[1]
SPEC = importlib.util.spec_from_file_location(
"audit_core_synthetic_sender_lane",
ROOT / "scripts" / "audit_core_synthetic_sender_lane.py",
)
assert SPEC and SPEC.loader
lane = importlib.util.module_from_spec(SPEC)
SPEC.loader.exec_module(lane)
CONTRACT = "RAILIANCE-WP-0024-T02-SYNTHETIC-01"
def z(value: datetime) -> str:
return value.astimezone(UTC).replace(microsecond=0).isoformat().replace("+00:00", "Z")
def receipt(now: datetime, **sender: str) -> dict:
start = now - timedelta(minutes=1)
end = start + timedelta(minutes=15)
return {
"procedure": lane.LEASE.PROCEDURE,
"task_id": lane.LEASE.TASK_ID,
"status": "approved",
"approval_id": "t02-window-1",
"window": {"start": z(start), "end": z(end)},
"abort_operator": "operator:example",
"owners": {
owner: {"acknowledged": True, "message_id": f"msg-{owner}"}
for owner in ("audit-core", "rapp-postgres", "railiance-platform")
},
"synthetic_load": {"contract_id": CONTRACT, "driver_revision": "sha256:" + "0" * 64},
"synthetic_sender": {
"identity": "t02-synthetic-1",
"source": "railiance-t02-synthetic",
"tenant": "tenant:trial:t02-synthetic-1",
"expires_at": z(end),
**sender,
},
}
class FakeOperator:
def __init__(self, registry: list) -> None:
self.registry_value = registry
self.writes: list[list] = []
def registry(self) -> list:
return json.loads(json.dumps(self.registry_value))
def write_registry(self, registry: list) -> None:
self.writes.append(registry)
self.registry_value = json.loads(json.dumps(registry))
class SyntheticSenderLaneTests(unittest.TestCase):
def setUp(self) -> None:
self.temp = tempfile.TemporaryDirectory()
self.dir = Path(self.temp.name)
os.chmod(self.dir, 0o700)
self.now = datetime.now(UTC)
def tearDown(self) -> None:
self.temp.cleanup()
def write_receipt(self, document: dict) -> Path:
path = self.dir / "receipt.json"
path.write_text(json.dumps(document), encoding="utf-8")
return path
def test_receipt_yields_exactly_one_scoped_expiring_identity(self) -> None:
_, spec = lane.load_receipt(self.write_receipt(receipt(self.now)), self.now, require_open_window=True)
entry = lane.registry_entry(spec, "bearer")
self.assertEqual(["railiance-t02-synthetic"], entry["sources"])
self.assertEqual(["tenant:trial:t02-synthetic-1"], entry["tenants"])
self.assertTrue(entry["may_write"] and entry["may_read"])
self.assertLessEqual(lane.LEASE.parse_time(entry["expires_at"]), lane.LEASE.parse_time(receipt(self.now)["window"]["end"]))
def test_receipt_rejections(self) -> None:
end = lane.LEASE.parse_time(receipt(self.now)["window"]["end"])
bad = [
{"source": "*"},
{"source": "user-engine"},
{"tenant": "*"},
{"tenant": "tenant:friendly:binky"},
{"identity": "user-engine"},
{"expires_at": z(end + timedelta(seconds=1))},
{"expires_at": "2026-10-14T10:00:00+00:00"},
]
for override in bad:
with self.subTest(override=override):
with self.assertRaises(lane.LaneError):
lane.sender_spec(receipt(self.now, **override))
missing = receipt(self.now)
del missing["synthetic_sender"]
with self.assertRaises(lane.LaneError):
lane.sender_spec(missing)
extra = receipt(self.now)
extra["synthetic_sender"]["tokens"] = ["x"]
with self.assertRaises(lane.LaneError):
lane.sender_spec(extra)
wrong_contract = receipt(self.now)
wrong_contract["synthetic_load"]["contract_id"] = "OTHER"
with self.assertRaises(lane.LaneError):
lane.sender_spec(wrong_contract)
def test_pending_example_receipt_is_refused(self) -> None:
with self.assertRaises(lane.LEASE.ProcedureError):
lane.load_receipt(ROOT / "docs" / "audit-core-database-lease-approval.example.json", self.now, require_open_window=False)
def test_projection_time_gate(self) -> None:
document = receipt(self.now)
spec = lane.sender_spec(document)
lane.assert_projection_time(document, spec, self.now)
start = lane.LEASE.parse_time(document["window"]["start"])
for when in (start - timedelta(seconds=1), start + lane.PROJECT_CUTOFF + timedelta(seconds=1)):
with self.assertRaises(lane.LaneError):
lane.assert_projection_time(document, spec, when)
short = receipt(self.now, expires_at=z(self.now + timedelta(minutes=5)))
with self.assertRaises(lane.LaneError):
lane.assert_projection_time(short, lane.sender_spec(short), self.now)
def test_registry_add_refuses_overlap_and_remove_is_exact(self) -> None:
spec = lane.sender_spec(receipt(self.now))
production = [{"name": "user-engine", "tokens": ["p"], "sources": ["user-engine"], "tenants": ["*"]}]
updated = lane.add_identity(production, spec, "bearer")
self.assertEqual(2, len(updated))
with self.assertRaises(lane.LaneError):
lane.add_identity(updated, spec, "again")
shared_source = [{"name": "x", "sources": [spec["source"]], "tenants": ["tenant:trial:x"]}]
self.assertTrue(lane.registry_conflicts(shared_source, spec))
cleaned, removed = lane.remove_identity(updated, spec)
self.assertEqual((production, 1), (cleaned, removed))
self.assertEqual((production, 0), lane.remove_identity(production, spec))
hijacked = [{"name": spec["name"], "sources": ["user-engine"], "tenants": ["*"]}]
with self.assertRaises(lane.LaneError):
lane.remove_identity(hijacked, spec)
def test_bearer_file_is_new_0600_outside_git(self) -> None:
target = self.dir / "t02.bearer"
lane.write_bearer(target, "value")
self.assertEqual(0o600, target.stat().st_mode & 0o777)
lane.check_bearer_path(target, must_exist=True)
with self.assertRaises(lane.LaneError):
lane.write_bearer(target, "value")
with self.assertRaises(lane.LaneError):
lane.check_bearer_path(ROOT / "t02.bearer", must_exist=False)
with self.assertRaises(lane.LaneError):
lane.check_bearer_path(Path("relative.bearer"), must_exist=False)
open_dir = self.dir / "open"
open_dir.mkdir(mode=0o755)
os.chmod(open_dir, 0o755)
with self.assertRaises(lane.LaneError):
lane.check_bearer_path(open_dir / "t02.bearer", must_exist=False)
def test_cleanup_requires_matching_evidence_or_abort_reason(self) -> None:
approval = receipt(self.now)
with self.assertRaises(lane.LaneError):
lane.assert_evidence(None, None, approval)
evidence = self.dir / "evidence.json"
evidence.write_text(json.dumps({
"procedure": lane.LEASE.PROCEDURE, "approval_id": "other",
"completed_at": z(self.now), "secret_values_observed": False,
}), encoding="utf-8")
with self.assertRaises(lane.LaneError):
lane.assert_evidence(evidence, None, approval)
evidence.write_text(json.dumps({
"procedure": lane.LEASE.PROCEDURE, "approval_id": approval["approval_id"],
"completed_at": z(self.now), "secret_values_observed": False,
}), encoding="utf-8")
self.assertEqual("evidence-recorded", lane.assert_evidence(evidence, None, approval)["mode"])
self.assertEqual("abort", lane.assert_evidence(None, "readiness never dropped", approval)["mode"])
def test_cleanup_registry_is_idempotent_and_restarts_only_on_change(self) -> None:
spec = lane.sender_spec(receipt(self.now))
operator = FakeOperator(lane.add_identity([], spec, "bearer"))
with mock.patch.object(lane.CREDS, "force_sender_sync_and_restart") as restart:
self.assertEqual(1, lane._cleanup_registry(operator, spec))
self.assertEqual(0, lane._cleanup_registry(operator, spec))
self.assertEqual(1, restart.call_count)
self.assertEqual([], operator.registry_value)
def test_project_rolls_back_registry_and_bearer_on_restart_failure(self) -> None:
approval_path = self.write_receipt(receipt(self.now))
approval = json.loads(approval_path.read_text())
bearer = self.dir / "t02.bearer"
operator = FakeOperator([{"name": "user-engine", "tokens": ["p"], "sources": ["user-engine"], "tenants": ["*"]}])
args = argparse.Namespace(
approval=approval_path, confirm=lane.confirm_token(approval), bearer_file=bearer,
remote="railiance01", token_file=self.dir / "unused",
)
green = {"automated_checks_passed": True, "ready": "1/1", "pod_uid": "uid-1"}
restart = mock.Mock(side_effect=[lane.CREDS.ProcedureError("rollout failed"), None])
with mock.patch.object(lane.PREFLIGHT, "Remote"), \
mock.patch.object(lane, "cluster_state", return_value=green), \
mock.patch.object(lane.LEASE, "attest_chain_state"), \
mock.patch.object(lane.CREDS, "Operator", return_value=operator), \
mock.patch.object(lane.CREDS, "force_sender_sync_and_restart", restart):
with self.assertRaises(lane.CREDS.ProcedureError):
lane.project(args)
self.assertFalse(bearer.exists())
self.assertEqual(["user-engine"], [entry["name"] for entry in operator.registry_value])
self.assertEqual(2, restart.call_count)
def test_project_requires_exact_confirmation_and_no_bearer_on_stdout(self) -> None:
approval_path = self.write_receipt(receipt(self.now))
stdout, stderr = io.StringIO(), io.StringIO()
with redirect_stdout(stdout), redirect_stderr(stderr):
code = lane.main(["project", "--approval", str(approval_path), "--confirm", "wrong",
"--bearer-file", str(self.dir / "b")])
self.assertEqual(1, code)
self.assertIn("--confirm audit-core-synthetic-sender:t02-window-1:attended", stderr.getvalue())
self.assertEqual("", stdout.getvalue())
def test_offline_preflight_is_value_safe(self) -> None:
stdout = io.StringIO()
with redirect_stdout(stdout):
code = lane.main(["preflight", "--offline", "--approval", str(self.write_receipt(receipt(self.now))),
"--bearer-file", str(self.dir / "t02.bearer")])
self.assertEqual(0, code)
result = json.loads(stdout.getvalue())
self.assertEqual("projection-window-open", result["time_state"])
self.assertFalse(result["secret_values_observed"])
self.assertNotIn("tokens", json.dumps(result))
if __name__ == "__main__":
unittest.main()