secrets-engine/src/secrets_engine/decision_check.py

"""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 ipaddress
import json
from pathlib import Path
from urllib.parse import urlsplit
from typing import Any, Callable
from urllib.error import HTTPError, URLError
from urllib.request import HTTPRedirectHandler, ProxyHandler, Request, build_opener

from secrets_engine.errors import DecisionError
from secrets_engine.decision_signature import strict_json, load_trust, authenticate_response
from secrets_engine.openbao import read_strict_token_file

_MAX_BODY = 512 * 1024

#: Suffixes that name an in-cluster Service and must never be dialled from here.
_CLUSTER_SUFFIXES = (".svc.cluster.local", ".svc", ".cluster.local")


def _is_loopback(host: str) -> bool:
    try:
        return ipaddress.ip_address(host.strip("[]")).is_loopback
    except ValueError:
        return False


def require_supported_pdp_address(base_url: str) -> None:
    """Require a literal loopback origin before reading the caller credential.

    The operator must independently bind the port-forward to the admitted
    Kubernetes cluster and named PDP pod. URL validation alone cannot establish
    the identity of a process listening on a local port. Hostnames, redirects
    and ambient proxies must not redirect the credential away from that port.
    """
    try:
        url = urlsplit(base_url)
        host = (url.hostname or "").lower()
        valid = (
            bool(host) and url.scheme in {"http", "https"}
            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 any(ord(char) <= 32 or ord(char) == 127 for char in base_url)
        )
    except ValueError:
        valid = False
    if not valid:
        raise DecisionError("access-engine check URL is missing or invalid")
    if any(host.rstrip(".").endswith(suffix) for suffix in _CLUSTER_SUFFIXES):
        raise DecisionError(
            "access-engine check URL is an in-cluster Service name; use the "
            "owner-bound literal loopback port-forward (docs/pdp-access-path.md)"
        )
    if not _is_loopback(host):
        raise DecisionError(
            "access-engine check URL host is not loopback in literal IP form; "
            "use the owner-bound 127.0.0.1 or [::1] port-forward "
            "(docs/pdp-access-path.md)"
        )


class _NoRedirect(HTTPRedirectHandler):
    def redirect_request(self, req, fp, code, msg, headers, newurl):
        return None


def _pdp_urlopen(request, *, timeout):
    # This is a local tunnel: never inherit an environment/system proxy or a
    # process-global opener. No redirect, including same-origin, is a decision.
    return build_opener(ProxyHandler({}), _NoRedirect()).open(request, timeout=timeout)


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],
    trust_file: Path | None = None,
    verifier: Path | None = None,
    allow_unsigned_test_fixture: bool = False,
    timeout_seconds: float = 3,
    opener: Callable[..., Any] | None = None,
) -> 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.
    """
    require_signature = not allow_unsigned_test_fixture
    if not base_url or not base_url.startswith(("http://", "https://")):
        raise DecisionError("access-engine check URL is missing or invalid")
    require_supported_pdp_address(base_url)
    trust = load_trust(trust_file) if require_signature else None
    if require_signature and (verifier is None or not Path(verifier).is_absolute() or not Path(verifier).is_file()):
        raise DecisionError("access-engine signature verifier is not installed")
    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 or _pdp_urlopen)(http_request, timeout=timeout_seconds) as response:
            if getattr(response, "status", 200) != 200:
                raise DecisionError("access-engine check did not return a decision")
            body = response.read(_MAX_BODY + 1)
            if len(body) > _MAX_BODY:
                raise DecisionError("access-engine check body is too large")
            payload = strict_json(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 (ValueError, UnicodeDecodeError, RecursionError) 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")
    if require_signature:
        return authenticate_response(body, trust=trust, verifier=verifier)
    return payload