tegwick
be29c28100
Assistant: codex Assistant-Model: gpt-5.6-sol Assistant-Session: 01a0291a-1e87-7151-9934-fcbfe3f65eb1
307 lines
12 KiB
Python
307 lines
12 KiB
Python
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()
|