secrets-engine/src/secrets_engine/service_auth.py

"""KeyCape service-JWT exchange shared by two separately validated profiles.

The OpenBao login profile and approval consumer profile have distinct clients,
audiences and scopes; both belong to tenant:platform. Provider selection never falls back to another
identity. Token parsing is a claim preflight, not signature verification: the
receiving OpenBao or approval-engine service verifies RS256 before accepting it.
"""
from __future__ import annotations

import base64
import binascii
import json
import hashlib
import ipaddress
import os
import stat
import subprocess
from dataclasses import dataclass, field
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Callable
from urllib.error import HTTPError, URLError
from urllib.parse import urlencode, urlsplit
from urllib.request import Request, urlopen

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

CLIENT_ID = "secrets-engine-openbao"
SUBJECT = "service:secrets-engine"
PRINCIPAL_TYPE = "service"
TENANT = "tenant:platform"
ROLE = "secrets-engine"
SCOPE = "openbao:login"
MAX_TOKEN_SECONDS = 15 * 60
RENEW_WINDOW_SECONDS = 3 * 60


@dataclass(frozen=True)
class KeyCapeServiceAuthConfig:
    token_url: str
    issuer: str
    # The OpenBao profile takes no file: its secret is read in memory from
    # OpenBao per command (OpenBaoOidcClientSecretSource). The approval
    # profile subclass still uses a consumer file.
    client_secret_file: Path | None = None
    client_id: str = CLIENT_ID
    subject: str = SUBJECT
    audience: str = CLIENT_ID
    tenant: str = TENANT
    required_role: str = ROLE
    scope: str = SCOPE
    timeout_seconds: float = 10.0
    max_future_iat_seconds: int = field(default=60, init=False)

    def __post_init__(self) -> None:
        if not self.token_url.startswith("https://"):
            raise BackendError("KeyCape token URL must use HTTPS")
        if not self.issuer.startswith("https://"):
            raise BackendError("KeyCape issuer must use HTTPS")
        if self.client_id != CLIENT_ID or self.audience != CLIENT_ID:
            raise BackendError("KeyCape service client/audience must match accepted contract")
        if (
            self.subject != SUBJECT
            or self.tenant != TENANT
            or self.required_role != ROLE
            or self.scope != SCOPE
        ):
            raise BackendError("KeyCape service identity claims must match accepted contract")
        if self.timeout_seconds <= 0:
            raise BackendError("KeyCape timeout must be positive")
        if type(self) is KeyCapeServiceAuthConfig and self.client_secret_file is not None:
            raise BackendError(
                "secrets-engine-openbao client secret file source is refused; "
                "the secret is read in memory from OpenBao per command"
            )


@dataclass(frozen=True)
class ServiceJWT:
    token: str = field(repr=False)
    issued_at: int
    expires_at: int
    claims: dict[str, Any] = field(repr=False)

    def needs_renewal(self, now: datetime | None = None) -> bool:
        current = int((now or datetime.now(timezone.utc)).timestamp())
        return self.expires_at - current <= RENEW_WINDOW_SECONDS


def _decode_segment(value: str, label: str) -> dict[str, Any]:
    try:
        padded = value + "=" * (-len(value) % 4)
        decoded = base64.urlsafe_b64decode(padded.encode("ascii"))
        parsed = json.loads(decoded)
    except (UnicodeEncodeError, binascii.Error, json.JSONDecodeError) as e:
        raise BackendError(f"KeyCape JWT has invalid {label}") from e
    if not isinstance(parsed, dict):
        raise BackendError(f"KeyCape JWT {label} must be an object")
    return parsed


def preflight_service_jwt(
    token: str,
    config: KeyCapeServiceAuthConfig,
    *,
    now: datetime | None = None,
) -> ServiceJWT:
    """Validate non-cryptographic JWT shape/claims before resource-server use."""
    parts = token.split(".")
    if len(parts) != 3 or not all(parts):
        raise BackendError("KeyCape access token is not a compact JWT")
    header = _decode_segment(parts[0], "header")
    claims = _decode_segment(parts[1], "payload")
    if header.get("alg") != "RS256":
        raise BackendError("KeyCape JWT algorithm is not RS256")

    exact_claims = {
        "iss": config.issuer,
        "sub": config.subject,
        "aud": config.audience,
        "principal_type": PRINCIPAL_TYPE,
        "tenant": config.tenant,
        "groups": [],
    }
    for name, expected in exact_claims.items():
        if claims.get(name) != expected:
            raise BackendError(f"KeyCape JWT claim '{name}' does not match contract")
    if claims.get("roles") != [config.required_role]:
        raise BackendError("KeyCape JWT roles do not match contract")
    if claims.get("scope") != config.scope:
        raise BackendError("KeyCape JWT scope does not match contract")

    assurance = claims.get("assurance")
    if not isinstance(assurance, dict) or (
        assurance.get("level") != "aal1"
        or assurance.get("methods") != ["client_secret"]
        or assurance.get("mfa") is not False
        or assurance.get("source") != "key-cape"
    ):
        raise BackendError("KeyCape JWT assurance does not match contract")

    issued_at = claims.get("iat")
    expires_at = claims.get("exp")
    if (
        not isinstance(issued_at, int)
        or isinstance(issued_at, bool)
        or not isinstance(expires_at, int)
        or isinstance(expires_at, bool)
    ):
        raise BackendError("KeyCape JWT iat/exp must be integer timestamps")
    current = int((now or datetime.now(timezone.utc)).timestamp())
    if issued_at > current + config.max_future_iat_seconds:
        raise BackendError("KeyCape JWT issue time is in the future")
    if expires_at <= current:
        raise BackendError("KeyCape JWT has expired")
    if expires_at <= issued_at or expires_at - issued_at > MAX_TOKEN_SECONDS:
        raise BackendError("KeyCape JWT lifetime exceeds accepted 15-minute bound")
    return ServiceJWT(
        token=token,
        issued_at=issued_at,
        expires_at=expires_at,
        claims=dict(claims),
    )


