diff --git a/.gitignore b/.gitignore index e4e0199..9baf29a 100644 --- a/.gitignore +++ b/.gitignore @@ -3,3 +3,6 @@ .claude/* !.claude/rules/ !.claude/rules/*.md + +__pycache__/ +*.pyc diff --git a/Makefile b/Makefile new file mode 100644 index 0000000..e8ec0d4 --- /dev/null +++ b/Makefile @@ -0,0 +1,6 @@ +.PHONY: check test + +check: test + +test: + PYTHONDONTWRITEBYTECODE=1 python3 -m unittest discover -s tests -v diff --git a/docs/security-zones_v0.1.md b/docs/security-zones_v0.1.md index 9df455e..8bbc402 100644 --- a/docs/security-zones_v0.1.md +++ b/docs/security-zones_v0.1.md @@ -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: guarantees: - authoritative-workload-identity diff --git a/tests/test_resolve_zones.py b/tests/test_resolve_zones.py new file mode 100644 index 0000000..9a407ab --- /dev/null +++ b/tests/test_resolve_zones.py @@ -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() diff --git a/tools/resolve_zones.py b/tools/resolve_zones.py new file mode 100644 index 0000000..4274316 --- /dev/null +++ b/tools/resolve_zones.py @@ -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())