tegwick
7ce58ae638
Assistant: codex Assistant-Model: gpt-5.6-sol Assistant-Session: 01a0291a-1e87-7151-9934-fcbfe3f65eb1
92 lines
3.5 KiB
Python
92 lines
3.5 KiB
Python
"""Caller identity for ops-warden's outbound flex-auth policy calls.
|
|
|
|
flex-auth's `flex-auth-ops-warden` pin (FLEX-WP-0016) authenticates the *caller*
|
|
before it evaluates the request: `Authorization: Bearer <token>` is passed to a
|
|
Kubernetes TokenReview, and `resource.system: ops-warden` is bound to the
|
|
principal `system:serviceaccount:ops-warden:ops-warden`. Until ops-warden sends
|
|
that header, the pin logs `caller authentication warning` and can only run in
|
|
`warn` mode — which is why enforcing caller authentication is a separate gate.
|
|
|
|
This module resolves the token at call time and hands it straight to the request.
|
|
Nothing is cached to disk, logged, or echoed: ops-warden carries the value, it
|
|
does not hold it (ADR-0002).
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import subprocess
|
|
|
|
from warden.config import CallerAuthConfig
|
|
|
|
|
|
class CallerIdentityError(Exception):
|
|
"""Raised when a caller token was configured but could not be obtained."""
|
|
|
|
|
|
def resolve_caller_token(cfg: CallerAuthConfig) -> str | None:
|
|
"""Return the bearer token for flex-auth, or None when mode is ``none``.
|
|
|
|
Raises CallerIdentityError when a token was configured but is unavailable.
|
|
The token itself never appears in an exception message.
|
|
"""
|
|
mode = cfg.mode
|
|
if mode == "none":
|
|
return None
|
|
|
|
if mode == "file":
|
|
if cfg.token_path is None:
|
|
raise CallerIdentityError("caller_auth mode 'file' has no token_path")
|
|
try:
|
|
token = cfg.token_path.read_text()
|
|
except OSError as e:
|
|
raise CallerIdentityError(
|
|
f"caller token file unreadable: {cfg.token_path} ({e.strerror})"
|
|
) from e
|
|
elif mode == "env":
|
|
token = os.environ.get(cfg.token_env, "")
|
|
if not token.strip():
|
|
raise CallerIdentityError(
|
|
f"caller token env {cfg.token_env} is unset or empty"
|
|
)
|
|
elif mode == "command":
|
|
if not cfg.command:
|
|
raise CallerIdentityError("caller_auth mode 'command' has no command")
|
|
try:
|
|
result = subprocess.run(
|
|
cfg.command,
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=30,
|
|
check=False,
|
|
)
|
|
except FileNotFoundError as e:
|
|
raise CallerIdentityError(
|
|
f"caller token command not found: {cfg.command[0]}"
|
|
) from e
|
|
except subprocess.TimeoutExpired as e:
|
|
raise CallerIdentityError("caller token command timed out") from e
|
|
if result.returncode != 0:
|
|
stderr = (result.stderr or "").strip().splitlines()
|
|
detail = stderr[-1] if stderr else f"exit {result.returncode}"
|
|
raise CallerIdentityError(f"caller token command failed: {detail}")
|
|
token = result.stdout
|
|
else:
|
|
raise CallerIdentityError(f"unsupported caller_auth mode {mode!r}")
|
|
|
|
token = token.strip()
|
|
if not token:
|
|
raise CallerIdentityError(f"caller_auth mode {mode!r} produced an empty token")
|
|
if any(ch.isspace() for ch in token):
|
|
# flex-auth rejects a bearer token containing whitespace outright.
|
|
raise CallerIdentityError(
|
|
f"caller_auth mode {mode!r} produced a token containing whitespace"
|
|
)
|
|
return token
|
|
|
|
|
|
def caller_auth_headers(cfg: CallerAuthConfig) -> dict[str, str]:
|
|
"""Headers to attach to a flex-auth /v1/check call ({} when unauthenticated)."""
|
|
token = resolve_caller_token(cfg)
|
|
if token is None:
|
|
return {}
|
|
return {"Authorization": f"Bearer {token}"}
|