secrets-engine/src/secrets_engine/openbao.py

"""OpenBao backend adapter.

Thin wrapper over the `bao` CLI. Isolated here so the rest of the engine speaks
in lanes/plans, not in OpenBao endpoint quirks (FR: "isolate backend adapter").

Named engine auth providers (no implicit fallback) live in ``engine_auth``.
``OpenBaoClient.resolve`` only materializes an already-selected bootstrap file
or environment token.

This adapter NEVER returns a secret value to its callers except through the
narrow `read_field_present()` (boolean) and the exec-delivery path, which writes
straight into a child process and never logs.
"""
from __future__ import annotations

import hashlib
import json
import os
import re
import shutil
import stat
import subprocess
import tempfile
from contextlib import contextmanager
from dataclasses import dataclass, field
from pathlib import Path

from secrets_engine.errors import BackendError, ProvisioningError
from secrets_engine.safe_paths import containing_git_worktree


def read_strict_token_file(path: Path, *, purpose: str = "token") -> str:
    """Read token material only from a mode-0600 path outside Git worktrees."""
    if not path.exists():
        raise ProvisioningError(f"{purpose} file not found: {path}")
    st = path.stat()
    if st.st_mode & 0o077:
        raise ProvisioningError(
            f"{purpose} file {path} is group/other-accessible "
            f"(mode {oct(st.st_mode & 0o777)}); must be 0600"
        )
    # Refuse a token file living inside a Git worktree.
    worktree = containing_git_worktree(path)
    if worktree is not None:
        raise ProvisioningError(
            f"{purpose} file {path} is inside a Git worktree ({worktree}); "
            "store it outside any repo"
        )
    token = path.read_text(encoding="utf-8").strip()
    if not token:
        raise ProvisioningError(f"{purpose} file {path} is empty")
    return token


def _check_token_file(path: Path) -> str:
    """Compatibility wrapper for bootstrap authentication input."""
    return read_strict_token_file(path, purpose="bootstrap token")


def accessor_fingerprint(accessor: str) -> str:
    """Non-secret handle for an accessor the caller already holds."""
    return hashlib.sha256(accessor.encode("utf-8")).hexdigest()[:12]


@dataclass(frozen=True)
class WrappedResponse:
    """One OpenBao response-wrapped payload. The wrap token is secret."""

    wrap_token: str = field(repr=False)
    accessor_fingerprint: str
    ttl: str
    creation_path: str

    def evidence(self) -> dict[str, object]:
        payload: dict[str, object] = {
            "wrap_handle": self.accessor_fingerprint,
            "wrap_ttl": self.ttl,
            "creation_path": self.creation_path,
        }
        return payload


@dataclass
class ScopedTokenSession:
    """One AppRole login token that revokes itself on close."""

    client: "OpenBaoClient"
    accessor_fingerprint: str
    closed: bool = False
    revocation_attempted: bool = False
    revocation_succeeded: bool = False
    # Value-free evidence of the in-memory client-secret read, if any.
    source_evidence: dict[str, object] = field(default_factory=dict)

    def evidence(self) -> dict[str, object]:
        """Return non-secret lifecycle evidence for this issued session."""
        return {
            **self.source_evidence,
            "session_handle": self.accessor_fingerprint,
            "established": True,
            "revocation_attempted": self.revocation_attempted,
            "revocation_succeeded": self.revocation_succeeded,
        }

    def close(self) -> None:
        if self.closed:
            return
        self.revocation_attempted = True
        try:
            self.client._run_ok(["token", "revoke", "-self"])
            self.revocation_succeeded = True
        finally:
            # Drop the credential from the reusable client object even when
            # backend cleanup fails; TTL/use limits remain the final backstop.
            self.client.token = ""
            self.closed = True


@contextmanager
def _unchanged(env: dict[str, str]):
    yield env


@contextmanager
def _no_ambient_token(env: dict[str, str]):
    """Strip env tokens and point the CLI at a token helper that returns none."""
    helper = shutil.which("true")
    if not helper:
        raise BackendError("cannot disable the bao token helper; refusing login")
    run_env = {
        k: v for k, v in env.items() if k not in ("BAO_TOKEN", "VAULT_TOKEN")
    }
    with tempfile.TemporaryDirectory(prefix="se-bao-anon-") as tmp:
        config = Path(tmp) / "cli.hcl"
        config.write_text(f'token_helper = "{helper}"\n', encoding="utf-8")
        run_env["BAO_CONFIG_PATH"] = str(config)
        run_env["VAULT_CONFIG_PATH"] = str(config)
        yield run_env


