74 lines
2.8 KiB
Python
74 lines
2.8 KiB
Python
|
"""access-engine (flex-auth) Check client — step 2 of GH-DEC-2026-003.
|
||
|
|
|
||
|
|
This engine consumes a decision; it never renders one. The DecisionEnvelope
|
||
|
|
returned here is validated by ``authorization.validate_decision_envelope``
|
||
|
|
against the exact proposed action before it can satisfy the production stance.
|
||
|
|
|
||
|
|
No estate-wide PDP exists by design: flex-auth runs per-consumer cluster-local
|
||
|
|
pins, so the address is per-deployment configuration and its absence fails
|
||
|
|
production closed.
|
||
|
|
"""
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import json
|
||
|
|
from pathlib import Path
|
||
|
|
from typing import Any, Callable
|
||
|
|
from urllib.error import HTTPError, URLError
|
||
|
|
from urllib.request import Request, urlopen
|
||
|
|
|
||
|
|
from secrets_engine.errors import DecisionError
|
||
|
|
from secrets_engine.openbao import read_strict_token_file
|
||
|
|
|
||
|
|
_MAX_BODY = 512 * 1024
|
||
|
|
|
||
|
|
|
||
|
|
def _status_message(status: int) -> str:
|
||
|
|
if status in (401, 403):
|
||
|
|
return "access-engine refused the caller"
|
||
|
|
if status == 404:
|
||
|
|
return "access-engine has no such endpoint"
|
||
|
|
if status == 503:
|
||
|
|
return "access-engine is unavailable"
|
||
|
|
return f"access-engine returned HTTP {status}"
|
||
|
|
|
||
|
|
|
||
|
|
def check_decision(
|
||
|
|
*,
|
||
|
|
base_url: str,
|
||
|
|
token_file: Path,
|
||
|
|
request: dict[str, Any],
|
||
|
|
timeout_seconds: float = 3,
|
||
|
|
opener: Callable[..., Any] = urlopen,
|
||
|
|
) -> dict[str, Any]:
|
||
|
|
"""POST /v1/check. Any non-200, non-JSON, or transport failure fails closed.
|
||
|
|
|
||
|
|
Silence is never permission: an unreachable PDP raises rather than
|
||
|
|
returning a permissive default.
|
||
|
|
"""
|
||
|
|
if not base_url or not base_url.startswith(("http://", "https://")):
|
||
|
|
raise DecisionError("access-engine check URL is missing or invalid")
|
||
|
|
token = read_strict_token_file(Path(token_file), purpose="access-engine credential")
|
||
|
|
encoded = json.dumps(request).encode("utf-8")
|
||
|
|
http_request = Request(
|
||
|
|
base_url.rstrip("/") + "/v1/check",
|
||
|
|
data=encoded,
|
||
|
|
method="POST",
|
||
|
|
)
|
||
|
|
http_request.add_header("Authorization", f"Bearer {token}")
|
||
|
|
http_request.add_header("Content-Type", "application/json")
|
||
|
|
http_request.add_header("Accept", "application/json")
|
||
|
|
try:
|
||
|
|
with opener(http_request, timeout=timeout_seconds) as response:
|
||
|
|
if getattr(response, "status", 200) != 200:
|
||
|
|
raise DecisionError("access-engine check did not return a decision")
|
||
|
|
payload = json.loads(response.read(_MAX_BODY).decode("utf-8"))
|
||
|
|
except HTTPError as e:
|
||
|
|
raise DecisionError(f"access-engine check refused: {_status_message(e.code)}") from e
|
||
|
|
except URLError as e:
|
||
|
|
raise DecisionError("access-engine is unreachable for check") from e
|
||
|
|
except json.JSONDecodeError as e:
|
||
|
|
raise DecisionError("access-engine check returned a non-JSON body") from e
|
||
|
|
if not isinstance(payload, dict):
|
||
|
|
raise DecisionError("access-engine check returned a non-object body")
|
||
|
|
return payload
|