Transport = Callable[..., Any]


@dataclass(frozen=True)
class SecretFileCheck:
    """One value-free consumer-file check. Never carries file contents."""

    name: str
    ok: bool
    detail: str = ""


def check_client_secret_file(path: Path) -> list[SecretFileCheck]:
    """Inspect a confidential-client secret file without exposing its value.

    The consumer copy must be an absolute, canonical (no symlink component),
    regular, non-empty file owned by the effective user with no group/other
    permission bits, in a parent directory owned by the same user with no
    group/other permission bits, outside every Git worktree. Stops at the first
    structural failure; later checks are then not reported as passed.
    """
    checks: list[SecretFileCheck] = []

    def add(name: str, ok: bool, detail: str = "") -> bool:
        checks.append(SecretFileCheck(name, ok, detail))
        return ok

    raw = Path(path)
    if not add("absolute", raw.is_absolute(), "" if raw.is_absolute() else "path must be absolute"):
        return checks
    try:
        canonical = raw.resolve(strict=True)
    except (OSError, RuntimeError):
        add("exists", False, "file not found or unresolvable")
        return checks
    add("exists", True)
    if not add(
        "canonical",
        canonical == raw,
        "" if canonical == raw else "path has a symlink or non-canonical component",
    ):
        return checks
    st = os.lstat(raw)
    if not add(
        "regular_file",
        stat.S_ISREG(st.st_mode),
        "" if stat.S_ISREG(st.st_mode) else "not a regular file",
    ):
        return checks
    euid = os.geteuid()
    add("file_owner", st.st_uid == euid, "" if st.st_uid == euid else "not owned by the running user")
    mode = stat.S_IMODE(st.st_mode)
    add("file_mode", not mode & 0o077, "" if not mode & 0o077 else f"group/other-accessible mode {oct(mode)}; must be 0600")
    pst = os.lstat(raw.parent)
    add("parent_owner", pst.st_uid == euid, "" if pst.st_uid == euid else "parent not owned by the running user")
    pmode = stat.S_IMODE(pst.st_mode)
    add(
        "parent_mode",
        not pmode & 0o077,
        "" if not pmode & 0o077 else f"group/other-accessible parent mode {oct(pmode)}; must be 0700",
    )
    worktree = containing_git_worktree(raw)
    add("outside_git_worktree", worktree is None, "" if worktree is None else "inside a Git worktree")
    add("non_empty", st.st_size > 0, "" if st.st_size > 0 else "file is empty")
    return checks


def read_client_secret_file(path: Path, *, purpose: str = "KeyCape client secret") -> str:
    """Read the consumer secret only after every value-free check passes."""
    failed = [c for c in check_client_secret_file(path) if not c.ok]
    if failed:
        reasons = "; ".join(f"{c.name}: {c.detail}" for c in failed)
        raise ProvisioningError(f"{purpose} file {path} refused ({reasons})")
    value = Path(path).read_text(encoding="utf-8").strip()
    if not value or any(ch.isspace() for ch in value):
        raise ProvisioningError(
            f"{purpose} file {path} refused (must hold exactly one non-empty token)"
        )
    return value


