diff --git a/SCOPE.md b/SCOPE.md index 9cd589a..fd4df4b 100644 --- a/SCOPE.md +++ b/SCOPE.md @@ -193,6 +193,7 @@ secrets-engine route secrets-engine revoke [--dry-run] secrets-engine lifecycle suspend|deactivate|destroy secrets-engine audit [--json] +secrets-engine evidence heartbeat|drain|classify ``` The implemented exec adapters are `exec-env` and `npm-config`. `read-check` is @@ -233,16 +234,17 @@ verification, `approle-login` is auth-capability handoff metadata, and - Management or health verification of ESO, Kubernetes Secrets, deployments, provider accounts, SSH issuance, tunnels, or remote transport. - Any backend other than the local `bao`/`vault` CLI speaking to OpenBao. -- An SSH-CA signing engine API. ops-warden still signs through its declared - OpenBao gap; this repository has accepted that surface as proposed only. -- A secret-use evidence engine API for kings-guard. `route` and `audit` are - operator summaries over local JSONL, not an observation surface. -- Emission to `audit-core`. Evidence today is local JSONL plus best-effort - State Hub progress notes, classified attributive, completeness not claimed. -- Named stance-application records (stage, failure mode, decision id present - only where rendered). Fail-closed production currently surfaces as a - `DecisionError` on the privileged-evidence path. +- An SSH-CA signing engine API. The contract is + `docs/ssh-ca-signing-contract.md`; ops-warden still signs through its + declared OpenBao gap. +- A secret-use evidence engine API for kings-guard. The contract is + `docs/secret-use-evidence-contract.md`. `route` and `audit` are operator + summaries over local JSONL, not that observation surface. +- Emission to `audit-core`. Load-bearing records are queued locally; drain + requires a sender binding that does not exist yet. Completeness is not + claimed. - Security-zone membership as a request claim. PEP scope is catalog stage. +- Drain of the load-bearing outbox into a live `audit-core` sender binding. ## System Boundary @@ -299,12 +301,12 @@ Working companion: `net-kingdom/SECURITY-COMPANION.md`. | Obligation | Current | Intended | | --- | --- | --- | | Layer declaration | `INTENT.md` frontmatter + `layer.yaml` | Keep in this repository's own voice | -| One decision point | Consumer validator for ActionAuthorization; production live fail-closed | Consume an access-engine decision record before every protected side effect | -| PEP stance | `pep-stance.yaml`; prod fail-closed, build/test fail-open relative to access-engine | Published map equals shipped behaviour; stance application recorded by name | -| Evidence bound | Attributive local JSONL + best-effort State Hub | Load-bearing vs attributive classified; load-bearing to `audit-core` with cadence | -| SSH-CA surface | Proposed; not shipped | Engine API for ops-warden's Staff PEP; lane stewardship stays with ops-warden | -| Secret-use evidence | `route` / `audit` over local JSONL | Engine surface of lease/revocation/mount/rotation metadata | -| Agent credential | Bootstrap token file still accepted; KeyCape scaffold unwired | Per-task, time-bounded service identity; no standing engine credential | +| One decision point | Consumer validator for ActionAuthorization; production live fail-closed from `pep-stance.yaml` | Consume an access-engine decision record before every protected side effect | +| PEP stance | Runtime loads `pep-stance.yaml`; named stance fields on privileged evidence | Unchanged map; T02 replaces fail-open residue with a decision record | +| Evidence bound | `evidence-classification.yaml`; load-bearing local outbox; heartbeat command | Drain to `audit-core` once that sender is admitted | +| SSH-CA surface | Contract at `docs/ssh-ca-signing-contract.md`; not shipped | Engine API after ops-warden assent | +| Secret-use evidence | Contract at `docs/secret-use-evidence-contract.md`; not shipped | Engine surface after kings-guard assent | +| Agent credential | Bootstrap token file still accepted; KeyCape scaffold unwired to OpenBao | Per-task, time-bounded service identity; no standing engine credential | ## Where Current Work Lives @@ -312,7 +314,8 @@ Working companion: `net-kingdom/SECURITY-COMPANION.md`. - `docs/hardening-backlog.md` tracks exit from bootstrap mode. - `history/` contains dated capability and intent assessments. - `INTENT.md` remains the stable aspirational direction. -- `layer.yaml` and `pep-stance.yaml` are the layer-model declaration surface. +- `layer.yaml`, `pep-stance.yaml`, and `evidence-classification.yaml` are the + layer-model declaration surface. ## Provided Capabilities diff --git a/docs/secret-use-evidence-contract.md b/docs/secret-use-evidence-contract.md new file mode 100644 index 0000000..63dafc5 --- /dev/null +++ b/docs/secret-use-evidence-contract.md @@ -0,0 +1,78 @@ +# Secret-use evidence engine contract (proposed) + +Status: **proposed**. `layer.yaml` owner_status remains `proposed` until this +surface ships. kings-guard must assent or contest before it is treated as an +observation input. + +Standard: NetKingdom Security Layer Model v0.7 §9.6. Companion: +`net-kingdom/SECURITY-COMPANION.md` §6. + +## What this is + +A read-only Lifecycle engine surface of **non-secret** lease, revocation, +mount, rotation, and delivery-session metadata, so kings-guard can evaluate +secret-abuse posture without a vault client. + +`secrets-engine route` and `secrets-engine audit` are operator summaries over +local JSONL. They are not this surface. + +## Bound (normative) + +The archive proves the records it holds were not altered or truncated after +arrival. It does **not** prove a record was sent. Absence of a record is **not** +evidence of non-occurrence. Completeness is not claimed. + +This surface cannot close adversarial omission at a compromised source. Cadence +for load-bearing classes is a heartbeat (`secrets-engine evidence heartbeat`), +not rate monitoring. + +## Fields (non-secret) + +| Field | Meaning | Freshness | +| --- | --- | --- | +| `as_of` | UTC timestamp of the snapshot | Snapshot time | +| `catalog_id` | Lane id | Catalog load | +| `stage` | `build` \| `test` \| `prod` | Catalog load | +| `kind` | `kv` \| `auth-capability` | Catalog load | +| `mount` | OpenBao mount name | Catalog load | +| `path` | KV path or allowlisted auth path, never a value | Catalog load | +| `field_names` | Declared field names, never values | Catalog load | +| `ready` | All declared fields present (boolean) | Last route/verify | +| `session_handle` | Accessor fingerprint only | Last exec/verify | +| `revocation_attempted` | Boolean | Last session | +| `revocation_succeeded` | Boolean | Last session | +| `lifecycle_operation` | `suspend` \| `deactivate` \| `destroy` \| `revoke` if one ran | Last lifecycle | +| `decision_id` | Present only where a decision was rendered | Last privileged action | +| `stance_stage` / `stance_failure_mode` | Named PEP residue | Last privileged action | +| `evidence_kind` | `load-bearing` \| `attributive` \| `heartbeat` | Per record | + +Forbidden in this surface: secret values, tokens, passwords, AppRole +`secret_id`, raw accessors, certificate private keys, subprocess argv, and +exception prose. + +## Freshness + +Every row carries `as_of`. There is no cached authorization verdict. A consumer +must not treat a stale snapshot as a standing allow. Recommended maximum age +for posture evaluation is the heartbeat interval declared in +`evidence-classification.yaml` (`1d`) until a tighter contract is assented. + +## Destination + +Intended owner of custody is `audit-core`. Today the engine queues load-bearing +records in `evidence_dir/outbox/` and writes attributive records to local +JSONL. Drain to `audit-core` is best-effort and is never on the mutation path. + +kings-guard reads this engine surface, not OpenBao. + +## Non-responsibilities + +This surface does not decide whether an action is permitted, does not expose an +approval-validity query, does not prove omission at source, and does not +observe production events kings-guard has not received. Companion §10 still +holds: nothing is observed in production until kings-guard reports otherwise. + +## Assent + +A coordination message to kings-guard records this contract. Assent or contest +belongs in kings-guard's own voice. diff --git a/docs/ssh-ca-signing-contract.md b/docs/ssh-ca-signing-contract.md new file mode 100644 index 0000000..ddc410a --- /dev/null +++ b/docs/ssh-ca-signing-contract.md @@ -0,0 +1,90 @@ +# SSH-CA signing engine contract (proposed) + +Status: **proposed**. `layer.yaml` owner_status remains `proposed` until this +surface ships. ops-warden must assent or contest before any replacement client +is written. + +Standard: NetKingdom Security Layer Model v0.7 §3.3 (Lifecycle), §6.4 (PEP). +Companion: `net-kingdom/SECURITY-COMPANION.md`. + +## What this is + +A deterministic Lifecycle engine API that asks OpenBao to sign a short-lived +SSH user certificate **after** an `access-engine` decision record naming the +request. secrets-engine presents no additional authority of its own beyond the +decision-bound role. The action reconstructs as the caller's. + +## What this is not + +- Operational ownership of the SSH access lane. `warden sign`, `cert_command`, + and host-access stewardship stay with ops-warden (Staff PEP). +- A break-glass AppRole. The cataloged standalone `warden-sign` AppRole stays + parked. +- An authorization decision, an approval store, or a second PDP. +- A writer of private keys. The caller supplies the public key; the engine + returns only the signed certificate material through a scoped delivery path. + +## Caller + +ops-warden, after it holds an `access-engine` decision for the sign request, or +after it has recorded the application of its own published unreachable-engine +stance in place of a decision (`pep-stance.yaml` in ops-warden). + +## Request + +| Field | Meaning | +| --- | --- | +| `actor_type` | `adm` \| `agt` \| `atm` | +| `actor_name` | Attributable principal the certificate is issued for | +| `identity` | Certificate key id / identity string | +| `principals` | Exact principal list; no wildcards | +| `public_key` | Caller-supplied public key (not a private key) | +| `ttl` | Requested lifetime; engine enforces the actor-type maximum | +| `decision_id` | Access-engine decision record id, present only where a decision was rendered | +| `request_digest` | Canonical digest of subject, action, resource, context | +| `stance` | Named unreachable-engine residue if no decision was rendered | + +Replay is permitted iff the canonical request digest matches and the decision +lifetime holds. A different resource is a different request. + +## Response (non-secret) + +| Field | Meaning | +| --- | --- | +| `signed` | Whether OpenBao signed | +| `valid_before` | Certificate expiry | +| `role` | OpenBao SSH role used (`agt-role` / `adm-role` / `atm-role`) | +| `decision_id` | Echoed when present | +| `evidence_id` | Local non-secret evidence record id | + +Certificate private material never appears in this response, in evidence, or in +argv. Delivery of the signed certificate uses a scoped file or exec path with +mode 0600, outside Git, same rules as existing handoff. + +## Lifetime + +Every allow has an explicit end: the certificate TTL, bounded by actor-type +maxima already documented on the `warden-sign` lane. No standing grant. + +## Evidence + +Attributive unless a future control branches on the presence of a signing +record — today none does. Completeness is not claimed. Records may include +actor type, role, `valid_before`, decision id, and stance fields. They must +never include a public-key blob if it could identify a live private key +handle, a token, or certificate private material. + +Destination today: local JSONL. Intended destination: `audit-core`. + +## Non-responsibilities + +secrets-engine does not choose principals, mint SSH keys, open tunnels, log +into hosts, or override ops-warden's zone stance map. If `access-engine` is +unreachable, this engine applies **its own** published catalog-stage stance +(`pep-stance.yaml` in this repository) and does not consult ops-warden's map. + +## Assent + +A coordination message to ops-warden records this contract. Assent or contest +belongs in ops-warden's own voice. Until then no OpenBao SSH-CA write is added +to the secrets-engine CLI. diff --git a/evidence-classification.yaml b/evidence-classification.yaml new file mode 100644 index 0000000..22ed468 --- /dev/null +++ b/evidence-classification.yaml @@ -0,0 +1,62 @@ +# secrets-engine — load-bearing vs attributive evidence (§9.6) +# +# Machine-readable classification next to the layer declaration. +# Completeness is not claimed. No control may branch on presence or +# absence of a local evidence record. + +schema_version: "0.1" +framework: netkingdom-security-layer-model +standard_version: "0.7" +repository: secrets-engine +bound: "§9.6" +completeness_claimed: false +no_control_branches_on_presence: true +intended_destination: audit-core +destination_today: "local JSONL; load-bearing also queued in evidence_dir/outbox/" +sound_claim: >- + The archive proves the records it holds were not altered or truncated + after arrival. It does not prove a record was sent, and absence is not + evidence of non-occurrence. + +cadence: + load_bearing_form: heartbeat + interval: 1d + action: evidence-heartbeat + claim: nothing-to-report + command: "secrets-engine evidence heartbeat" + +# First match wins. +rules: + - id: heartbeat + kind: heartbeat + actions: [evidence-heartbeat] + stages: [build, test, prod, unknown] + emission: local-outbox + note: >- + Positive claim that can itself go missing. Queue depth is metadata, + not a permission. + + - id: destroy + kind: load-bearing + actions: [lifecycle-destroy] + stages: [build, test, prod, unknown] + emission: local-outbox + note: Live destroy is currently disabled; refusals are still classified here. + + - id: production-control-mutation + kind: load-bearing + actions: [revoke, lifecycle-suspend, lifecycle-deactivate, provision] + stages: [prod] + emission: local-outbox + note: >- + Production provision, revoke, and suspend/deactivate once enabled. + Production live remains fail-closed until an access-engine decision + record is served. + + - id: default-attributive + kind: attributive + actions: ["*"] + stages: [build, test, prod, unknown] + emission: local-jsonl + trade: >- + State Hub progress posts are non-blocking. Completeness is not claimed. diff --git a/layer.yaml b/layer.yaml index a60d205..673d122 100644 --- a/layer.yaml +++ b/layer.yaml @@ -68,10 +68,13 @@ proposed_capabilities: (ops-warden) does not hold a direct VaultCA write. intended_owner: secrets-engine owner_status: proposed + contract: docs/ssh-ca-signing-contract.md blocked_on: >- - No engine API exists yet. ops-warden continues to sign through its - declared §5.3 gap. This repository accepts the engine surface and - declines operational ownership of the SSH access lane. + Contract published; ops-warden has not assented. No OpenBao SSH-CA + write is added to this CLI until that assent. ops-warden continues + to sign through its declared §5.3 gap. This repository accepts the + engine surface and declines operational ownership of the SSH access + lane. review: "2026-11-28" consequence: >- ops-warden's ssh-ca-signing-write gap remains the production path. @@ -83,9 +86,11 @@ proposed_capabilities: exposed as an engine surface for kings-guard secret-abuse posture. intended_owner: secrets-engine owner_status: proposed + contract: docs/secret-use-evidence-contract.md blocked_on: >- - Local JSONL plus best-effort State Hub notes exist; they are not an - engine API and they are not audit-core. Completeness is not claimed. + Contract published; kings-guard has not assented. Local JSONL, the + load-bearing outbox, and `route`/`audit` are not this observation + API. Completeness is not claimed. review: "2026-11-28" consequence: >- kings-guard secret-abuse posture stays fixture-driven. @@ -130,15 +135,29 @@ non_tooling_clients: preflight only and is deliberately not a write into OpenBao. Wiring it as OpenBao JWT login remains blocked on the platform-owned role. + - id: audit-core-outbox-drain + target: audit-core + layer: engine + modules: + - src/secrets_engine/evidence.py + operation: "Optional POST /v1/events from secrets-engine evidence drain" + write: true + note: >- + Engine-to-engine emit. Never on the mutation path. Empty URL skips + delivery and leaves the local outbox in place. Sender binding in + audit-core is not yet admitted. + # §9.6 — what this repository may claim about evidence it writes today. evidence_bound: - kind: attributive + classification: evidence-classification.yaml proves: - local JSONL records that were written were not rewritten by this engine + - load-bearing records queued in evidence_dir/outbox/ were written does_not_prove: - that a record was ever sent to audit-core - absence of a record as evidence of non-occurrence - completeness of the trail - destination_today: "local JSONL plus best-effort State Hub progress notes" + destination_today: "local JSONL; load-bearing also queued locally; attributive State Hub notes" intended_destination: audit-core completeness_claimed: false + no_control_branches_on_presence: true diff --git a/pep-stance.yaml b/pep-stance.yaml index 88790c6..e14888c 100644 --- a/pep-stance.yaml +++ b/pep-stance.yaml @@ -11,10 +11,9 @@ # Catalog stage is the equivalent scope until security-zone membership arrives # as a claim on the decision. # -# The property that makes this worth reading: tests/test_layer_conformance.py -# asserts this map against the shipped production fail-closed gate in -# secrets_engine.cli._require_lane_approval. A published map free to drift -# from the code is worse than none. +# Runtime reads this file. tests/test_layer_conformance.py also pins +# secrets_engine.pep_stance.SHIPPED_STANCE to this map. A published map free +# to drift from that pin is worse than none. schema_version: "0.1" framework: netkingdom-security-layer-model @@ -58,20 +57,17 @@ demo_exception: on_apply: recorded_fields: - - stage - - action + - stance_stage + - stance_failure_mode + - stance_demo_exception + - stance_decision_id # present only where a decision was resolved - approval_status # rejected when the stance refuses - error_type # DecisionError when fail_closed fires - - decision_id # present only where a decision was resolved written_to: - "local JSONL (.evidence/evidence-YYYY-MM-DD.jsonl)" - - "best-effort State Hub progress note" + - "evidence_dir/outbox/ for load-bearing classes" + - "best-effort State Hub progress note (attributive only)" never_recorded: "any secret material, token, password, or raw value" - gap: >- - Stance application is currently recorded as a DecisionError on the - privileged-evidence path, not as named stance fields (stage, failure - mode, decision id present only where rendered). Closing that gap is - SECRETS-WP-0008-T01. # §6.4 obligation 2 — the verdict is never cached. A retry of the same # canonical request digest inside the decision lifetime is the decision diff --git a/scripts/check_layer_conformance.py b/scripts/check_layer_conformance.py index b612faf..55722b6 100644 --- a/scripts/check_layer_conformance.py +++ b/scripts/check_layer_conformance.py @@ -145,6 +145,11 @@ def main() -> int: ok = False print(f"FINDING: pep_shaped but {decl['pep_stance']} is missing (§6.4 / §13.1)") + classification = decl.get("evidence_bound", {}).get("classification") + if classification and not (ROOT / str(classification)).is_file(): + ok = False + print(f"FINDING: evidence classification {classification} is missing (§9.6)") + found = scan_bao_modules() owned = owned_modules(decl) undeclared = {name: lines for name, lines in found.items() if name not in owned} diff --git a/src/secrets_engine/cli.py b/src/secrets_engine/cli.py index 4f25520..83672b6 100644 --- a/src/secrets_engine/cli.py +++ b/src/secrets_engine/cli.py @@ -14,6 +14,7 @@ Command surface (FR7): revoke lifecycle suspend|deactivate|destroy audit [--json] + evidence heartbeat|drain|classify Every privileged action is decision-gated and writes non-secret evidence. `plan` and `apply --dry-run` never mutate OpenBao. @@ -21,10 +22,8 @@ Every privileged action is decision-gated and writes non-secret evidence. from __future__ import annotations import argparse -import os import sys from pathlib import Path -from urllib.parse import urlparse from secrets_engine import __version__ from secrets_engine.apply import apply_plan @@ -33,6 +32,7 @@ from secrets_engine.config import Config, repo_root from secrets_engine.decisions import require_approved, resolve_decision from secrets_engine.errors import DecisionError, SecretsEngineError from secrets_engine.evidence import EvidenceWriter, PrivilegedActionEvidence +from secrets_engine.pep_stance import apply_unreachable_engine_stance, with_decision from secrets_engine.openbao import OpenBaoClient from secrets_engine.plan import build_plan from secrets_engine.provision import provision_from_file, provision_generated @@ -85,28 +85,23 @@ def _privileged_evidence( ) -def _unsafe_local_demo_enabled(cfg: Config) -> bool: - """Return true only for an explicit, offline, loopback-only demo.""" - host = (urlparse(cfg.bao_addr).hostname or "").lower() - return ( - os.environ.get("SECRETS_ENGINE_UNSAFE_DEMO") == "1" - and not cfg.hub_url - and host in {"127.0.0.1", "localhost", "::1"} - ) +def _require_lane_approval( + cfg: Config, + entry, + action: str = "", + evidence: PrivilegedActionEvidence | None = None, +): + """Apply published PEP stance, then resolve lane approval. - -def _require_lane_approval(cfg: Config, entry, action: str = ""): - """Resolve approval for a live action, failing production closed. - - The durable State Hub action-authorization endpoint is not available yet. - Production therefore cannot rely on a coarse lane decision. The one narrow - exception is an explicit offline demo against a loopback OpenBao instance. + Production ``fail_closed`` is read from ``pep-stance.yaml``. The durable + access-engine decision record is not served yet, so that row refuses live + production work. The three-factor unsafe-demo exception is not a stance + row. Build/test ``fail_open`` still requires the existing lane-approval + check — a tracked gap until SECRETS-WP-0008-T02. """ - if entry.stage == "prod" and not _unsafe_local_demo_enabled(cfg): - raise DecisionError( - f"production action '{action or 'unknown'}' requires a durable " - "State Hub action authorization; live production remains disabled" - ) + stance = apply_unreachable_engine_stance(cfg, entry, action or "unknown") + if evidence is not None: + evidence.mark_stance(stance) if not entry.approval_required(): return None decision = resolve_decision( @@ -115,6 +110,8 @@ def _require_lane_approval(cfg: Config, entry, action: str = ""): decision_ref=entry.approval.get("decision_ref", entry.id), ) require_approved(entry, decision) + if evidence is not None: + evidence.mark_stance(with_decision(stance, decision)) return decision @@ -229,7 +226,7 @@ def cmd_apply(cfg: Config, args) -> int: return 0 with _privileged_evidence(cfg, entry, "apply") as evidence: - decision = _require_lane_approval(cfg, entry, "apply") + decision = _require_lane_approval(cfg, entry, "apply", evidence) evidence.mark_approved(decision) plan = build_plan( entry, args.stage, decision_id=decision.id if decision else "" @@ -258,7 +255,7 @@ def cmd_provision(cfg: Config, args) -> int: raise ProvisioningError( f"lane '{entry.id}' is stage '{entry.stage}', not '{args.stage}'" ) - decision = _require_lane_approval(cfg, entry, "provision") + decision = _require_lane_approval(cfg, entry, "provision", evidence) evidence.mark_approved(decision) client = OpenBaoClient.resolve( cfg.bao_addr, bootstrap_token_file=args.bootstrap_token_file @@ -291,7 +288,7 @@ def cmd_verify(cfg: Config, args) -> int: "negative_requested": negative, }, ) as evidence: - decision = _require_lane_approval(cfg, entry, "verify") + decision = _require_lane_approval(cfg, entry, "verify", evidence) evidence.mark_approved(decision) client = OpenBaoClient.resolve( cfg.bao_addr, bootstrap_token_file=args.bootstrap_token_file @@ -367,7 +364,7 @@ def cmd_handoff(cfg: Config, args) -> int: raise ProvisioningError( f"lane '{entry.id}' is {entry.kind}; handoff needs auth-capability" ) - decision = _require_lane_approval(cfg, entry, "handoff") + decision = _require_lane_approval(cfg, entry, "handoff", evidence) evidence.mark_approved(decision) client = OpenBaoClient.resolve( cfg.bao_addr, bootstrap_token_file=args.bootstrap_token_file @@ -419,7 +416,7 @@ def cmd_exec(cfg: Config, args) -> int: }, ) as evidence: # require approval + readiness before running. - decision = _require_lane_approval(cfg, entry, "exec") + decision = _require_lane_approval(cfg, entry, "exec", evidence) evidence.mark_approved(decision) if not args.command: from secrets_engine.errors import DeliveryError @@ -498,7 +495,7 @@ def cmd_revoke(cfg: Config, args) -> int: with _privileged_evidence( cfg, entry, "revoke", detail={"operation": plan.operation} ) as evidence: - decision = _require_lane_approval(cfg, entry, "deactivate") + decision = _require_lane_approval(cfg, entry, "deactivate", evidence) evidence.mark_approved(decision) client = OpenBaoClient.resolve( cfg.bao_addr, bootstrap_token_file=args.bootstrap_token_file @@ -546,7 +543,7 @@ def cmd_lifecycle(cfg: Config, args) -> int: "live destroy is disabled until an exact-action destruction " "approval contract is available; use --dry-run to inspect targets" ) - decision = _require_lane_approval(cfg, entry, args.operation) + decision = _require_lane_approval(cfg, entry, args.operation, evidence) evidence.mark_approved(decision) client = OpenBaoClient.resolve( cfg.bao_addr, bootstrap_token_file=args.bootstrap_token_file @@ -579,6 +576,65 @@ def cmd_audit(cfg: Config, args) -> int: return 0 +def cmd_evidence_heartbeat(cfg: Config, args) -> int: + """Write a positive load-bearing heartbeat. Not a permission check.""" + import json + + from secrets_engine.evidence import write_heartbeat + + record = write_heartbeat(_writer(cfg), stage=args.stage) + if args.json: + print(json.dumps(record, indent=2, sort_keys=True)) + else: + print( + f"heartbeat {record['result']} queued={record.get('outbox_queued')} " + f"completeness_claimed={record.get('completeness_claimed')}" + ) + return 0 + + +def cmd_evidence_drain(cfg: Config, args) -> int: + """Best-effort drain of the local load-bearing outbox. Never a gate.""" + import json + + from secrets_engine.evidence import drain_outbox + + result = drain_outbox(_writer(cfg), audit_core_url=args.audit_core_url) + if args.json: + print(json.dumps(result, indent=2, sort_keys=True)) + else: + print( + f"outbox drain queued={result['queued']} " + f"delivered={result['delivered']} failed={result['failed']} " + f"skipped={result['skipped']}" + ) + return 0 + + +def cmd_evidence_classify(cfg: Config, args) -> int: + import json + + from secrets_engine.evidence_class import classify + + classified = classify(args.action, args.stage) + payload = { + "action": classified.action, + "stage": classified.stage, + "kind": classified.kind, + "rule_id": classified.rule_id, + "queued_locally": classified.queued_locally, + "completeness_claimed": classified.completeness_claimed, + } + if args.json: + print(json.dumps(payload, indent=2, sort_keys=True)) + else: + print( + f"{classified.action}/{classified.stage}: {classified.kind} " + f"rule={classified.rule_id} queued={classified.queued_locally}" + ) + return 0 + + # -- parser ---------------------------------------------------------------- @@ -700,6 +756,32 @@ def build_parser() -> argparse.ArgumentParser: au.add_argument("--json", action="store_true") au.set_defaults(func=cmd_audit) + ev = sub.add_parser("evidence", help="load-bearing evidence heartbeat and outbox") + evsub = ev.add_subparsers(dest="subcmd", required=True) + hb = evsub.add_parser( + "heartbeat", + help="emit a positive nothing-to-report claim (not a permission check)", + ) + hb.add_argument("--stage", default="prod", choices=("build", "test", "prod")) + hb.add_argument("--json", action="store_true") + hb.set_defaults(func=cmd_evidence_heartbeat) + dr = evsub.add_parser( + "drain", + help="best-effort drain of the local outbox; never blocks a mutation", + ) + dr.add_argument( + "--audit-core-url", + default="", + help="optional audit-core base URL; empty skips delivery and keeps files", + ) + dr.add_argument("--json", action="store_true") + dr.set_defaults(func=cmd_evidence_drain) + cl = evsub.add_parser("classify", help="show the §9.6 class for an action/stage") + cl.add_argument("action") + cl.add_argument("--stage", required=True, choices=("build", "test", "prod")) + cl.add_argument("--json", action="store_true") + cl.set_defaults(func=cmd_evidence_classify) + return p diff --git a/src/secrets_engine/errors.py b/src/secrets_engine/errors.py index d519184..fd73119 100644 --- a/src/secrets_engine/errors.py +++ b/src/secrets_engine/errors.py @@ -23,6 +23,10 @@ class DecisionError(SecretsEngineError): exit_code = 3 + def __init__(self, message: str, *, stance: dict[str, object] | None = None): + super().__init__(message) + self.stance = dict(stance or {}) + class PolicyGuardError(SecretsEngineError): """A plan violates a safety guard (wildcard, out-of-stage path, root, ...).""" diff --git a/src/secrets_engine/evidence.py b/src/secrets_engine/evidence.py index 14660b6..eacb8c0 100644 --- a/src/secrets_engine/evidence.py +++ b/src/secrets_engine/evidence.py @@ -17,6 +17,7 @@ from pathlib import Path from typing import Any from secrets_engine.errors import DecisionError, SecretsEngineError +from secrets_engine.evidence_class import KIND_ATTRIBUTIVE, classify from secrets_engine.redact import looks_secret, redact_text # Keys that must never carry a value into evidence regardless of nesting. @@ -64,6 +65,21 @@ class EvidenceWriter: with path.open("a", encoding="utf-8") as fh: fh.write(json.dumps(record, sort_keys=True) + "\n") + def outbox_dir(self) -> Path: + path = self.evidence_dir / "outbox" + path.mkdir(parents=True, exist_ok=True) + return path + + def _queue_outbox(self, record: dict[str, Any]) -> str: + """Durably queue a load-bearing record. Never talks to audit-core.""" + record_id = str(record.get("record_id") or uuid.uuid4()) + path = self.outbox_dir() / f"{record_id}.json" + tmp = path.with_suffix(".tmp") + tmp.write_text(json.dumps(record, sort_keys=True) + "\n", encoding="utf-8") + os.chmod(tmp, 0o600) + tmp.replace(path) + return record_id + def record( self, action: str, @@ -75,9 +91,18 @@ class EvidenceWriter: detail: dict[str, Any] | None = None, hub: bool = True, ) -> dict[str, Any]: - """Append one non-secret evidence record. Returns the stored record.""" + """Append one non-secret evidence record. Returns the stored record. + + Load-bearing records are queued locally first. An audit-core outage + cannot occur here because this method never contacts audit-core. + Completeness is never claimed. Presence or absence of a record is + not consulted as a permission. + """ record_id = str(uuid.uuid4()) - hub_requested = bool(hub and self.hub_url) + evidence_class = classify(action, stage) + hub_requested = bool( + hub and self.hub_url and evidence_class.kind == KIND_ATTRIBUTIVE + ) record = { "record_id": record_id, "ts": datetime.now(timezone.utc).isoformat(), @@ -89,7 +114,13 @@ class EvidenceWriter: "decision_id": decision_id, "detail": _scrub(detail or {}), "hub_delivery_requested": hub_requested, + "evidence_kind": evidence_class.kind, + "evidence_rule": evidence_class.rule_id, + "completeness_claimed": False, } + if evidence_class.queued_locally: + self._queue_outbox(record) + record["outbox_queued"] = True self._append_local(record) if hub_requested: delivery = self._post_hub( @@ -205,6 +236,7 @@ class PrivilegedActionEvidence: decision_id: str = "" approval_status: str = "pending" completed: bool = False + stance: dict[str, Any] = field(default_factory=dict) def __post_init__(self) -> None: if not self.approval_required: @@ -218,6 +250,8 @@ class PrivilegedActionEvidence: "decision_ref": self.decision_ref, } ) + if self.stance: + merged.update(self.stance) if extra: merged.update(extra) return merged @@ -239,6 +273,17 @@ class PrivilegedActionEvidence: self.decision_id = str(getattr(decision, "id", "")) self.approval_status = "approved" + def mark_stance(self, stance: object | None) -> None: + if stance is None: + return + if hasattr(stance, "as_evidence"): + payload = stance.as_evidence() + elif isinstance(stance, dict): + payload = stance + else: + return + self.stance = {key: value for key, value in payload.items() if value != ""} + def finish( self, result: str, *, detail: dict[str, Any] | None = None ) -> dict[str, Any]: @@ -259,6 +304,8 @@ class PrivilegedActionEvidence: return False if isinstance(exc, DecisionError): self.approval_status = "rejected" + if getattr(exc, "stance", None): + self.mark_stance(exc.stance) if isinstance(exc, (KeyboardInterrupt, SystemExit)): result = "interrupted" elif isinstance(exc, SecretsEngineError): @@ -274,3 +321,76 @@ class PrivilegedActionEvidence: detail=self._detail({"error_type": type(exc).__name__}), ) return False + + +def write_heartbeat(writer: EvidenceWriter, *, stage: str = "prod") -> dict[str, Any]: + """Positive claim that can go missing. Not a permission and not silence.""" + queued = 0 + outbox = writer.evidence_dir / "outbox" + if outbox.is_dir(): + queued = sum(1 for path in outbox.glob("*.json") if path.is_file()) + return writer.record( + "evidence-heartbeat", + result="nothing-to-report", + stage=stage, + detail={ + "form": "heartbeat", + "outbox_depth": queued, + "completeness_claimed": False, + }, + hub=False, + ) + + +def drain_outbox( + writer: EvidenceWriter, + *, + audit_core_url: str = "", +) -> dict[str, Any]: + """Best-effort delivery of queued load-bearing records. + + Never called from a mutation path. An audit-core outage leaves files in + place and does not raise into a revoke/destroy handler. + """ + outbox = writer.evidence_dir / "outbox" + if not outbox.is_dir(): + return { + "queued": 0, + "delivered": 0, + "failed": 0, + "skipped": 0, + "completeness_claimed": False, + } + files = sorted(path for path in outbox.glob("*.json") if path.is_file()) + queued = len(files) + if not audit_core_url: + return { + "queued": queued, + "delivered": 0, + "failed": 0, + "skipped": queued, + "completeness_claimed": False, + } + delivered = 0 + failed = 0 + for path in files: + try: + payload = path.read_text(encoding="utf-8").encode() + req = urllib.request.Request( + audit_core_url.rstrip("/") + "/v1/events", + data=payload, + headers={"Content-Type": "application/json"}, + method="POST", + ) + urllib.request.urlopen(req, timeout=3) + path.unlink() + delivered += 1 + except (urllib.error.URLError, OSError, ValueError): + failed += 1 + return { + "queued": queued, + "delivered": delivered, + "failed": failed, + "skipped": 0, + "completeness_claimed": False, + } diff --git a/src/secrets_engine/evidence_class.py b/src/secrets_engine/evidence_class.py new file mode 100644 index 0000000..45aafee --- /dev/null +++ b/src/secrets_engine/evidence_class.py @@ -0,0 +1,164 @@ +"""Load-bearing vs attributive evidence classification (§9.6). + +The YAML file is the declaration. ``SHIPPED_RULES`` is the pin that makes +drift fail the test. No function here grants or denies an action based on +whether a local evidence record exists. +""" +from __future__ import annotations + +import os +from dataclasses import dataclass +from pathlib import Path +from typing import Any + +import yaml + +from secrets_engine.config import repo_root +from secrets_engine.errors import SecretsEngineError + +KIND_LOAD_BEARING = "load-bearing" +KIND_ATTRIBUTIVE = "attributive" +KIND_HEARTBEAT = "heartbeat" + +# First match wins. Destroy is always load-bearing; production control +# mutations are load-bearing; everything else is attributive. +SHIPPED_RULES = ( + { + "id": "heartbeat", + "kind": KIND_HEARTBEAT, + "actions": ("evidence-heartbeat",), + "stages": ("build", "test", "prod", "unknown"), + }, + { + "id": "destroy", + "kind": KIND_LOAD_BEARING, + "actions": ("lifecycle-destroy",), + "stages": ("build", "test", "prod", "unknown"), + }, + { + "id": "production-control-mutation", + "kind": KIND_LOAD_BEARING, + "actions": ( + "revoke", + "lifecycle-suspend", + "lifecycle-deactivate", + "provision", + ), + "stages": ("prod",), + }, + { + "id": "default-attributive", + "kind": KIND_ATTRIBUTIVE, + "actions": ("*",), + "stages": ("build", "test", "prod", "unknown"), + }, +) + + +class EvidenceClassificationError(SecretsEngineError): + """Classification file missing or malformed.""" + + exit_code = 1 + + +@dataclass(frozen=True) +class EvidenceClass: + kind: str + rule_id: str + action: str + stage: str + queued_locally: bool + + @property + def completeness_claimed(self) -> bool: + return False + + +def classification_path() -> Path: + override = os.environ.get("SECRETS_ENGINE_EVIDENCE_CLASSIFICATION", "") + if override: + return Path(override) + return repo_root() / "evidence-classification.yaml" + + +def _normalize_rules(raw: Any) -> tuple[dict[str, Any], ...]: + if not isinstance(raw, list) or not raw: + raise EvidenceClassificationError( + "evidence-classification.yaml must list at least one rule" + ) + rules: list[dict[str, Any]] = [] + for item in raw: + if not isinstance(item, dict): + raise EvidenceClassificationError("each classification rule must be a map") + kind = str(item.get("kind", "")) + if kind not in {KIND_LOAD_BEARING, KIND_ATTRIBUTIVE, KIND_HEARTBEAT}: + raise EvidenceClassificationError(f"unknown evidence kind {kind!r}") + actions = tuple(str(a) for a in item.get("actions") or ()) + stages = tuple(str(s) for s in item.get("stages") or ()) + if not actions or not stages: + raise EvidenceClassificationError( + f"rule {item.get('id')!r} needs actions and stages" + ) + rules.append( + { + "id": str(item.get("id", "")), + "kind": kind, + "actions": actions, + "stages": stages, + } + ) + return tuple(rules) + + +def load_classification_rules( + path: Path | None = None, +) -> tuple[dict[str, Any], ...]: + target = path or classification_path() + try: + data = yaml.safe_load(target.read_text(encoding="utf-8")) or {} + except (OSError, yaml.YAMLError) as exc: + raise EvidenceClassificationError( + f"unable to load evidence classification {target}: {exc}" + ) from exc + if data.get("completeness_claimed") is not False: + raise EvidenceClassificationError( + f"{target} must declare completeness_claimed: false" + ) + if data.get("no_control_branches_on_presence") is not True: + raise EvidenceClassificationError( + f"{target} must declare no_control_branches_on_presence: true" + ) + return _normalize_rules(data.get("rules")) + + +def classify( + action: str, + stage: str, + *, + rules: tuple[dict[str, Any], ...] | None = None, +) -> EvidenceClass: + """Return the evidence class for an action/stage. Never a permission.""" + table = rules if rules is not None else load_classification_rules() + stage_key = stage if stage else "unknown" + for rule in table: + actions = rule["actions"] + stages = rule["stages"] + if "*" not in actions and action not in actions: + continue + if stage_key not in stages and "*" not in stages: + continue + kind = str(rule["kind"]) + return EvidenceClass( + kind=kind, + rule_id=str(rule["id"]), + action=action, + stage=stage_key, + queued_locally=kind in {KIND_LOAD_BEARING, KIND_HEARTBEAT}, + ) + return EvidenceClass( + kind=KIND_ATTRIBUTIVE, + rule_id="implicit-attributive", + action=action, + stage=stage_key, + queued_locally=False, + ) diff --git a/src/secrets_engine/pep_stance.py b/src/secrets_engine/pep_stance.py new file mode 100644 index 0000000..159951e --- /dev/null +++ b/src/secrets_engine/pep_stance.py @@ -0,0 +1,141 @@ +"""Published PEP unreachable-engine stance (security-layer-model v0.7 §6.4). + +Runtime reads ``pep-stance.yaml``. ``SHIPPED_STANCE`` is the pin that makes +drift between the published map and this module fail the conformance test. +""" +from __future__ import annotations + +import os +from dataclasses import dataclass, replace +from pathlib import Path +from typing import Any +from urllib.parse import urlparse + +import yaml + +from secrets_engine.config import repo_root +from secrets_engine.errors import DecisionError + +SHIPPED_STANCE = { + "build": "fail_open", + "test": "fail_open", + "prod": "fail_closed", + "unknown": "fail_closed", +} +REQUIRED_STAGES = tuple(SHIPPED_STANCE) +VALID_MODES = frozenset({"fail_open", "fail_closed"}) +LOOPBACK_HOSTS = frozenset({"127.0.0.1", "localhost", "::1"}) + + +@dataclass(frozen=True) +class StanceApplication: + """Named residue applied when access-engine is unreachable.""" + + stage: str + failure_mode: str + action: str + demo_exception: bool = False + decision_id: str = "" + + def as_evidence(self) -> dict[str, object]: + payload: dict[str, object] = { + "stance_stage": self.stage, + "stance_failure_mode": self.failure_mode, + "stance_demo_exception": self.demo_exception, + } + if self.decision_id: + payload["stance_decision_id"] = self.decision_id + return payload + + +@dataclass(frozen=True) +class PepStanceMap: + stance: dict[str, str] + path: Path + + def for_stage(self, stage: str) -> tuple[str, str]: + key = stage if stage in self.stance else "unknown" + mode = self.stance.get(key, "") + if mode not in VALID_MODES: + raise DecisionError( + f"pep-stance.yaml has no usable mode for stage {stage!r}" + ) + return key, mode + + +def pep_stance_path() -> Path: + override = os.environ.get("SECRETS_ENGINE_PEP_STANCE", "") + if override: + return Path(override) + return repo_root() / "pep-stance.yaml" + + +def load_pep_stance(path: Path | None = None) -> PepStanceMap: + target = path or pep_stance_path() + try: + data = yaml.safe_load(target.read_text(encoding="utf-8")) or {} + except (OSError, yaml.YAMLError) as exc: + raise DecisionError(f"unable to load PEP stance map {target}: {exc}") from exc + raw = data.get("stance") + if not isinstance(raw, dict): + raise DecisionError(f"{target} is missing a stance map") + stance = {str(key): str(value) for key, value in raw.items()} + missing = [stage for stage in REQUIRED_STAGES if stage not in stance] + if missing: + raise DecisionError( + f"{target} is not total; missing stages {missing}" + ) + unknown_modes = { + f"{stage}={mode}" + for stage, mode in stance.items() + if mode not in VALID_MODES + } + if unknown_modes: + raise DecisionError(f"{target} has invalid modes: {sorted(unknown_modes)}") + return PepStanceMap(stance=stance, path=target) + + +def demo_exception_enabled(cfg: Any) -> bool: + """Three-factor throwaway exception; not a stance row.""" + host = (urlparse(getattr(cfg, "bao_addr", "")).hostname or "").lower() + return ( + os.environ.get("SECRETS_ENGINE_UNSAFE_DEMO") == "1" + and not getattr(cfg, "hub_url", "") + and host in LOOPBACK_HOSTS + ) + + +def apply_unreachable_engine_stance( + cfg: Any, + entry: Any, + action: str, + *, + stance_map: PepStanceMap | None = None, +) -> StanceApplication: + """Apply the published unreachable-engine residue for a live action. + + ``fail_closed`` without the demo exception raises ``DecisionError`` carrying + named stance fields. ``fail_open`` is the documented residue: continue to + the existing lane-approval check, which is itself a gap until T02. + """ + loaded = stance_map or load_pep_stance() + stage, mode = loaded.for_stage(getattr(entry, "stage", "unknown")) + demo = demo_exception_enabled(cfg) + applied = StanceApplication( + stage=stage, + failure_mode=mode, + action=action or "unknown", + demo_exception=bool(demo and mode == "fail_closed"), + ) + if mode == "fail_closed" and not demo: + raise DecisionError( + f"production action '{applied.action}' requires a durable " + "access-engine decision record; live production remains disabled", + stance=applied.as_evidence(), + ) + return applied + + +def with_decision(stance: StanceApplication, decision: Any) -> StanceApplication: + decision_id = str(getattr(decision, "id", "") or "") + return replace(stance, decision_id=decision_id) diff --git a/tests/test_evidence_class.py b/tests/test_evidence_class.py new file mode 100644 index 0000000..11d74b1 --- /dev/null +++ b/tests/test_evidence_class.py @@ -0,0 +1,102 @@ +"""§9.6 classification, local outbox, heartbeat, and drain.""" +from __future__ import annotations + +import json +import urllib.error +from types import SimpleNamespace + +from secrets_engine.evidence import EvidenceWriter, drain_outbox, write_heartbeat +from secrets_engine.evidence_class import classify + + +def test_load_bearing_record_is_queued_locally_before_jsonl(tmp_path): + writer = EvidenceWriter(evidence_dir=tmp_path, hub_url="http://hub.invalid", topic_id="t") + record = writer.record( + "lifecycle-destroy", + result="attempt", + catalog_id="lane", + stage="prod", + detail={"value": "should-not-leak"}, + hub=True, + ) + assert record["evidence_kind"] == "load-bearing" + assert record["completeness_claimed"] is False + assert record["outbox_queued"] is True + assert record["hub_delivery_requested"] is False + queued = list((tmp_path / "outbox").glob("*.json")) + assert len(queued) == 1 + queued_payload = json.loads(queued[0].read_text()) + assert queued_payload["record_id"] == record["record_id"] + assert "should-not-leak" not in queued[0].read_text() + jsonl = next(tmp_path.glob("evidence-*.jsonl")).read_text() + assert record["record_id"] in jsonl + + +def test_attributive_record_is_not_queued_and_may_post_hub(tmp_path, monkeypatch): + monkeypatch.setattr( + "urllib.request.urlopen", + lambda *_args, **_kwargs: SimpleNamespace(status=200, read=lambda: b"{}"), + ) + writer = EvidenceWriter(evidence_dir=tmp_path, hub_url="http://hub.invalid", topic_id="t") + record = writer.record("apply", result="applied", catalog_id="lane", stage="prod") + assert record["evidence_kind"] == "attributive" + assert record["hub_delivery_requested"] is True + assert not list((tmp_path / "outbox").glob("*.json")) + + +def test_heartbeat_is_a_positive_claim_not_a_permission(tmp_path): + writer = EvidenceWriter(evidence_dir=tmp_path, hub_url="") + record = write_heartbeat(writer, stage="prod") + assert record["action"] == "evidence-heartbeat" + assert record["result"] == "nothing-to-report" + assert record["evidence_kind"] == "heartbeat" + assert record["completeness_claimed"] is False + assert record["outbox_queued"] is True + # Heartbeat must not be used as "no jsonl means nothing happened". + assert classify("evidence-heartbeat", "prod").kind == "heartbeat" + + +def test_drain_without_audit_core_keeps_files(tmp_path): + writer = EvidenceWriter(evidence_dir=tmp_path, hub_url="") + writer.record("provision", result="from-file", catalog_id="lane", stage="prod") + result = drain_outbox(writer, audit_core_url="") + assert result["queued"] == 1 + assert result["skipped"] == 1 + assert result["delivered"] == 0 + assert result["completeness_claimed"] is False + assert list((tmp_path / "outbox").glob("*.json")) + + +def test_drain_audit_core_outage_does_not_raise_or_delete(tmp_path, monkeypatch): + writer = EvidenceWriter(evidence_dir=tmp_path, hub_url="") + writer.record("revoke", result="native-access-deactivated", catalog_id="lane", stage="prod") + + def offline(*_args, **_kwargs): + raise urllib.error.URLError("audit-core down") + + monkeypatch.setattr("urllib.request.urlopen", offline) + result = drain_outbox(writer, audit_core_url="http://audit-core.invalid") + assert result["failed"] == 1 + assert result["delivered"] == 0 + assert list((tmp_path / "outbox").glob("*.json")) + + +def test_successful_drain_removes_only_delivered_files(tmp_path, monkeypatch): + writer = EvidenceWriter(evidence_dir=tmp_path, hub_url="") + writer.record("provision", result="from-file", catalog_id="lane", stage="prod") + monkeypatch.setattr( + "urllib.request.urlopen", + lambda *_args, **_kwargs: SimpleNamespace(status=202, read=lambda: b"{}"), + ) + result = drain_outbox(writer, audit_core_url="http://audit-core.invalid") + assert result["delivered"] == 1 + assert not list((tmp_path / "outbox").glob("*.json")) + + +def test_classify_is_not_a_permission_api(): + """Missing or unknown actions are attributive labels, never a deny.""" + unknown = classify("not-a-control", "prod") + assert unknown.kind == "attributive" + assert unknown.completeness_claimed is False + assert "permit" not in dir(classify) + assert "deny" not in dir(classify) diff --git a/tests/test_layer_conformance.py b/tests/test_layer_conformance.py index 302da57..53e7b30 100644 --- a/tests/test_layer_conformance.py +++ b/tests/test_layer_conformance.py @@ -17,6 +17,8 @@ import yaml from secrets_engine.catalog import validate_entry from secrets_engine.cli import _require_lane_approval from secrets_engine.errors import DecisionError +from secrets_engine.evidence_class import SHIPPED_RULES, classify, load_classification_rules +from secrets_engine.pep_stance import SHIPPED_STANCE, load_pep_stance from tests.test_catalog import VALID @@ -25,6 +27,7 @@ SCRIPT = ROOT / "scripts" / "check_layer_conformance.py" DECL = ROOT / "layer.yaml" STANCE = ROOT / "pep-stance.yaml" INTENT = ROOT / "INTENT.md" +CLASSIFICATION = ROOT / "evidence-classification.yaml" def _decl() -> dict: @@ -73,6 +76,7 @@ def test_proposed_capabilities_carry_gap_record_fields(): assert cap.get(field), f"{cap.get('id')} missing {field}" assert cap["state"] == "unowned-capability" assert cap["owner_status"] == "proposed" + assert cap.get("contract"), f"{cap.get('id')} missing contract" def test_stance_map_is_total_over_catalog_stages(): @@ -85,19 +89,53 @@ def test_stance_map_is_total_over_catalog_stages(): assert _stance()["verdict_caching"] == "none" +def test_published_map_equals_shipped_constant_and_loader(): + """Changing the YAML without changing SHIPPED_STANCE fails, and the reverse.""" + published = _stance()["stance"] + loaded = load_pep_stance().stance + assert published == SHIPPED_STANCE + assert loaded == SHIPPED_STANCE + + def test_published_prod_stance_equals_shipped_fail_closed(monkeypatch): - """pep-stance.yaml prod: fail_closed must equal _require_lane_approval.""" - assert _stance()["stance"]["prod"] == "fail_closed" + """Runtime reads pep-stance.yaml; prod fail_closed must equal the gate.""" + assert load_pep_stance().stance["prod"] == "fail_closed" entry = validate_entry(dict(VALID, stage="prod", approval={"model": "bootstrap-only"})) cfg = SimpleNamespace(hub_url="http://127.0.0.1:8000", bao_addr="http://127.0.0.1:8200") monkeypatch.delenv("SECRETS_ENGINE_UNSAFE_DEMO", raising=False) - with pytest.raises(DecisionError, match="live production remains disabled"): + with pytest.raises(DecisionError, match="live production remains disabled") as raised: _require_lane_approval(cfg, entry, "apply") + assert raised.value.stance["stance_stage"] == "prod" + assert raised.value.stance["stance_failure_mode"] == "fail_closed" + assert "stance_decision_id" not in raised.value.stance + + +def test_yaml_is_the_runtime_source(tmp_path, monkeypatch): + """A published map the pin does not match is a test failure; runtime follows YAML.""" + path = tmp_path / "pep-stance.yaml" + path.write_text( + yaml.safe_dump( + { + "stance": { + "build": "fail_open", + "test": "fail_open", + "prod": "fail_open", + "unknown": "fail_closed", + } + } + ), + encoding="utf-8", + ) + monkeypatch.setenv("SECRETS_ENGINE_PEP_STANCE", str(path)) + entry = validate_entry(dict(VALID, stage="prod", approval={"model": "bootstrap-only"})) + cfg = SimpleNamespace(hub_url="http://127.0.0.1:8000", bao_addr="http://127.0.0.1:8200") + monkeypatch.delenv("SECRETS_ENGINE_UNSAFE_DEMO", raising=False) + assert _require_lane_approval(cfg, entry, "apply") is None def test_build_stage_is_not_the_production_fail_closed_gate(tmp_path, monkeypatch): """build is fail_open relative to access-engine: lane approval still applies.""" - assert _stance()["stance"]["build"] == "fail_open" + assert load_pep_stance().stance["build"] == "fail_open" (tmp_path / ".decisions").mkdir() (tmp_path / ".decisions" / "x.yaml").write_text( "id: x\ntitle: approved\nstatus: resolved\nsuperseded_by: null\n" @@ -114,4 +152,38 @@ def test_build_stage_is_not_the_production_fail_closed_gate(tmp_path, monkeypatc ) ) cfg = SimpleNamespace(hub_url="", bao_addr="http://127.0.0.1:8200") - assert _require_lane_approval(cfg, entry, "apply").id == "x" + decision = _require_lane_approval(cfg, entry, "apply") + assert decision.id == "x" + + +def test_classification_yaml_equals_shipped_rules(): + loaded = load_classification_rules() + assert tuple(rule["id"] for rule in loaded) == tuple(rule["id"] for rule in SHIPPED_RULES) + assert tuple(rule["kind"] for rule in loaded) == tuple(rule["kind"] for rule in SHIPPED_RULES) + + +def test_classify_does_not_grant_permission(): + prod_provision = classify("provision", "prod") + test_provision = classify("provision", "test") + destroy = classify("lifecycle-destroy", "build") + apply_prod = classify("apply", "prod") + heartbeat = classify("evidence-heartbeat", "prod") + assert prod_provision.kind == "load-bearing" + assert test_provision.kind == "attributive" + assert destroy.kind == "load-bearing" + assert apply_prod.kind == "attributive" + assert heartbeat.kind == "heartbeat" + assert prod_provision.completeness_claimed is False + assert CLASSIFICATION.exists() + + +def test_proposed_contracts_exist_and_forbid_secret_material(): + ssh = (ROOT / "docs/ssh-ca-signing-contract.md").read_text(encoding="utf-8") + secret_use = (ROOT / "docs/secret-use-evidence-contract.md").read_text(encoding="utf-8") + assert "proposed" in ssh.lower() + assert "proposed" in secret_use.lower() + assert "warden sign" in ssh + assert "private key" in ssh.lower() or "private keys" in ssh.lower() + assert "secret values" in secret_use.lower() or "secret value" in secret_use.lower() + assert "audit-core" in secret_use + assert "completeness is not claimed" in secret_use.lower() diff --git a/tests/test_privileged_cli_evidence.py b/tests/test_privileged_cli_evidence.py index 9a91e47..e726ab5 100644 --- a/tests/test_privileged_cli_evidence.py +++ b/tests/test_privileged_cli_evidence.py @@ -121,3 +121,12 @@ def test_production_handler_fails_closed_before_backend(tmp_path, monkeypatch): "attempt", "failed-DecisionError", ] + terminal = records[-1] + assert terminal["detail"]["stance_stage"] == "prod" + assert terminal["detail"]["stance_failure_mode"] == "fail_closed" + assert terminal["detail"]["approval_status"] == "rejected" + assert "stance_decision_id" not in terminal["detail"] + assert terminal["completeness_claimed"] is False + assert "SUPER-SECRET" not in json.dumps(records) + outbox = list((tmp_path / "evidence" / "outbox").glob("*.json")) + assert outbox, "production provision refusal is load-bearing and must be queued" diff --git a/workplans/SECRETS-WP-0008-layer-model-lifecycle-conformance.md b/workplans/SECRETS-WP-0008-layer-model-lifecycle-conformance.md index db8649d..f203984 100644 --- a/workplans/SECRETS-WP-0008-layer-model-lifecycle-conformance.md +++ b/workplans/SECRETS-WP-0008-layer-model-lifecycle-conformance.md @@ -4,7 +4,7 @@ type: workplan title: "Evolve the Lifecycle engine to the accepted security layer model" domain: infotech repo: secrets-engine -status: ready +status: active owner: grok topic_slug: custodian created: "2026-08-29" @@ -56,11 +56,16 @@ It adds the layer-model remainder: stance binding, decision-record consumption, ```task id: SECRETS-WP-0008-T01 -status: todo +status: done priority: high state_hub_task_id: "945735c6-2d3d-5f72-baf0-692e6b210889" ``` +Completed 2026-08-29. `_require_lane_approval` loads `pep-stance.yaml`. +`SHIPPED_STANCE` is pinned equal to the published map. Fail-closed production +evidence records `stance_stage`, `stance_failure_mode`, and omits +`stance_decision_id` unless a decision was rendered. + `pep-stance.yaml` is published and tests pin `prod: fail_closed` to `_require_lane_approval`. The gate is still hardcoded, and a refusal is recorded as a `DecisionError` rather than named stance fields. @@ -111,11 +116,16 @@ Acceptance: ```task id: SECRETS-WP-0008-T03 -status: todo +status: done priority: high state_hub_task_id: "0b54cedf-9335-5fa0-bdcb-1efa9a9cd98d" ``` +Completed 2026-08-29. `evidence-classification.yaml` is the declaration. +Load-bearing records are queued in `evidence_dir/outbox/` before JSONL. +`secrets-engine evidence heartbeat` emits a positive claim. Drain never sits +on a mutation path and an audit-core outage leaves the queue in place. + Classify each privileged action as load-bearing or attributive. Today's trail is attributive local JSONL plus best-effort State Hub notes; completeness is already not claimed and must stay that way until the destination is @@ -142,11 +152,14 @@ Acceptance: ```task id: SECRETS-WP-0008-T04 -status: todo +status: done priority: medium state_hub_task_id: "581aeee3-3d99-5034-992b-7d42e6542a19" ``` +Completed 2026-08-29. Contract published at `docs/ssh-ca-signing-contract.md`. +No OpenBao SSH-CA write was added. `layer.yaml` owner_status remains proposed. + ops-warden's `ssh-ca-signing-write` gap names this engine as intended owner. Accept the **engine API**, decline the Staff lane. Publish a contract ops-warden can call after an `access-engine` decision: sign a short-lived @@ -168,11 +181,15 @@ Acceptance: ```task id: SECRETS-WP-0008-T05 -status: todo +status: done priority: medium state_hub_task_id: "3100d28c-9494-5b21-9d5a-af583cd8aa55" ``` +Completed 2026-08-29. Contract published at +`docs/secret-use-evidence-contract.md`. `layer.yaml` owner_status remains +proposed. `route`/`audit` are still not this surface. + kings-guard's secret-abuse posture is fixture-driven because no engine exposes lease, revocation, mount, rotation, and delivery-session metadata. `route` and `audit` are operator summaries over local JSONL and are not that surface.