"""Backends that resolve ``op://...`` URIs (or env vars) into credentials.
There is **one** real backend in the MVP: ``OpCLIBackend``. It works on
both laptops (biometric-unlocked desktop app session) and unattended
hosts (``OP_SERVICE_ACCOUNT_TOKEN`` set in env) — same ``op read``
shell-out, the auth difference is purely environmental.
``EnvBackend`` is for tests and CI only — when ``R2_ACCESS_KEY_ID`` is
already in env we short-circuit to direct env-var reads.
"""
from __future__ import annotations
import os
import shutil
import subprocess
from typing import Protocol
from constellation_utils.secrets.models import CloudflareSecrets, R2Secrets
[docs]
class ConstellationAuthError(RuntimeError):
"""Raised when no auth backend is available or a read fails.
Messages aim to be actionable: what was expected, what was found,
suggested next step.
"""
class _Backend(Protocol):
def read_r2(self, uri_map: dict[str, str]) -> R2Secrets: ...
def read_cloudflare(self, uri_map: dict[str, str]) -> CloudflareSecrets: ...
[docs]
class OpCLIBackend:
"""Resolve credentials by shelling out to ``op read``.
Works for both laptop biometric sessions and unattended hosts that
have ``OP_SERVICE_ACCOUNT_TOKEN`` set — the ``op`` CLI picks up the
token from env automatically.
"""
[docs]
def read_one(self, uri: str) -> str:
try:
result = subprocess.run(
["op", "read", uri],
capture_output=True,
check=True,
text=True,
)
except FileNotFoundError as exc:
raise ConstellationAuthError(
"the `op` CLI is not on PATH.\n"
" expected : `op` available in $PATH\n"
" fix : install with `brew install --cask 1password-cli`,\n"
" then run `op signin` (laptop) or set\n"
" OP_SERVICE_ACCOUNT_TOKEN (unattended)."
) from exc
except subprocess.CalledProcessError as exc:
stderr = (exc.stderr or "").strip()
raise ConstellationAuthError(
f"`op read {uri}` failed.\n"
f" reason : {stderr or 'no stderr from op'}\n"
f" fix : verify the item exists and your account "
f"(or service-account token) has read access to it.\n"
f" if you just installed `op`, run `op signin` first."
) from exc
return result.stdout.strip()
[docs]
def read_r2(self, uri_map: dict[str, str]) -> R2Secrets:
resolved = {field: self.read_one(uri) for field, uri in uri_map.items()}
return R2Secrets.model_validate(resolved)
[docs]
def read_cloudflare(self, uri_map: dict[str, str]) -> CloudflareSecrets:
resolved = {field: self.read_one(uri) for field, uri in uri_map.items()}
return CloudflareSecrets.model_validate(resolved)
[docs]
class EnvBackend:
"""Direct env-var reads. Tests/CI only — never the production path."""
R2_ENV_MAP = {
"endpoint": "R2_ENDPOINT",
"access_key_id": "R2_ACCESS_KEY_ID",
"secret_access_key": "R2_SECRET_ACCESS_KEY",
"bucket": "R2_BUCKET",
"region": "R2_REGION",
}
CLOUDFLARE_ENV_MAP = {
"api_token": "CLOUDFLARE_API_TOKEN",
"account_id": "CLOUDFLARE_ACCOUNT_ID",
}
[docs]
def read_r2(self, uri_map: dict[str, str]) -> R2Secrets:
# uri_map is ignored — env-var reads have a fixed mapping.
del uri_map
values: dict[str, str] = {}
missing: list[str] = []
for field, env_name in self.R2_ENV_MAP.items():
val = os.environ.get(env_name)
if val is None:
# `region` has a default; everything else is required.
if field == "region":
continue
missing.append(env_name)
else:
values[field] = val
if missing:
raise ConstellationAuthError(
"EnvBackend selected but required env vars are missing.\n"
f" missing : {', '.join(missing)}\n"
" fix : set them, or unset R2_ACCESS_KEY_ID to fall\n"
" back to the `op` CLI backend."
)
return R2Secrets.model_validate(values)
[docs]
def read_cloudflare(self, uri_map: dict[str, str]) -> CloudflareSecrets:
del uri_map
values: dict[str, str] = {}
missing: list[str] = []
for field, env_name in self.CLOUDFLARE_ENV_MAP.items():
val = os.environ.get(env_name)
if val is None:
missing.append(env_name)
else:
values[field] = val
if missing:
raise ConstellationAuthError(
"EnvBackend selected but required Cloudflare env vars are missing.\n"
f" missing : {', '.join(missing)}\n"
" fix : set them, or unset CLOUDFLARE_API_TOKEN to fall\n"
" back to the `op` CLI backend."
)
return CloudflareSecrets.model_validate(values)
_ENV_BACKEND_TRIGGERS = ("R2_ACCESS_KEY_ID", "CLOUDFLARE_API_TOKEN")
[docs]
def select_backend() -> _Backend:
"""Pick a backend based on the current env.
Order:
1. ``R2_ACCESS_KEY_ID`` or ``CLOUDFLARE_API_TOKEN`` set →
EnvBackend (tests/CI escape hatch).
2. ``op`` on PATH → OpCLIBackend (laptops + rigs).
3. else → ConstellationAuthError.
"""
override = os.environ.get("CONSTELLATION_SECRETS_BACKEND", "").lower()
if override == "env":
return EnvBackend()
if override == "op_cli":
return OpCLIBackend()
if override and override != "auto":
raise ConstellationAuthError(
f"unknown CONSTELLATION_SECRETS_BACKEND={override!r}\n"
" expected : one of 'auto', 'op_cli', 'env' (or unset)\n"
" fix : unset the var or pick a valid backend."
)
if any(trigger in os.environ for trigger in _ENV_BACKEND_TRIGGERS):
return EnvBackend()
if shutil.which("op") is not None:
return OpCLIBackend()
raise ConstellationAuthError(
"no auth backend available.\n"
" expected : `op` CLI on PATH, or R2_*/CLOUDFLARE_* env vars set\n"
" fix : `brew install --cask 1password-cli && op signin`,\n"
" or set R2_ACCESS_KEY_ID + R2_SECRET_ACCESS_KEY +\n"
" R2_ENDPOINT + R2_BUCKET (R2 tests),\n"
" or CLOUDFLARE_API_TOKEN + CLOUDFLARE_ACCOUNT_ID\n"
" (Cloudflare API tests)."
)