feat: compile security zone declarations

Assistant: codex
Assistant-Model: gpt-5.6-sol
Assistant-Session: 01a0291a-1e87-7151-9934-fcbfe3f65eb1
This commit is contained in:
tegwick 2026-08-22 14:19:07 +02:00
parent a51039330c
commit 9b6ada7c89
5 changed files with 433 additions and 9 deletions

3
.gitignore vendored
View file

@ -3,3 +3,6 @@
.claude/*
!.claude/rules/
!.claude/rules/*.md
__pycache__/
*.pyc

6
Makefile Normal file
View file

@ -0,0 +1,6 @@
.PHONY: check test
check: test
test:
PYTHONDONTWRITEBYTECODE=1 python3 -m unittest discover -s tests -v

View file

@ -83,11 +83,12 @@ adds one non-monotone continuity band required by ops-warden `ADR-0006`.
| `z1-operational` | `M1` promotion evidence | Production secret handling for at most `internal` data | Early operational workloads with an owned front door |
| `z2-protected` | `M2` promotion evidence | `prod`; at most `confidential` data | Production workloads requiring review, SLO/on-call, and incident readiness |
| `z3-critical` | `M3` promotion evidence | `prod`; at most `restricted` data | Critical or regulated workloads with the strongest normal failure behavior |
| `z3-continuity` | `M3` plus dependency/recovery evidence | `prod`; at most `restricted` data | Foundational access or recovery workloads where fail-closed dependency cycles would cause an outage |
| `z2-continuity` | `M2` plus dependency/recovery evidence | `prod`; at most `confidential` data | Foundational access or recovery workloads where fail-closed dependency cycles would cause an outage |
`z3-continuity` is a sibling of `z3-critical`, not a lower rung. It exists
`z2-continuity` is a sibling of `z2-protected`, not a relaxation of its
admission floor. It exists
because enforcement stance is not monotone: a tunnel or credential-issuance
path can require M3 scrutiny while deliberately remaining fail-open for one
path can require production scrutiny while deliberately remaining fail-open for one
availability-sensitive control. Placement on a reef is never evidence for any
zone.
@ -146,7 +147,7 @@ owns the other rows and every PEP failure mode.
| `z1-operational` | `advisory` / `fail_open` | `enforced` / `fail_closed` | `advisory`; evaluator failure cannot produce `autonomous` |
| `z2-protected` | `enforced` / `fail_open` | `enforced` / `fail_closed` | `enforced`; minimum `founder_required` when the zone rule matches |
| `z3-critical` | `enforced` / `fail_closed` | `enforced` / `fail_closed` | `enforced`; minimum `founder_required` when the zone rule matches |
| `z3-continuity` | `enforced` / `fail_open` | `enforced` / `fail_closed` | `enforced`; minimum `founder_required` when the zone rule matches |
| `z2-continuity` | `enforced` / `fail_open` | `enforced` / `fail_closed` | `enforced`; minimum `founder_required` when the zone rule matches |
| `unknown` | `advisory` / `fail_open` under the versioned `build` profile | `enforced` / `fail_closed` | `enforced`; never `autonomous` from zone evidence |
The `unknown` pre-sign treatment is an explicit organization-build policy, not
@ -159,7 +160,7 @@ does not extend to disclosing high-risk credentials. For a missing lane risk:
- `z0-experimental` may resolve to `standard` only when admission proves the
lane can expose synthetic material exclusively;
- `z1-operational` and `z2-protected` resolve to at least `high`;
- `z3-critical` and `z3-continuity` resolve to `critical`, treated by the
- `z3-critical` resolves to `critical`, treated by the
boundary as at least `high`; and
- unresolved membership resolves to at least `high`.
@ -201,12 +202,16 @@ tenancy:
# tenancy-posture_v0.1 content omitted
zones:
standard: security-zones_v0.1
membership: z3-continuity
membership: z2-continuity
responsible_party: ops-bridge
justification: foundational tunnel path must retain availability under PDP loss
context:
maturity: M2
criticality: high
data_classification: confidential
evidence:
- ref: docs/evidence/ops-bridge-tunnel-zone.md
supports: [M3, continuity-dependency, recovery]
supports: [M2, continuity-dependency, recovery]
reviewed: "2026-08-22"
review_due: "2026-11-22"
```
@ -220,6 +225,10 @@ Required fields are:
- `membership` — one zone id from §4;
- `responsible_party` — the party answering for this membership;
- `justification` — why the zone fits the workload's actual context;
- `context` — the evidenced `maturity`, `criticality`, and
`data_classification` used for admission. A managed workload's latter two
values must agree with its resolved rapp projection; `n/a` requires an
evidence-backed reason;
- `evidence` — one or more references and the admission facts each supports;
- `reviewed` and `review_due` — ISO dates, with review due after review.
@ -240,9 +249,9 @@ workload_ref:
name: ops-bridge-tunnel
deployable: null # optional for a managed deployable
identity_binding: ssh-certificate/ops-warden/agt-ops-bridge
declared_zone: z3-continuity
declared_zone: z2-continuity
admission: satisfied
effective_zone: z3-continuity
effective_zone: z2-continuity
membership_revision: sha256:<digest>
guarantees:
- authoritative-workload-identity

128
tests/test_resolve_zones.py Normal file
View file

@ -0,0 +1,128 @@
from pathlib import Path
from tempfile import TemporaryDirectory
import unittest
import yaml
from tools.resolve_zones import 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",
},
}
class ResolveZonesTest(unittest.TestCase):
def resolve(self, document):
with TemporaryDirectory() as directory:
path = Path(directory) / "tenancy.yaml"
path.write_text(yaml.safe_dump(document, sort_keys=False))
return resolve_paths([path])
def test_satisfied_membership_selects_zone_control_profile(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.assertTrue(record["membership_revision"].startswith("sha256:"))
def test_below_floor_is_unsatisfied_and_uses_unknown_profile(self):
result = self.resolve(declaration(maturity="M1"))
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")
def test_zone_below_context_floor_is_unsatisfied_even_with_high_maturity(self):
document = declaration(zone="z1-operational", maturity="M2")
result = self.resolve(document)
self.assertTrue(result["ok"])
record = result["records"][0]
self.assertEqual(record["admission"], "unsatisfied")
self.assertEqual(record["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"]
)
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_membership_returns_unknown_without_inference(self):
document = declaration()
document.pop("zones")
document.pop("workload_identity")
result = self.resolve(document)
self.assertTrue(result["ok"])
self.assertEqual(result["records"][0]["effective_zone"], "unknown")
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"])
if __name__ == "__main__":
unittest.main()

278
tools/resolve_zones.py Normal file
View file

@ -0,0 +1,278 @@
#!/usr/bin/env python3
"""Resolve security-zone declarations without guessing workload identity."""
from __future__ import annotations
import argparse
import hashlib
import json
import sys
from datetime import date
from pathlib import Path
from typing import Any, Iterable
import yaml
MATURITY_RANK = {"M0": 0, "M1": 1, "M2": 2, "M3": 3}
CRITICALITY_FLOOR = {"low": 0, "medium": 1, "high": 2, "critical": 3}
DATACLASS_FLOOR = {
"synthetic": 0,
"internal": 1,
"confidential": 2,
"restricted": 3,
}
ZONE_FLOOR = {
"z0-experimental": 0,
"z1-operational": 1,
"z2-protected": 2,
"z2-continuity": 2,
"z3-critical": 3,
}
CONTROL_PROFILE = {
"z0-experimental": {
"flex-auth/pre-sign": ("advisory", "fail_open"),
"ops-warden/agent-high-risk-read": ("enforced", "fail_closed"),
"ops-warden/plan-zone-rule": ("advisory", "fail_closed"),
},
"z1-operational": {
"flex-auth/pre-sign": ("advisory", "fail_open"),
"ops-warden/agent-high-risk-read": ("enforced", "fail_closed"),
"ops-warden/plan-zone-rule": ("advisory", "fail_closed"),
},
"z2-protected": {
"flex-auth/pre-sign": ("enforced", "fail_open"),
"ops-warden/agent-high-risk-read": ("enforced", "fail_closed"),
"ops-warden/plan-zone-rule": ("enforced", "fail_closed"),
},
"z2-continuity": {
"flex-auth/pre-sign": ("enforced", "fail_open"),
"ops-warden/agent-high-risk-read": ("enforced", "fail_closed"),
"ops-warden/plan-zone-rule": ("enforced", "fail_closed"),
},
"z3-critical": {
"flex-auth/pre-sign": ("enforced", "fail_closed"),
"ops-warden/agent-high-risk-read": ("enforced", "fail_closed"),
"ops-warden/plan-zone-rule": ("enforced", "fail_closed"),
},
"unknown": {
"flex-auth/pre-sign": ("advisory", "fail_open"),
"ops-warden/agent-high-risk-read": ("enforced", "fail_closed"),
"ops-warden/plan-zone-rule": ("enforced", "fail_closed"),
},
}
class DeclarationError(ValueError):
"""A declaration violates the security-zones_v0.1 contract."""
def _required(mapping: dict[str, Any], key: str, where: str) -> Any:
value = mapping.get(key)
if value is None or value == "" or value == []:
raise DeclarationError(f"{where}.{key} is required")
return value
def _parse_date(value: Any, where: str) -> date:
if isinstance(value, date):
return value
try:
return date.fromisoformat(str(value))
except ValueError as exc:
raise DeclarationError(f"{where} must be an ISO date") from exc
def _services(document: dict[str, Any]) -> Iterable[dict[str, Any]]:
services = document.get("services")
if services is not None:
if not isinstance(services, list) or not services:
raise DeclarationError("services must be a non-empty list")
if "zones" in document or "workload_identity" in document:
raise DeclarationError(
"multi-service declarations keep zones and workload_identity per service"
)
yield from services
return
yield document
def _validate_identity(service: str, identity: Any) -> dict[str, Any]:
if not isinstance(identity, dict):
raise DeclarationError(f"{service}.workload_identity must be a mapping")
name = str(_required(identity, "name", f"{service}.workload_identity"))
if name != service:
raise DeclarationError(
f"{service}.workload_identity.name must equal service, got {name!r}"
)
_required(identity, "kind", f"{service}.workload_identity")
_required(identity, "responsible_repo", f"{service}.workload_identity")
bindings = _required(
identity, "identity_bindings", f"{service}.workload_identity"
)
if not isinstance(bindings, list):
raise DeclarationError(
f"{service}.workload_identity.identity_bindings must be a list"
)
for index, binding in enumerate(bindings):
where = f"{service}.workload_identity.identity_bindings[{index}]"
if not isinstance(binding, dict):
raise DeclarationError(f"{where} must be a mapping")
for key in ("scheme", "authority", "subject", "principal_type"):
_required(binding, key, where)
if binding["principal_type"] not in {"service", "agent"}:
raise DeclarationError(f"{where}.principal_type must be service or agent")
return identity
def _admission(service: str, zones: dict[str, Any]) -> tuple[str, str]:
membership = str(_required(zones, "membership", f"{service}.zones"))
if membership not in ZONE_FLOOR:
raise DeclarationError(f"{service}.zones.membership is unknown: {membership!r}")
context = _required(zones, "context", f"{service}.zones")
if not isinstance(context, dict):
raise DeclarationError(f"{service}.zones.context must be a mapping")
maturity = str(_required(context, "maturity", f"{service}.zones.context"))
criticality = str(
_required(context, "criticality", f"{service}.zones.context")
)
dataclass = str(
_required(context, "data_classification", f"{service}.zones.context")
)
if maturity not in MATURITY_RANK:
raise DeclarationError(f"{service}.zones.context.maturity is invalid")
if criticality not in CRITICALITY_FLOOR:
raise DeclarationError(f"{service}.zones.context.criticality is invalid")
if dataclass == "public":
return "unknown", "public_data_classification_floor_unresolved"
if dataclass == "n/a":
_required(
context,
"data_classification_reason",
f"{service}.zones.context",
)
data_floor = 0
elif dataclass in DATACLASS_FLOOR:
data_floor = DATACLASS_FLOOR[dataclass]
else:
raise DeclarationError(
f"{service}.zones.context.data_classification is invalid"
)
zone_rank = ZONE_FLOOR[membership]
context_rank = max(CRITICALITY_FLOOR[criticality], data_floor)
if zone_rank < context_rank:
return "unsatisfied", f"{membership}_below_M{context_rank}_context_floor"
if MATURITY_RANK[maturity] < zone_rank:
return "unsatisfied", f"{maturity}_below_M{zone_rank}_zone_floor"
if membership == "z2-continuity":
supported = {
str(fact)
for item in zones["evidence"]
if isinstance(item, dict)
for fact in item.get("supports", [])
}
required = {"continuity-dependency", "recovery"}
if not required.issubset(supported):
return "unsatisfied", "continuity_evidence_incomplete"
return "satisfied", "admission_floor_met"
def resolve_service(service_entry: dict[str, Any], source: str) -> dict[str, Any]:
service = str(_required(service_entry, "service", source))
zones = service_entry.get("zones")
if zones is None:
return {
"workload_id": service,
"declared_zone": None,
"admission": "unknown",
"admission_reason": "zone_membership_absent",
"effective_zone": "unknown",
"membership_revision": None,
"controls": _controls("unknown"),
"source": source,
}
identity = _validate_identity(service, service_entry.get("workload_identity"))
if not isinstance(zones, dict):
raise DeclarationError(f"{service}.zones must be a mapping")
if zones.get("standard") != "security-zones_v0.1":
raise DeclarationError(
f"{service}.zones.standard must be security-zones_v0.1"
)
for key in (
"responsible_party",
"justification",
"evidence",
"reviewed",
"review_due",
):
_required(zones, key, f"{service}.zones")
if not isinstance(zones["evidence"], list):
raise DeclarationError(f"{service}.zones.evidence must be a list")
reviewed = _parse_date(zones["reviewed"], f"{service}.zones.reviewed")
review_due = _parse_date(zones["review_due"], f"{service}.zones.review_due")
if review_due <= reviewed:
raise DeclarationError(f"{service}.zones.review_due must be after reviewed")
admission, reason = _admission(service, zones)
membership = str(zones["membership"])
effective = membership if admission == "satisfied" else "unknown"
revision_input = json.dumps(
{"workload_identity": identity, "zones": zones},
sort_keys=True,
separators=(",", ":"),
default=str,
).encode()
revision = "sha256:" + hashlib.sha256(revision_input).hexdigest()
return {
"workload_id": service,
"declared_zone": membership,
"admission": admission,
"admission_reason": reason,
"effective_zone": effective,
"membership_revision": revision,
"guarantees": [
"authoritative-workload-identity",
"explicit-zone-membership",
"non-inferred-resolution",
"enforcement-time-exception-expiry",
],
"controls": _controls(effective),
"source": source,
}
def _controls(zone: str) -> list[dict[str, str]]:
return [
{"id": control, "stance": stance, "failure_mode": failure}
for control, (stance, failure) in CONTROL_PROFILE[zone].items()
]
def resolve_paths(paths: Iterable[Path]) -> dict[str, Any]:
records: list[dict[str, Any]] = []
errors: list[dict[str, str]] = []
for path in paths:
try:
document = yaml.safe_load(path.read_text()) or {}
if not isinstance(document, dict):
raise DeclarationError("document must be a mapping")
for entry in _services(document):
if not isinstance(entry, dict):
raise DeclarationError("service entry must be a mapping")
records.append(resolve_service(entry, str(path)))
except (OSError, yaml.YAMLError, DeclarationError) as exc:
errors.append({"source": str(path), "error": str(exc)})
return {"ok": not errors, "standard": "security-zones_v0.1", "records": records, "errors": errors}
def main() -> int:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("paths", nargs="+", type=Path)
args = parser.parse_args()
result = resolve_paths(args.paths)
print(json.dumps(result, indent=2, sort_keys=True))
return 0 if result["ok"] else 1
if __name__ == "__main__":
sys.exit(main())