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 os
import stat
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
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
    client_secret_file: Path
    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")


@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)

    def exchange(self, *, now: datetime | None = None) -> ServiceJWT:
        """Perform one client-credentials exchange; never retry or fall back."""
        secret = read_client_secret_file(self.config.client_secret_file)
        basic = base64.b64encode(
            f"{self.config.client_id}:{secret}".encode("utf-8")
        ).decode("ascii")
        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)