From 72d3327c28f7b5f71606101cc6d9d0d3e25c4de0 Mon Sep 17 00:00:00 2001 From: tegwick Date: Wed, 2 Sep 2026 13:09:10 +0200 Subject: [PATCH] Add native rotate and persistent lane overlay states rotate replaces one declared KV field through the merge-safe patch path and never prints the value. Overlay states active/suspended/deactivated/ compromised live under the evidence directory. compromise/reactivate and successful suspend/deactivate/revoke update that overlay; exec/wrap/ handoff/provision refuse non-active lanes. Provider-side rotation stays with rotation.owner. Production remains fail-closed. Assistant: grok Assistant-Session: 01a05f07-ae72-7781-9fcb-19efd61add00 --- SCOPE.md | 13 ++- docs/cli.md | 13 +++ docs/hardening-backlog.md | 15 ++- evidence-classification.yaml | 2 +- layer.yaml | 3 + src/secrets_engine/cli.py | 161 +++++++++++++++++++++++++-- src/secrets_engine/evidence_class.py | 3 + src/secrets_engine/lane_state.py | 137 +++++++++++++++++++++++ src/secrets_engine/rotate.py | 43 +++++++ tests/test_lane_state.py | 91 +++++++++++++++ tests/test_layer_conformance.py | 4 + tests/test_rotate.py | 131 ++++++++++++++++++++++ 12 files changed, 596 insertions(+), 20 deletions(-) create mode 100644 src/secrets_engine/lane_state.py create mode 100644 src/secrets_engine/rotate.py create mode 100644 tests/test_lane_state.py create mode 100644 tests/test_rotate.py diff --git a/SCOPE.md b/SCOPE.md index 49cd73a..91b057f 100644 --- a/SCOPE.md +++ b/SCOPE.md @@ -179,8 +179,11 @@ high-risk lanes. These operations do not manage external workload delivery. `session revoke` revokes an already-issued token accessor or lease id the operator already -holds; evidence is fingerprint-only. There is currently no rotation command, -compromised state, or persistent/reversible lane state machine. +holds; evidence is fingerprint-only. `rotate` replaces one declared KV field +through the merge-safe write path. Persistent overlay states +(`active`/`suspended`/`deactivated`/`compromised`) live under the evidence +directory; they do not recreate OpenBao objects and do not rotate provider +credentials. ## CLI Surface @@ -222,10 +225,8 @@ metadata. `secrets-engine wrap` implements response-wrapped operator handoff. - A service API, daemon, UI, queue, scheduler, or remote multi-user service. - Platform JWT mount/role materialization for the implemented KeyCape service-auth / `service-jwt` provider. -- Provider-side rotation or coordinated multi-consumer rollout. -- First-class rotate, compromise, reactivate, lease-status, or audit report - commands; lifecycle operations currently execute plans without persistent - lane state. +- Provider-side / workload consumer rotation. Overlay lane state is local and + non-secret only; it is not an OpenBao-side state machine. - Resolution of a durable access-engine decision record / State Hub ActionAuthorization and wiring its validated approval threshold to each production handler. The consumer validator exists; the serving endpoint does diff --git a/docs/cli.md b/docs/cli.md index b2d4d77..94530a9 100644 --- a/docs/cli.md +++ b/docs/cli.md @@ -53,6 +53,7 @@ secrets-engine decision inspect secrets-engine plan --stage secrets-engine apply --stage [--dry-run] [--bootstrap-token-file F] secrets-engine provision --stage --field NAME (--from-file F | --generate) +secrets-engine rotate --stage --field NAME --from-file F [--dry-run] secrets-engine verify [--field NAME] [--positive] [--negative] [--negative-token-file F] secrets-engine handoff --stage --role-id-file F --secret-id-file F secrets-engine wrap --out F [--ttl 15m] @@ -64,6 +65,9 @@ secrets-engine session revoke (--accessor-file F | --lease-id-file F) [--stage s secrets-engine lifecycle suspend [--dry-run] secrets-engine lifecycle deactivate [--dry-run] secrets-engine lifecycle destroy [--dry-run] [--confirm-destroy ] +secrets-engine lifecycle compromise [--reason TEXT] [--dry-run] +secrets-engine lifecycle reactivate [--dry-run] +secrets-engine state show [--json] secrets-engine audit [--json] secrets-engine secret-use snapshot [--catalog-id ID] [--json] ``` @@ -117,6 +121,15 @@ All three preserve externally managed auth and workload delivery. The legacy `revoke` command is a compatibility alias for safe native deactivation, never KV destruction. +`rotate` replaces one declared KV field through the same merge-safe patch as +provision. It does not roll ESO/consumers or revoke a provider credential; +that remains `rotation.owner`. Auth-capability lanes use wrap/handoff instead. + +`lifecycle compromise` / `reactivate` and successful suspend/deactivate/revoke +write a non-secret overlay under the evidence directory. `state show` reads it. +exec/wrap/handoff/provision refuse non-active overlay states; rotate is still +allowed on compromised lanes. Overlay state does not recreate OpenBao objects. + `exec --mode exec-file` writes the selected field to a mode-0600 temp file and sets `{FIELD}_FILE` to that path for the child only. The file is overwritten and unlinked after the child exits. The value is not copied into the child diff --git a/docs/hardening-backlog.md b/docs/hardening-backlog.md index 2d1a426..ad9f57f 100644 --- a/docs/hardening-backlog.md +++ b/docs/hardening-backlog.md @@ -66,11 +66,16 @@ contents in this repo. ## H4 โ€” Rotation & lifecycle states -- Implement `rotate` and persistent `compromised` / `deactivated` lane states - with evidence. Explicit suspend/deactivate/destroy plans now exist, and - ordinary `revoke` safely aliases native AppRole/policy deactivation, but lane - state and coordinated provider/workload rotation remain outstanding. Live - destroy remains disabled until exact-action authorization is available. +- Implemented native `rotate`: merge-safe KV field replace from a mode-0600 + file, sibling preservation, no argv values. Auth-capability lanes stay on + wrap/handoff. Provider/workload consumer rollout remains the catalog + `rotation.owner` duty. +- Implemented persistent overlay states `active` / `suspended` / + `deactivated` / `compromised` under `evidence_dir/lane-state/` (non-secret, + not Git). `lifecycle compromise` / `reactivate` and successful + suspend/deactivate/revoke update the overlay. exec/wrap/handoff/provision + refuse non-active lanes; rotate remains allowed on compromised. +- Live destroy remains disabled until exact-action authorization is available. ## H4a โ€” Known-accessor operator command diff --git a/evidence-classification.yaml b/evidence-classification.yaml index b499744..2c646cb 100644 --- a/evidence-classification.yaml +++ b/evidence-classification.yaml @@ -45,7 +45,7 @@ rules: - id: production-control-mutation kind: load-bearing - actions: [revoke, lifecycle-suspend, lifecycle-deactivate, provision, session-revoke, wrap] + actions: [revoke, lifecycle-suspend, lifecycle-deactivate, provision, session-revoke, wrap, rotate, lifecycle-compromise, lifecycle-reactivate] stages: [prod] emission: local-outbox note: >- diff --git a/layer.yaml b/layer.yaml index c18662f..16ae2d4 100644 --- a/layer.yaml +++ b/layer.yaml @@ -60,6 +60,9 @@ protected_actions: - lifecycle-destroy - session-revoke - wrap + - rotate + - lifecycle-compromise + - lifecycle-reactivate # ยง13 proposed capabilities. Owner status is proposed, not assented, until # the surface exists in this repository's own contract. diff --git a/src/secrets_engine/cli.py b/src/secrets_engine/cli.py index df8a724..215b97a 100644 --- a/src/secrets_engine/cli.py +++ b/src/secrets_engine/cli.py @@ -7,6 +7,7 @@ Command surface (FR7): plan --stage apply --stage [--dry-run] [--auth auto] [--bootstrap-token-file F] provision --stage (--from-file F | --generate) --field NAME + rotate --stage --field NAME --from-file F [--dry-run] verify [--positive] [--negative] [--field NAME] [--negative-token-file F] handoff --stage --role-id-file F --secret-id-file F wrap --out F [--ttl 15m] @@ -14,7 +15,8 @@ Command surface (FR7): route [--json] revoke session revoke (--accessor-file F | --lease-id-file F) [--stage stage] - lifecycle suspend|deactivate|destroy + lifecycle suspend|deactivate|destroy|compromise|reactivate + state show [--json] audit [--json] secret-use snapshot [--catalog-id ID] [--json] evidence heartbeat|drain|classify @@ -42,6 +44,13 @@ 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.engine_auth import login_service_jwt, select_engine_auth +from secrets_engine.lane_state import ( + load_lane_state, + operation_state, + require_delivery_state, + require_provision_state, + save_lane_state, +) from secrets_engine.pep_stance import apply_unreachable_engine_stance, with_decision from secrets_engine.openbao import OpenBaoClient, accessor_fingerprint, read_strict_token_file from secrets_engine.plan import build_plan @@ -304,6 +313,7 @@ def cmd_provision(cfg: Config, args) -> int: ) decision = _require_lane_approval(cfg, entry, "provision", evidence) evidence.mark_approved(decision) + require_provision_state(cfg.evidence_dir, entry.id) with _open_backend(cfg, args, evidence) as client: if args.generate: f = provision_generated(client, entry, field) @@ -318,6 +328,49 @@ def cmd_provision(cfg: Config, args) -> int: return 0 +def cmd_rotate(cfg: Config, args) -> int: + from secrets_engine.rotate import render_rotate_plan, rotate_from_file + + entry = get_entry(cfg.catalog_dir, args.catalog_id) + field = args.field or (entry.fields[0] if entry.fields else "") + if args.dry_run: + print(render_rotate_plan(entry, field)) + print("\n(dry-run: no OpenBao mutation performed)") + return 0 + with _privileged_evidence( + cfg, entry, "rotate", detail={"field": field} + ) as evidence: + if args.stage != entry.stage: + from secrets_engine.errors import ProvisioningError + + raise ProvisioningError( + f"lane '{entry.id}' is stage '{entry.stage}', not '{args.stage}'" + ) + decision = _require_lane_approval(cfg, entry, "rotate", evidence) + evidence.mark_approved(decision) + with _open_backend(cfg, args, evidence) as client: + f = rotate_from_file(client, entry, field, Path(args.from_file)) + record = save_lane_state( + cfg.evidence_dir, + entry.id, + load_lane_state(cfg.evidence_dir, entry.id).state, + operation="rotate", + ) + print( + f"rotated lane '{entry.id}' field '{f}' โ€” value not displayed " + f"(state={record.state})" + ) + evidence.finish( + "rotated", + detail={ + "field": f, + "lane_state": record.state, + "rotation_owner": str((entry.rotation or {}).get("owner") or ""), + }, + ) + return 0 + + def cmd_verify(cfg: Config, args) -> int: entry = get_entry(cfg.catalog_dir, args.catalog_id) fields = [args.field] if args.field else list(entry.fields) @@ -406,6 +459,7 @@ def cmd_handoff(cfg: Config, args) -> int: ) decision = _require_lane_approval(cfg, entry, "handoff", evidence) evidence.mark_approved(decision) + require_delivery_state(cfg.evidence_dir, entry.id, "handoff") with _open_backend(cfg, args, evidence) as client: result = write_approle_handoff( client, @@ -445,6 +499,7 @@ def cmd_wrap(cfg: Config, args) -> int: ) as evidence: decision = _require_lane_approval(cfg, entry, "wrap", evidence) evidence.mark_approved(decision) + require_delivery_state(cfg.evidence_dir, entry.id, "wrap") with _open_backend(cfg, args, evidence) as client: result = write_wrapped_handoff( client, entry, out_file=Path(args.out), ttl=args.ttl @@ -487,6 +542,7 @@ def cmd_exec(cfg: Config, args) -> int: # require approval + readiness before running. decision = _require_lane_approval(cfg, entry, "exec", evidence) evidence.mark_approved(decision) + require_delivery_state(cfg.evidence_dir, entry.id, "exec") if not args.command: from secrets_engine.errors import DeliveryError @@ -611,16 +667,73 @@ def cmd_revoke(cfg: Config, args) -> int: result = apply_lifecycle_plan(client, plan) print(plan.render()) print(result.render()) + record = save_lane_state( + cfg.evidence_dir, entry.id, "deactivated", operation="revoke" + ) evidence.finish( "native-access-deactivated", detail={ "applied": list(result.applied), "preserved": list(result.preserved), + "lane_state": record.state, }, ) return 0 +def _cmd_lifecycle_state(cfg: Config, args, entry) -> int: + mapped = operation_state(args.operation) + reason = getattr(args, "reason", "") or "" + current = load_lane_state(cfg.evidence_dir, entry.id) + if args.dry_run: + print( + f"Lane state plan for '{entry.id}': {current.state} -> {mapped} " + f"(operation={args.operation})" + ) + print("\n(dry-run: no state file written)") + return 0 + with _privileged_evidence( + cfg, + entry, + f"lifecycle-{args.operation}", + detail={"operation": args.operation, "from_state": current.state}, + ) as evidence: + decision = _require_lane_approval(cfg, entry, args.operation, evidence) + evidence.mark_approved(decision) + record = save_lane_state( + cfg.evidence_dir, + entry.id, + mapped or current.state, + operation=args.operation, + reason=reason, + ) + print(f"lane '{entry.id}' state {current.state} -> {record.state}") + evidence.finish( + record.state, + detail={"lane_state": record.state, "from_state": current.state}, + ) + return 0 + + +def cmd_state_show(cfg: Config, args) -> int: + import json + + entry = get_entry(cfg.catalog_dir, args.catalog_id) + record = load_lane_state(cfg.evidence_dir, entry.id) + payload = record.as_dict() + if args.json: + print(json.dumps(payload, indent=2, sort_keys=True)) + else: + print( + f"lane: {payload['catalog_id']} state={payload['state']} " + f"last={payload.get('last_operation') or '-'} " + f"at={payload.get('updated_at') or '-'}" + ) + if payload.get("reason"): + print(f"reason: {payload['reason']}") + return 0 + + def cmd_lifecycle(cfg: Config, args) -> int: from secrets_engine.lifecycle import ( apply_lifecycle_plan, @@ -629,6 +742,8 @@ def cmd_lifecycle(cfg: Config, args) -> int: ) entry = get_entry(cfg.catalog_dir, args.catalog_id) + if args.operation in {"compromise", "reactivate"}: + return _cmd_lifecycle_state(cfg, args, entry) plan = build_lifecycle_plan(entry, args.operation) if args.dry_run: print(plan.render()) @@ -655,15 +770,21 @@ def cmd_lifecycle(cfg: Config, args) -> int: evidence.mark_approved(decision) with _open_backend(cfg, args, evidence) as client: result = apply_lifecycle_plan(client, plan) + mapped = operation_state(args.operation) + record = None + if mapped: + record = save_lane_state( + cfg.evidence_dir, entry.id, mapped, operation=args.operation + ) print(plan.render()) print(result.render()) - evidence.finish( - "applied", - detail={ - "applied": list(result.applied), - "preserved": list(result.preserved), - }, - ) + detail = { + "applied": list(result.applied), + "preserved": list(result.preserved), + } + if record is not None: + detail["lane_state"] = record.state + evidence.finish("applied", detail=detail) return 0 @@ -823,6 +944,15 @@ def build_parser() -> argparse.ArgumentParser: add_token_arg(pr) pr.set_defaults(func=cmd_provision) + rot = sub.add_parser("rotate", help="replace a declared KV field without printing it") + rot.add_argument("catalog_id") + rot.add_argument("--stage", required=True, choices=("build", "test", "prod")) + rot.add_argument("--field", default=None) + rot.add_argument("--from-file", required=True, help="mode-0600 file holding the new value") + rot.add_argument("--dry-run", action="store_true") + add_token_arg(rot) + rot.set_defaults(func=cmd_rotate) + ve = sub.add_parser("verify", help="positive/negative verification (no value printed)") ve.add_argument("catalog_id") ve.add_argument("--field", default=None) @@ -909,6 +1039,8 @@ def build_parser() -> argparse.ArgumentParser: ("suspend", "stop new native logins; preserve policy and KV custody"), ("deactivate", "remove native AppRole/policy; preserve KV custody"), ("destroy", "irreversibly delete KV metadata after deactivation"), + ("compromise", "mark the lane compromised; block delivery until reactivate"), + ("reactivate", "clear compromised/suspended/deactivated overlay state"), ): lp = lcsub.add_parser(operation, help=help_text) lp.add_argument("catalog_id") @@ -919,9 +1051,22 @@ def build_parser() -> argparse.ArgumentParser: default="", help="exact catalog id required for live irreversible deletion", ) + if operation == "compromise": + lp.add_argument( + "--reason", + default="", + help="non-secret operator note, max 200 characters", + ) add_token_arg(lp) lp.set_defaults(func=cmd_lifecycle) + st = sub.add_parser("state", help="show persistent non-secret lane lifecycle state") + stsub = st.add_subparsers(dest="subcmd", required=True) + stshow = stsub.add_parser("show", help="show overlay state for a catalog lane") + stshow.add_argument("catalog_id") + stshow.add_argument("--json", action="store_true") + stshow.set_defaults(func=cmd_state_show) + su = sub.add_parser("secret-use", help="read-only secret-use evidence surface") susub = su.add_subparsers(dest="subcmd", required=True) snap = susub.add_parser( diff --git a/src/secrets_engine/evidence_class.py b/src/secrets_engine/evidence_class.py index 51acaa3..936a609 100644 --- a/src/secrets_engine/evidence_class.py +++ b/src/secrets_engine/evidence_class.py @@ -45,6 +45,9 @@ SHIPPED_RULES = ( "provision", "session-revoke", "wrap", + "rotate", + "lifecycle-compromise", + "lifecycle-reactivate", ), "stages": ("prod",), }, diff --git a/src/secrets_engine/lane_state.py b/src/secrets_engine/lane_state.py new file mode 100644 index 0000000..4f90dd7 --- /dev/null +++ b/src/secrets_engine/lane_state.py @@ -0,0 +1,137 @@ +"""Persistent non-secret lane lifecycle state. + +State lives under the evidence directory, never in Git, and never holds a +secret value. It does not recreate OpenBao objects; ``apply`` remains the +metadata path. Delivery commands consult this overlay and fail closed. +""" +from __future__ import annotations + +from dataclasses import dataclass +from datetime import datetime, timezone +from pathlib import Path +import yaml + +from secrets_engine.errors import DecisionError, PolicyGuardError +from secrets_engine.redact import looks_secret, redact_text + +STATES = ("active", "suspended", "deactivated", "compromised") +DELIVERY_BLOCKED = frozenset({"suspended", "deactivated", "compromised"}) +PROVISION_BLOCKED = frozenset({"suspended", "deactivated", "compromised"}) + + +@dataclass(frozen=True) +class LaneState: + catalog_id: str + state: str + updated_at: str = "" + last_operation: str = "" + reason: str = "" + + def as_dict(self) -> dict[str, str]: + payload = { + "catalog_id": self.catalog_id, + "state": self.state, + "updated_at": self.updated_at, + "last_operation": self.last_operation, + } + if self.reason: + payload["reason"] = self.reason + return payload + + +def state_dir(evidence_dir: Path) -> Path: + return Path(evidence_dir) / "lane-state" + + +def state_path(evidence_dir: Path, catalog_id: str) -> Path: + return state_dir(evidence_dir) / f"{catalog_id}.yaml" + + +def load_lane_state(evidence_dir: Path, catalog_id: str) -> LaneState: + path = state_path(evidence_dir, catalog_id) + if not path.is_file(): + return LaneState(catalog_id=catalog_id, state="active") + try: + data = yaml.safe_load(path.read_text(encoding="utf-8")) or {} + except (OSError, yaml.YAMLError) as exc: + raise PolicyGuardError(f"unable to load lane state for '{catalog_id}'") from exc + if not isinstance(data, dict): + raise PolicyGuardError(f"lane state for '{catalog_id}' is invalid") + state = str(data.get("state") or "active") + if state not in STATES: + raise PolicyGuardError(f"lane '{catalog_id}' has unknown state '{state}'") + return LaneState( + catalog_id=str(data.get("catalog_id") or catalog_id), + state=state, + updated_at=str(data.get("updated_at") or ""), + last_operation=str(data.get("last_operation") or ""), + reason=str(data.get("reason") or ""), + ) + + +def save_lane_state( + evidence_dir: Path, + catalog_id: str, + state: str, + *, + operation: str, + reason: str = "", + now: datetime | None = None, +) -> LaneState: + if state not in STATES: + raise PolicyGuardError(f"unknown lane state '{state}'") + cleaned = _clean_reason(reason) + record = LaneState( + catalog_id=catalog_id, + state=state, + updated_at=(now or datetime.now(timezone.utc)).astimezone(timezone.utc).isoformat(), + last_operation=operation, + reason=cleaned, + ) + path = state_path(evidence_dir, catalog_id) + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(yaml.safe_dump(record.as_dict(), sort_keys=True), encoding="utf-8") + return record + + +def _clean_reason(reason: str) -> str: + text = (reason or "").strip() + if not text: + return "" + if len(text) > 200: + raise PolicyGuardError("lane-state reason must be at most 200 characters") + if looks_secret(text) or redact_text(text) != text: + raise PolicyGuardError("lane-state reason must not contain secret-like material") + return text + + +def require_delivery_state(evidence_dir: Path, catalog_id: str, action: str) -> LaneState: + """Refuse exec/wrap/handoff when the lane is not active.""" + current = load_lane_state(evidence_dir, catalog_id) + if current.state in DELIVERY_BLOCKED: + raise DecisionError( + f"lane '{catalog_id}' is {current.state}; " + f"refusing {action} until lifecycle reactivate" + ) + return current + + +def require_provision_state(evidence_dir: Path, catalog_id: str) -> LaneState: + current = load_lane_state(evidence_dir, catalog_id) + if current.state in PROVISION_BLOCKED: + hint = "rotate" if current.state == "compromised" else "lifecycle reactivate" + raise DecisionError( + f"lane '{catalog_id}' is {current.state}; refusing provision; use {hint}" + ) + return current + + +def operation_state(operation: str) -> str | None: + """Return the state persisted after a successful lifecycle operation.""" + return { + "suspend": "suspended", + "deactivate": "deactivated", + "compromise": "compromised", + "reactivate": "active", + "revoke": "deactivated", + }.get(operation) diff --git a/src/secrets_engine/rotate.py b/src/secrets_engine/rotate.py new file mode 100644 index 0000000..c0aac5e --- /dev/null +++ b/src/secrets_engine/rotate.py @@ -0,0 +1,43 @@ +"""Native KV rotation: replace a declared field without touching siblings. + +This updates OpenBao custody only. Catalog ``rotation.owner`` remains the +provider/workload owner; this engine does not roll consumers or revoke a +provider credential. Auth-capability lanes rotate via wrap/handoff, not here. +""" +from __future__ import annotations + +from pathlib import Path + +from secrets_engine.catalog import CatalogEntry +from secrets_engine.errors import ProvisioningError +from secrets_engine.openbao import OpenBaoClient +from secrets_engine.provision import provision_from_file + + +def render_rotate_plan(entry: CatalogEntry, field: str) -> str: + owner = str((entry.rotation or {}).get("owner") or "") + lines = [ + f"Rotate plan for lane '{entry.id}'", + f" field: {field}", + f" target: {entry.mount}/{entry.path}", + f" owner: {owner or '-'}", + " siblings: preserved (CAS patch)", + " workload delivery: not mutated", + ] + return "\n".join(lines) + + +def rotate_from_file( + client: OpenBaoClient, entry: CatalogEntry, field: str, file_path: Path +) -> str: + """Replace one declared KV field. Returns the field name only.""" + if not entry.stores_kv_value(): + raise ProvisioningError( + f"lane '{entry.id}' is {entry.kind}; rotate native KV via this " + "command, or wrap/handoff for auth-capability material" + ) + if field not in entry.fields: + raise ProvisioningError( + f"field '{field}' not declared in lane '{entry.id}' fields {entry.fields}" + ) + return provision_from_file(client, entry, field, file_path) diff --git a/tests/test_lane_state.py b/tests/test_lane_state.py new file mode 100644 index 0000000..b608f28 --- /dev/null +++ b/tests/test_lane_state.py @@ -0,0 +1,91 @@ +import copy +from types import SimpleNamespace + +import pytest +import yaml + +from secrets_engine.catalog import validate_entry +from secrets_engine.config import Config +from secrets_engine.errors import DecisionError, PolicyGuardError +from secrets_engine.lane_state import ( + load_lane_state, + require_delivery_state, + require_provision_state, + save_lane_state, +) +from tests.test_catalog import VALID + + +def _cfg(tmp_path): + return Config( + catalog_dir=tmp_path, + policy_dir=tmp_path, + evidence_dir=tmp_path / "evidence", + hub_url="", + bao_addr="http://127.0.0.1:8200", + topic_id="test-topic", + ) + + +def test_missing_state_is_active(tmp_path): + record = load_lane_state(tmp_path / "evidence", "test-lane") + assert record.state == "active" + assert record.last_operation == "" + + +def test_compromise_blocks_delivery_and_provision(tmp_path): + evidence = tmp_path / "evidence" + save_lane_state(evidence, "test-lane", "compromised", operation="compromise") + with pytest.raises(DecisionError, match="compromised"): + require_delivery_state(evidence, "test-lane", "exec") + with pytest.raises(DecisionError, match="rotate"): + require_provision_state(evidence, "test-lane") + path = evidence / "lane-state" / "test-lane.yaml" + dumped = path.read_text(encoding="utf-8") + assert yaml.safe_load(dumped)["state"] == "compromised" + assert "npm_" not in dumped + + +def test_reason_rejects_secret_like_material(tmp_path): + with pytest.raises(PolicyGuardError, match="secret-like"): + save_lane_state( + tmp_path / "evidence", + "test-lane", + "compromised", + operation="compromise", + reason="npm_abcdefghijklmnop", + ) + + +def test_exec_refuses_compromised_lane(tmp_path, monkeypatch): + from secrets_engine import cli + + entry = validate_entry(copy.deepcopy(VALID)) + save_lane_state(tmp_path / "evidence", entry.id, "compromised", operation="compromise") + monkeypatch.setattr(cli, "get_entry", lambda *_args: entry) + monkeypatch.setattr(cli, "_require_lane_approval", lambda *_args, **_kwargs: None) + monkeypatch.setattr( + cli.OpenBaoClient, + "resolve", + lambda *_args, **_kwargs: pytest.fail("backend must not be reached"), + ) + args = SimpleNamespace( + catalog=entry.id, + field="api_token", + mode="exec-env", + command=["true"], + bootstrap_token_file=None, + auth="auto", + ) + with pytest.raises(DecisionError, match="compromised"): + cli.cmd_exec(_cfg(tmp_path), args) + + +def test_state_show_defaults_active(tmp_path, monkeypatch): + from secrets_engine import cli + + entry = validate_entry(copy.deepcopy(VALID)) + monkeypatch.setattr(cli, "get_entry", lambda *_args: entry) + args = SimpleNamespace(catalog_id=entry.id, json=True) + rc = cli.cmd_state_show(_cfg(tmp_path), args) + assert rc == 0 diff --git a/tests/test_layer_conformance.py b/tests/test_layer_conformance.py index d2226c4..2cf97a9 100644 --- a/tests/test_layer_conformance.py +++ b/tests/test_layer_conformance.py @@ -173,6 +173,8 @@ def test_classify_does_not_grant_permission(): heartbeat = classify("evidence-heartbeat", "prod") session_revoke = classify("session-revoke", "prod") wrap = classify("wrap", "prod") + rotate = classify("rotate", "prod") + compromise = classify("lifecycle-compromise", "prod") assert prod_provision.kind == "load-bearing" assert test_provision.kind == "attributive" assert destroy.kind == "load-bearing" @@ -180,6 +182,8 @@ def test_classify_does_not_grant_permission(): assert heartbeat.kind == "heartbeat" assert session_revoke.kind == "load-bearing" assert wrap.kind == "load-bearing" + assert rotate.kind == "load-bearing" + assert compromise.kind == "load-bearing" assert prod_provision.completeness_claimed is False assert CLASSIFICATION.exists() diff --git a/tests/test_rotate.py b/tests/test_rotate.py new file mode 100644 index 0000000..1c38abf --- /dev/null +++ b/tests/test_rotate.py @@ -0,0 +1,131 @@ +import copy +from types import SimpleNamespace + +import pytest + +from secrets_engine.catalog import validate_entry +from secrets_engine.config import Config +from secrets_engine.errors import DecisionError, ProvisioningError +from secrets_engine.lane_state import load_lane_state, save_lane_state +from secrets_engine.rotate import render_rotate_plan, rotate_from_file +from tests.test_catalog import VALID + + +class RecordingPatchClient: + def __init__(self): + self.calls = [] + + def ensure_kv_mount(self, mount): + self.calls.append(("ensure", mount)) + + def kv_patch_fields(self, mount, path, values): + self.calls.append(("patch", mount, path, list(values))) + return 1 + + +def _cfg(tmp_path): + return Config( + catalog_dir=tmp_path, + policy_dir=tmp_path, + evidence_dir=tmp_path / "evidence", + hub_url="", + bao_addr="http://127.0.0.1:8200", + topic_id="test-topic", + ) + + +def test_rotate_plan_is_non_secret(): + entry = validate_entry(copy.deepcopy(VALID)) + text = render_rotate_plan(entry, "api_token") + assert "api_token" in text + assert "test/team/thing" in text + assert "SUPER" not in text + + +def test_rotate_patches_declared_field_only(tmp_path): + entry = validate_entry(copy.deepcopy(VALID)) + value = tmp_path / "new.value" + value.write_text("replacement-value", encoding="utf-8") + value.chmod(0o600) + client = RecordingPatchClient() + field = rotate_from_file(client, entry, "api_token", value) + assert field == "api_token" + assert client.calls[-1][0] == "patch" + assert client.calls[-1][3] == ["api_token"] + + +def test_rotate_rejects_auth_capability(): + from tests.test_auth_capability import AUTH + + entry = validate_entry(copy.deepcopy(AUTH)) + with pytest.raises(ProvisioningError, match="auth-capability"): + rotate_from_file(object(), entry, "api_token", __import__("pathlib").Path("/tmp/x")) + + +def test_rotate_allowed_when_compromised(tmp_path, monkeypatch): + from secrets_engine import cli + + entry = validate_entry(copy.deepcopy(VALID)) + save_lane_state( + tmp_path / "evidence", entry.id, "compromised", operation="compromise" + ) + monkeypatch.setattr(cli, "get_entry", lambda *_args: entry) + monkeypatch.setattr(cli, "_require_lane_approval", lambda *_args, **_kwargs: None) + monkeypatch.setattr( + "secrets_engine.rotate.rotate_from_file", + lambda *_args, **_kwargs: "api_token", + ) + + class _Client: + pass + + monkeypatch.setattr( + cli, "_open_backend", lambda *_a, **_k: __import__("contextlib").nullcontext(_Client()) + ) + args = SimpleNamespace( + catalog_id=entry.id, + stage=entry.stage, + field="api_token", + from_file="/tmp/new.value", + dry_run=False, + bootstrap_token_file=None, + auth="auto", + ) + rc = cli.cmd_rotate(_cfg(tmp_path), args) + assert rc == 0 + assert load_lane_state(tmp_path / "evidence", entry.id).state == "compromised" + assert load_lane_state(tmp_path / "evidence", entry.id).last_operation == "rotate" + + +def test_production_rotate_fails_closed(tmp_path, monkeypatch): + from secrets_engine import cli + + data = copy.deepcopy(VALID) + data.update(stage="prod", approval={"model": "decision", "decision_ref": "x"}) + entry = validate_entry(data) + monkeypatch.setattr(cli, "get_entry", lambda *_args: entry) + monkeypatch.delenv("SECRETS_ENGINE_UNSAFE_DEMO", raising=False) + monkeypatch.setattr( + cli.OpenBaoClient, + "resolve", + lambda *_args, **_kwargs: pytest.fail("backend must not be reached"), + ) + args = SimpleNamespace( + catalog_id=entry.id, + stage="prod", + field="api_token", + from_file="/tmp/new.value", + dry_run=False, + bootstrap_token_file=None, + auth="auto", + ) + cfg = Config( + catalog_dir=tmp_path, + policy_dir=tmp_path, + evidence_dir=tmp_path / "evidence", + hub_url="http://127.0.0.1:8000", + bao_addr="http://127.0.0.1:8200", + topic_id="test-topic", + ) + with pytest.raises(DecisionError, match="production action 'rotate'"): + cli.cmd_rotate(cfg, args)