state-hub/api/services/forge_credential.py

"""Resolve the forge read credential (STATE-WP-0084-T03, MASON-WP-0003-T05).

Three sources, tried in order: a mounted file, an environment variable, and
OpenBao via Kubernetes auth. Production uses the third — this cluster has no
agent injector and no secrets-store CSI driver, so the pod authenticates with
its projected ServiceAccount token and reads the KV path itself, which is what
`MASON-WP-0003-T02` built the Kubernetes auth role for. The first two exist so
the code is runnable and testable outside the cluster.

**Absent is a supported state, never an error.** A hub with no credential still
derives every public repository; only private ones become unreadable, and
`STATE-WP-0084-T01` already reports that as its own named condition rather than
as records that stopped deriving. So every failure here — no configuration, no
network, OpenBao down, permission denied — resolves to `None`. Raising would
convert "cannot read nine repositories" into "the whole pass failed".
"""

from __future__ import annotations

import logging
import os
import time
from pathlib import Path

import httpx

logger = logging.getLogger(__name__)

TOKEN_ENV = "FORGE_READ_TOKEN"
TOKEN_FILE_ENV = "FORGE_READ_TOKEN_FILE"
OPENBAO_ADDR_ENV = "OPENBAO_ADDR"
OPENBAO_ROLE_ENV = "OPENBAO_K8S_ROLE"
OPENBAO_JWT_PATH_ENV = "OPENBAO_K8S_TOKEN_PATH"
OPENBAO_AUTH_MOUNT_ENV = "OPENBAO_K8S_AUTH_MOUNT"
SECRET_PATH_ENV = "FORGE_READ_SECRET_PATH"
SECRET_KEY_ENV = "FORGE_READ_SECRET_KEY"

# Long enough that a fleet reset of 121 repositories does not re-authenticate
# 121 times; short enough that a rotated token is picked up without a redeploy,
# which is what MASON-WP-0003-T02 requires of this lane.
CACHE_TTL_SECONDS = 300.0

_cache: tuple[float, str | None] | None = None


def reset_cache() -> None:
    global _cache
    _cache = None


def _from_file() -> str | None:
    path = os.environ.get(TOKEN_FILE_ENV)
    if not path:
        return None
    try:
        return Path(path).read_text(encoding="utf-8").strip() or None
    except OSError:
        # Deliberately not falling through to the environment: a broken mount
        # that silently used a stale value would look like success.
        logger.warning("forge credential: token file %s is unreadable", path)
        return None


def _from_env() -> str | None:
    return (os.environ.get(TOKEN_ENV) or "").strip() or None


def _from_openbao() -> str | None:
    addr = (os.environ.get(OPENBAO_ADDR_ENV) or "").strip().rstrip("/")
    secret_path = (os.environ.get(SECRET_PATH_ENV) or "").strip().strip("/")
    role = (os.environ.get(OPENBAO_ROLE_ENV) or "").strip()
    jwt_path = os.environ.get(OPENBAO_JWT_PATH_ENV) or "/var/run/secrets/openbao/token"
    mount = (os.environ.get(OPENBAO_AUTH_MOUNT_ENV) or "kubernetes").strip("/")
    key = (os.environ.get(SECRET_KEY_ENV) or "token").strip()
    if not (addr and secret_path and role):
        return None
    try:
        jwt = Path(jwt_path).read_text(encoding="utf-8").strip()
    except OSError:
        logger.warning("forge credential: no ServiceAccount token at %s", jwt_path)
        return None
    try:
        with httpx.Client(timeout=10.0) as client:
            login = client.post(
                f"{addr}/v1/auth/{mount}/login", json={"role": role, "jwt": jwt}
            )
            login.raise_for_status()
            client_token = login.json()["auth"]["client_token"]
            read = client.get(
                f"{addr}/v1/{secret_path}", headers={"X-Vault-Token": client_token}
            )
            read.raise_for_status()
            data = read.json()["data"]
            # KV v2 nests the payload under a second "data"; v1 does not.
            if isinstance(data.get("data"), dict):
                data = data["data"]
    except (httpx.HTTPError, KeyError, ValueError) as exc:
        # Never include the response body: a failed KV read can echo content.
        logger.warning("forge credential: OpenBao lookup failed (%s)", type(exc).__name__)
        return None
    value = data.get(key)
    if not isinstance(value, str) or not value.strip():
        logger.warning("forge credential: key %r absent at the KV path", key)
        return None
    return value.strip()


def forge_read_token(*, use_cache: bool = True) -> str | None:
    """The forge read credential, or `None` if this instance has none."""
    global _cache
    now = time.monotonic()
    if use_cache and _cache is not None and now - _cache[0] < CACHE_TTL_SECONDS:
        return _cache[1]
    if os.environ.get(TOKEN_FILE_ENV):
        # Configured to use a file means *that* file and nothing else. Falling
        # back would let a broken mount quietly resolve to a stale environment
        # value that nobody knows is in use.
        token = _from_file()
    else:
        token = _from_env() or _from_openbao()
    _cache = (now, token)
    return token