Implement reproducible S1 handoff contracts
Assistant: codex Assistant-Model: gpt-5.6-sol Assistant-Session: 01a02994-7685-7940-bf34-3555b8256018
This commit is contained in:
parent
c8cb1c8edf
commit
b93af8cc78
44 changed files with 2035 additions and 342 deletions
13
tests/fixtures/inventory/invalid-mixed.yaml
vendored
Normal file
13
tests/fixtures/inventory/invalid-mixed.yaml
vendored
Normal file
|
|
@ -0,0 +1,13 @@
|
|||
schema_version: "1.0"
|
||||
servers:
|
||||
- name: unsafe-mixed-host
|
||||
provider: hosteurope
|
||||
lifecycle_mode: adopted
|
||||
ip: 192.0.2.10
|
||||
ssh_user: admin
|
||||
baseline_profile: ufw-managed
|
||||
provisioning:
|
||||
server_type: cpx21
|
||||
region: nbg1
|
||||
image: ubuntu-24.04
|
||||
role: test
|
||||
13
tests/fixtures/inventory/valid-hetzner.yaml
vendored
Normal file
13
tests/fixtures/inventory/valid-hetzner.yaml
vendored
Normal file
|
|
@ -0,0 +1,13 @@
|
|||
schema_version: "1.0"
|
||||
servers:
|
||||
- name: fixture-hetzner-01
|
||||
provider: hetzner
|
||||
lifecycle_mode: provider-managed
|
||||
ssh_user: admin
|
||||
baseline_profile: ufw-managed
|
||||
provisioning:
|
||||
server_type: cpx21
|
||||
region: nbg1
|
||||
image: ubuntu-24.04
|
||||
role: test
|
||||
labels: [fixture]
|
||||
62
tests/test_baseline_contract.py
Normal file
62
tests/test_baseline_contract.py
Normal file
|
|
@ -0,0 +1,62 @@
|
|||
from __future__ import annotations
|
||||
|
||||
import copy
|
||||
import json
|
||||
import subprocess
|
||||
import sys
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
sys.path.insert(0, str(ROOT / "scripts"))
|
||||
|
||||
from baseline_contract import ( # noqa: E402
|
||||
BaselineError,
|
||||
load_spec,
|
||||
profile_hostvars,
|
||||
validate_repo_consumers,
|
||||
validate_spec,
|
||||
)
|
||||
|
||||
|
||||
class BaselineContractTests(unittest.TestCase):
|
||||
def setUp(self) -> None:
|
||||
self.payload = load_spec(ROOT / "spec" / "server-baseline.yaml")
|
||||
|
||||
def test_profiles_resolve_distinct_firewall_controls(self) -> None:
|
||||
managed = profile_hostvars(self.payload, "ufw-managed")
|
||||
external = profile_hostvars(self.payload, "external-firewall")
|
||||
self.assertTrue(managed["ufw_manage"])
|
||||
self.assertIn("ufw", managed["baseline_required_services"])
|
||||
self.assertFalse(external["ufw_manage"])
|
||||
self.assertNotIn("ufw", external["baseline_required_services"])
|
||||
self.assertIn("replacement_control", external["baseline_firewall"])
|
||||
|
||||
def test_governed_package_mismatch_fails(self) -> None:
|
||||
broken = copy.deepcopy(self.payload)
|
||||
broken["defaults"]["packages"].remove("htop")
|
||||
with self.assertRaisesRegex(BaselineError, "htop"):
|
||||
validate_spec(broken)
|
||||
|
||||
def test_governed_ssh_weakening_fails(self) -> None:
|
||||
broken = copy.deepcopy(self.payload)
|
||||
broken["defaults"]["ssh_directives"]["PermitRootLogin"] = "yes"
|
||||
with self.assertRaisesRegex(BaselineError, "SSH"):
|
||||
validate_spec(broken)
|
||||
|
||||
def test_ansible_and_goss_consume_shared_variables(self) -> None:
|
||||
validate_repo_consumers(ROOT)
|
||||
|
||||
def test_dynamic_inventory_resolves_each_host_profile(self) -> None:
|
||||
raw = subprocess.check_output(
|
||||
[sys.executable, str(ROOT / "ansible" / "inventory_from_yaml.py")],
|
||||
text=True,
|
||||
cwd=ROOT,
|
||||
)
|
||||
hostvars = json.loads(raw)["_meta"]["hostvars"]
|
||||
self.assertEqual("external-firewall", hostvars["CoulombCore"]["baseline_profile"])
|
||||
self.assertEqual("ufw-managed", hostvars["Railiance01"]["baseline_profile"])
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
64
tests/test_handoff.py
Normal file
64
tests/test_handoff.py
Normal file
|
|
@ -0,0 +1,64 @@
|
|||
from __future__ import annotations
|
||||
|
||||
import sys
|
||||
import unittest
|
||||
from datetime import datetime, timezone
|
||||
from pathlib import Path
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
sys.path.insert(0, str(ROOT / "scripts"))
|
||||
|
||||
from s1_handoff import build_receipt # noqa: E402
|
||||
from s1_receipt import validate_receipt # noqa: E402
|
||||
|
||||
|
||||
HOSTS = [
|
||||
{"name": "Railiance01", "baseline_profile": "ufw-managed"},
|
||||
{"name": "CoulombCore", "baseline_profile": "external-firewall"},
|
||||
]
|
||||
|
||||
|
||||
class HandoffTests(unittest.TestCase):
|
||||
def test_all_hosts_and_evidence_produce_pass(self) -> None:
|
||||
receipt = build_receipt(
|
||||
revision="3734a1c",
|
||||
inventory_digest="a" * 64,
|
||||
hosts=HOSTS,
|
||||
results={"Railiance01": 0, "CoulombCore": 0},
|
||||
observed_at=datetime(2026, 8, 23, tzinfo=timezone.utc),
|
||||
freshness_hours=24,
|
||||
evidence=[
|
||||
{"kind": "goss-tap", "sha256": "b" * 64},
|
||||
{"kind": "goss-tap", "sha256": "c" * 64},
|
||||
],
|
||||
)
|
||||
self.assertEqual("pass", receipt["status"])
|
||||
validate_receipt(receipt)
|
||||
|
||||
def test_one_failed_host_fails_aggregate(self) -> None:
|
||||
receipt = build_receipt(
|
||||
revision="3734a1c",
|
||||
inventory_digest="a" * 64,
|
||||
hosts=HOSTS,
|
||||
results={"Railiance01": 0, "CoulombCore": 1},
|
||||
observed_at=datetime(2026, 8, 23, tzinfo=timezone.utc),
|
||||
freshness_hours=24,
|
||||
evidence=[],
|
||||
)
|
||||
self.assertEqual("fail", receipt["status"])
|
||||
|
||||
def test_dry_run_cannot_claim_pass(self) -> None:
|
||||
receipt = build_receipt(
|
||||
revision="3734a1c",
|
||||
inventory_digest="a" * 64,
|
||||
hosts=HOSTS,
|
||||
results=None,
|
||||
observed_at=datetime(2026, 8, 23, tzinfo=timezone.utc),
|
||||
freshness_hours=24,
|
||||
evidence=[],
|
||||
)
|
||||
self.assertEqual("not-run", receipt["status"])
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
64
tests/test_inventory_contract.py
Normal file
64
tests/test_inventory_contract.py
Normal file
|
|
@ -0,0 +1,64 @@
|
|||
from __future__ import annotations
|
||||
|
||||
import copy
|
||||
import sys
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
import yaml
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
sys.path.insert(0, str(ROOT / "scripts"))
|
||||
|
||||
from inventory_contract import ( # noqa: E402
|
||||
InventoryError,
|
||||
load_inventory,
|
||||
managed_hetzner_servers,
|
||||
validate_inventory,
|
||||
)
|
||||
|
||||
|
||||
class InventoryContractTests(unittest.TestCase):
|
||||
def test_current_adopted_inventory_has_no_managed_hetzner_hosts(self) -> None:
|
||||
payload = load_inventory(ROOT / "inventory" / "servers.yaml")
|
||||
self.assertEqual([], managed_hetzner_servers(payload))
|
||||
self.assertEqual(
|
||||
{"CoulombCore", "Railiance01"},
|
||||
{server["name"] for server in payload["servers"]},
|
||||
)
|
||||
|
||||
def test_valid_hetzner_fixture_is_selected(self) -> None:
|
||||
payload = load_inventory(
|
||||
ROOT / "tests" / "fixtures" / "inventory" / "valid-hetzner.yaml"
|
||||
)
|
||||
self.assertEqual(
|
||||
["fixture-hetzner-01"],
|
||||
[server["name"] for server in managed_hetzner_servers(payload)],
|
||||
)
|
||||
|
||||
def test_adopted_host_cannot_carry_provisioning_fields(self) -> None:
|
||||
with self.assertRaisesRegex(InventoryError, "must not carry"):
|
||||
load_inventory(
|
||||
ROOT / "tests" / "fixtures" / "inventory" / "invalid-mixed.yaml"
|
||||
)
|
||||
|
||||
def test_managed_host_requires_complete_provisioning(self) -> None:
|
||||
payload = yaml.safe_load(
|
||||
(ROOT / "tests" / "fixtures" / "inventory" / "valid-hetzner.yaml").read_text()
|
||||
)
|
||||
broken = copy.deepcopy(payload)
|
||||
del broken["servers"][0]["provisioning"]["server_type"]
|
||||
with self.assertRaisesRegex(InventoryError, "server_type"):
|
||||
validate_inventory(broken)
|
||||
|
||||
def test_managed_host_cannot_have_static_ip(self) -> None:
|
||||
payload = yaml.safe_load(
|
||||
(ROOT / "tests" / "fixtures" / "inventory" / "valid-hetzner.yaml").read_text()
|
||||
)
|
||||
payload["servers"][0]["ip"] = "192.0.2.20"
|
||||
with self.assertRaisesRegex(InventoryError, "addresses come from provider"):
|
||||
validate_inventory(payload)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
79
tests/test_secret_and_receipt_contracts.py
Normal file
79
tests/test_secret_and_receipt_contracts.py
Normal file
|
|
@ -0,0 +1,79 @@
|
|||
from __future__ import annotations
|
||||
|
||||
import copy
|
||||
import json
|
||||
import sys
|
||||
import unittest
|
||||
import uuid
|
||||
from pathlib import Path
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
sys.path.insert(0, str(ROOT / "scripts"))
|
||||
|
||||
from check_secret_paths import is_encrypted_content, is_protected_path # noqa: E402
|
||||
from s1_receipt import ReceiptError, load_receipt, validate_receipt # noqa: E402
|
||||
from sops_rotation import rotation_plan # noqa: E402
|
||||
|
||||
|
||||
class SecretAndReceiptContractTests(unittest.TestCase):
|
||||
def test_inventory_secret_paths_are_protected(self) -> None:
|
||||
self.assertTrue(is_protected_path("secrets/provider.yaml"))
|
||||
self.assertTrue(is_protected_path("inventory/group_vars/secrets.sops.yaml"))
|
||||
self.assertFalse(is_protected_path("inventory/group_vars/all.yaml"))
|
||||
|
||||
def test_plaintext_and_empty_files_fail(self) -> None:
|
||||
self.assertFalse(is_encrypted_content("secrets/example.yaml", "value: clear"))
|
||||
self.assertFalse(is_encrypted_content("secrets/example.age", ""))
|
||||
self.assertTrue(is_encrypted_content("secrets/example.yaml", "sops:\n age: []\n"))
|
||||
|
||||
def test_synthetic_chain_validates(self) -> None:
|
||||
load_receipt(ROOT / "docs" / "evidence" / "s1-receipts" / "synthetic-provisioning-chain.json")
|
||||
|
||||
def test_incomplete_passing_chain_fails(self) -> None:
|
||||
payload = json.loads(
|
||||
(ROOT / "docs" / "evidence" / "s1-receipts" / "synthetic-provisioning-chain.json").read_text()
|
||||
)
|
||||
payload["phases"][2]["status"] = "not-run"
|
||||
with self.assertRaisesRegex(ReceiptError, "every phase"):
|
||||
validate_receipt(payload)
|
||||
|
||||
def test_secret_shaped_receipt_content_fails(self) -> None:
|
||||
payload = json.loads(
|
||||
(ROOT / "docs" / "evidence" / "s1-receipts" / "synthetic-provisioning-chain.json").read_text()
|
||||
)
|
||||
payload["token"] = "not-even-a-real-value"
|
||||
with self.assertRaisesRegex(ReceiptError, "secret-shaped key"):
|
||||
validate_receipt(payload)
|
||||
|
||||
payload.pop("token")
|
||||
payload["provider_access_token"] = "redacted-is-still-not-allowed"
|
||||
with self.assertRaisesRegex(ReceiptError, "secret-shaped key"):
|
||||
validate_receipt(payload)
|
||||
|
||||
def test_passing_verification_without_evidence_fails(self) -> None:
|
||||
payload = {
|
||||
"schema_version": "1.0",
|
||||
"receipt_id": str(uuid.uuid4()),
|
||||
"event_type": "verification",
|
||||
"synthetic": False,
|
||||
"created_at": "2026-08-23T09:30:00Z",
|
||||
"source_revision": "3734a1c",
|
||||
"inventory_sha256": "a" * 64,
|
||||
"status": "pass",
|
||||
"hosts": ["Railiance01"],
|
||||
"profiles": {"Railiance01": "ufw-managed"},
|
||||
"observed_at": "2026-08-23T09:30:00Z",
|
||||
"fresh_until": "2026-08-24T09:30:00Z",
|
||||
"evidence": [],
|
||||
}
|
||||
with self.assertRaisesRegex(ReceiptError, "host evidence"):
|
||||
validate_receipt(payload)
|
||||
|
||||
def test_current_sops_recipient_metadata_matches_policy(self) -> None:
|
||||
plan = rotation_plan(ROOT)
|
||||
self.assertTrue(plan)
|
||||
self.assertFalse(any(item["changed"] for item in plan))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Loading…
Add table
Add a link
Reference in a new issue