coulomb/zone-engine
4
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat: harden zone reference contracts

Browse source 
Assistant: codex
Assistant-Model: gpt-5.6-sol
Assistant-Session: 01a0291a-1e87-7151-9934-fcbfe3f65eb1
This commit is contained in:
tegwick 2026-08-23 12:27:13 +02:00
parent bed5c3b53a
commit be29c28100
19 changed files with 1712 additions and 454 deletions
Download patch file Download diff file Expand all files Collapse all files

63
tests/test_canon_lineage.py Normal file
View file

@ -0,0 +1,63 @@
from pathlib import Path
from tempfile import TemporaryDirectory
import unittest
from tools.check_canon_lineage import check_lineage
CANON = b"""---
id: netkingdom-security-zones-v0.1
status: proposed
---
# Fixture canon
"""
class CanonLineageTest(unittest.TestCase):
def test_matching_artifact_passes_and_mutation_fails(self):
import hashlib
with TemporaryDirectory() as directory:
root = Path(directory)
path = root / "canon" / "standards" / "security-zones_v0.1.md"
path.parent.mkdir(parents=True)
path.write_bytes(CANON)
manifest = {
"standard": "canon-lineage_v0.1",
"artifact": "security-zones_v0.1",
"publication_owner": "net-kingdom",
"canonical_path": "canon/standards/security-zones_v0.1.md",
"canonical_revision": "fixture",
"canonical_status": "proposed",
"canonical_sha256": hashlib.sha256(CANON).hexdigest(),
}
result = check_lineage(manifest, root, verify_revision=False)
self.assertTrue(result["ok"])
path.write_bytes(CANON + b"changed\n")
result = check_lineage(manifest, root, verify_revision=False)
self.assertFalse(result["ok"])
self.assertIn("content hash changed", result["errors"][0])
def test_lifecycle_change_is_detected_separately(self):
import hashlib
accepted = CANON.replace(b"proposed", b"accepted")
with TemporaryDirectory() as directory:
root = Path(directory)
path = root / "standard.md"
path.write_bytes(accepted)
manifest = {
"standard": "canon-lineage_v0.1",
"canonical_path": "standard.md",
"canonical_revision": "fixture",
"canonical_status": "proposed",
"canonical_sha256": hashlib.sha256(accepted).hexdigest(),
}
result = check_lineage(manifest, root, verify_revision=False)
self.assertFalse(result["ok"])
self.assertIn("lifecycle changed", result["errors"][0])
if __name__ == "__main__":
unittest.main()

217
tests/test_resolve_zones.py
View file

@ -1,10 +1,16 @@
from copy import deepcopy
from pathlib import Path
from tempfile import TemporaryDirectory
import unittest
import yaml
from tools.resolve_zones import resolve_paths
from tools.resolve_zones import (
PROFILE_ZONES,
compare_snapshots,
resolve_manifest,
resolve_paths,
)
def declaration(*, service="flex-auth", zone="z2-protected", maturity="M2"):
@ -44,45 +50,120 @@ def declaration(*, service="flex-auth", zone="z2-protected", maturity="M2"):
}
def control_profile():
def zones(enforced):
return {
zone: {
"stance": "enforced" if zone in enforced else "advisory",
"failure_mode": "fail_closed" if zone == "z3-critical" else "fail_open",
}
for zone in PROFILE_ZONES
}
return {
"standard": "security-zone-control-profile_v0.1",
"profile_id": "netkingdom-build",
"version": "flex-auth@policy-v2+ops-warden@zone-v1",
"controls": {
"flex-auth/pre-sign": {
"policy_owner": "flex-auth",
"pep_owner": "ops-warden",
"policy_ref": "flex-auth/examples/ops-warden/policy_package.md@v2",
"zones": zones(
{"z2-protected", "z2-continuity", "z3-critical"}
),
}
},
}
class ResolveZonesTest(unittest.TestCase):
def resolve(self, document):
def resolve(
self,
document,
*,
source_revision="fixture@abc123",
workload_refs=None,
profile=None,
previous=None,
):
with TemporaryDirectory() as directory:
path = Path(directory) / "tenancy.yaml"
path.write_text(yaml.safe_dump(document, sort_keys=False))
return resolve_paths([path])
return resolve_paths(
[path],
source_revision=source_revision,
workload_refs=workload_refs,
profile=profile,
previous=previous,
)
def test_satisfied_membership_selects_zone_control_profile(self):
def test_satisfied_membership_is_source_bound_without_implicit_controls(self):
result = self.resolve(declaration())
self.assertTrue(result["ok"])
record = result["records"][0]
self.assertEqual(record["admission"], "satisfied")
self.assertEqual(record["effective_zone"], "z2-protected")
pre_sign = next(c for c in record["controls"] if c["id"] == "flex-auth/pre-sign")
self.assertEqual(pre_sign["stance"], "enforced")
self.assertEqual(pre_sign["failure_mode"], "fail_open")
self.assertNotIn("controls", record)
self.assertEqual(record["workload_ref"]["applicability"], "applicable")
self.assertEqual(record["workload_ref"]["name"], "flex-auth")
self.assertTrue(record["membership_revision"].startswith("sha256:"))
self.assertIn("source-revision-bound-membership", record["guarantees"])
def test_below_floor_is_unsatisfied_and_uses_unknown_profile(self):
result = self.resolve(declaration(maturity="M1"))
def test_explicit_profile_projects_owner_and_version_provenance(self):
result = self.resolve(declaration(), profile=control_profile())
self.assertTrue(result["ok"])
record = result["records"][0]
self.assertEqual(
record["control_profile"],
{
"id": "netkingdom-build",
"version": "flex-auth@policy-v2+ops-warden@zone-v1",
},
)
self.assertEqual(record["controls"][0]["policy_owner"], "flex-auth")
self.assertEqual(record["controls"][0]["pep_owner"], "ops-warden")
self.assertEqual(record["controls"][0]["stance"], "enforced")
self.assertTrue(record["controls"][0]["policy_ref"])
def test_invalid_profile_fails_projection_but_preserves_membership(self):
profile = control_profile()
profile["controls"]["flex-auth/pre-sign"]["zones"].pop("unknown")
result = self.resolve(declaration(), profile=profile)
self.assertFalse(result["ok"])
self.assertEqual(result["records"][0]["admission"], "satisfied")
self.assertNotIn("controls", result["records"][0])
self.assertIn("must be total", result["profile_errors"][0])
def test_ownerless_and_unqualified_profiles_are_rejected(self):
profile = control_profile()
profile["controls"]["pre-sign"] = profile["controls"].pop(
"flex-auth/pre-sign"
)
result = self.resolve(declaration(), profile=profile)
self.assertFalse(result["ok"])
self.assertIn("owner-qualified", result["profile_errors"][0])
def test_below_floor_is_unsatisfied_and_profile_uses_unknown(self):
result = self.resolve(declaration(maturity="M1"), profile=control_profile())
self.assertTrue(result["ok"])
record = result["records"][0]
self.assertEqual(record["admission"], "unsatisfied")
self.assertEqual(record["effective_zone"], "unknown")
pre_sign = next(c for c in record["controls"] if c["id"] == "flex-auth/pre-sign")
self.assertEqual(pre_sign["stance"], "advisory")
self.assertEqual(record["controls"][0]["stance"], "advisory")
def test_zone_below_context_floor_is_unsatisfied_even_with_high_maturity(self):
document = declaration(zone="z1-operational", maturity="M2")
result = self.resolve(document)
result = self.resolve(declaration(zone="z1-operational", maturity="M2"))
self.assertTrue(result["ok"])
record = result["records"][0]
self.assertEqual(record["admission"], "unsatisfied")
self.assertEqual(record["admission_reason"], "z1-operational_below_M2_context_floor")
self.assertEqual(result["records"][0]["admission"], "unsatisfied")
self.assertEqual(
result["records"][0]["admission_reason"],
"z1-operational_below_M2_context_floor",
)
def test_continuity_zone_requires_dependency_and_recovery_evidence(self):
document = declaration(zone="z2-continuity")
result = self.resolve(document)
self.assertTrue(result["ok"])
self.assertEqual(result["records"][0]["admission"], "unsatisfied")
document["zones"]["evidence"][0]["supports"].extend(
["continuity-dependency", "recovery"]
@ -101,13 +182,60 @@ class ResolveZonesTest(unittest.TestCase):
"public_data_classification_floor_unresolved",
)
def test_missing_membership_returns_unknown_without_inference(self):
def test_missing_identity_and_membership_never_infer_from_service_or_path(self):
document = declaration()
document.pop("zones")
document.pop("workload_identity")
result = self.resolve(document)
record = result["records"][0]
self.assertTrue(result["ok"])
self.assertEqual(result["records"][0]["effective_zone"], "unknown")
self.assertIsNone(record["workload_id"])
self.assertIsNone(record["workload_ref"]["name"])
self.assertEqual(record["admission_reason"], "workload_reference_unresolved")
def test_managed_reference_must_match_authoritative_identity(self):
ref = {
"flex-auth": {
"applicability": "applicable",
"rapp_id": "rapp-flex-auth",
"name": "flex-auth",
"deployable": "flex-auth",
}
}
result = self.resolve(declaration(), workload_refs=ref)
self.assertTrue(result["ok"])
self.assertEqual(result["records"][0]["workload_ref"]["rapp_id"], "rapp-flex-auth")
ref["flex-auth"]["name"] = "guessed-from-repo"
result = self.resolve(declaration(), workload_refs=ref)
self.assertFalse(result["ok"])
self.assertIn("must equal workload_identity.name", result["errors"][0]["error"])
def test_manifest_represents_explicit_not_applicable_subject(self):
with TemporaryDirectory() as directory:
root = Path(directory)
declaration_path = root / "tenancy.yaml"
declaration_path.write_text(yaml.safe_dump(declaration()))
manifest = {
"standard": "zone-resolver-input_v0.1",
"sources": [
{
"path": "tenancy.yaml",
"source_revision": "flex-auth@abc123",
}
],
"subjects": [
{
"subject_id": "human-operator",
"source": "ops-warden/catalog@abc123",
"workload_ref": {"applicability": "not-applicable"},
}
],
}
result = resolve_manifest(manifest, base_dir=root)
by_id = {record["subject_id"]: record for record in result["records"]}
self.assertTrue(result["ok"])
self.assertEqual(by_id["human-operator"]["admission"], "not-applicable")
self.assertIsNone(by_id["human-operator"]["effective_zone"])
def test_zone_requires_identity_bound_to_service(self):
document = declaration()
@ -123,6 +251,57 @@ class ResolveZonesTest(unittest.TestCase):
self.assertFalse(result["ok"])
self.assertIn("data_classification_reason is required", result["errors"][0]["error"])
def test_missing_source_revision_never_claims_a_bound_digest(self):
result = self.resolve(declaration(), source_revision=None)
record = result["records"][0]
self.assertTrue(result["ok"])
self.assertIsNone(record["membership_revision"])
self.assertEqual(record["membership_revision_reason"], "source_revision_absent")
self.assertNotIn("source-revision-bound-membership", record["guarantees"])
def test_revision_is_order_independent_and_source_sensitive(self):
document = declaration()
document["workload_identity"]["identity_bindings"].append(
{
"scheme": "spiffe",
"authority": "railiance01",
"subject": "spiffe://railiance01/ns/flex-auth/sa/flex-auth",
"principal_type": "service",
}
)
document["zones"]["evidence"].append(
{"ref": "second", "supports": ["on-call", "M2"]}
)
first = self.resolve(document)["records"][0]["membership_revision"]
reordered = deepcopy(document)
reordered["workload_identity"]["identity_bindings"].reverse()
reordered["zones"]["evidence"].reverse()
reordered["zones"]["evidence"][0]["supports"].reverse()
second = self.resolve(reordered)["records"][0]["membership_revision"]
changed_source = self.resolve(
document, source_revision="fixture@different"
)["records"][0]["membership_revision"]
self.assertEqual(first, second)
self.assertNotEqual(first, changed_source)
def test_snapshot_reports_add_remove_and_change(self):
before = self.resolve(declaration(service="old-service"))
current_document = {
"services": [
declaration(service="old-service", zone="z3-critical", maturity="M3"),
declaration(service="new-service"),
]
}
current = self.resolve(current_document, previous=before)
self.assertEqual(current["changes"]["added"], ["new-service"])
self.assertEqual(current["changes"]["removed"], [])
self.assertEqual(
[item["subject_id"] for item in current["changes"]["changed"]],
["old-service"],
)
removed = compare_snapshots([], current)
self.assertEqual(removed["removed"], ["new-service", "old-service"])
if __name__ == "__main__":
unittest.main()

