diff --git a/GOAL.md b/GOAL.md index 5e77825..a4e811f 100644 --- a/GOAL.md +++ b/GOAL.md @@ -1,7 +1,7 @@ --- repo: zone-engine repo_flavor: project -project_status: complete +project_status: active started: "2026-08-19" --- @@ -59,3 +59,7 @@ expires on its own. Archive when the standard is canon, the declarations are live, and either a runtime exists with an owner or the decision that none is needed is recorded. + +`ZONE-WP-0002` is the final reference-contract hardening and retirement pass. +It does not reopen the no-runtime decision; it makes the retained artifacts +truthful enough for an explicit archive-or-maintain decision. diff --git a/Makefile b/Makefile index e8ec0d4..a739323 100644 --- a/Makefile +++ b/Makefile @@ -1,6 +1,16 @@ -.PHONY: check test +.PHONY: canon-lineage check exception-fixtures reference-fixtures test -check: test +check: test reference-fixtures exception-fixtures test: PYTHONDONTWRITEBYTECODE=1 python3 -m unittest discover -s tests -v + +reference-fixtures: + PYTHONDONTWRITEBYTECODE=1 python3 tools/resolve_zones.py --manifest fixtures/manifests/reference.yaml --control-profile profiles/netkingdom-build-v0.1.yaml >/dev/null + +exception-fixtures: + PYTHONDONTWRITEBYTECODE=1 python3 tools/check_zone_exceptions.py fixtures/exceptions/valid-active.yaml --policy fixtures/exceptions/policy.yaml --at 2026-08-23T10:00:00Z >/dev/null + ! PYTHONDONTWRITEBYTECODE=1 python3 tools/check_zone_exceptions.py fixtures/exceptions/invalid-overlap.yaml --policy fixtures/exceptions/policy.yaml --at 2026-08-23T11:30:00Z >/dev/null + +canon-lineage: + PYTHONDONTWRITEBYTECODE=1 python3 tools/check_canon_lineage.py --manifest docs/canon-lineage.yaml --canon-root "$(CANON_ROOT)" diff --git a/docs/canon-lineage.yaml b/docs/canon-lineage.yaml new file mode 100644 index 0000000..bee856e --- /dev/null +++ b/docs/canon-lineage.yaml @@ -0,0 +1,9 @@ +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: 104d5c6 +canonical_status: proposed +canonical_sha256: 32e71e9c0d6946bb14099eb66193822da26f199de0e11d8488464207f3bd9906 +verified_at: "2026-08-23" +local_representation: pointer diff --git a/docs/security-zones_v0.1.md b/docs/security-zones_v0.1.md index eb2047c..4170758 100644 --- a/docs/security-zones_v0.1.md +++ b/docs/security-zones_v0.1.md @@ -1,341 +1,34 @@ --- -id: security-zones_v0.1 -type: standard -status: draft +id: security-zones_v0.1-source-pointer +type: standard-source-pointer +status: superseded owner: zone-engine publication_owner: net-kingdom -date: "2026-08-22" +canonical_path: net-kingdom/canon/standards/security-zones_v0.1.md +canonical_revision: 104d5c6 +canonical_status: proposed +canonical_sha256: 32e71e9c0d6946bb14099eb66193822da26f199de0e11d8488464207f3bd9906 +updated: "2026-08-23" --- -# NetKingdom Security Zones v0.1 +# Security Zones v0.1 — non-authoritative source pointer -## 1. Purpose +This file is **not the published standard**. The authoritative text is +`net-kingdom/canon/standards/security-zones_v0.1.md`, owned and published by +net-kingdom. Its reviewed lineage is pinned in `docs/canon-lineage.yaml` and can +be checked with: -A security zone is a named workload-admission standard. It answers which -scrutiny a workload has qualified for; control-owner policy then answers what a -particular control does in that zone. A zone is not a repository label, a -credential lane, a network segment, a reef, or a temporary exception. - -This standard is a sibling of `tenancy-posture_v0.1`. It owns zone identity, -membership, admission, resolution, and the time-boxed exception lifecycle. -`flex-auth` remains the only PDP for decisions it renders. Every other control -continues to be owned and evaluated at its existing enforcement point. - -## 2. Authority and conformance - -Authority is split deliberately: - -| Fact or rule | Authority | -| --- | --- | -| Workload identity and requested membership | Workload's responsible repo | -| Zone identities and admission standards | This standard, owned by zone-engine | -| Per-zone stance | Owner of the control that renders the effect | -| Failure behavior when a dependency is unavailable | Owner of the PEP | -| Publication | net-kingdom canon | - -Conformance is **accuracy, not altitude**. A lower zone declared with evidence -is conformant. A stricter zone that the workload cannot evidence is not. -Changing membership to evade a control is not an exception mechanism. - -The workload is the sole **zone** policy subject. It includes independently -governed application, automation, maintenance, and operational/control-plane -execution units. The workload executing a tunnel, credential broker, policy -service, or maintenance operation requires identity; the tunnel, grant, lane, -pattern, action, human, or agent does not become a workload merely because a -control observes it. Those retain native typed identity and caller/resource -context. - -## 3. Resolution is authoritative - -The stable `workload_identity.name` is the join key. Runtime principals, -resources, credential lanes, and controls reference it explicitly. A resolver -MUST NOT infer membership from repository ownership, paths, lane types, actor -classes, environment, criticality, reefs, organization posture, or naming -conventions. - -For a managed deployable, the authoritative cross-repository reference is the -Repo Manager v1 tuple `(rapp_id, workload_identity.name)`, optionally narrowed -by `deployable`. A catalog also declares whether its subject is workload -applicable. Resolution returns both the reference result and admission result: - -- `satisfied` — the workload has an authoritative identity and evidence meeting - the declared zone's floor; `effective_zone` is the declared zone; -- `unsatisfied` — identity and membership are declared, but evidence is below - the admission floor; `effective_zone` is `unknown`; -- `unknown` — identity, membership, or a required floor/evidence input cannot be - resolved for a workload-applicable subject; `effective_zone` is `unknown`; -- `not-applicable` — the owning catalog explicitly establishes that the subject - is an action, actor, lane, pattern, or resource rather than a workload; no - zone is resolved and the control uses that subject's native identity. - -`unknown` is a result, not a zone. A control MUST declare an explicit treatment -for it. That treatment may deny, escalate, or use a reviewed build-stage rule; -it never manufactures membership and never grants an exception. - -## 4. Zone catalog - -The initial catalog follows the estate's existing `M0`–`M3` maturity ladder and -adds one non-monotone continuity band required by ops-warden `ADR-0006`. - -| Zone id | Admission floor | Environment/data boundary | Purpose | -| --- | --- | --- | --- | -| `z0-experimental` | `M0` promotion evidence | Synthetic only; no real credential or user data | Experiments where advisory controls preserve iteration speed | -| `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 | -| `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 | - -`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 production scrutiny while deliberately remaining fail-open for one -availability-sensitive control. Placement on a reef is never evidence for any -zone. - -### 4.1 Context floor - -The declared zone must be at least as strict as the workload context requires. -The context floor is the maximum of every resolved input: - -| `criticality` | Minimum maturity | -| --- | --- | -| `low` | `M0` | -| `medium` | `M1` | -| `high` | `M2` | -| `critical` | `M3` | - -Data-class floors are consumed from the canon-owned classification mapping; the -current `synthetic → M0`, `internal → M1`, `confidential → M2`, and -`restricted → M3` mapping is already used by ops-warden. An absent or unresolved -floor makes admission `unknown`. In particular, `public` is disclosure policy -and `synthetic` is data origin: they are not aliases, and this standard does not -invent a floor while info-tech-canon's ruling is pending. - -`organization_posture` may select a versioned control profile, but it never -changes identity, membership, or admission. - -## 5. Stance and failure-mode model - -Every owner-qualified control publishes a total mapping over all zone ids plus -`unknown`. There is no implicit default. - -- `enforced` — the control's deny or escalation effect changes the outcome; -- `advisory` — the control evaluates fully and records the outcome it would - have produced, but does not block; -- `exempt` — the control is deliberately not evaluated for this zone and emits - the versioned exemption reason. - -Failure mode is separate and PEP-side: - -- `fail_closed` — an unavailable or invalid evaluator produces the control's - safe blocking/escalation outcome; -- `fail_open` — the PEP proceeds and records that the control was unavailable. - -`exempt` has no failure mode. A local control with no remote dependency uses -`fail_closed` to describe evaluator errors. Changing stance is a policy-package -change owned by the control owner, not a membership change. - -### 5.1 Initial build-stage control profile - -This profile is the v0.1 proposal for the first consumer. It is not stored in a -workload's `zones:` declaration. flex-auth owns the pre-sign rows; ops-warden -owns the other rows and every PEP failure mode. - -| Zone | flex-auth pre-sign stance / ops-warden PEP | agent high-risk read boundary | `warden plan` zone rule | -| --- | --- | --- | --- | -| `z0-experimental` | `advisory` / `fail_open` | `enforced` / `fail_closed` | `advisory`; evaluator failure cannot produce `autonomous` | -| `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 | -| `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 -a permissive membership default. It must change through a versioned control -profile when the organization posture graduates. - -The agent read boundary stays enforced in every zone: build-stage permissiveness -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` resolves to `critical`, treated by the - boundary as at least `high`; and -- unresolved membership resolves to at least `high`. - -An explicit grade always remains preferable. `accepted` is an acceptance record, -not a risk grade. - -## 6. Declaration in `tenancy.yaml` - -For a single-service declaration, `zones:` is a sibling of `tenancy:` and -`workload_identity`. For a `services:` declaration, both `workload_identity` and -`zones` occur inside the same service entry. A multi-service file MUST NOT use a -top-level `zones:` block. - -Every managed running deployable has an authoritative -`rapp-*/declarations/rapp.yaml`. Its `workload_identity.declaration_ref` points -to that declaration, and consuming catalogs reference it using the Repo Manager -v1 tuple. A pre-rapp deployable is migration debt and resolves `unknown`. An -independently governed operational execution unit that is not a managed -deployable may declare directly in its responsible repo's `tenancy.yaml`; this -does not turn a human action or infrastructure resource into a fictional rapp or -workload. - -```yaml -schema_version: "0.1" -framework: netkingdom-tenancy-posture -service: ops-bridge-tunnel -role: operational-access-path -workload_identity: - name: ops-bridge-tunnel - kind: operational-control-plane - responsible_repo: ops-bridge - identity_bindings: - - scheme: ssh-certificate - authority: ops-warden - subject: agt-ops-bridge - principal_type: agent - environment: prod -tenancy: - # tenancy-posture_v0.1 content omitted -zones: - standard: security-zones_v0.1 - 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: [M2, continuity-dependency, recovery] - reviewed: "2026-08-22" - review_due: "2026-11-22" +```bash +make canon-lineage CANON_ROOT=/path/to/net-kingdom ``` -The `zones:` block contains only membership evidence. It MUST NOT contain -control stance, failure mode, organization posture, or exceptions. +The original zone-engine draft remains available in Git history through +revision `a510393`. Local decisions and evidence remain in: -Required fields are: +- `docs/estate-partition-2026-08-19.md`; +- `docs/exception-lifecycle-2026-08-22.md`; and +- `docs/evidence/security-zone-adoption-2026-08-22.md`. -- `standard` — exactly `security-zones_v0.1`; -- `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. - -Permanent membership changes are reviewed source changes. A change to a lower -floor also records its reason and approver in the change review. Temporary -relaxation uses an exception and never changes `membership`. - -## 7. Compilation and resolved view - -Compilation produces a workload-addressable resolved record. At minimum it -contains: - -```yaml -workload_id: ops-bridge-tunnel -workload_ref: - applicability: applicable - rapp_id: null # required for a managed deployable - name: ops-bridge-tunnel - deployable: null # optional for a managed deployable -identity_binding: ssh-certificate/ops-warden/agt-ops-bridge -declared_zone: z2-continuity -admission: satisfied -effective_zone: z2-continuity -membership_revision: sha256: -guarantees: - - authoritative-workload-identity - - explicit-zone-membership - - non-inferred-resolution - - enforcement-time-exception-expiry -controls: - - id: flex-auth/pre-sign - policy_owner: flex-auth - stance: enforced - pep_owner: ops-warden - failure_mode: fail_open - policy_ref: -``` - -The `membership_revision` covers the authoritative workload binding, `zones:` -block, and source revision. Control results include their policy/profile version -and any active exception id and expiry. This is the machine-readable answer to -“which zone is this workload in, and what applies there?” It may be compiled -into existing consumer artifacts; it is not a synchronous zone-engine lookup. - -For managed deployables, compilation consumes the exact Repo Manager reference -projection: - -```yaml -workload_ref: - applicability: applicable - rapp_id: rapp-issue-core - name: issue-core - deployable: issue-core # optional -``` - -The owning catalog uses `applicability: not-applicable` for a native non-workload -subject. Absence of either applicability or an expected reference is `unknown`, -not `not-applicable`. Zone-engine consumes these outcomes; it does not parse a -path or repository name to repair them. - -For flex-auth's pre-sign control, the governed workload is the target of the -certificate or grant, so the compiler writes `workload_id`, `security_zone`, -`security_zone_admission`, and `security_zone_revision` on the resource -attributes. Caller identity remains in the subject. A control that governs the -requesting workload must declare that role explicitly and use separately named -caller-workload attributes. - -The dormant `trust_zone: platform` constant is not security-zone membership and -MUST be retired before adoption. The new concept is named `security_zone`; the -two fields must not coexist as competing zone sources. - -## 8. Membership-change observability - -A membership change becomes effective only through a reviewed declaration and -a newly compiled artifact. The compiler emits the source and membership -revision, rejects ambiguous identities, and reports additions, removals, and -changes against the preceding snapshot. Controls expose the membership revision -in their decision or verdict evidence. - -A zone that can be widened by editing an unversioned runtime label is not -conformant. - -## 9. Time-boxed exceptions - -The normative lifecycle is the T04 decision in -`docs/exception-lifecycle-2026-08-22.md`: only the control owner's designated -authority grants a named-workload, named-zone, named-control relaxation within a -declared maximum duration. Enforcement applies it only for -`not_before <= now < not_after`; invalid or unevaluable records are inactive, -expiry restores the base rule automatically, and no minted credential, lease, -or session may outlive the exception. - -Exceptions live with the versioned control policy or PEP configuration and are -evaluated where their effects occur. This requires no zone-engine runtime. - -## 10. Adoption - -Net-kingdom published this standard at revision `337484a`. Adoption requires: - -1. flex-auth and ops-warden accept the initial control profile or publish a - versioned replacement with total zone and `unknown` coverage; -2. at least two workload owners declare authoritative identities and zones; -3. a third consumer compiles or reads the resolved view; and -4. ops-warden retires `policy.enabled` and the dormant `trust_zone` constant in - the same migration. - -All four gates were met on 2026-08-22. The exact consumer revisions, tests, -resolved membership digests, and live caller decision are recorded in -`docs/evidence/security-zone-adoption-2026-08-22.md` in the owning zone-engine -repository. +Publication and adoption do not silently promote a canon artifact from +`proposed` to `accepted`. Consumers must read the lifecycle recorded by canon, +not infer one from this repository. diff --git a/fixtures/exceptions/invalid-overlap.yaml b/fixtures/exceptions/invalid-overlap.yaml new file mode 100644 index 0000000..680d666 --- /dev/null +++ b/fixtures/exceptions/invalid-overlap.yaml @@ -0,0 +1,30 @@ +standard: security-zone-exceptions_v0.1 +exceptions: + - exception_id: zone-exc-fixture-001 + security_zone: z2-protected + control: flex-auth/pre-sign + workloads: [issue-core] + base: {stance: enforced, failure_mode: fail_closed} + relaxation: {stance: advisory} + justification: first overlapping fixture + 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@fixture-1 + - exception_id: zone-exc-fixture-002 + security_zone: z2-protected + control: flex-auth/pre-sign + workloads: [issue-core] + base: {stance: enforced, failure_mode: fail_closed} + relaxation: {stance: advisory} + justification: second overlapping fixture + requested_by: issue-core + granted_by: ops-warden/security-owner + issued_at: "2026-08-23T09:30:00Z" + not_before: "2026-08-23T11:00:00Z" + not_after: "2026-08-23T12:30:00Z" + maximum_duration_policy: ops-warden/security-zone-exceptions@1 + change_ref: ops-warden@fixture-2 diff --git a/fixtures/exceptions/policy.yaml b/fixtures/exceptions/policy.yaml new file mode 100644 index 0000000..0d7282e --- /dev/null +++ b/fixtures/exceptions/policy.yaml @@ -0,0 +1,8 @@ +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 diff --git a/fixtures/exceptions/valid-active.yaml b/fixtures/exceptions/valid-active.yaml new file mode 100644 index 0000000..51ee8b6 --- /dev/null +++ b/fixtures/exceptions/valid-active.yaml @@ -0,0 +1,19 @@ +standard: security-zone-exceptions_v0.1 +exceptions: + - exception_id: zone-exc-fixture-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 fixture + 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@fixture + durable_authorities: + - id: ssh-cert:fixture + not_after: "2026-08-23T11:00:00Z" diff --git a/fixtures/manifests/adoption.yaml b/fixtures/manifests/adoption.yaml new file mode 100644 index 0000000..83e9b00 --- /dev/null +++ b/fixtures/manifests/adoption.yaml @@ -0,0 +1,12 @@ +standard: zone-resolver-input_v0.1 +sources: + - path: ../../../flex-auth/tenancy.yaml + source_revision: flex-auth@00a7bc46090ceb71dc23fef0530629f18398d3b1 + - path: ../../../ops-warden/tenancy.yaml + source_revision: ops-warden@c8fa02adf0ab9ef699508141f099bc71c38e76f3 +subjects: + - subject_id: human-operator + source: ops-warden/catalog@c8fa02adf0ab9ef699508141f099bc71c38e76f3 + source_revision: ops-warden@c8fa02adf0ab9ef699508141f099bc71c38e76f3 + workload_ref: + applicability: not-applicable diff --git a/fixtures/manifests/reference.yaml b/fixtures/manifests/reference.yaml new file mode 100644 index 0000000..fd6328a --- /dev/null +++ b/fixtures/manifests/reference.yaml @@ -0,0 +1,16 @@ +standard: zone-resolver-input_v0.1 +sources: + - path: tenancy.yaml + source_revision: fixture-owner@abc123 + workload_refs: + fixture-workload: + applicability: applicable + rapp_id: rapp-fixture + name: fixture-workload + deployable: fixture-workload +subjects: + - subject_id: human-operator + source: fixture-catalog@abc123 + source_revision: fixture-catalog@abc123 + workload_ref: + applicability: not-applicable diff --git a/fixtures/manifests/tenancy.yaml b/fixtures/manifests/tenancy.yaml new file mode 100644 index 0000000..400145e --- /dev/null +++ b/fixtures/manifests/tenancy.yaml @@ -0,0 +1,28 @@ +schema_version: "0.1" +framework: netkingdom-tenancy-posture +service: fixture-workload +role: fixture +workload_identity: + name: fixture-workload + kind: managed-deployable + responsible_repo: fixture-owner + identity_bindings: + - scheme: kubernetes-service-account + authority: railiance01 + subject: system:serviceaccount:fixture:fixture-workload + principal_type: service +tenancy: {} +zones: + standard: security-zones_v0.1 + membership: z2-protected + responsible_party: fixture-owner + justification: canonical resolver fixture + context: + maturity: M2 + criticality: high + data_classification: internal + evidence: + - ref: fixture + supports: [M2] + reviewed: "2026-08-23" + review_due: "2026-11-23" diff --git a/history/2026-08-23-canon-lineage-and-lifecycle.md b/history/2026-08-23-canon-lineage-and-lifecycle.md new file mode 100644 index 0000000..9c43a71 --- /dev/null +++ b/history/2026-08-23-canon-lineage-and-lifecycle.md @@ -0,0 +1,28 @@ +# Canon lineage and lifecycle ruling — 2026-08-23 + +## Evidence reviewed + +- net-kingdom `337484a` added the canonical security-zone standard with + `status: proposed`. +- net-kingdom `104d5c6` recorded completion of all four adoption gates while + deliberately retaining `status: proposed`. +- The sibling `tenancy-posture_v0.1` standard is also published as + `status: proposed`; publication is therefore not evidence of acceptance. +- The canonical security-zone content at `104d5c6` has SHA-256 + `32e71e9c0d6946bb14099eb66193822da26f199de0e11d8488464207f3bd9906`. + +## Ruling applied + +The canon owner's checked-in lifecycle is authoritative: v0.1 remains +`proposed` until net-kingdom explicitly changes it. zone-engine does not promote +the standard by inference and does not maintain a second normative copy. + +The former local draft at `docs/security-zones_v0.1.md` is replaced by a +non-authoritative pointer. `docs/canon-lineage.yaml` pins canonical owner, path, +revision, lifecycle, and content hash. `tools/check_canon_lineage.py` verifies +both the reviewed revision and the current checkout, so either content drift or +a lifecycle change requires a reviewed lineage update. + +This closes the duplicate-authority shape without changing canon. Any future +status promotion is a net-kingdom source change and a trigger to refresh the +lineage record here if this repository is retained. diff --git a/profiles/netkingdom-build-v0.1.yaml b/profiles/netkingdom-build-v0.1.yaml new file mode 100644 index 0000000..edddaed --- /dev/null +++ b/profiles/netkingdom-build-v0.1.yaml @@ -0,0 +1,37 @@ +standard: security-zone-control-profile_v0.1 +profile_id: netkingdom-build-v0.1 +version: flex-auth@e521e7b+ops-warden@7ce58ae +controls: + flex-auth/pre-sign: + policy_owner: flex-auth + pep_owner: ops-warden + policy_ref: flex-auth@e521e7b:examples/ops-warden/policy_package.md + zones: + z0-experimental: {stance: advisory, failure_mode: fail_open} + z1-operational: {stance: advisory, failure_mode: fail_open} + z2-protected: {stance: enforced, failure_mode: fail_open} + z2-continuity: {stance: enforced, failure_mode: fail_open} + z3-critical: {stance: enforced, failure_mode: fail_closed} + unknown: {stance: advisory, failure_mode: fail_open} + ops-warden/agent-high-risk-read: + policy_owner: ops-warden + pep_owner: ops-warden + policy_ref: ops-warden@7ce58ae:src/warden/routing/models.py + zones: + z0-experimental: {stance: enforced, failure_mode: fail_closed} + z1-operational: {stance: enforced, failure_mode: fail_closed} + z2-protected: {stance: enforced, failure_mode: fail_closed} + z2-continuity: {stance: enforced, failure_mode: fail_closed} + z3-critical: {stance: enforced, failure_mode: fail_closed} + unknown: {stance: enforced, failure_mode: fail_closed} + ops-warden/plan-zone-rule: + policy_owner: ops-warden + pep_owner: ops-warden + policy_ref: security-zones_v0.1@104d5c6:section-5.1 + zones: + z0-experimental: {stance: advisory, failure_mode: fail_closed} + z1-operational: {stance: advisory, failure_mode: fail_closed} + z2-protected: {stance: enforced, failure_mode: fail_closed} + z2-continuity: {stance: enforced, failure_mode: fail_closed} + z3-critical: {stance: enforced, failure_mode: fail_closed} + unknown: {stance: enforced, failure_mode: fail_closed} diff --git a/tests/test_canon_lineage.py b/tests/test_canon_lineage.py new file mode 100644 index 0000000..e885eb8 --- /dev/null +++ b/tests/test_canon_lineage.py @@ -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() diff --git a/tests/test_resolve_zones.py b/tests/test_resolve_zones.py index 9a407ab..4fae26a 100644 --- a/tests/test_resolve_zones.py +++ b/tests/test_resolve_zones.py @@ -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() diff --git a/tests/test_zone_exceptions.py b/tests/test_zone_exceptions.py new file mode 100644 index 0000000..a28ebc5 --- /dev/null +++ b/tests/test_zone_exceptions.py @@ -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() diff --git a/tools/check_canon_lineage.py b/tools/check_canon_lineage.py new file mode 100644 index 0000000..586ed25 --- /dev/null +++ b/tools/check_canon_lineage.py @@ -0,0 +1,125 @@ +#!/usr/bin/env python3 +"""Verify a local canon-lineage record against an authoritative checkout.""" + +from __future__ import annotations + +import argparse +import hashlib +import json +import subprocess +import sys +from pathlib import Path +from typing import Any, Mapping + +import yaml + + +LINEAGE_STANDARD = "canon-lineage_v0.1" + + +class LineageError(ValueError): + """The lineage manifest or canonical artifact is inconsistent.""" + + +def _required(mapping: Mapping[str, Any], key: str) -> Any: + value = mapping.get(key) + if value is None or value == "": + raise LineageError(f"lineage.{key} is required") + return value + + +def _sha256(content: bytes) -> str: + return hashlib.sha256(content).hexdigest() + + +def _frontmatter(content: bytes) -> dict[str, Any]: + text = content.decode() + if not text.startswith("---\n") or "\n---\n" not in text[4:]: + raise LineageError("canonical artifact requires YAML frontmatter") + raw = text.split("\n---\n", 1)[0][4:] + value = yaml.safe_load(raw) or {} + if not isinstance(value, dict): + raise LineageError("canonical frontmatter must be a mapping") + return value + + +def check_lineage( + manifest: Any, + canon_root: Path, + *, + verify_revision: bool = True, +) -> dict[str, Any]: + if not isinstance(manifest, Mapping): + raise LineageError("lineage manifest must be a mapping") + if manifest.get("standard") != LINEAGE_STANDARD: + raise LineageError(f"lineage.standard must be {LINEAGE_STANDARD}") + relative = Path(str(_required(manifest, "canonical_path"))) + if relative.is_absolute() or ".." in relative.parts: + raise LineageError("canonical_path must stay below canon_root") + revision = str(_required(manifest, "canonical_revision")) + expected_hash = str(_required(manifest, "canonical_sha256")) + expected_status = str(_required(manifest, "canonical_status")) + canonical_path = canon_root / relative + try: + content = canonical_path.read_bytes() + except OSError as exc: + raise LineageError(f"cannot read canonical artifact: {exc}") from exc + actual_hash = _sha256(content) + frontmatter = _frontmatter(content) + errors: list[str] = [] + if actual_hash != expected_hash: + errors.append( + f"canonical content hash changed: expected {expected_hash}, got {actual_hash}" + ) + if str(frontmatter.get("status")) != expected_status: + errors.append( + "canonical lifecycle changed: " + f"expected {expected_status}, got {frontmatter.get('status')}" + ) + revision_hash = None + if verify_revision: + completed = subprocess.run( + ["git", "-C", str(canon_root), "show", f"{revision}:{relative.as_posix()}"], + check=False, + capture_output=True, + ) + if completed.returncode != 0: + errors.append( + f"cannot read canonical artifact at revision {revision}: " + + completed.stderr.decode().strip() + ) + else: + revision_hash = _sha256(completed.stdout) + if revision_hash != expected_hash: + errors.append( + f"revision {revision} content does not match canonical_sha256" + ) + return { + "ok": not errors, + "artifact": manifest.get("artifact"), + "publication_owner": manifest.get("publication_owner"), + "canonical_path": relative.as_posix(), + "canonical_revision": revision, + "canonical_status": frontmatter.get("status"), + "canonical_sha256": actual_hash, + "revision_sha256": revision_hash, + "errors": errors, + } + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--manifest", required=True, type=Path) + parser.add_argument("--canon-root", required=True, type=Path) + args = parser.parse_args() + try: + manifest = yaml.safe_load(args.manifest.read_text()) or {} + result = check_lineage(manifest, args.canon_root) + except (OSError, yaml.YAMLError, LineageError) as exc: + result = {"ok": False, "errors": [str(exc)]} + print(json.dumps(result, indent=2, sort_keys=True)) + return 0 if result["ok"] else 1 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/tools/check_zone_exceptions.py b/tools/check_zone_exceptions.py new file mode 100644 index 0000000..471d9d4 --- /dev/null +++ b/tools/check_zone_exceptions.py @@ -0,0 +1,358 @@ +#!/usr/bin/env python3 +"""Validate security-zone exception records at an explicit instant.""" + +from __future__ import annotations + +import argparse +import json +import sys +from datetime import datetime, timezone +from pathlib import Path +from typing import Any, Mapping + +import yaml + +try: + from tools.resolve_zones import ZONE_FLOOR +except ModuleNotFoundError: # direct ``python tools/...`` execution + from resolve_zones import ZONE_FLOOR + + +POLICY_STANDARD = "security-zone-exception-policy_v0.1" +RECORD_STANDARD = "security-zone-exceptions_v0.1" + + +class ExceptionConformanceError(ValueError): + """The exception input or policy is structurally unusable.""" + + +def _required(mapping: Mapping[str, Any], key: str, where: str) -> Any: + value = mapping.get(key) + if value is None or value == "" or value == []: + raise ExceptionConformanceError(f"{where}.{key} is required") + return value + + +def _instant(value: Any, where: str) -> datetime: + text = str(value) + if text.endswith("Z"): + text = text[:-1] + "+00:00" + try: + parsed = datetime.fromisoformat(text) + except ValueError as exc: + raise ExceptionConformanceError(f"{where} must be an ISO timestamp") from exc + if parsed.tzinfo is None: + raise ExceptionConformanceError(f"{where} must include a timezone") + return parsed.astimezone(timezone.utc) + + +def _policy(policy: Any) -> dict[str, Any]: + if not isinstance(policy, Mapping): + raise ExceptionConformanceError("exception policy must be a mapping") + if policy.get("standard") != POLICY_STANDARD: + raise ExceptionConformanceError( + f"exception policy standard must be {POLICY_STANDARD}" + ) + policy_id = str(_required(policy, "policy_id", "policy")) + version = str(_required(policy, "version", "policy")) + controls = _required(policy, "controls", "policy") + if not isinstance(controls, Mapping) or not controls: + raise ExceptionConformanceError("policy.controls must be a non-empty mapping") + normalized: dict[str, Any] = {} + for control_id, control in controls.items(): + where = f"policy.controls.{control_id}" + if not isinstance(control, Mapping): + raise ExceptionConformanceError(f"{where} must be a mapping") + if "/" not in str(control_id): + raise ExceptionConformanceError(f"{where} id must be owner-qualified") + authorities = _required(control, "grant_authorities", where) + if not isinstance(authorities, list) or not all( + isinstance(authority, str) and authority for authority in authorities + ): + raise ExceptionConformanceError( + f"{where}.grant_authorities must be a non-empty string list" + ) + maximum = _required(control, "maximum_duration_seconds", where) + if not isinstance(maximum, int) or maximum <= 0: + raise ExceptionConformanceError( + f"{where}.maximum_duration_seconds must be a positive integer" + ) + normalized[str(control_id)] = { + "grant_authorities": set(authorities), + "maximum_duration_seconds": maximum, + } + return { + "policy_id": policy_id, + "version": version, + "policy_ref": f"{policy_id}@{version}", + "controls": normalized, + } + + +def _validate_relaxation(record: Mapping[str, Any], errors: list[str]) -> None: + base = record.get("base") + relaxation = record.get("relaxation") + if not isinstance(base, Mapping) or not isinstance(relaxation, Mapping): + errors.append("base and relaxation must be mappings") + return + changed = False + if "stance" in relaxation: + base_stance = base.get("stance") + relaxed_stance = relaxation.get("stance") + allowed = { + "enforced": {"advisory", "exempt"}, + "advisory": {"exempt"}, + } + if relaxed_stance not in allowed.get(base_stance, set()): + errors.append("relaxation.stance must strictly relax the base stance") + else: + changed = True + if "failure_mode" in relaxation: + if base.get("failure_mode") != "fail_closed" or relaxation.get( + "failure_mode" + ) != "fail_open": + errors.append( + "failure-mode relaxation must change fail_closed to fail_open" + ) + else: + changed = True + if not changed and not errors: + errors.append("relaxation must change stance or failure_mode") + + +def _record_result( + record: Any, + policy: Mapping[str, Any], + at: datetime, +) -> dict[str, Any]: + errors: list[str] = [] + if not isinstance(record, Mapping): + return { + "exception_id": None, + "valid": False, + "active": False, + "state": "invalid", + "errors": ["exception record must be a mapping"], + } + exception_id = record.get("exception_id") + for key in ( + "exception_id", + "security_zone", + "control", + "workloads", + "base", + "relaxation", + "justification", + "requested_by", + "granted_by", + "issued_at", + "not_before", + "not_after", + "maximum_duration_policy", + "change_ref", + ): + value = record.get(key) + if value is None or value == "" or value == () or value == []: + errors.append(f"{key} is required") + zone = record.get("security_zone") + if zone not in ZONE_FLOOR: + errors.append("security_zone must be a named zone") + control_id = record.get("control") + control_policy = policy["controls"].get(control_id) + if control_policy is None: + errors.append("control is absent from the owner exception policy") + workloads = record.get("workloads") + if not isinstance(workloads, list) or not workloads: + errors.append("workloads must be a non-empty list") + workloads = [] + elif any( + not isinstance(workload, str) + or not workload + or workload in {"*", "unknown"} + for workload in workloads + ): + errors.append("workloads must contain exact resolved workload ids") + elif len(set(workloads)) != len(workloads): + errors.append("workloads must not contain duplicates") + _validate_relaxation(record, errors) + + issued = before = after = None + for key in ("issued_at", "not_before", "not_after"): + try: + parsed = _instant(record.get(key), key) + if key == "issued_at": + issued = parsed + elif key == "not_before": + before = parsed + else: + after = parsed + except ExceptionConformanceError as exc: + errors.append(str(exc)) + if issued and before and after: + if issued > before: + errors.append("issued_at must be at or before not_before") + if before >= after: + errors.append("not_before must be before not_after") + if control_policy and (after - before).total_seconds() > control_policy[ + "maximum_duration_seconds" + ]: + errors.append("exception duration exceeds owner maximum") + if control_policy and record.get("granted_by") not in control_policy[ + "grant_authorities" + ]: + errors.append("granted_by is not a designated control authority") + if record.get("maximum_duration_policy") != policy["policy_ref"]: + errors.append("maximum_duration_policy does not match evaluated owner policy") + + durable = record.get("durable_authorities") or [] + if not isinstance(durable, list): + errors.append("durable_authorities must be a list") + else: + for index, authority in enumerate(durable): + if not isinstance(authority, Mapping) or not authority.get("id"): + errors.append(f"durable_authorities[{index}] requires id and not_after") + continue + try: + authority_after = _instant( + authority.get("not_after"), + f"durable_authorities[{index}].not_after", + ) + except ExceptionConformanceError as exc: + errors.append(str(exc)) + continue + if after and authority_after > after: + errors.append( + f"durable_authorities[{index}] outlives the exception" + ) + valid = not errors + active = bool(valid and before and after and before <= at < after) + if not valid: + state = "invalid" + elif at < before: + state = "future" + elif at >= after: + state = "expired" + else: + state = "active" + return { + "exception_id": exception_id, + "control": control_id, + "workloads": sorted(workloads), + "not_before": before.isoformat() if before else None, + "not_after": after.isoformat() if after else None, + "valid": valid, + "active": active, + "state": state, + "errors": errors, + } + + +def _overlap(left: Mapping[str, Any], right: Mapping[str, Any]) -> bool: + if left.get("control") != right.get("control"): + return False + if not set(left.get("workloads", [])).intersection(right.get("workloads", [])): + return False + if not all((left.get("not_before"), left.get("not_after"), right.get("not_before"), right.get("not_after"))): + return False + left_before = _instant(left["not_before"], "left.not_before") + left_after = _instant(left["not_after"], "left.not_after") + right_before = _instant(right["not_before"], "right.not_before") + right_after = _instant(right["not_after"], "right.not_after") + return max(left_before, right_before) < min(left_after, right_after) + + +def evaluate_exceptions( + document: Any, + policy_document: Any, + *, + at: datetime, +) -> dict[str, Any]: + policy = _policy(policy_document) + if not isinstance(document, Mapping) or document.get("standard") != RECORD_STANDARD: + raise ExceptionConformanceError( + f"exception document standard must be {RECORD_STANDARD}" + ) + records = document.get("exceptions") + if not isinstance(records, list): + raise ExceptionConformanceError("exceptions must be a list") + results = [_record_result(record, policy, at) for record in records] + ids: dict[str, list[int]] = {} + for index, result in enumerate(results): + if result["exception_id"]: + ids.setdefault(str(result["exception_id"]), []).append(index) + for exception_id, indexes in ids.items(): + if len(indexes) > 1: + for index in indexes: + results[index]["errors"].append( + f"duplicate exception_id {exception_id}" + ) + results[index].update(valid=False, active=False, state="invalid") + source_by_id = { + str(record.get("exception_id")): record + for record in records + if isinstance(record, Mapping) and record.get("exception_id") + } + for index, record in enumerate(records): + if not isinstance(record, Mapping) or not record.get("renews"): + continue + renewed = str(record["renews"]) + if renewed == str(record.get("exception_id")) or renewed not in source_by_id: + results[index]["errors"].append( + "renews must name a different existing exception id" + ) + results[index].update(valid=False, active=False, state="invalid") + for left in range(len(records)): + for right in range(left + 1, len(records)): + if not results[left]["valid"] or not results[right]["valid"]: + continue + try: + overlapping = _overlap(records[left], records[right]) + except ExceptionConformanceError: + overlapping = False + if overlapping: + for index in (left, right): + results[index]["errors"].append( + f"overlaps exception {results[right if index == left else left]['exception_id']}" + ) + results[index].update(valid=False, active=False, state="invalid") + return { + "ok": all(result["valid"] for result in results), + "standard": RECORD_STANDARD, + "evaluated_at": at.astimezone(timezone.utc).isoformat(), + "policy": {"id": policy["policy_id"], "version": policy["version"]}, + "results": sorted(results, key=lambda result: str(result["exception_id"])), + "active_exception_ids": sorted( + str(result["exception_id"]) + for result in results + if result["active"] + ), + } + + +def _load(path: Path) -> dict[str, Any]: + value = yaml.safe_load(path.read_text()) or {} + if not isinstance(value, dict): + raise ExceptionConformanceError(f"{path} must contain a mapping") + return value + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("exceptions", type=Path) + parser.add_argument("--policy", required=True, type=Path) + parser.add_argument("--at", required=True) + args = parser.parse_args() + try: + result = evaluate_exceptions( + _load(args.exceptions), + _load(args.policy), + at=_instant(args.at, "--at"), + ) + except (OSError, yaml.YAMLError, ExceptionConformanceError) as exc: + result = {"ok": False, "errors": [str(exc)], "results": []} + print(json.dumps(result, indent=2, sort_keys=True)) + return 0 if result["ok"] else 1 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/tools/resolve_zones.py b/tools/resolve_zones.py index 4274316..cf8f387 100644 --- a/tools/resolve_zones.py +++ b/tools/resolve_zones.py @@ -9,11 +9,14 @@ import json import sys from datetime import date from pathlib import Path -from typing import Any, Iterable +from typing import Any, Iterable, Mapping import yaml +STANDARD = "security-zones_v0.1" +INPUT_STANDARD = "zone-resolver-input_v0.1" +PROFILE_STANDARD = "security-zone-control-profile_v0.1" MATURITY_RANK = {"M0": 0, "M1": 1, "M2": 2, "M3": 3} CRITICALITY_FLOOR = {"low": 0, "medium": 1, "high": 2, "critical": 3} DATACLASS_FLOOR = { @@ -29,52 +32,49 @@ ZONE_FLOOR = { "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"), - }, -} +PROFILE_ZONES = frozenset({*ZONE_FLOOR, "unknown"}) class DeclarationError(ValueError): """A declaration violates the security-zones_v0.1 contract.""" -def _required(mapping: dict[str, Any], key: str, where: str) -> Any: +class ProfileError(ValueError): + """A control profile lacks authoritative, total provenance.""" + + +def _required(mapping: Mapping[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 _canonical(value: Any) -> Any: + """Return a stable, mapping- and list-order-independent JSON value.""" + + if isinstance(value, Mapping): + return {str(key): _canonical(value[key]) for key in sorted(value)} + if isinstance(value, list): + items = [_canonical(item) for item in value] + return sorted( + items, + key=lambda item: json.dumps( + item, sort_keys=True, separators=(",", ":"), default=str + ), + ) + if isinstance(value, date): + return value.isoformat() + return value + + +def _digest(value: Any) -> str: + encoded = json.dumps( + _canonical(value), sort_keys=True, separators=(",", ":"), default=str + ).encode() + return "sha256:" + hashlib.sha256(encoded).hexdigest() + + def _parse_date(value: Any, where: str) -> date: if isinstance(value, date): return value @@ -84,7 +84,7 @@ def _parse_date(value: Any, where: str) -> date: raise DeclarationError(f"{where} must be an ISO date") from exc -def _services(document: dict[str, Any]) -> Iterable[dict[str, Any]]: +def _services(document: Mapping[str, Any]) -> Iterable[dict[str, Any]]: services = document.get("services") if services is not None: if not isinstance(services, list) or not services: @@ -95,7 +95,7 @@ def _services(document: dict[str, Any]) -> Iterable[dict[str, Any]]: ) yield from services return - yield document + yield dict(document) def _validate_identity(service: str, identity: Any) -> dict[str, Any]: @@ -111,10 +111,11 @@ def _validate_identity(service: str, identity: Any) -> dict[str, Any]: bindings = _required( identity, "identity_bindings", f"{service}.workload_identity" ) - if not isinstance(bindings, list): + if not isinstance(bindings, list) or not bindings: raise DeclarationError( - f"{service}.workload_identity.identity_bindings must be a list" + f"{service}.workload_identity.identity_bindings must be a non-empty list" ) + seen: set[tuple[str, str, str, str]] = set() for index, binding in enumerate(bindings): where = f"{service}.workload_identity.identity_bindings[{index}]" if not isinstance(binding, dict): @@ -123,15 +124,97 @@ def _validate_identity(service: str, identity: Any) -> dict[str, Any]: _required(binding, key, where) if binding["principal_type"] not in {"service", "agent"}: raise DeclarationError(f"{where}.principal_type must be service or agent") + identity_key = tuple( + str(binding[key]) + for key in ("scheme", "authority", "subject", "principal_type") + ) + if identity_key in seen: + raise DeclarationError(f"{where} duplicates an identity binding") + seen.add(identity_key) return identity -def _admission(service: str, zones: dict[str, Any]) -> tuple[str, str]: +def _binding_refs(identity: Mapping[str, Any] | None) -> list[str]: + if not identity: + return [] + return sorted( + "/".join( + str(binding[key]) for key in ("scheme", "authority", "subject") + ) + for binding in identity["identity_bindings"] + ) + + +def _validate_workload_ref( + service: str, + workload_ref: Any, + identity: Mapping[str, Any] | None, +) -> dict[str, Any]: + if workload_ref is None: + if identity is None: + return { + "applicability": "applicable", + "rapp_id": None, + "name": None, + "deployable": None, + } + return { + "applicability": "applicable", + "rapp_id": None, + "name": str(identity["name"]), + "deployable": None, + } + if not isinstance(workload_ref, Mapping): + raise DeclarationError(f"{service}.workload_ref must be a mapping") + applicability = str( + _required(workload_ref, "applicability", f"{service}.workload_ref") + ) + if applicability not in {"applicable", "not-applicable"}: + raise DeclarationError( + f"{service}.workload_ref.applicability must be applicable or not-applicable" + ) + rapp_id = workload_ref.get("rapp_id") or None + name = workload_ref.get("name") or None + deployable = workload_ref.get("deployable") or None + if applicability == "not-applicable": + if any(value is not None for value in (rapp_id, name, deployable)): + raise DeclarationError( + f"{service}.workload_ref not-applicable must not carry a workload tuple" + ) + if identity is not None: + raise DeclarationError( + f"{service} cannot be both an authoritative workload and not-applicable" + ) + return { + "applicability": applicability, + "rapp_id": None, + "name": None, + "deployable": None, + } + if deployable is not None and rapp_id is None: + raise DeclarationError( + f"{service}.workload_ref.deployable requires rapp_id" + ) + if name is not None: + name = str(name) + if identity is not None and name is not None and name != identity["name"]: + raise DeclarationError( + f"{service}.workload_ref.name must equal workload_identity.name" + ) + return { + "applicability": applicability, + "rapp_id": str(rapp_id) if rapp_id is not None else None, + "name": name, + "deployable": str(deployable) if deployable is not None else None, + } + + +def _admission(service: str, zones: Mapping[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): + if not isinstance(context, Mapping): raise DeclarationError(f"{service}.zones.context must be a mapping") maturity = str(_required(context, "maturity", f"{service}.zones.context")) criticality = str( @@ -147,11 +230,7 @@ def _admission(service: str, zones: dict[str, Any]) -> tuple[str, str]: if dataclass == "public": return "unknown", "public_data_classification_floor_unresolved" if dataclass == "n/a": - _required( - context, - "data_classification_reason", - f"{service}.zones.context", - ) + _required(context, "data_classification_reason", f"{service}.zones.context") data_floor = 0 elif dataclass in DATACLASS_FLOOR: data_floor = DATACLASS_FLOOR[dataclass] @@ -169,7 +248,7 @@ def _admission(service: str, zones: dict[str, Any]) -> tuple[str, str]: supported = { str(fact) for item in zones["evidence"] - if isinstance(item, dict) + if isinstance(item, Mapping) for fact in item.get("supports", []) } required = {"continuity-dependency", "recovery"} @@ -178,27 +257,91 @@ def _admission(service: str, zones: dict[str, Any]) -> tuple[str, str]: return "satisfied", "admission_floor_met" -def resolve_service(service_entry: dict[str, Any], source: str) -> dict[str, Any]: +def _base_record( + service: str, + source: str, + source_revision: str | None, + workload_ref: Mapping[str, Any], + identity: Mapping[str, Any] | None, +) -> dict[str, Any]: + return { + "subject_id": service, + "workload_id": workload_ref.get("name"), + "workload_ref": dict(workload_ref), + "identity_bindings": _binding_refs(identity), + "declared_zone": None, + "admission": "unknown", + "admission_reason": "zone_membership_absent", + "effective_zone": "unknown", + "membership_revision": None, + "membership_revision_reason": "source_revision_absent" + if source_revision is None + else "zone_membership_absent", + "guarantees": ["non-inferred-resolution"], + "source": source, + "source_revision": source_revision, + } + + +def resolve_service( + service_entry: dict[str, Any], + source: str, + *, + source_revision: str | None = None, + workload_ref: Mapping[str, Any] | None = None, +) -> dict[str, Any]: service = str(_required(service_entry, "service", source)) + identity_value = service_entry.get("workload_identity") + identity = ( + _validate_identity(service, identity_value) + if identity_value is not None + else None + ) + reference = _validate_workload_ref(service, workload_ref, identity) + record = _base_record(service, source, source_revision, reference, identity) + + if reference["applicability"] == "not-applicable": + record.update( + { + "workload_id": None, + "admission": "not-applicable", + "admission_reason": "catalog_declared_not_applicable", + "effective_zone": None, + "membership_revision_reason": "not_applicable", + "guarantees": [ + "catalog-declared-not-applicable", + "non-inferred-resolution", + ], + } + ) + return record + + if reference["name"] is None: + record["admission_reason"] = "workload_reference_unresolved" + record["membership_revision_reason"] = "workload_reference_unresolved" + return record + + record["workload_id"] = reference["name"] + if identity is None: + if service_entry.get("zones") is not None: + raise DeclarationError( + f"{service}.workload_identity is required beside zones" + ) + record["admission_reason"] = "workload_identity_unresolved" + record["membership_revision_reason"] = "workload_identity_unresolved" + record["guarantees"].append("explicit-workload-reference") + return record + + record["guarantees"].extend( + ["authoritative-workload-identity", "explicit-workload-reference"] + ) 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")) + return record 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" - ) + if zones.get("standard") != STANDARD: + raise DeclarationError(f"{service}.zones.standard must be {STANDARD}") for key in ( "responsible_party", "justification", @@ -207,8 +350,8 @@ def resolve_service(service_entry: dict[str, Any], source: str) -> dict[str, Any "review_due", ): _required(zones, key, f"{service}.zones") - if not isinstance(zones["evidence"], list): - raise DeclarationError(f"{service}.zones.evidence must be a list") + if not isinstance(zones["evidence"], list) or not zones["evidence"]: + raise DeclarationError(f"{service}.zones.evidence must be a non-empty 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: @@ -216,60 +359,405 @@ def resolve_service(service_entry: dict[str, Any], source: str) -> dict[str, Any 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() + revision = None + revision_reason = "source_revision_absent" + if source_revision: + revision = _digest( + { + "source_revision": source_revision, + "workload_ref": reference, + "workload_identity": identity, + "zones": zones, + } + ) + revision_reason = "source_bound" + record.update( + { + "declared_zone": membership, + "admission": admission, + "admission_reason": reason, + "effective_zone": effective, + "membership_revision": revision, + "membership_revision_reason": revision_reason, + "guarantees": sorted( + set( + record["guarantees"] + + ["explicit-zone-membership"] + + (["source-revision-bound-membership"] if revision else []) + ) + ), + } + ) + return record + + +def validate_control_profile(profile: Any) -> dict[str, Any]: + if not isinstance(profile, Mapping): + raise ProfileError("control profile must be a mapping") + if profile.get("standard") != PROFILE_STANDARD: + raise ProfileError(f"control profile standard must be {PROFILE_STANDARD}") + profile_id = profile.get("profile_id") + version = profile.get("version") + if not profile_id or not version: + raise ProfileError("control profile requires profile_id and version") + controls = profile.get("controls") + if not isinstance(controls, Mapping) or not controls: + raise ProfileError("control profile controls must be a non-empty mapping") + normalized: dict[str, Any] = { + "standard": PROFILE_STANDARD, + "profile_id": str(profile_id), + "version": str(version), + "controls": {}, + } + for control_id in sorted(controls): + definition = controls[control_id] + where = f"controls.{control_id}" + if not isinstance(definition, Mapping): + raise ProfileError(f"{where} must be a mapping") + policy_owner = definition.get("policy_owner") + pep_owner = definition.get("pep_owner") + policy_ref = definition.get("policy_ref") + if not policy_owner or not pep_owner or not policy_ref: + raise ProfileError( + f"{where} requires policy_owner, pep_owner, and policy_ref" + ) + if "/" not in str(control_id) or not str(control_id).startswith( + f"{policy_owner}/" + ): + raise ProfileError( + f"{where} id must be owner-qualified by policy_owner" + ) + mappings = definition.get("zones") + if not isinstance(mappings, Mapping): + raise ProfileError(f"{where}.zones must be a mapping") + supplied = set(mappings) + if supplied != PROFILE_ZONES: + missing = sorted(PROFILE_ZONES - supplied) + extra = sorted(supplied - PROFILE_ZONES) + raise ProfileError( + f"{where}.zones must be total; missing={missing}, extra={extra}" + ) + normalized_zones: dict[str, dict[str, Any]] = {} + for zone in sorted(PROFILE_ZONES): + rule = mappings[zone] + if not isinstance(rule, Mapping): + raise ProfileError(f"{where}.zones.{zone} must be a mapping") + stance = rule.get("stance") + failure_mode = rule.get("failure_mode") + if stance not in {"enforced", "advisory", "exempt"}: + raise ProfileError(f"{where}.zones.{zone}.stance is invalid") + if stance == "exempt": + if failure_mode not in {None, ""}: + raise ProfileError( + f"{where}.zones.{zone} exempt must not have failure_mode" + ) + failure_mode = None + elif failure_mode not in {"fail_open", "fail_closed"}: + raise ProfileError( + f"{where}.zones.{zone}.failure_mode is invalid" + ) + normalized_zones[zone] = { + "stance": stance, + "failure_mode": failure_mode, + } + normalized["controls"][str(control_id)] = { + "policy_owner": str(policy_owner), + "pep_owner": str(pep_owner), + "policy_ref": str(policy_ref), + "zones": normalized_zones, + } + return normalized + + +def project_controls(record: dict[str, Any], profile: Mapping[str, Any]) -> None: + zone = record.get("effective_zone") + if zone not in PROFILE_ZONES: + return + record["control_profile"] = { + "id": profile["profile_id"], + "version": profile["version"], + } + record["controls"] = [] + for control_id, definition in profile["controls"].items(): + rule = definition["zones"][zone] + record["controls"].append( + { + "id": control_id, + "policy_owner": definition["policy_owner"], + "pep_owner": definition["pep_owner"], + "policy_ref": definition["policy_ref"], + "stance": rule["stance"], + "failure_mode": rule["failure_mode"], + } + ) + + +def compare_snapshots( + records: Iterable[Mapping[str, Any]], + previous: Mapping[str, Any] | None, +) -> dict[str, Any]: + current_by_id = {str(record["subject_id"]): record for record in records} + previous_records = previous.get("records", []) if previous else [] + previous_by_id = { + str(record["subject_id"]): record + for record in previous_records + if isinstance(record, Mapping) and record.get("subject_id") + } + current_ids = set(current_by_id) + previous_ids = set(previous_by_id) + changed: list[dict[str, Any]] = [] + for subject_id in sorted(current_ids & previous_ids): + current = current_by_id[subject_id] + prior = previous_by_id[subject_id] + fields = ( + "workload_ref", + "identity_bindings", + "declared_zone", + "admission", + "effective_zone", + "membership_revision", + ) + if any(_canonical(current.get(key)) != _canonical(prior.get(key)) for key in fields): + changed.append( + { + "subject_id": subject_id, + "before_revision": prior.get("membership_revision"), + "after_revision": current.get("membership_revision"), + } + ) 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, + "baseline": "previous" if previous is not None else "initial", + "added": sorted(current_ids - previous_ids), + "removed": sorted(previous_ids - current_ids), + "changed": changed, } -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]: +def _resolve_documents( + sources: Iterable[tuple[Path, str | None, Mapping[str, Any]]], + *, + profile: Mapping[str, Any] | None = None, + previous: Mapping[str, Any] | None = None, +) -> dict[str, Any]: records: list[dict[str, Any]] = [] errors: list[dict[str, str]] = [] - for path in paths: + profile_errors: list[str] = [] + validated_profile: Mapping[str, Any] | None = None + if profile is not None: + try: + validated_profile = validate_control_profile(profile) + except ProfileError as exc: + profile_errors.append(str(exc)) + for path, source_revision, workload_refs in sources: try: document = yaml.safe_load(path.read_text()) or {} - if not isinstance(document, dict): + if not isinstance(document, Mapping): 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))) + service = str(_required(entry, "service", str(path))) + record = resolve_service( + entry, + str(path), + source_revision=source_revision, + workload_ref=workload_refs.get(service), + ) + if validated_profile is not None: + project_controls(record, validated_profile) + records.append(record) 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} + records.sort(key=lambda record: str(record["subject_id"])) + duplicate_ids = sorted( + subject_id + for subject_id in {record["subject_id"] for record in records} + if sum(record["subject_id"] == subject_id for record in records) > 1 + ) + if duplicate_ids: + errors.append( + { + "source": "resolved-records", + "error": f"duplicate subject ids: {duplicate_ids}", + } + ) + return { + "ok": not errors and not profile_errors, + "standard": STANDARD, + "records": records, + "changes": compare_snapshots(records, previous), + "errors": errors, + "profile_errors": profile_errors, + } + + +def resolve_paths( + paths: Iterable[Path], + *, + source_revision: str | None = None, + source_revisions: Mapping[str, str] | None = None, + workload_refs: Mapping[str, Any] | None = None, + profile: Mapping[str, Any] | None = None, + previous: Mapping[str, Any] | None = None, +) -> dict[str, Any]: + revisions = source_revisions or {} + refs = workload_refs or {} + sources = [ + ( + path, + revisions.get(str(path), source_revision), + refs, + ) + for path in paths + ] + return _resolve_documents(sources, profile=profile, previous=previous) + + +def resolve_manifest( + manifest: Mapping[str, Any], + *, + base_dir: Path, + profile: Mapping[str, Any] | None = None, + previous: Mapping[str, Any] | None = None, +) -> dict[str, Any]: + if manifest.get("standard") != INPUT_STANDARD: + raise DeclarationError(f"manifest.standard must be {INPUT_STANDARD}") + entries = manifest.get("sources") + if not isinstance(entries, list) or not entries: + raise DeclarationError("manifest.sources must be a non-empty list") + sources: list[tuple[Path, str | None, Mapping[str, Any]]] = [] + for index, entry in enumerate(entries): + if not isinstance(entry, Mapping): + raise DeclarationError(f"manifest.sources[{index}] must be a mapping") + path_value = _required(entry, "path", f"manifest.sources[{index}]") + path = Path(str(path_value)) + if not path.is_absolute(): + path = base_dir / path + source_revision = entry.get("source_revision") + refs = entry.get("workload_refs") or {} + if not isinstance(refs, Mapping): + raise DeclarationError( + f"manifest.sources[{index}].workload_refs must be a mapping" + ) + sources.append( + ( + path, + str(source_revision) if source_revision else None, + refs, + ) + ) + result = _resolve_documents(sources, profile=profile, previous=previous) + subjects = manifest.get("subjects") or [] + if not isinstance(subjects, list): + raise DeclarationError("manifest.subjects must be a list") + for index, subject in enumerate(subjects): + if not isinstance(subject, Mapping): + raise DeclarationError(f"manifest.subjects[{index}] must be a mapping") + subject_id = str( + _required(subject, "subject_id", f"manifest.subjects[{index}]") + ) + reference = _validate_workload_ref( + subject_id, subject.get("workload_ref"), None + ) + if reference["applicability"] != "not-applicable": + raise DeclarationError( + f"manifest.subjects[{index}] is only for explicit not-applicable subjects" + ) + record = _base_record( + subject_id, + str(subject.get("source") or "manifest.subjects"), + str(subject["source_revision"]) if subject.get("source_revision") else None, + reference, + None, + ) + record.update( + { + "workload_id": None, + "admission": "not-applicable", + "admission_reason": "catalog_declared_not_applicable", + "effective_zone": None, + "membership_revision_reason": "not_applicable", + "guarantees": [ + "catalog-declared-not-applicable", + "non-inferred-resolution", + ], + } + ) + result["records"].append(record) + result["records"].sort(key=lambda record: str(record["subject_id"])) + subject_ids = [str(record["subject_id"]) for record in result["records"]] + duplicates = sorted( + subject_id + for subject_id in set(subject_ids) + if subject_ids.count(subject_id) > 1 + ) + if duplicates: + result["errors"].append( + { + "source": "manifest", + "error": f"duplicate subject ids: {duplicates}", + } + ) + result["ok"] = False + result["changes"] = compare_snapshots(result["records"], previous) + return result + + +def _load_mapping(path: Path, where: str) -> dict[str, Any]: + try: + value = yaml.safe_load(path.read_text()) or {} + except (OSError, yaml.YAMLError) as exc: + raise DeclarationError(f"could not read {where}: {exc}") from exc + if not isinstance(value, dict): + raise DeclarationError(f"{where} must be a mapping") + return value def main() -> int: parser = argparse.ArgumentParser(description=__doc__) - parser.add_argument("paths", nargs="+", type=Path) + parser.add_argument("paths", nargs="*", type=Path) + parser.add_argument("--manifest", type=Path) + parser.add_argument("--source-revision") + parser.add_argument("--control-profile", type=Path) + parser.add_argument("--previous", type=Path) args = parser.parse_args() - result = resolve_paths(args.paths) + if bool(args.manifest) == bool(args.paths): + parser.error("provide either declaration paths or --manifest") + try: + profile = ( + _load_mapping(args.control_profile, "control profile") + if args.control_profile + else None + ) + previous = ( + _load_mapping(args.previous, "previous snapshot") + if args.previous + else None + ) + if args.manifest: + manifest = _load_mapping(args.manifest, "manifest") + result = resolve_manifest( + manifest, + base_dir=args.manifest.parent, + profile=profile, + previous=previous, + ) + else: + result = resolve_paths( + args.paths, + source_revision=args.source_revision, + profile=profile, + previous=previous, + ) + except DeclarationError as exc: + result = { + "ok": False, + "standard": STANDARD, + "records": [], + "changes": {"baseline": "initial", "added": [], "removed": [], "changed": []}, + "errors": [{"source": "input", "error": str(exc)}], + "profile_errors": [], + } print(json.dumps(result, indent=2, sort_keys=True)) return 0 if result["ok"] else 1 diff --git a/workplans/ZONE-WP-0002-reference-contract-hardening-and-retirement.md b/workplans/ZONE-WP-0002-reference-contract-hardening-and-retirement.md index fc8a5a6..8a9c414 100644 --- a/workplans/ZONE-WP-0002-reference-contract-hardening-and-retirement.md +++ b/workplans/ZONE-WP-0002-reference-contract-hardening-and-retirement.md @@ -4,7 +4,7 @@ type: workplan title: "Harden the reference contract and make retirement decidable" domain: infotech repo: zone-engine -status: ready +status: active owner: zone-engine assignee: codex lane: yellow @@ -82,7 +82,7 @@ must arrive before T03 and T07 can be marked done. ```task id: ZONE-WP-0002-T01 -status: todo +status: progress priority: high quality_dor: DoR-Ok quality_dor_at: "2026-08-23"