"""Named engine OpenBao authentication. No implicit fallback. Steady-state is the reviewed KeyCape service identity plus a platform-owned OpenBao JWT login. Bootstrap token files and ``BAO_TOKEN`` remain explicit providers. A service-jwt failure never reads those providers. """ from __future__ import annotations import re import shutil from dataclasses import dataclass from pathlib import Path from typing import Any import yaml from secrets_engine.errors import BackendError from secrets_engine.openbao import OpenBaoClient, ScopedTokenSession from secrets_engine.service_auth import KeyCapeServiceAuthConfig, KeyCapeServiceAuthProvider _NAME_RE = re.compile(r"^[A-Za-z0-9._-]+$") PROVIDERS = ("auto", "service-jwt", "bootstrap", "env") @dataclass(frozen=True) class JwtLoginContract: """Non-secret OpenBao JWT login coordinates published by the platform owner.""" mount: str role: str bound_issuer: str path: Path @dataclass(frozen=True) class AuthSelection: provider: str bootstrap_token_file: str | Path | None = None break_glass: bool = False def _optional_path(value: object) -> Path | None: if value in (None, ""): return None return Path(str(value)) def jwt_login_contract_path(cfg: Any) -> Path | None: return _optional_path(getattr(cfg, "openbao_jwt_login_file", None)) def load_jwt_login_contract(cfg: Any) -> JwtLoginContract: path = jwt_login_contract_path(cfg) if path is None: raise BackendError( "service-jwt requires SECRETS_ENGINE_OPENBAO_JWT_LOGIN; " "the platform JWT mount/role contract is not published" ) if not path.is_file(): raise BackendError( "service-jwt OpenBao JWT login contract file is missing" ) try: data = yaml.safe_load(path.read_text(encoding="utf-8")) or {} except (OSError, yaml.YAMLError) as exc: raise BackendError("unable to load OpenBao JWT login contract") from exc if not isinstance(data, dict): raise BackendError("OpenBao JWT login contract must be a mapping") mount = str(data.get("mount") or "") role = str(data.get("role") or "") issuer = str(data.get("bound_issuer") or "") if not _NAME_RE.fullmatch(mount) or not _NAME_RE.fullmatch(role): raise BackendError("OpenBao JWT login mount/role is invalid") if not issuer.startswith("https://"): raise BackendError("OpenBao JWT login bound_issuer must use HTTPS") return JwtLoginContract(mount=mount, role=role, bound_issuer=issuer, path=path) def keycape_config(cfg: Any) -> KeyCapeServiceAuthConfig: token_url = str(getattr(cfg, "keycape_token_url", "") or "") issuer = str(getattr(cfg, "keycape_issuer", "") or "") secret = _optional_path(getattr(cfg, "keycape_client_secret_file", None)) if not token_url or not issuer or secret is None: raise BackendError( "service-jwt requires KeyCape token URL, issuer, and client-secret file" ) return KeyCapeServiceAuthConfig( token_url=token_url, issuer=issuer, client_secret_file=secret, ) def jwt_contract_configured(cfg: Any) -> bool: path = jwt_login_contract_path(cfg) return path is not None def select_engine_auth(cfg: Any, args: Any) -> AuthSelection: """Choose exactly one provider. Never chain JWT failure into bootstrap/env.""" requested = str(getattr(args, "auth", "auto") or "auto") if requested not in PROVIDERS: raise BackendError(f"unknown engine auth provider '{requested}'") bootstrap = getattr(args, "bootstrap_token_file", None) jwt_intended = jwt_contract_configured(cfg) if requested == "service-jwt" or (requested == "auto" and jwt_intended): if bootstrap: raise BackendError( "service-jwt does not accept --bootstrap-token-file; no fallback" ) return AuthSelection(provider="service-jwt") if requested == "bootstrap" or bootstrap: if requested == "env": raise BackendError("env auth does not use --bootstrap-token-file") if not bootstrap: raise BackendError("bootstrap auth requires --bootstrap-token-file") return AuthSelection( provider="bootstrap", bootstrap_token_file=bootstrap, break_glass=True, ) return AuthSelection(provider="env") def login_service_jwt(cfg: Any) -> ScopedTokenSession: """Mint a short-lived OpenBao token from KeyCape. No parent token, no fallback.""" contract = load_jwt_login_contract(cfg) kcfg = keycape_config(cfg) if contract.bound_issuer != kcfg.issuer: raise BackendError("OpenBao JWT login issuer does not match KeyCape issuer") service_jwt = KeyCapeServiceAuthProvider(kcfg).exchange() bao_bin = shutil.which("bao") or shutil.which("vault") or "" anon = OpenBaoClient(addr=cfg.bao_addr, token="", bao_bin=bao_bin) try: return anon.login_jwt(contract.mount, contract.role, service_jwt.token) finally: # Drop the KeyCape JWT from this frame; OpenBao verifies the signature. del service_jwt