Source code for constellation_utils.secrets

"""Public secrets API.

Top-level functions return frozen Pydantic models holding the resolved
credential values. Each function is cached per-process so repeated calls
are free.

Auth flows:
- Laptops: `op` CLI uses the desktop app's biometric-unlocked session.
- Rigs / unattended: same `op` CLI consumes ``OP_SERVICE_ACCOUNT_TOKEN``
  from the environment without a desktop app.
- Tests / CI: if ``R2_ACCESS_KEY_ID`` is already set in env, the env
  backend short-circuits the op CLI entirely.

Profile selection: ``CONSTELLATION_PROFILE=testing|production``
(default ``testing``).
"""

from __future__ import annotations

from functools import lru_cache

from constellation_utils.secrets._backends import (
    ConstellationAuthError,
    select_backend,
)
from constellation_utils.secrets._config import load_profile
from constellation_utils.secrets.models import CloudflareSecrets, R2Secrets

__all__ = [
    "r2_raw_rw",
    "r2_raw_ro",
    "r2_assets_rw",
    "r2_assets_ro",
    "cloudflare",
    "R2Secrets",
    "CloudflareSecrets",
    "ConstellationAuthError",
]




@lru_cache(maxsize=1)
def r2_raw_rw() -> R2Secrets:
    """Return read-write credentials for the constellation-data (raw recordings) bucket.

    Reads from 1Password via the ``op`` CLI by default, or from
    ``R2_*`` env vars in test/CI environments. Cached for the
    lifetime of the process.

    Raises ``ConstellationAuthError`` if no auth backend is available
    (e.g. ``op`` is not installed and ``R2_ACCESS_KEY_ID`` is not set).
    """
    cfg = load_profile()
    backend = select_backend()
    return backend.read_r2(cfg["r2_raw_rw"])





@lru_cache(maxsize=1)
def r2_raw_ro() -> R2Secrets:
    """Return read-only credentials for the constellation-data (raw recordings) bucket.

    Reads from 1Password via the ``op`` CLI by default. Cached for the
    lifetime of the process.
    """
    cfg = load_profile()
    backend = select_backend()
    return backend.read_r2(cfg["r2_raw_ro"])





@lru_cache(maxsize=1)
def r2_assets_rw() -> R2Secrets:
    """Return read-write credentials for the constellation-assets bucket.

    Reads from 1Password via the ``op`` CLI by default. Cached for the
    lifetime of the process.
    """
    cfg = load_profile()
    backend = select_backend()
    return backend.read_r2(cfg["r2_assets_rw"])





@lru_cache(maxsize=1)
def r2_assets_ro() -> R2Secrets:
    """Return read-only credentials for the constellation-assets bucket.

    Reads from 1Password via the ``op`` CLI by default. Cached for the
    lifetime of the process.
    """
    cfg = load_profile()
    backend = select_backend()
    return backend.read_r2(cfg["r2_assets_ro"])





@lru_cache(maxsize=1)
def cloudflare() -> CloudflareSecrets:
    """Return the Cloudflare account API credentials for the current profile.

    Reads from 1Password via the ``op`` CLI by default, or from
    ``CLOUDFLARE_API_TOKEN`` + ``CLOUDFLARE_ACCOUNT_ID`` env vars in
    test/CI environments. Cached for the lifetime of the process.

    Raises ``ConstellationAuthError`` if the ``cloudflare:`` block is
    missing from the active profile's YAML, or if no auth backend is
    available.
    """
    cfg = load_profile()
    if "cloudflare" not in cfg:
        raise ConstellationAuthError(
            "the active profile is missing a `cloudflare:` block.\n"
            "  fix : add it to secrets.testing.yaml / secrets.production.yaml,\n"
            "        mapping `api_token` and `account_id` to op:// URIs."
        )
    backend = select_backend()
    return backend.read_cloudflare(cfg["cloudflare"])