"""Non-destructive lifecycle plans for native secrets-engine access. The legacy KV revoke path deleted all KV metadata while leaving delivery auth behind. That is not revocation and is not reversible. Ordinary revoke now means *deactivate native access*: remove only policy/AppRole objects managed by secrets-engine and preserve OpenBao custody plus externally managed workload delivery. """ from __future__ import annotations from dataclasses import dataclass from secrets_engine.catalog import CatalogEntry from secrets_engine.errors import PolicyGuardError from secrets_engine.openbao import OpenBaoClient LIFECYCLE_OPERATIONS = ("suspend", "deactivate", "destroy") @dataclass(frozen=True) class LifecycleAction: kind: str target: str mutation: bool reason: str mount: str = "" path: str = "" def render(self) -> str: mode = "mutation" if self.mutation else "no mutation" return f" [{self.kind}] {self.target} ({mode}; {self.reason})" @dataclass(frozen=True) class LifecyclePlan: catalog_id: str operation: str actions: tuple[LifecycleAction, ...] def render(self) -> str: lines = [ f"Lifecycle plan for lane '{self.catalog_id}'", f" operation: {self.operation}", " actions:", ] lines.extend(action.render() for action in self.actions) return "\n".join(lines) @dataclass(frozen=True) class LifecycleResult: applied: tuple[str, ...] preserved: tuple[str, ...] def render(self) -> str: lines = [f" applied: {target}" for target in self.applied] lines.extend(f" preserved: {target}" for target in self.preserved) return "\n".join(lines) or " (nothing to do)" def build_lifecycle_plan(entry: CatalogEntry, operation: str) -> LifecyclePlan: """Plan an explicit lifecycle operation with no hidden mutations.""" if operation not in LIFECYCLE_OPERATIONS: raise PolicyGuardError( f"unknown lifecycle operation '{operation}'; allowed {LIFECYCLE_OPERATIONS}" ) if operation == "destroy" and not entry.stores_kv_value(): raise PolicyGuardError( f"lane '{entry.id}' is {entry.kind}; it has no KV custody to destroy" ) actions: list[LifecycleAction] = [] if entry.manages_delivery_auth: actions.append( LifecycleAction( "delete-approle", entry.role_name, True, "stop new native delivery logins", ) ) if operation in {"deactivate", "destroy"}: actions.append( LifecycleAction( "delete-policy", entry.policy_name, True, "remove native consumer policy", ) ) else: actions.append( LifecycleAction( "preserve-policy", entry.policy_name, False, "suspension keeps policy for reviewed rollback", ) ) elif entry.has_delivery_auth: actions.extend( ( LifecycleAction( "preserve-external-approle", entry.role_name, False, "delivery auth is externally managed", ), LifecycleAction( "preserve-external-policy", entry.policy_name, False, "delivery auth is externally managed", ), ) ) else: actions.append( LifecycleAction( "no-native-delivery-auth", entry.id, False, "lane declares no native delivery auth", ) ) if entry.stores_kv_value(): if operation == "destroy": actions.append( LifecycleAction( "delete-kv-metadata", f"{entry.mount}/{entry.path}", True, "irreversibly delete all KV versions and metadata", mount=entry.mount, path=entry.path, ) ) else: actions.append( LifecycleAction( "preserve-kv-custody", f"{entry.mount}/{entry.path}", False, "suspend/deactivate never delete KV metadata or values", ) ) if entry.workload_delivery: actions.append( LifecycleAction( "preserve-workload-delivery", entry.id, False, "workload delivery is externally managed", ) ) return LifecyclePlan( catalog_id=entry.id, operation=operation, actions=tuple(actions), ) def build_native_deactivation_plan(entry: CatalogEntry) -> LifecyclePlan: """Compatibility plan for ordinary revoke: safe native deactivation.""" return build_lifecycle_plan(entry, "deactivate") def require_destroy_confirmation(entry: CatalogEntry, confirmation: str) -> None: """Require an exact lane id before an irreversible custody operation.""" if confirmation != entry.id: raise PolicyGuardError( "destroy requires --confirm-destroy with the exact catalog id; " "no custody mutation performed" ) def apply_lifecycle_plan( client: OpenBaoClient, plan: LifecyclePlan ) -> LifecycleResult: """Apply exactly the mutations described by a lifecycle plan.""" applied: list[str] = [] preserved: list[str] = [] for action in plan.actions: if not action.mutation: preserved.append(action.target) continue if action.kind == "delete-approle": client.delete_approle(action.target) elif action.kind == "delete-policy": client.delete_policy(action.target) elif action.kind == "delete-kv-metadata": client.kv_delete_metadata(action.mount, action.path) else: # defensive: a rendered plan must not grow hidden mutations raise ValueError(f"unsupported lifecycle mutation '{action.kind}'") applied.append(action.target) return LifecycleResult(applied=tuple(applied), preserved=tuple(preserved))