from copy import deepcopy from pathlib import Path from tempfile import TemporaryDirectory import unittest import yaml from tools.resolve_zones import ( PROFILE_ZONES, compare_snapshots, resolve_manifest, resolve_paths, ) def declaration(*, service="flex-auth", zone="z2-protected", maturity="M2"): return { "schema_version": "0.1", "framework": "netkingdom-tenancy-posture", "service": service, "role": "policy-decision-point", "workload_identity": { "name": service, "kind": "platform-service", "responsible_repo": service, "identity_bindings": [ { "scheme": "kubernetes-service-account", "authority": "railiance01", "subject": f"system:serviceaccount:{service}:{service}", "principal_type": "service", } ], }, "tenancy": {}, "zones": { "standard": "security-zones_v0.1", "membership": zone, "responsible_party": service, "justification": "fixture", "context": { "maturity": maturity, "criticality": "high", "data_classification": "internal", }, "evidence": [{"ref": "fixture", "supports": [maturity]}], "reviewed": "2026-08-22", "review_due": "2026-11-22", }, } 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, *, 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], source_revision=source_revision, workload_refs=workload_refs, profile=profile, previous=previous, ) 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") 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_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") self.assertEqual(record["controls"][0]["stance"], "advisory") def test_zone_below_context_floor_is_unsatisfied_even_with_high_maturity(self): result = self.resolve(declaration(zone="z1-operational", maturity="M2")) self.assertTrue(result["ok"]) 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.assertEqual(result["records"][0]["admission"], "unsatisfied") document["zones"]["evidence"][0]["supports"].extend( ["continuity-dependency", "recovery"] ) result = self.resolve(document) self.assertEqual(result["records"][0]["admission"], "satisfied") def test_public_floor_stays_unknown(self): document = declaration() document["zones"]["context"]["data_classification"] = "public" result = self.resolve(document) self.assertTrue(result["ok"]) self.assertEqual(result["records"][0]["admission"], "unknown") self.assertEqual( result["records"][0]["admission_reason"], "public_data_classification_floor_unresolved", ) 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.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() document["workload_identity"]["name"] = "guessed-from-path" result = self.resolve(document) self.assertFalse(result["ok"]) self.assertIn("must equal service", result["errors"][0]["error"]) def test_n_a_data_classification_requires_reason(self): document = declaration() document["zones"]["context"]["data_classification"] = "n/a" result = self.resolve(document) 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()