@dataclass
class OpenBaoClient:
    addr: str
    token: str = ""
    bao_bin: str = ""

    @classmethod
    def resolve(
        cls, addr: str, *, bootstrap_token_file: str | Path | None = None
    ) -> "OpenBaoClient":
        token = ""
        if bootstrap_token_file:
            token = _check_token_file(Path(bootstrap_token_file))
        else:
            token = os.environ.get("BAO_TOKEN", os.environ.get("VAULT_TOKEN", ""))
        bao_bin = shutil.which("bao") or shutil.which("vault") or ""
        return cls(addr=addr, token=token, bao_bin=bao_bin)

    # -- low level ---------------------------------------------------------

    def _run(
        self, args: list[str], *, stdin: str | None = None, anonymous: bool = False
    ) -> subprocess.CompletedProcess:
        if not self.bao_bin:
            raise BackendError(
                "no 'bao' (or 'vault') CLI on PATH; cannot reach OpenBao backend"
            )
        env = dict(os.environ)
        env["BAO_ADDR"] = self.addr
        env["VAULT_ADDR"] = self.addr
        if self.token:
            env["BAO_TOKEN"] = self.token
            env["VAULT_TOKEN"] = self.token
        with _no_ambient_token(env) if anonymous else _unchanged(env) as run_env:
            try:
                return subprocess.run(
                    [self.bao_bin, *args],
                    input=stdin,
                    env=run_env,
                    capture_output=True,
                    text=True,
                    timeout=30,
                )
            except FileNotFoundError as e:
                raise BackendError(f"backend binary not runnable: {e}") from e
            except subprocess.TimeoutExpired as e:
                raise BackendError(f"backend call timed out: {' '.join(args)}") from e

    def _run_ok(
        self, args: list[str], *, stdin: str | None = None, anonymous: bool = False
    ) -> str:
        proc = self._run(args, stdin=stdin, anonymous=anonymous)
        if proc.returncode != 0:
            # stderr from bao does not contain the secret value for these calls.
            raise BackendError(
                f"bao {' '.join(args[:2])} failed (exit {proc.returncode}): "
                f"{proc.stderr.strip() or proc.stdout.strip()}"
            )
        return proc.stdout

    def _run_ok_with_json_file(
        self, args: list[str], payload: dict, *, anonymous: bool = False
    ) -> str:
        """Run a CLI operation with JSON from a strict temporary file reference."""
        fd, temp_name = tempfile.mkstemp(prefix="se-bao-input-", suffix=".json")
        temp_path = Path(temp_name)
        try:
            os.fchmod(fd, 0o600)
            with os.fdopen(fd, "w", encoding="utf-8") as fh:
                fd = -1
                json.dump(payload, fh)
                fh.write("\n")
            if anonymous:
                return self._run_ok([*args, f"@{temp_path}"], anonymous=True)
            return self._run_ok([*args, f"@{temp_path}"])
        finally:
            if fd >= 0:
                os.close(fd)
            try:
                temp_path.unlink()
            except FileNotFoundError:
                pass

    # -- health / capabilities --------------------------------------------

    def is_reachable(self) -> bool:
        if not self.bao_bin:
            return False
        proc = self._run(["status", "-format=json"])
        # status returns non-zero when sealed but still reachable; treat any
        # parseable JSON as reachable.
        try:
            json.loads(proc.stdout or "{}")
            return True
        except json.JSONDecodeError:
            return proc.returncode == 0

    # -- policies ----------------------------------------------------------

    def write_policy(self, name: str, hcl: str) -> None:
        self._run_ok(["policy", "write", name, "-"], stdin=hcl)

    def read_policy(self, name: str) -> str | None:
        proc = self._run(["policy", "read", name])
        if proc.returncode != 0:
            return None
        return proc.stdout

    # -- approle -----------------------------------------------------------

    def ensure_approle_enabled(self) -> None:
        proc = self._run(["auth", "list", "-format=json"])
        if proc.returncode == 0:
            try:
                methods = json.loads(proc.stdout)
                if "approle/" in methods:
                    return
            except json.JSONDecodeError:
                pass
        enable = self._run(["auth", "enable", "approle"])
        if enable.returncode == 0:
            return
        stderr = enable.stderr or ""
        if "already in use" in stderr:
            return
        if "permission denied" in stderr.lower():
            return
        raise BackendError(f"could not enable approle: {stderr.strip()}")

    def write_approle(
        self,
        role_name: str,
        policies: list[str],
        ttl: str = "30m",
        *,
        max_ttl: str | None = None,
        secret_id_ttl: str | None = None,
        secret_id_num_uses: int = 0,
        token_num_uses: int = 0,
    ) -> None:
        args = [
            "write",
            f"auth/approle/role/{role_name}",
            f"token_policies={','.join(policies)}",
            f"token_ttl={ttl}",
            f"token_max_ttl={max_ttl or ttl}",
            f"secret_id_num_uses={secret_id_num_uses}",
            f"token_num_uses={token_num_uses}",
        ]
        if secret_id_ttl:
            args.append(f"secret_id_ttl={secret_id_ttl}")
        self._run_ok(args)

    def approle_exists(self, role_name: str) -> bool:
        proc = self._run(["read", f"auth/approle/role/{role_name}"])
        return proc.returncode == 0

    def read_approle_role_id(self, role_name: str) -> str:
        return self._run_ok(
            ["read", "-field=role_id", f"auth/approle/role/{role_name}/role-id"]
        ).strip()

    def create_approle_secret_id(self, role_name: str) -> str:
        return self._run_ok(
            ["write", "-field=secret_id", "-f", f"auth/approle/role/{role_name}/secret-id"]
        ).strip()

    def create_approle_session(self, role_name: str) -> ScopedTokenSession:
        """Login through AppRole without putting role/secret ids in argv."""
        role_id = self.read_approle_role_id(role_name)
        secret_id = self.create_approle_secret_id(role_name)
        try:
            output = self._run_ok_with_json_file(
                ["write", "-format=json", "auth/approle/login"],
                {"role_id": role_id, "secret_id": secret_id},
            )
        finally:
            del secret_id
        try:
            auth = json.loads(output)["auth"]
            token = str(auth["client_token"])
            accessor = str(auth.get("accessor", ""))
        except (json.JSONDecodeError, KeyError, TypeError) as e:
            raise BackendError("malformed AppRole login response") from e
        if not token:
            raise BackendError("AppRole login returned an empty token")
        fingerprint = accessor_fingerprint(accessor)
        scoped = OpenBaoClient(addr=self.addr, token=token, bao_bin=self.bao_bin)
        del token
        return ScopedTokenSession(
            client=scoped,
            accessor_fingerprint=fingerprint,
        )

    @contextmanager
    def approle_session(self, role_name: str):
        """Yield a scoped client and revoke its token on every exit path."""
        session = self.create_approle_session(role_name)
        try:
            yield session
        finally:
            session.close()

    def token_capabilities(self, path: str, *, token: str) -> list[str]:
        """Return token capabilities for a path without returning any secret value."""
        client = OpenBaoClient(addr=self.addr, token=token, bao_bin=self.bao_bin)
        out = client._run_ok(["token", "capabilities", "-format=json", path])
        try:
            data = json.loads(out)
        except json.JSONDecodeError:
            return [line.strip() for line in out.splitlines() if line.strip()]
        if isinstance(data, list):
            return [str(item) for item in data]
        if isinstance(data, dict):
            caps = data.get("capabilities", [])
            if isinstance(caps, list):
                return [str(item) for item in caps]
        return []

    def delete_policy(self, name: str) -> None:
        self._run_ok(["policy", "delete", name])

    def delete_approle(self, role_name: str) -> None:
        self._run_ok(["delete", f"auth/approle/role/{role_name}"])

    def login_jwt(self, mount: str, role: str, jwt: str) -> ScopedTokenSession:
        """Login through JWT auth without putting the JWT in argv.

        The login carries no ambient token: neither BAO_TOKEN/VAULT_TOKEN nor a
        CLI token helper is consulted, so a service-jwt login never presents an
        env or bootstrap credential alongside the JWT.
        """
        output = self._run_ok_with_json_file(
            ["write", "-format=json", f"auth/{mount}/login"],
            {"role": role, "jwt": jwt},
            anonymous=True,
        )
        try:
            auth = json.loads(output)["auth"]
            token = str(auth["client_token"])
            accessor = str(auth.get("accessor", ""))
        except (json.JSONDecodeError, KeyError, TypeError) as e:
            raise BackendError("malformed JWT login response") from e
        if not token:
            raise BackendError("JWT login returned an empty token")
        fingerprint = accessor_fingerprint(accessor)
        scoped = OpenBaoClient(addr=self.addr, token=token, bao_bin=self.bao_bin)
        del token
        return ScopedTokenSession(client=scoped, accessor_fingerprint=fingerprint)

    def revoke_accessor(self, accessor: str) -> None:
        """Revoke a token the operator already holds, by accessor only."""
        if not accessor or any(ch.isspace() for ch in accessor):
            raise BackendError("token accessor is missing or invalid")
        self._run_ok(["token", "revoke", "-accessor", accessor])

    def revoke_lease(self, lease_id: str) -> None:
        """Revoke a lease the operator already holds, by lease id only."""
        if not lease_id or any(ch.isspace() for ch in lease_id):
            raise BackendError("lease id is missing or invalid")
        self._run_ok(["lease", "revoke", lease_id])

    def _parse_wrap_response(self, stdout: str, *, ttl: str, creation_path: str) -> WrappedResponse:
        # Verify native wrapping metadata. Never reconstruct a purported source
        # path or TTL from the caller when the backend omitted or changed it.
        try:
            payload = json.loads(stdout)
        except (json.JSONDecodeError, TypeError):
            raise BackendError("wrapped response is not JSON") from None
        if not isinstance(payload, dict):
            raise BackendError("wrapped response is invalid")
        wrap = payload.get("wrap_info")
        if not isinstance(wrap, dict) or not wrap.get("token"):
            raise BackendError("response wrapping was not applied")
        if payload.get("data") or payload.get("auth"):
            raise BackendError("wrapped response unexpectedly included unwrapped material")
        match = re.fullmatch(r"([1-9][0-9]*)([smh])", ttl)
        seconds = int(match[1]) * {"s": 1, "m": 60, "h": 3600}[match[2]] if match else 0
        if not 0 < seconds <= 900:
            raise BackendError("wrap TTL is outside the supported bound")
        token, accessor = wrap.get("token"), wrap.get("accessor")
        if (not isinstance(token, str) or not token or any(ch.isspace() for ch in token)
                or not isinstance(accessor, str) or not accessor
                or type(wrap.get("ttl")) is not int or wrap["ttl"] != seconds
                or wrap.get("creation_path") != creation_path):
            raise BackendError("native wrap identity, TTL or creation path unverified")
        return WrappedResponse(
            wrap_token=token,
            accessor_fingerprint=accessor_fingerprint(accessor),
            ttl=str(seconds),
            creation_path=wrap["creation_path"],
        )

    def wrap_kv_get(self, mount: str, path: str, *, ttl: str) -> WrappedResponse:
        """Wrap a KV read. Fail if OpenBao returns the secret unwrapped."""
        target = f"{mount}/{path}"
        proc = self._run(["kv", "get", f"-wrap-ttl={ttl}", "-format=json", target])
        if proc.returncode != 0:
            raise BackendError("wrapped KV read failed")
        return self._parse_wrap_response(proc.stdout, ttl=ttl, creation_path=f"{mount}/data/{path}")

    def wrap_approle_secret_id(self, role_name: str, *, ttl: str) -> WrappedResponse:
        """Wrap a single-use AppRole secret_id. Fail if returned unwrapped."""
        target = f"auth/approle/role/{role_name}/secret-id"
        proc = self._run(
            ["write", f"-wrap-ttl={ttl}", "-format=json", "-force", target]
        )
        if proc.returncode != 0:
            raise BackendError("wrapped AppRole secret-id mint failed")
        return self._parse_wrap_response(proc.stdout, ttl=ttl, creation_path=target)

    # -- KV v2 -------------------------------------------------------------

    def kv_mount_exists(self, mount: str) -> bool:
        proc = self._run(["secrets", "list", "-format=json"])
        if proc.returncode != 0:
            return False
        try:
            return f"{mount}/" in json.loads(proc.stdout)
        except json.JSONDecodeError:
            return False

    def ensure_kv_mount(self, mount: str) -> None:
        if self.kv_mount_exists(mount):
            return
        enable = self._run(["secrets", "enable", "-path", mount, "kv-v2"])
        if enable.returncode == 0:
            return
        stderr = enable.stderr or ""
        if "already in use" in stderr:
            return
        # Stage roles (e.g. secrets-engine-prod) cannot manage sys/mounts. When the
        # caller also cannot list mounts, a permission-denied enable means the mount
        # was operator-preprovisioned — continue to policy/approle apply.
        if "permission denied" in stderr.lower():
            return
        raise BackendError(f"could not enable kv at {mount}: {stderr.strip()}")

    def kv_current_version(self, mount: str, path: str) -> int:
        """Return current KV v2 version from metadata, or 0 when absent.

        Metadata contains no secret values. Permission/backend errors fail closed
        rather than being misread as a new path.
        """
        proc = self._run(
            ["kv", "metadata", "get", "-format=json", f"-mount={mount}", path]
        )
        if proc.returncode != 0:
            message = f"{proc.stderr}\n{proc.stdout}".lower()
            if "no value found" in message or "not found" in message or "404" in message:
                return 0
            raise BackendError(
                f"bao kv metadata get failed (exit {proc.returncode}): "
                f"{proc.stderr.strip() or proc.stdout.strip()}"
            )
        try:
            version = json.loads(proc.stdout).get("data", {}).get("current_version", 0)
            return int(version)
        except (json.JSONDecodeError, TypeError, ValueError) as e:
            raise BackendError("malformed KV metadata response") from e

    def kv_patch_fields(
        self,
        mount: str,
        path: str,
        values: dict[str, str],
        *,
        expected_version: int | None = None,
    ) -> int:
        """Atomically update declared fields without exposing values in argv.

        Existing paths use server-side HTTP PATCH with CAS, preserving every
        unmentioned sibling. New paths use CAS=0 put. Values are written to a
        mode-0600 temporary JSON input file referenced by path and removed in a
        finally block. The returned integer is the version used as the CAS base.
        """
        if not values:
            raise BackendError("refusing empty KV field update")
        if any(not isinstance(k, str) or not k for k in values):
            raise BackendError("KV field update contains an invalid field name")
        version = (
            self.kv_current_version(mount, path)
            if expected_version is None
            else expected_version
        )
        if version < 0:
            raise BackendError("expected KV version must be non-negative")

        if version == 0:
            args = [
                "kv",
                "put",
                f"-mount={mount}",
                "-cas=0",
                path,
            ]
        else:
            args = [
                "kv",
                "patch",
                f"-mount={mount}",
                "-method=patch",
                f"-cas={version}",
                path,
            ]
        self._run_ok_with_json_file(args, values)
        return version

    def kv_metadata_exists(self, mount: str, path: str) -> bool:
        proc = self._run(["kv", "metadata", "get", "-format=json", f"{mount}/{path}"])
        return proc.returncode == 0

    def kv_fields_present(
        self,
        mount: str,
        path: str,
        fields: list[str] | tuple[str, ...],
        *,
        token: str | None = None,
    ) -> dict[str, bool]:
        """Return declared-field presence after one read, never field values.

        If ``token`` is given, the read is attempted as that scoped token. A
        denied or malformed read reports every requested field as absent; no
        response data is returned to the caller.
        """
        requested = tuple(dict.fromkeys(fields))
        if not requested:
            return {}
        client = self
        if token is not None:
            client = OpenBaoClient(addr=self.addr, token=token, bao_bin=self.bao_bin)
        proc = client._run(["kv", "get", "-format=json", f"{mount}/{path}"])
        if proc.returncode != 0:
            return {field: False for field in requested}
        try:
            doc = json.loads(proc.stdout)
        except json.JSONDecodeError:
            return {field: False for field in requested}
        data = doc.get("data", {}).get("data", {})
        if not isinstance(data, dict):
            return {field: False for field in requested}
        return {field: field in data and bool(data[field]) for field in requested}

    def kv_field_present(
        self, mount: str, path: str, field: str, *, token: str | None = None
    ) -> bool:
        """Compatibility wrapper for a single non-secret presence result."""
        return self.kv_fields_present(mount, path, [field], token=token)[field]

    def kv_can_read(self, mount: str, path: str, *, token: str) -> bool:
        """True iff `token` is permitted to read the path at all (no value used)."""
        client = OpenBaoClient(addr=self.addr, token=token, bao_bin=self.bao_bin)
        proc = client._run(["kv", "get", "-format=json", f"{mount}/{path}"])
        return proc.returncode == 0

    def kv_delete_metadata(self, mount: str, path: str) -> None:
        self._run_ok(["kv", "metadata", "delete", f"{mount}/{path}"])