tegwick
7688445184
Assistant: codex Assistant-Model: gpt-5.6-luna Assistant-Session: 01a07ff8-19d0-7820-b4d0-1353833cb7fc
113 lines
4.7 KiB
Python
113 lines
4.7 KiB
Python
"""The approval consumer's own KeyCape identity, distinct from OpenBao login.
|
|
|
|
Access tokens stay in memory for one request. Claim preflight is not signature
|
|
verification or an authorization decision; approval-engine verifies the JWT.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass, field
|
|
from pathlib import Path
|
|
from typing import Any
|
|
from urllib.parse import urlsplit
|
|
from urllib.request import HTTPRedirectHandler, build_opener
|
|
|
|
from secrets_engine.errors import BackendError, DecisionError
|
|
from secrets_engine.openbao import read_strict_token_file
|
|
from secrets_engine.service_auth import KeyCapeServiceAuthConfig, KeyCapeServiceAuthProvider
|
|
|
|
|
|
class _NoRedirect(HTTPRedirectHandler):
|
|
def redirect_request(self, req, fp, code, msg, headers, newurl):
|
|
# urllib otherwise forwards Basic/Bearer headers on a redirected GET.
|
|
return None
|
|
|
|
|
|
def credential_urlopen(request, *, timeout):
|
|
return build_opener(_NoRedirect()).open(request, timeout=timeout)
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class KeyCapeApprovalAuthConfig(KeyCapeServiceAuthConfig):
|
|
client_id: str = "secrets-engine-approval"
|
|
audience: str = "approval-engine"
|
|
tenant: str = "tenant:platform"
|
|
scope: str = "approval:read"
|
|
max_future_iat_seconds: int = field(default=30, init=False)
|
|
|
|
def __post_init__(self) -> None:
|
|
# Deliberately do not relax the separate OpenBao profile's validator.
|
|
if (
|
|
self.client_id != "secrets-engine-approval"
|
|
or self.audience != "approval-engine"
|
|
or self.subject != "service:secrets-engine"
|
|
or self.tenant != "tenant:platform"
|
|
or self.required_role != "secrets-engine"
|
|
or self.scope not in {"approval:read", "approval:consume"}
|
|
or self.max_future_iat_seconds != 30
|
|
):
|
|
raise BackendError("KeyCape approval identity does not match accepted contract")
|
|
try:
|
|
issuer = urlsplit(self.issuer)
|
|
valid = (
|
|
issuer.scheme == "https" and bool(issuer.hostname)
|
|
and issuer.username is None and issuer.password is None
|
|
and not issuer.query and not issuer.fragment
|
|
and issuer.path in {"", "/"}
|
|
and self.token_url == self.issuer.rstrip("/") + "/token"
|
|
and issuer.port != 0
|
|
)
|
|
except ValueError:
|
|
valid = False
|
|
if not valid:
|
|
raise BackendError("KeyCape approval exchange requires the issuer's HTTPS /token")
|
|
if self.timeout_seconds <= 0:
|
|
raise BackendError("KeyCape timeout must be positive")
|
|
|
|
|
|
def approval_auth_configured(cfg: Any) -> bool:
|
|
token_file = getattr(cfg, "approval_token_file", None)
|
|
secret_file = getattr(cfg, "approval_client_secret_file", None)
|
|
if token_file and secret_file:
|
|
raise DecisionError("approval auth requires one provider; token and client-secret files conflict")
|
|
return bool(token_file or secret_file)
|
|
|
|
|
|
def require_approval_address(base_url: str) -> None:
|
|
"""The first CLI consumer uses TLS or the owner's literal loopback tunnel.
|
|
|
|
A loopback address alone does not establish the tunnel's target identity;
|
|
the owner must bind kubectl port-forward to the admitted cluster and pod.
|
|
"""
|
|
try:
|
|
url = urlsplit(base_url)
|
|
host = (url.hostname or "").rstrip(".").lower()
|
|
valid = (
|
|
bool(host) and url.username is None and url.password is None
|
|
and not url.query and not url.fragment and url.path in {"", "/"}
|
|
and url.port != 0
|
|
and not host.endswith((".svc", ".svc.cluster.local", ".cluster.local"))
|
|
and (url.scheme == "https" or (
|
|
url.scheme == "http" and host in {"127.0.0.1", "::1"}
|
|
))
|
|
)
|
|
except ValueError:
|
|
valid = False
|
|
if not valid:
|
|
raise DecisionError("approval service auth requires HTTPS or an owner-bound literal loopback tunnel")
|
|
|
|
|
|
def approval_token(cfg: Any, *, scope: str) -> str:
|
|
"""Select one explicit provider. Never fall back after exchange failure."""
|
|
if not approval_auth_configured(cfg):
|
|
raise DecisionError("approval credential is unconfigured")
|
|
secret_file = getattr(cfg, "approval_client_secret_file", None)
|
|
if secret_file:
|
|
require_approval_address(str(getattr(cfg, "approval_url", "") or ""))
|
|
config = KeyCapeApprovalAuthConfig(
|
|
token_url=str(getattr(cfg, "keycape_token_url", "") or ""),
|
|
issuer=str(getattr(cfg, "keycape_issuer", "") or ""),
|
|
client_secret_file=Path(secret_file),
|
|
scope=scope,
|
|
)
|
|
return KeyCapeServiceAuthProvider(config, transport=credential_urlopen).exchange().token
|
|
return read_strict_token_file(Path(cfg.approval_token_file), purpose="approval credential")
|