agent-credential-broker/agent_credential_broker/backends/openbao.py

"""OpenBao (Vault-compatible) HTTP API backend -- stdlib only.

Talks to the KV v2 engine at ``<openbao_mount>`` over plain
``urllib.request``; there is no ``bao`` CLI dependency.

    OpenBao --(HTTPS API)--> broker --> adapters

Layout (design doc sections 3/4/6):

    <mount>/data/<prefix>/<ref>      secret fields {username, password,
                                     totp, uris, notes, ...}
    <mount>/metadata/<prefix>/<ref>  custom_metadata: agent.* policy
                                     fields + reserved ``label``/``uris``

Key property: the ref index is built exclusively from the *metadata*
endpoints -- ``data/*`` is only ever read by ``get_item`` on demand, so
unlock + list_metadata move zero plaintext secrets through the broker.

Authentication (non-interactive by design -- the ``password`` argument
of ``unlock()`` is IGNORED for this backend; the 0600 credential files
are the secret material, same model as client_id_file/client_secret_file):

    1. AGENT_BROKER_OPENBAO_TOKEN (or the env var named by
       ``openbao_token_env``)  -> verified via auth/token/lookup-self
    2. openbao_token_file (0600) -> verified the same way
    3. openbao_role_id_file + openbao_secret_id_file (0600)
       -> POST /v1/auth/approle/login -> auth.client_token

The token lives only in memory. Renewable tokens are renewed shortly
before expiry; a 401/403 mid-session triggers one transparent AppRole
re-login, then BROKER_LOCKED. lock()/logout() call
auth/token/revoke-self and drop all in-memory material.

Recommended AppRole policy (minimum):

    path "<mount>/data/<prefix>/*"     { capabilities = ["read"] }
    path "<mount>/metadata/<prefix>/*" { capabilities = ["read","list"] }
"""

from __future__ import annotations

import http.client
import json
import os
import ssl
import time
import urllib.error
import urllib.request

from .. import errors
from ..config import read_secret_file
from ..errors import BrokerError, backend_error
from ..items import kv_metadata_projection, normalize_kv_item
from ..sanitize import sanitize_text
from . import VaultBackend

_RENEW_MARGIN_S = 60        # renew a renewable token when <60s remain
_LIST_MAX_DEPTH = 8         # guard for recursive metadata listing


class _ApiError(Exception):
    """Sanitized HTTP-layer failure (status + safe detail only)."""

    def __init__(self, status: int, detail: str):
        super().__init__(f"HTTP {status}: {detail}")
        self.status = status
        self.detail = detail