@dataclass(frozen=True)
class KeyCapeServiceAuthProvider:
    config: KeyCapeServiceAuthConfig
    transport: Transport = field(default=urlopen, repr=False, compare=False)
    # In-memory secret source. Required for the OpenBao profile; the approval
    # profile reads its consumer file when this is None.
    secret_source: Callable[[], str] | None = field(default=None, repr=False, compare=False)

    def exchange(self, *, now: datetime | None = None) -> ServiceJWT:
        """Perform one client-credentials exchange; never retry or fall back."""
        if self.secret_source is not None:
            secret = self.secret_source()
        elif type(self.config) is KeyCapeServiceAuthConfig or self.config.client_secret_file is None:
            raise BackendError(
                "secrets-engine-openbao requires the in-memory OpenBao secret source; "
                "no file fallback"
            )
        else:
            secret = read_client_secret_file(self.config.client_secret_file)
        basic = base64.b64encode(
            f"{self.config.client_id}:{secret}".encode("utf-8")
        ).decode("ascii")
        del secret
        body = urlencode(
            {"grant_type": "client_credentials", "scope": self.config.scope}
        ).encode("ascii")
        request = Request(
            self.config.token_url,
            data=body,
            method="POST",
            headers={
                "Authorization": f"Basic {basic}",
                "Content-Type": "application/x-www-form-urlencoded",
                "Accept": "application/json",
            },
        )
        try:
            response = self.transport(request, timeout=self.config.timeout_seconds)
            with response:
                status = getattr(response, "status", 200)
                raw = response.read(65537)
                if len(raw) > 65536:
                    raise BackendError("KeyCape token response exceeds size bound")
        except HTTPError as e:
            raise BackendError(f"KeyCape token exchange failed with HTTP {e.code}") from e
        except (URLError, TimeoutError, OSError) as e:
            raise BackendError("KeyCape token exchange failed") from e
        if status != 200:
            raise BackendError(f"KeyCape token exchange failed with HTTP {status}")
        try:
            payload = json.loads(raw)
        except (UnicodeDecodeError, json.JSONDecodeError) as e:
            raise BackendError("KeyCape token response is not valid JSON") from e
        if not isinstance(payload, dict):
            raise BackendError("KeyCape token response must be an object")
        if payload.get("id_token") or payload.get("refresh_token"):
            raise BackendError("KeyCape service exchange returned a forbidden extra token")
        token = payload.get("access_token")
        if not isinstance(token, str) or not token:
            raise BackendError("KeyCape token response has no access token")
        if str(payload.get("token_type", "")).lower() != "bearer":
            raise BackendError("KeyCape token response type is not Bearer")
        expires_in = payload.get("expires_in")
        if (
            not isinstance(expires_in, int)
            or isinstance(expires_in, bool)
            or not 0 < expires_in <= MAX_TOKEN_SECONDS
        ):
            raise BackendError("KeyCape token response lifetime is outside contract")
        return preflight_service_jwt(token, self.config, now=now)


# -- in-memory OpenBao source for the secrets-engine-openbao profile ---------
#
# ops-mason plan keycape-secrets-engine-openbao-client-custody (658cdbc),
# founder direction 2026-10-06: the value lives only in OpenBao. Per command an
# attended OIDC login on one narrow role yields a 5m/4-use operator token that
# is captured from a pipe, used once in a request header for the KV GET, and
# revoked with revoke-self. Nothing is written to file, env, argv or cache.

OIDC_MOUNT = "netkingdom"
OIDC_ROLE = "keycape-secrets-engine-openbao-operator-workload-kv-read"
KV_MOUNT = "platform"
KV_PATH = "workloads/keycape/secrets-engine-openbao"
KV_FIELD = "CLIENT_SECRET"
SOURCE_NAME = "openbao-oidc-in-memory"
_AMBIENT_TOKEN_ENV = ("BAO_TOKEN", "VAULT_TOKEN")
_MAX_KV_RESPONSE = 65536

Runner = Callable[[list[str], dict[str, str]], "tuple[int, bytes]"]


def _run_login(argv: list[str], env: dict[str, str]) -> tuple[int, bytes]:
    """Run the attended login. stdout is a pipe into memory; stderr (the
    browser prompt, no token) stays on the operator's terminal."""
    proc = subprocess.run(  # noqa: S603 - fixed argv, no shell
        argv,
        env=env,
        stdin=subprocess.DEVNULL,
        stdout=subprocess.PIPE,
        stderr=None,
        check=False,
    )
    return proc.returncode, proc.stdout


def _require_bao_address(addr: str) -> str:
    try:
        parts = urlsplit(addr)
        host = parts.hostname or ""
        port_ok = parts.port != 0
    except ValueError:
        raise BackendError("OpenBao address is invalid for the in-memory secret read") from None
    if not host or parts.username or parts.password or not port_ok:
        raise BackendError("OpenBao address is invalid for the in-memory secret read")
    try:
        loopback = ipaddress.ip_address(host).is_loopback
    except ValueError:
        loopback = host == "localhost"
    if parts.scheme == "https" or (parts.scheme == "http" and loopback):
        return addr.rstrip("/")
    raise BackendError("OpenBao address must be HTTPS or a literal loopback forward")