151
tests/test_zone_exceptions.py Normal file
View file

@ -0,0 +1,151 @@
from copy import deepcopy
from datetime import datetime, timezone
import unittest
from tools.check_zone_exceptions import evaluate_exceptions
def policy():
return {
"standard": "security-zone-exception-policy_v0.1",
"policy_id": "ops-warden/security-zone-exceptions",
"version": "1",
"controls": {
"flex-auth/pre-sign": {
"grant_authorities": ["ops-warden/security-owner"],
"maximum_duration_seconds": 7200,
}
},
}
def exception_record():
return {
"exception_id": "zone-exc-001",
"security_zone": "z2-protected",
"control": "flex-auth/pre-sign",
"workloads": ["issue-core"],
"base": {"stance": "enforced", "failure_mode": "fail_closed"},
"relaxation": {"stance": "advisory"},
"justification": "bounded migration",
"requested_by": "issue-core",
"granted_by": "ops-warden/security-owner",
"issued_at": "2026-08-23T09:00:00Z",
"not_before": "2026-08-23T10:00:00Z",
"not_after": "2026-08-23T12:00:00Z",
"maximum_duration_policy": "ops-warden/security-zone-exceptions@1",
"change_ref": "ops-warden@abc123",
"durable_authorities": [
{"id": "ssh-cert:123", "not_after": "2026-08-23T11:00:00Z"}
],
}
def evaluate(record, at="2026-08-23T10:00:00+00:00"):
return evaluate_exceptions(
{
"standard": "security-zone-exceptions_v0.1",
"exceptions": [record],
},
policy(),
at=datetime.fromisoformat(at),
)
class ExceptionConformanceTest(unittest.TestCase):
def test_not_before_is_inclusive(self):
result = evaluate(exception_record())
self.assertTrue(result["ok"])
self.assertEqual(result["active_exception_ids"], ["zone-exc-001"])
self.assertEqual(result["results"][0]["state"], "active")
def test_not_after_is_exclusive(self):
result = evaluate(exception_record(), "2026-08-23T12:00:00+00:00")
self.assertTrue(result["ok"])
self.assertFalse(result["results"][0]["active"])
self.assertEqual(result["results"][0]["state"], "expired")
def test_future_record_is_valid_but_inactive(self):
result = evaluate(exception_record(), "2026-08-23T09:30:00+00:00")
self.assertTrue(result["ok"])
self.assertEqual(result["results"][0]["state"], "future")
def test_wrong_grant_authority_is_invalid_and_inactive(self):
record = exception_record()
record["granted_by"] = "workload/self"
result = evaluate(record)
self.assertFalse(result["ok"])
self.assertFalse(result["results"][0]["active"])
self.assertIn("designated", result["results"][0]["errors"][0])
def test_duration_beyond_owner_maximum_is_invalid(self):
record = exception_record()
record["not_after"] = "2026-08-23T12:00:01Z"
result = evaluate(record)
self.assertFalse(result["ok"])
self.assertTrue(
any(
"exceeds owner maximum" in error
for error in result["results"][0]["errors"]
)
)
def test_wildcard_and_unknown_workloads_are_forbidden(self):
for workload in ("*", "unknown"):
with self.subTest(workload=workload):
record = exception_record()
record["workloads"] = [workload]
result = evaluate(record)
self.assertFalse(result["ok"])
self.assertIn("exact resolved", result["results"][0]["errors"][0])
def test_durable_authority_must_not_outlive_exception(self):
record = exception_record()
record["durable_authorities"][0]["not_after"] = "2026-08-23T12:00:01Z"
result = evaluate(record)
self.assertFalse(result["ok"])
self.assertIn("outlives", result["results"][0]["errors"][0])
def test_failure_mode_can_only_relax_closed_to_open(self):
record = exception_record()
record["relaxation"] = {"failure_mode": "fail_open"}
result = evaluate(record)
self.assertTrue(result["ok"])
record["base"]["failure_mode"] = "fail_open"
result = evaluate(record)
self.assertFalse(result["ok"])
self.assertIn("fail_closed", result["results"][0]["errors"][0])
def test_overlapping_grants_for_same_control_and_workload_are_rejected(self):
first = exception_record()
second = deepcopy(first)
second["exception_id"] = "zone-exc-002"
document = {
"standard": "security-zone-exceptions_v0.1",
"exceptions": [first, second],
}
result = evaluate_exceptions(
document,
policy(),
at=datetime(2026, 8, 23, 10, tzinfo=timezone.utc),
)
self.assertFalse(result["ok"])
self.assertTrue(all("overlaps" in item["errors"][0] for item in result["results"]))
def test_renewal_requires_a_new_existing_id(self):
record = exception_record()
record["renews"] = record["exception_id"]
result = evaluate(record)
self.assertFalse(result["ok"])
self.assertIn("different existing", result["results"][0]["errors"][0])
def test_timezone_is_required(self):
record = exception_record()
record["not_after"] = "2026-08-23T12:00:00"
result = evaluate(record)
self.assertFalse(result["ok"])
self.assertIn("timezone", result["results"][0]["errors"][0])
if __name__ == "__main__":
unittest.main()