class OpenBaoBackend(VaultBackend):
    name = "openbao"

    def __init__(self, config):
        super().__init__(config)
        self._addr = config.openbao_addr
        self._mount = config.openbao_mount
        self._prefix = config.openbao_prefix
        self._timeout = config.request_timeout
        self._max_bytes = config.max_response_bytes
        self._token: str | None = None
        self._token_expires_at: float | None = None
        self._renewable = False
        self._meta: list[dict] = []
        # Live secrets, used ONLY to scrub error text before surfacing.
        self._secrets: tuple = ()
        self._ssl_ctx = self._build_ssl_ctx(config)

    # -- plumbing ------------------------------------------------------

    @staticmethod
    def _build_ssl_ctx(config):
        if not config.openbao_addr.startswith("https://"):
            return None
        if config.openbao_tls_skip_verify:
            ctx = ssl.create_default_context()
            ctx.check_hostname = False
            ctx.verify_mode = ssl.CERT_NONE
            return ctx
        return ssl.create_default_context(cafile=config.openbao_ca_cert or None)

    def _headers(self, token: str | None) -> dict:
        h = {"Accept": "application/json"}
        if token:
            h["X-Vault-Token"] = token
        if self.config.openbao_namespace:
            h["X-Vault-Namespace"] = self.config.openbao_namespace
        return h

    def _san(self, text: str) -> str:
        """Error text safe to surface: no token/role_id/secret_id echo."""
        return sanitize_text(text or "", self._secrets).strip()[:300]

    def _request(self, method: str, path: str, *, body=None,
                 token: str | None = None) -> dict:
        """One HTTP call. Returns parsed JSON on 2xx; raises _ApiError
        on HTTP errors (sanitized) and VAULT_BACKEND_ERROR on transport
        or parse failures. ``path`` may carry a query string."""
        url = f"{self._addr}/v1/{path.lstrip('/')}"
        data = None
        headers = self._headers(token)
        if body is not None:
            data = json.dumps(body).encode()
            headers["Content-Type"] = "application/json"
        req = urllib.request.Request(url, data=data, headers=headers,
                                     method=method)
        try:
            with urllib.request.urlopen(req, timeout=self._timeout,
                                        context=self._ssl_ctx) as resp:
                status = resp.status
                raw = resp.read(self._max_bytes + 1)
        except urllib.error.HTTPError as e:
            status = e.code
            try:
                raw = e.read(self._max_bytes + 1)
            except OSError:
                raw = b""
        except (urllib.error.URLError, OSError, ValueError,
                http.client.HTTPException) as e:
            raise backend_error(
                f"openbao request failed: {self._san(str(e)) or 'network error'}")
        if len(raw) > self._max_bytes:
            raise backend_error("openbao response exceeds max_response_bytes")
        if status >= 400:
            raise _ApiError(status, self._san(self._error_detail(raw)))
        if not raw:
            return {}
        try:
            parsed = json.loads(raw)
        except (json.JSONDecodeError, UnicodeDecodeError):
            raise backend_error("unparseable openbao response")
        if not isinstance(parsed, dict):
            raise backend_error("unexpected openbao response shape")
        return parsed

    @staticmethod
    def _error_detail(raw: bytes) -> str:
        """Vault error bodies are {"errors": ["..."]}; fall back to the
        (still sanitized) raw body when the shape differs."""
        try:
            errs = json.loads(raw).get("errors")
            if isinstance(errs, list) and errs:
                return "; ".join(str(x) for x in errs)
        except Exception:
            pass
        return raw.decode("utf-8", "replace")

    def _drop_token(self):
        self._token = None
        self._token_expires_at = None
        self._renewable = False
        self._secrets = ()
        self._meta = []

    # -- auth -----------------------------------------------------------

    def _env_token(self) -> str | None:
        return os.environ.get(self.config.openbao_token_env) or None

    def _approle_available(self) -> bool:
        c = self.config
        return (c.openbao_role_id_file.exists()
                and c.openbao_secret_id_file.exists())

    def configured(self) -> bool:
        c = self.config
        return bool(self._env_token()
                    or c.openbao_token_file.exists()
                    or self._approle_available())

    def _approle_login(self) -> None:
        """POST auth/approle/login. 4xx -> UNLOCK_FAILED (bad/insufficient
        credentials); anything else -> VAULT_BACKEND_ERROR."""
        c = self.config
        role_id = secret_id = None
        try:
            role_id = read_secret_file(c.openbao_role_id_file)
            secret_id = read_secret_file(c.openbao_secret_id_file)
            self._secrets = (role_id, secret_id)
            resp = self._request(
                "POST", "auth/approle/login",
                body={"role_id": role_id, "secret_id": secret_id})
        except _ApiError as e:
            if 400 <= e.status < 500:
                raise BrokerError(
                    "UNLOCK_FAILED",
                    f"openbao approle login failed ({e.status}): {e.detail}")
            raise backend_error(f"openbao approle login failed ({e.status})")
        finally:
            role_id = secret_id = None
        auth = resp.get("auth") or {}
        token = auth.get("client_token")
        if not token:
            self._secrets = ()
            raise backend_error("openbao approle login returned no client_token")
        self._token = token
        self._secrets = (token,)
        self._renewable = bool(auth.get("renewable"))
        lease = auth.get("lease_duration")
        self._token_expires_at = (time.time() + float(lease)) if lease else None

    def _token_file_or_env_login(self) -> bool:
        """Direct token from env var or 0600 file -> verify with
        auth/token/lookup-self. Returns False when no token source exists."""
        c = self.config
        token = self._env_token()
        if not token and c.openbao_token_file.exists():
            token = read_secret_file(c.openbao_token_file)
        if not token:
            return False
        self._secrets = (token,)
        try:
            resp = self._request("GET", "auth/token/lookup-self", token=token)
        except _ApiError as e:
            self._secrets = ()
            if 400 <= e.status < 500:
                raise BrokerError(
                    "UNLOCK_FAILED",
                    f"openbao token rejected ({e.status}): {e.detail}")
            raise backend_error(f"openbao token lookup failed ({e.status})")
        data = resp.get("data") or {}
        self._token = token
        self._renewable = bool(data.get("renewable"))
        ttl = data.get("ttl")
        self._token_expires_at = (time.time() + float(ttl)) if ttl else None
        return True

    def _try_relogin(self) -> bool:
        """One transparent AppRole re-login after a 401/403 (lazy
        re-login, per design). Never raises."""
        if not self._approle_available():
            return False
        try:
            self._approle_login()
            return True
        except BrokerError:
            return False

    def _renew_token(self) -> None:
        """renew-self before expiry. Auth failure drops the token ->
        locked; transient failures keep the token for the next call."""
        try:
            resp = self._request("POST", "auth/token/renew-self",
                                 token=self._token)
        except _ApiError as e:
            if e.status in (401, 403):
                self._drop_token()
                raise errors.locked()
            raise backend_error(f"openbao token renew failed ({e.status})")
        auth = resp.get("auth") or {}
        lease = auth.get("lease_duration")
        if lease:
            self._token_expires_at = time.time() + float(lease)
        self._renewable = bool(auth.get("renewable", self._renewable))

    def _require_token(self) -> str:
        if not self._token:
            raise errors.locked()
        if self._token_expires_at is not None:
            remaining = self._token_expires_at - time.time()
            if remaining <= 0:
                self._drop_token()
                raise errors.locked()
            if self._renewable and remaining < _RENEW_MARGIN_S:
                self._renew_token()
        return self._token

    def _authed(self, method: str, path: str, *, body=None,
                allow_404: bool = False, _retry: bool = True):
        """Authenticated API call with token lifecycle: renews near
        expiry, re-logs once on 401/403 via AppRole, else locks."""
        token = self._require_token()
        try:
            return self._request(method, path, body=body, token=token)
        except _ApiError as e:
            if e.status == 404 and allow_404:
                return None
            if e.status in (401, 403):
                if _retry and self._try_relogin():
                    return self._authed(method, path, body=body,
                                        allow_404=allow_404, _retry=False)
                self._drop_token()
                raise errors.locked()
            raise backend_error(
                f"openbao {method} {path} failed ({e.status}): {e.detail}")

    # -- VaultBackend ----------------------------------------------------

    def unlock(self, password: str) -> None:
        """Non-interactive by design: ``password`` is ignored (documented
        in the module docstring -- the 0600 files are the credential)."""
        self._drop_token()
        if self._token_file_or_env_login():
            pass
        elif self._approle_available():
            self._approle_login()
        else:
            raise errors.unconfigured(
                "openbao: no token file/env and no approle credential files")
        # Prefetch the ref index -- metadata endpoints only, never data/*.
        self._refresh_meta()

    def lock(self) -> None:
        token, self._token = self._token, None
        try:
            if token:
                self._request("POST", "auth/token/revoke-self", token=token)
        except (BrokerError, _ApiError):
            pass  # best-effort revoke; local state is dropped regardless
        finally:
            self._drop_token()

    def logout(self) -> None:
        """End the session exactly like lock(): revoke-self + drop."""
        self.lock()

    def list_metadata(self) -> list[dict]:
        return self._meta

    def _refresh_meta(self) -> None:
        meta = []
        for rel in self._list_keys(""):
            resp = self._authed(
                "GET", f"{self._mount}/metadata/{self._prefix}/{rel}",
                allow_404=True)
            cm = (((resp or {}).get("data") or {}).get("custom_metadata")) or {}
            meta.append(kv_metadata_projection(rel, cm))
        self._meta = meta

    def _list_keys(self, sub: str, _depth: int = 0) -> list[str]:
        """Recursive LIST over metadata/<prefix>/<sub>. Yields item ids
        relative to the prefix (``foo`` / ``sub/foo``). 404 -> empty."""
        if _depth > _LIST_MAX_DEPTH:
            return []
        rel = f"{self._prefix}/{sub}".rstrip("/")
        resp = self._authed("GET", f"{self._mount}/metadata/{rel}?list=true",
                            allow_404=True)
        keys = ((resp or {}).get("data") or {}).get("keys") or []
        out = []
        for k in keys:
            if not isinstance(k, str) or ".." in k:
                continue
            if k.endswith("/"):
                out += self._list_keys(f"{sub}{k}", _depth + 1)
            else:
                out.append(f"{sub}{k}")
        return out

    def get_item(self, item_id: str) -> dict:
        item_id = str(item_id or "")
        if not item_id or ".." in item_id or item_id.startswith("/"):
            raise errors.not_found(item_id)
        resp = self._authed(
            "GET", f"{self._mount}/data/{self._prefix}/{item_id}",
            allow_404=True)
        if resp is None:
            raise errors.not_found(item_id)
        return normalize_kv_item(item_id, resp.get("data") or {})