@dataclass
class OpenBaoOidcClientSecretSource:
    """Read CLIENT_SECRET in memory through one attended OIDC operator login.

    Evidence is value-free: source name, operator accessor fingerprint, and
    whether revoke-self was attempted and succeeded.
    """

    bao_addr: str
    bao_bin: str = "bao"
    oidc_mount: str = OIDC_MOUNT
    oidc_role: str = OIDC_ROLE
    kv_mount: str = KV_MOUNT
    kv_path: str = KV_PATH
    kv_field: str = KV_FIELD
    timeout_seconds: float = 10.0
    runner: Runner = field(default=_run_login, repr=False, compare=False)
    transport: Transport = field(default=urlopen, repr=False, compare=False)
    accessor_fingerprint: str = ""
    revocation_attempted: bool = False
    revocation_succeeded: bool = False

    def __post_init__(self) -> None:
        pinned = (OIDC_MOUNT, OIDC_ROLE, KV_MOUNT, KV_PATH, KV_FIELD)
        if (self.oidc_mount, self.oidc_role, self.kv_mount, self.kv_path, self.kv_field) != pinned:
            raise BackendError("OpenBao secret source coordinates do not match the reviewed plan")
        self.bao_addr = _require_bao_address(self.bao_addr)
        if not self.bao_bin:
            raise BackendError("bao CLI is required for the attended OIDC login")

    def evidence(self) -> dict[str, object]:
        return {
            "secret_source": SOURCE_NAME,
            "operator_session_handle": self.accessor_fingerprint,
            "operator_revocation_attempted": self.revocation_attempted,
            "operator_revocation_succeeded": self.revocation_succeeded,
        }

    def argv(self) -> list[str]:
        # Token never appears here; -no-store keeps it out of the token helper.
        return [
            self.bao_bin,
            "login",
            "-method=oidc",
            f"-path={self.oidc_mount}",
            "-no-store",
            "-format=json",
            f"role={self.oidc_role}",
        ]

    def _login_env(self) -> dict[str, str]:
        env = {k: v for k, v in os.environ.items() if k not in _AMBIENT_TOKEN_ENV}
        env["BAO_ADDR"] = self.bao_addr
        env.pop("VAULT_ADDR", None)
        return env

    def _login(self) -> str:
        code, out = self.runner(self.argv(), self._login_env())
        try:
            if code != 0:
                raise BackendError(f"attended OpenBao OIDC login failed (exit {code})")
            try:
                auth = json.loads(out)["auth"]
                token = str(auth["client_token"])
                accessor = str(auth.get("accessor", ""))
            except (UnicodeDecodeError, json.JSONDecodeError, KeyError, TypeError) as e:
                raise BackendError("malformed OpenBao OIDC login response") from e
        finally:
            del out
        if not token or any(ch.isspace() for ch in token):
            raise BackendError("OpenBao OIDC login returned no usable token")
        self.accessor_fingerprint = (
            hashlib.sha256(accessor.encode("utf-8")).hexdigest()[:12] if accessor else ""
        )
        return token

    def _call(self, method: str, path: str, token: str) -> tuple[int, bytes]:
        request = Request(
            f"{self.bao_addr}/v1/{path}",
            data=b"" if method == "POST" else None,
            method=method,
            headers={"X-Vault-Token": token, "Accept": "application/json"},
        )
        try:
            response = self.transport(request, timeout=self.timeout_seconds)
            with response:
                status = getattr(response, "status", 200)
                raw = response.read(_MAX_KV_RESPONSE + 1)
        except HTTPError as e:
            return e.code, b""
        except (URLError, TimeoutError, OSError) as e:
            raise BackendError(f"OpenBao request failed: {method} {path}") from e
        if len(raw) > _MAX_KV_RESPONSE:
            raise BackendError("OpenBao response exceeds size bound")
        return status, raw

    def _read_field(self, token: str) -> str:
        status, raw = self._call("GET", f"{self.kv_mount}/data/{self.kv_path}", token)
        if status != 200:
            raise BackendError(f"OpenBao KV read of the client secret failed with HTTP {status}")
        try:
            value = json.loads(raw)["data"]["data"][self.kv_field]
        except (UnicodeDecodeError, json.JSONDecodeError, KeyError, TypeError) as e:
            raise BackendError("OpenBao KV response has no client-secret field") from e
        finally:
            del raw
        if not isinstance(value, str) or not value or any(ch.isspace() for ch in value):
            raise BackendError("OpenBao client-secret field must hold exactly one token")
        return value

    def _revoke_self(self, token: str) -> None:
        self.revocation_attempted = True
        try:
            status, _ = self._call("POST", "auth/token/revoke-self", token)
        except BackendError:
            return
        self.revocation_succeeded = status in (200, 204)

    def __call__(self) -> str:
        token = self._login()
        try:
            return self._read_field(token)
        finally:
            self._revoke_self(token)
            del token