diff --git a/SCOPE.md b/SCOPE.md index 6ae7ca0..e569da0 100644 --- a/SCOPE.md +++ b/SCOPE.md @@ -110,6 +110,10 @@ cannot be recovered through that AppRole. event on a best-effort basis. Each requested State Hub delivery receives an append-only local `delivered`, `failed`, or `skipped-no-topic` companion record. +- Every live privileged CLI handler records an attempt before lane-approval + resolution and a terminal success, verification failure, rejection, + interruption, or typed backend/input failure. Failure evidence contains the + exception class and approval state, never exception prose. - `audit` summarizes allowlisted local lane evidence: action/result counts, canonical decision references, session cleanup outcomes, and State Hub delivery outcomes. It never re-emits arbitrary evidence detail. @@ -117,7 +121,7 @@ cannot be recovered through that AppRole. State Hub evidence delivery is not queued or transactional; the local receipt makes failure visible but does not replay it. Route and audit do not replace -authorization, OpenBao audit logs, or unrelated-identity denial evidence. +exact-action authorization or OpenBao audit logs. Treat their output as operational guidance, not complete attestation for high-risk lanes. diff --git a/docs/cli.md b/docs/cli.md index 5d6ca65..d19eb05 100644 --- a/docs/cli.md +++ b/docs/cli.md @@ -112,6 +112,13 @@ does not echo arbitrary JSONL detail. State Hub failures are recorded locally as append-only companion receipts; they are visible but are not automatically replayed. +Live `apply`, `provision`, `verify`, `handoff`, `exec`, `revoke`, `suspend`, and +`deactivate` share one evidence guard. It records an attempt before lane approval +is resolved and a terminal outcome on every normal or exceptional exit. Rejected +approval and backend/input failures record only approval state and exception +class, not exception text. This evidence describes what the CLI observed; it is +not an authorization decision and does not replace OpenBao audit logs. + ## Exit codes | Code | Meaning | diff --git a/src/secrets_engine/cli.py b/src/secrets_engine/cli.py index 08e2695..50720cb 100644 --- a/src/secrets_engine/cli.py +++ b/src/secrets_engine/cli.py @@ -30,7 +30,7 @@ from secrets_engine.catalog import get_entry, load_catalog 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 +from secrets_engine.evidence import EvidenceWriter, PrivilegedActionEvidence from secrets_engine.openbao import OpenBaoClient from secrets_engine.plan import build_plan from secrets_engine.provision import provision_from_file, provision_generated @@ -65,6 +65,24 @@ def _writer(cfg: Config) -> EvidenceWriter: return EvidenceWriter(evidence_dir=cfg.evidence_dir, hub_url=cfg.hub_url, topic_id=cfg.topic_id) +def _privileged_evidence( + cfg: Config, + entry, + action: str, + *, + detail: dict[str, object] | None = None, +) -> PrivilegedActionEvidence: + return PrivilegedActionEvidence( + writer=_writer(cfg), + action=action, + catalog_id=entry.id, + stage=entry.stage, + decision_ref=entry.approval.get("decision_ref", ""), + approval_required=entry.approval_required(), + detail=detail or {}, + ) + + def _require_lane_approval(cfg: Config, entry): """Resolve and enforce the lane approval for a privileged live action.""" if not entry.approval_required(): @@ -163,105 +181,153 @@ def cmd_plan(cfg: Config, args) -> int: def cmd_apply(cfg: Config, args) -> int: entry = _resolve_lane_and_decision(cfg, args.ref, args.stage) - decision = None - if entry.approval_required(): - try: - decision = resolve_decision( - hub_url=cfg.hub_url, repo_root=repo_root(), - decision_ref=entry.approval.get("decision_ref", args.ref), - ) - except DecisionError: - if not args.dry_run: - raise - if not args.dry_run: - require_approved(entry, decision) - plan = build_plan(entry, args.stage, decision_id=decision.id if decision else "") - w = _writer(cfg) if args.dry_run: + decision = None + if entry.approval_required(): + try: + decision = resolve_decision( + hub_url=cfg.hub_url, + repo_root=repo_root(), + decision_ref=entry.approval.get("decision_ref", args.ref), + ) + except DecisionError: + decision = None + plan = build_plan( + entry, args.stage, decision_id=decision.id if decision else "" + ) print(plan.render()) print("\n(dry-run: no OpenBao mutation performed)") - w.record("apply", result="dry-run", catalog_id=entry.id, stage=args.stage, - decision_id=decision.id if decision else "") + _writer(cfg).record( + "apply", + result="dry-run", + catalog_id=entry.id, + stage=args.stage, + decision_id=decision.id if decision else "", + ) return 0 - client = OpenBaoClient.resolve(cfg.bao_addr, bootstrap_token_file=args.bootstrap_token_file) - result = apply_plan(client, entry, plan) - print(result.render()) - w.record("apply", result="applied", catalog_id=entry.id, stage=args.stage, - decision_id=decision.id if decision else "", - detail={"applied": result.applied, "skipped": result.skipped}) + + with _privileged_evidence(cfg, entry, "apply") as evidence: + decision = _require_lane_approval(cfg, entry) + evidence.mark_approved(decision) + plan = build_plan( + entry, args.stage, decision_id=decision.id if decision else "" + ) + client = OpenBaoClient.resolve( + cfg.bao_addr, bootstrap_token_file=args.bootstrap_token_file + ) + result = apply_plan(client, entry, plan) + print(result.render()) + evidence.finish( + "applied", + detail={"applied": result.applied, "skipped": result.skipped}, + ) return 0 def cmd_provision(cfg: Config, args) -> int: entry = get_entry(cfg.catalog_dir, args.catalog_id) - 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) - client = OpenBaoClient.resolve(cfg.bao_addr, bootstrap_token_file=args.bootstrap_token_file) field = args.field or (entry.fields[0] if entry.fields else "") - if args.generate: - f = provision_generated(client, entry, field) - mode = "generated" - else: - f = provision_from_file(client, entry, field, Path(args.from_file)) - mode = "from-file" - print(f"provisioned lane '{entry.id}' field '{f}' ({mode}) — value not displayed") - _writer(cfg).record("provision", result=mode, catalog_id=entry.id, stage=entry.stage, - decision_id=decision.id if decision else "", detail={"field": f}) + with _privileged_evidence( + cfg, entry, "provision", 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) + evidence.mark_approved(decision) + client = OpenBaoClient.resolve( + cfg.bao_addr, bootstrap_token_file=args.bootstrap_token_file + ) + if args.generate: + f = provision_generated(client, entry, field) + mode = "generated" + else: + f = provision_from_file(client, entry, field, Path(args.from_file)) + mode = "from-file" + print( + f"provisioned lane '{entry.id}' field '{f}' ({mode}) — value not displayed" + ) + evidence.finish(mode, detail={"field": f}) return 0 def cmd_verify(cfg: Config, args) -> int: entry = get_entry(cfg.catalog_dir, args.catalog_id) - decision = _require_lane_approval(cfg, entry) - client = OpenBaoClient.resolve(cfg.bao_addr, bootstrap_token_file=args.bootstrap_token_file) fields = [args.field] if args.field else list(entry.fields) - if entry.stores_kv_value() and not fields: - from secrets_engine.errors import VerificationError - raise VerificationError(f"lane '{entry.id}' has no field to verify") positive = args.positive or not args.negative negative = args.negative or not args.positive - unrelated_token = None - if entry.stores_kv_value() and negative and args.negative_token_file: - from secrets_engine.openbao import read_strict_token_file - - unrelated_token = read_strict_token_file( - Path(args.negative_token_file), purpose="negative verification token" + with _privileged_evidence( + cfg, + entry, + "verify", + detail={ + "fields": fields, + "positive_requested": positive, + "negative_requested": negative, + }, + ) as evidence: + decision = _require_lane_approval(cfg, entry) + evidence.mark_approved(decision) + client = OpenBaoClient.resolve( + cfg.bao_addr, bootstrap_token_file=args.bootstrap_token_file ) - if entry.stores_kv_value(): - results = [] - if positive: - for field in fields: + if entry.stores_kv_value() and not fields: + from secrets_engine.errors import VerificationError + + raise VerificationError(f"lane '{entry.id}' has no field to verify") + unrelated_token = None + if entry.stores_kv_value() and negative and args.negative_token_file: + from secrets_engine.openbao import read_strict_token_file + + unrelated_token = read_strict_token_file( + Path(args.negative_token_file), + purpose="negative verification token", + ) + if entry.stores_kv_value(): + results = [] + if positive: + for field in fields: + results.extend( + run_verification( + client, entry, field, positive=True, negative=False + ) + ) + if negative: + # Denial is path-scoped; one real unrelated probe covers the path. results.extend( run_verification( - client, entry, field, positive=True, negative=False + client, + entry, + fields[0], + positive=False, + negative=True, + unrelated_token=unrelated_token, ) ) - if negative: - # Denial is path-scoped, so one probe covers every field on this path. - results.extend( - run_verification( - client, - entry, - fields[0], - positive=False, - negative=True, - unrelated_token=unrelated_token, - ) + else: + results = run_verification( + client, entry, "", positive=positive, negative=negative ) - else: - results = run_verification( - client, entry, "", positive=positive, negative=negative + rc = 0 + for result in results: + print(result.render()) + if not result.passed: + rc = 7 + evidence.writer.record( + "verify-check", + result=f"{result.check}:{'pass' if result.passed else 'fail'}", + catalog_id=entry.id, + stage=entry.stage, + decision_id=evidence.decision_id, + detail=result.detail, + ) + evidence.finish( + "pass" if rc == 0 else "verification-failed", + detail={"check_count": len(results)}, ) - rc = 0 - for r in results: - print(r.render()) - if not r.passed: - rc = 7 - _writer(cfg).record("verify", result=f"{r.check}:{'pass' if r.passed else 'fail'}", - catalog_id=entry.id, stage=entry.stage, - decision_id=decision.id if decision else "", detail=r.detail) return rc @@ -270,58 +336,76 @@ def cmd_handoff(cfg: Config, args) -> int: from secrets_engine.handoff import write_approle_handoff entry = get_entry(cfg.catalog_dir, args.catalog_id) - if args.stage != entry.stage: - raise ProvisioningError(f"lane '{entry.id}' is stage '{entry.stage}', not '{args.stage}'") - if entry.kind != "auth-capability": - raise ProvisioningError(f"lane '{entry.id}' is {entry.kind}; handoff needs auth-capability") - decision = _require_lane_approval(cfg, entry) - client = OpenBaoClient.resolve(cfg.bao_addr, bootstrap_token_file=args.bootstrap_token_file) - result = write_approle_handoff( - client, - entry, - role_id_file=Path(args.role_id_file), - secret_id_file=Path(args.secret_id_file), - ) - print(f"wrote AppRole handoff material for lane '{entry.id}' — secret_id not displayed") - print(f" role: {result.role_name}") - print(f" role_id_file: {result.role_id_file}") - print(f" secret_id_file: {result.secret_id_file}") - print(f" token_ttl: {result.token_ttl}") - print(f" secret_id_ttl: {result.secret_id_ttl}") - _writer(cfg).record( - "handoff", - result="secret-id-written", - catalog_id=entry.id, - stage=entry.stage, - decision_id=decision.id if decision else "", - detail={ - "role": result.role_name, - "role_id_file": result.role_id_file, - "secret_id_file": result.secret_id_file, - "token_ttl": result.token_ttl, - "secret_id_ttl": result.secret_id_ttl, - "secret_id_num_uses": result.secret_id_num_uses, - }, - ) + with _privileged_evidence(cfg, entry, "handoff") as evidence: + if args.stage != entry.stage: + raise ProvisioningError( + f"lane '{entry.id}' is stage '{entry.stage}', not '{args.stage}'" + ) + if entry.kind != "auth-capability": + raise ProvisioningError( + f"lane '{entry.id}' is {entry.kind}; handoff needs auth-capability" + ) + decision = _require_lane_approval(cfg, entry) + evidence.mark_approved(decision) + client = OpenBaoClient.resolve( + cfg.bao_addr, bootstrap_token_file=args.bootstrap_token_file + ) + result = write_approle_handoff( + client, + entry, + role_id_file=Path(args.role_id_file), + secret_id_file=Path(args.secret_id_file), + ) + print( + f"wrote AppRole handoff material for lane '{entry.id}' — " + "secret_id not displayed" + ) + print(f" role: {result.role_name}") + print(f" role_id_file: {result.role_id_file}") + print(f" secret_id_file: {result.secret_id_file}") + print(f" token_ttl: {result.token_ttl}") + print(f" secret_id_ttl: {result.secret_id_ttl}") + evidence.finish( + "secret-id-written", + detail={ + "role": result.role_name, + "role_id_file": result.role_id_file, + "secret_id_file": result.secret_id_file, + "token_ttl": result.token_ttl, + "secret_id_ttl": result.secret_id_ttl, + "secret_id_num_uses": result.secret_id_num_uses, + }, + ) return 0 def cmd_exec(cfg: Config, args) -> int: from secrets_engine.exec_delivery import exec_with_secret entry = get_entry(cfg.catalog_dir, args.catalog) - # require approval + readiness before running. - decision = _require_lane_approval(cfg, entry) - if not args.command: - from secrets_engine.errors import DeliveryError - raise DeliveryError("no command after '--'") - client = OpenBaoClient.resolve(cfg.bao_addr, bootstrap_token_file=args.bootstrap_token_file) field = args.field or (entry.fields[0] if entry.fields else "") - w = _writer(cfg) - w.record("exec", result="attempt", catalog_id=entry.id, stage=entry.stage, - decision_id=decision.id if decision else "", - detail={"command": args.command[0], "mode": args.mode}) session_detail: dict[str, object] = {} - try: + command_name = args.command[0] if args.command else "" + with _privileged_evidence( + cfg, + entry, + "exec", + detail={ + "command": command_name, + "mode": args.mode, + "field": field, + "session": session_detail, + }, + ) as evidence: + # require approval + readiness before running. + decision = _require_lane_approval(cfg, entry) + evidence.mark_approved(decision) + if not args.command: + from secrets_engine.errors import DeliveryError + + raise DeliveryError("no command after '--'") + client = OpenBaoClient.resolve( + cfg.bao_addr, bootstrap_token_file=args.bootstrap_token_file + ) rc = exec_with_secret( client, entry, @@ -330,23 +414,7 @@ def cmd_exec(cfg: Config, args) -> int: mode=args.mode, session_evidence=session_detail, ) - except SecretsEngineError as e: - w.record( - "exec", - result=f"failed-{type(e).__name__}", - catalog_id=entry.id, - stage=entry.stage, - decision_id=decision.id if decision else "", - detail={ - "command": args.command[0], - "mode": args.mode, - "session": session_detail, - }, - ) - raise - w.record("exec", result=f"exit-{rc}", catalog_id=entry.id, stage=entry.stage, - decision_id=decision.id if decision else "", - detail={"command": args.command[0], "session": session_detail}) + evidence.finish(f"exit-{rc}") return rc @@ -400,30 +468,29 @@ def cmd_revoke(cfg: Config, args) -> int: ) entry = get_entry(cfg.catalog_dir, args.catalog_id) - decision = None if args.dry_run else _require_lane_approval(cfg, entry) plan = build_native_deactivation_plan(entry) if args.dry_run: print(plan.render()) print("\n(dry-run: no OpenBao mutation performed)") return 0 - client = OpenBaoClient.resolve( - cfg.bao_addr, bootstrap_token_file=args.bootstrap_token_file - ) - result = apply_lifecycle_plan(client, plan) - print(plan.render()) - print(result.render()) - _writer(cfg).record( - "revoke", - result="native-access-deactivated", - catalog_id=entry.id, - stage=entry.stage, - decision_id=decision.id if decision else "", - detail={ - "operation": plan.operation, - "applied": list(result.applied), - "preserved": list(result.preserved), - }, - ) + with _privileged_evidence( + cfg, entry, "revoke", detail={"operation": plan.operation} + ) as evidence: + decision = _require_lane_approval(cfg, entry) + evidence.mark_approved(decision) + client = OpenBaoClient.resolve( + cfg.bao_addr, bootstrap_token_file=args.bootstrap_token_file + ) + result = apply_lifecycle_plan(client, plan) + print(plan.render()) + print(result.render()) + evidence.finish( + "native-access-deactivated", + detail={ + "applied": list(result.applied), + "preserved": list(result.preserved), + }, + ) return 0 @@ -435,41 +502,43 @@ def cmd_lifecycle(cfg: Config, args) -> int: ) entry = get_entry(cfg.catalog_dir, args.catalog_id) - if args.operation == "destroy" and not args.dry_run: - require_destroy_confirmation(entry, args.confirm_destroy) - # A lane-level approval is not an authorization to erase custody. Keep - # the destructive live path closed until T04 supplies a canonical, - # exact-action, expiring and dual-control-capable decision contract. - from secrets_engine.errors import PolicyGuardError - - raise PolicyGuardError( - "live destroy is disabled until an exact-action destruction " - "approval contract is available; use --dry-run to inspect targets" - ) - decision = None if args.dry_run else _require_lane_approval(cfg, entry) plan = build_lifecycle_plan(entry, args.operation) if args.dry_run: print(plan.render()) print("\n(dry-run: no OpenBao mutation performed)") return 0 - client = OpenBaoClient.resolve( - cfg.bao_addr, bootstrap_token_file=args.bootstrap_token_file - ) - result = apply_lifecycle_plan(client, plan) - print(plan.render()) - print(result.render()) - _writer(cfg).record( + with _privileged_evidence( + cfg, + entry, f"lifecycle-{args.operation}", - result="applied", - catalog_id=entry.id, - stage=entry.stage, - decision_id=decision.id if decision else "", - detail={ - "operation": plan.operation, - "applied": list(result.applied), - "preserved": list(result.preserved), - }, - ) + detail={"operation": plan.operation}, + ) as evidence: + if args.operation == "destroy": + require_destroy_confirmation(entry, args.confirm_destroy) + # A lane-level approval is not an authorization to erase custody. + # Keep the destructive live path closed until T04 supplies a + # canonical exact-action and dual-control-capable decision contract. + from secrets_engine.errors import PolicyGuardError + + raise PolicyGuardError( + "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) + evidence.mark_approved(decision) + client = OpenBaoClient.resolve( + cfg.bao_addr, bootstrap_token_file=args.bootstrap_token_file + ) + result = apply_lifecycle_plan(client, plan) + print(plan.render()) + print(result.render()) + evidence.finish( + "applied", + detail={ + "applied": list(result.applied), + "preserved": list(result.preserved), + }, + ) return 0 diff --git a/src/secrets_engine/evidence.py b/src/secrets_engine/evidence.py index eaf7882..5915e31 100644 --- a/src/secrets_engine/evidence.py +++ b/src/secrets_engine/evidence.py @@ -16,6 +16,7 @@ from datetime import datetime, timezone from pathlib import Path from typing import Any +from secrets_engine.errors import DecisionError, SecretsEngineError from secrets_engine.redact import looks_secret, redact_text # Keys that must never carry a value into evidence regardless of nesting. @@ -143,3 +144,88 @@ class EvidenceWriter: except (urllib.error.URLError, OSError, ValueError): # Hub being offline must never block secret work or leak anything. return "failed" + + +@dataclass +class PrivilegedActionEvidence: + """Record one privileged command's attempt, approval, and terminal result.""" + + writer: EvidenceWriter + action: str + catalog_id: str + stage: str + decision_ref: str = "" + approval_required: bool = False + detail: dict[str, Any] = field(default_factory=dict) + decision_id: str = "" + approval_status: str = "pending" + completed: bool = False + + def __post_init__(self) -> None: + if not self.approval_required: + self.approval_status = "not-required" + + def _detail(self, extra: dict[str, Any] | None = None) -> dict[str, Any]: + merged = dict(self.detail) + merged.update( + { + "approval_status": self.approval_status, + "decision_ref": self.decision_ref, + } + ) + if extra: + merged.update(extra) + return merged + + def __enter__(self) -> "PrivilegedActionEvidence": + self.writer.record( + self.action, + result="attempt", + catalog_id=self.catalog_id, + stage=self.stage, + detail=self._detail(), + ) + return self + + def mark_approved(self, decision: object | None) -> None: + if decision is None: + self.approval_status = "not-required" + return + self.decision_id = str(getattr(decision, "id", "")) + self.approval_status = "approved" + + def finish( + self, result: str, *, detail: dict[str, Any] | None = None + ) -> dict[str, Any]: + self.completed = True + return self.writer.record( + self.action, + result=result, + catalog_id=self.catalog_id, + stage=self.stage, + decision_id=self.decision_id, + detail=self._detail(detail), + ) + + def __exit__(self, exc_type, exc, _traceback) -> bool: + if exc_type is None: + if not self.completed: + self.finish("failed-incomplete") + return False + if isinstance(exc, DecisionError): + self.approval_status = "rejected" + if isinstance(exc, (KeyboardInterrupt, SystemExit)): + result = "interrupted" + elif isinstance(exc, SecretsEngineError): + result = f"failed-{type(exc).__name__}" + else: + result = "failed-unexpected" + self.writer.record( + self.action, + result=result, + catalog_id=self.catalog_id, + stage=self.stage, + decision_id=self.decision_id, + detail=self._detail({"error_type": type(exc).__name__}), + ) + return False diff --git a/tests/test_multifield_readiness.py b/tests/test_multifield_readiness.py index 2fa51dd..985ae8b 100644 --- a/tests/test_multifield_readiness.py +++ b/tests/test_multifield_readiness.py @@ -112,7 +112,14 @@ def test_verify_defaults_to_every_declared_field_and_one_path_denial( ("webhook_secret", True, False), ("api_token", False, True), ] - assert len(records) == 3 + assert len(records) == 5 + assert [record[1]["result"] for record in records] == [ + "attempt", + "positive:pass", + "positive:pass", + "negative:pass", + "pass", + ] def test_live_destroy_fails_before_approval_or_backend_until_action_contract( diff --git a/tests/test_privileged_cli_evidence.py b/tests/test_privileged_cli_evidence.py new file mode 100644 index 0000000..12bd626 --- /dev/null +++ b/tests/test_privileged_cli_evidence.py @@ -0,0 +1,98 @@ +import copy +import json +from types import SimpleNamespace + +import pytest + +from secrets_engine import cli +from secrets_engine.catalog import validate_entry +from secrets_engine.config import Config +from secrets_engine.errors import BackendError, DecisionError +from tests.test_catalog import VALID + + +def _config(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 _records(tmp_path): + path = next((tmp_path / "evidence").glob("evidence-*.jsonl")) + return [json.loads(line) for line in path.read_text().splitlines()] + + +def _provision_args(entry): + return SimpleNamespace( + catalog_id=entry.id, + stage=entry.stage, + field="api_token", + generate=False, + from_file="/tmp/test-value-file", + bootstrap_token_file=None, + ) + + +def test_provision_backend_exception_has_attempt_and_terminal_evidence( + tmp_path, monkeypatch +): + entry = validate_entry(copy.deepcopy(VALID)) + monkeypatch.setattr(cli, "get_entry", lambda *_args: entry) + monkeypatch.setattr(cli, "_require_lane_approval", lambda *_args: None) + monkeypatch.setattr(cli.OpenBaoClient, "resolve", lambda *_args, **_kwargs: object()) + monkeypatch.setattr( + cli, + "provision_from_file", + lambda *_args, **_kwargs: (_ for _ in ()).throw( + BackendError("fake-SUPER-SECRET-backend-message") + ), + ) + + with pytest.raises(BackendError): + cli.cmd_provision(_config(tmp_path), _provision_args(entry)) + + records = _records(tmp_path) + assert [record["result"] for record in records] == [ + "attempt", + "failed-BackendError", + ] + assert records[-1]["detail"]["approval_status"] == "not-required" + assert "fake-SUPER-SECRET" not in json.dumps(records) + + +def test_provision_decision_rejection_is_recorded_before_backend( + tmp_path, monkeypatch +): + data = copy.deepcopy(VALID) + data["approval"] = { + "model": "decision", + "decision_ref": "CCR-2026-0001", + } + entry = validate_entry(data) + monkeypatch.setattr(cli, "get_entry", lambda *_args: entry) + monkeypatch.setattr( + cli, + "_require_lane_approval", + lambda *_args: (_ for _ in ()).throw(DecisionError("not approved")), + ) + monkeypatch.setattr( + cli.OpenBaoClient, + "resolve", + lambda *_args, **_kwargs: pytest.fail("backend must not be reached"), + ) + + with pytest.raises(DecisionError): + cli.cmd_provision(_config(tmp_path), _provision_args(entry)) + + records = _records(tmp_path) + assert [record["result"] for record in records] == [ + "attempt", + "failed-DecisionError", + ] + assert records[-1]["detail"]["approval_status"] == "rejected" + assert records[-1]["detail"]["decision_ref"] == "CCR-2026-0001" diff --git a/tests/test_redact_evidence.py b/tests/test_redact_evidence.py index 0150c22..19132d9 100644 --- a/tests/test_redact_evidence.py +++ b/tests/test_redact_evidence.py @@ -2,7 +2,10 @@ import json import urllib.error from types import SimpleNamespace -from secrets_engine.evidence import EvidenceWriter, _scrub +import pytest + +from secrets_engine.evidence import EvidenceWriter, PrivilegedActionEvidence, _scrub +from secrets_engine.errors import BackendError, DecisionError from secrets_engine.redact import looks_secret, redact_text @@ -86,3 +89,53 @@ def test_evidence_records_hub_failure_without_raising(tmp_path, monkeypatch): ] assert lines[-1]["action"] == "evidence-delivery" assert lines[-1]["result"] == "failed" + + +def test_privileged_evidence_records_approval_and_backend_failure(tmp_path): + writer = EvidenceWriter(evidence_dir=tmp_path, hub_url="") + evidence = PrivilegedActionEvidence( + writer, + "provision", + "lane", + "prod", + decision_ref="CCR-2026-0001", + approval_required=True, + ) + + with pytest.raises(BackendError, match="fake backend failure"): + with evidence: + evidence.mark_approved(SimpleNamespace(id="e6381a56-3e55-4fac-b22c-63ee1c152ce8")) + raise BackendError("fake backend failure") + + lines = [ + json.loads(line) + for line in next(tmp_path.glob("evidence-*.jsonl")).read_text().splitlines() + ] + assert [line["result"] for line in lines] == ["attempt", "failed-BackendError"] + assert lines[-1]["detail"]["approval_status"] == "approved" + assert lines[-1]["detail"]["error_type"] == "BackendError" + assert "fake backend failure" not in json.dumps(lines) + + +def test_privileged_evidence_records_rejected_decision_without_message(tmp_path): + writer = EvidenceWriter(evidence_dir=tmp_path, hub_url="") + evidence = PrivilegedActionEvidence( + writer, + "apply", + "lane", + "prod", + decision_ref="CCR-2026-0001", + approval_required=True, + ) + + with pytest.raises(DecisionError): + with evidence: + raise DecisionError("operator prose must not enter evidence") + + lines = [ + json.loads(line) + for line in next(tmp_path.glob("evidence-*.jsonl")).read_text().splitlines() + ] + assert lines[-1]["result"] == "failed-DecisionError" + assert lines[-1]["detail"]["approval_status"] == "rejected" + assert "operator prose" not in json.dumps(lines) diff --git a/workplans/SECRETS-WP-0007-production-lifecycle-hardening.md b/workplans/SECRETS-WP-0007-production-lifecycle-hardening.md index 1dee317..8d9a50b 100644 --- a/workplans/SECRETS-WP-0007-production-lifecycle-hardening.md +++ b/workplans/SECRETS-WP-0007-production-lifecycle-hardening.md @@ -298,9 +298,13 @@ unrelated-identity input is now enforced as a strict mode-0600 out-of-repo token file; absence fails closed, and throwaway OpenBao integration deliberately adds an overlapping unrelated read policy and proves the check fails. Production identity selection/ownership, audit request correlation, queued/replayed State -Hub delivery, and complete privileged failure-path evidence remain outstanding. +Hub delivery, and exact-action authorization remain outstanding. All live +privileged handlers now share one attempt/terminal evidence guard: approval +rejection, backend/input exceptions, interruption, verification failure, and +success are recorded without exception prose. Tests prove decision and backend +failures stop before inappropriate backend work and exclude fake secret text. -The complete repository suite passes with 111 tests after these changes, +The complete repository suite passes with 115 tests after these changes, including throwaway OpenBao integration coverage. Make verification and routing truthful for multi-field and high-risk lanes: