approval-engine/approval_engine/pep.py

"""Fail-closed PEP client and sequencing harness for GH-DEC-2026-003."""

from __future__ import annotations

import json
import math
import os
import re
import stat
import subprocess
import sys
import time
from pathlib import Path
from typing import Any, Callable
from urllib.error import HTTPError, URLError
from urllib.parse import quote, urlsplit
from urllib.request import Request

from .binding import DIGEST_RE

_MAX_BODY = 256 * 1024
_ID_RE = re.compile(r"^[A-Za-z0-9:._-]+$")


class ApprovalProtocolError(RuntimeError):
    pass


def _status_message(status: int) -> str:
    if status == 409:
        return "approval consume conflict"
    if status == 404:
        return "approval not found"
    if status in {401, 403}:
        return "approval consume unauthorized"
    if status == 503:
        return "approval-engine unavailable"
    if status == 0:
        return "approval-engine unreachable"
    return f"approval endpoint returned status {status}"


class ApprovalHTTPClient:
    def __init__(
        self,
        base_url: str,
        token_file: str | Path,
        *,
        timeout_seconds: float = 3,
        opener: Callable[..., Any] | None = None,
    ) -> None:
        try:
            parsed = urlsplit(base_url)
            valid = (parsed.scheme in {"http", "https"} and parsed.hostname and
                     parsed.port != 0 and not parsed.username and not parsed.password and
                     not parsed.query and not parsed.fragment and
                     not any(ch.isspace() or ord(ch) < 32 for ch in base_url))
        except (TypeError, ValueError):
            valid = False
        if not valid:
            raise ApprovalProtocolError("approval-engine URL is missing or invalid")
        if isinstance(timeout_seconds, bool) or not isinstance(timeout_seconds, (int, float)) or not math.isfinite(timeout_seconds) or not 0 < timeout_seconds <= 60:
            raise ApprovalProtocolError("approval timeout must be finite and bounded")
        self.base_url = base_url.rstrip("/")
        self.token_file = Path(token_file)
        self.timeout_seconds = timeout_seconds
        self.opener = opener

    def _token(self) -> str:
        fd = None
        directory = None
        try:
            parts = self.token_file.absolute().parts
            if ".." in parts:
                raise ValueError("unsafe credential path")
            directory = os.open(parts[0], os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW)
            for part in parts[1:-1]:
                child = os.open(part, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW, dir_fd=directory)
                os.close(directory)
                directory = child
            fd = os.open(parts[-1], os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK, dir_fd=directory)
            info = os.fstat(fd)
            if (not stat.S_ISREG(info.st_mode) or info.st_uid not in {0, os.geteuid()}
                    or info.st_mode & 0o037 or info.st_size > 16384 or info.st_nlink != 1):
                raise ValueError("unsafe credential file")
            raw = os.read(fd, 16385)
            token = raw.decode("utf-8").strip()
            if len(raw) > 16384 or not token or any(ch.isspace() or ord(ch) < 33 or ord(ch) > 126 for ch in token):
                raise ValueError("invalid credential")
        except (OSError, UnicodeError, ValueError):
            raise ApprovalProtocolError("approval credential is unavailable") from None
        finally:
            if fd is not None:
                os.close(fd)
            if directory is not None:
                os.close(directory)
        return token

    @staticmethod
    def _decode(raw: str) -> dict[str, Any]:
        def pairs(items):
            result = {}
            for key, value in items:
                if key in result:
                    raise ValueError("duplicate JSON key")
                result[key] = value
            return result
        def constant(_value):
            raise ValueError("non-finite JSON value")
        def number(value):
            parsed = float(value)
            if not math.isfinite(parsed):
                raise ValueError("non-finite JSON number")
            return parsed
        try:
            result = json.loads(raw, object_pairs_hook=pairs, parse_constant=constant, parse_float=number)
        except (ValueError, RecursionError):
            raise ApprovalProtocolError("approval endpoint returned invalid JSON") from None
        if not isinstance(result, dict):
            raise ApprovalProtocolError("approval endpoint returned invalid payload")
        return result

    def _isolated_request(self, method, url, body, token, deadline):
        payload = json.dumps({"method": method, "url": url, "body": body,
                              "token": token, "timeout": self.timeout_seconds}, allow_nan=False).encode()
        if len(payload) > _MAX_BODY:
            raise ApprovalProtocolError("approval request is too large")
        remaining = deadline - time.monotonic()
        if remaining <= 0:
            raise ApprovalProtocolError("approval transport deadline exceeded")
        try:
            completed = subprocess.run(
                [sys.executable, "-I", str(Path(__file__).with_name("http_transport.py"))],
                input=payload, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL,
                timeout=remaining, check=False, env={},
            )
        except subprocess.TimeoutExpired:
            raise ApprovalProtocolError("approval transport deadline exceeded") from None
        except OSError:
            raise ApprovalProtocolError("approval-engine unreachable") from None
        if completed.returncode != 0 or len(completed.stdout) > 2 * _MAX_BODY:
            raise ApprovalProtocolError("approval transport failed")
        try:
            result = json.loads(completed.stdout)
        except (ValueError, UnicodeError):
            raise ApprovalProtocolError("approval transport failed") from None
        if not isinstance(result, dict):
            raise ApprovalProtocolError("approval transport failed")
        if result.get("error") == "status" and type(result.get("status")) is int:
            raise ApprovalProtocolError(_status_message(result["status"]))
        if result.get("error") == "content":
            raise ApprovalProtocolError("approval endpoint returned invalid content")
        if not isinstance(result.get("body"), str):
            raise ApprovalProtocolError("approval-engine unreachable")
        return self._decode(result["body"])

    def _request(
        self, method: str, path: str, body: dict[str, Any] | None = None
    ) -> dict[str, Any]:
        deadline = time.monotonic() + self.timeout_seconds
        if not isinstance(path, str) or not path.startswith("/") or path.startswith("//") or any(ch.isspace() or ord(ch) < 32 for ch in path):
            raise ApprovalProtocolError("approval request path is invalid")
        url = self.base_url + path
        if urlsplit(url)[:2] != urlsplit(self.base_url)[:2] or urlsplit(url).fragment:
            raise ApprovalProtocolError("approval request origin is invalid")
        token = self._token()
        try:
            text_body = None if body is None else json.dumps(body, allow_nan=False)
        except (ValueError, TypeError):
            raise ApprovalProtocolError("approval request is invalid") from None
        if self.opener is None:
            return self._isolated_request(method, url, text_body, token, deadline)
        # Explicit owner-controlled injection for test adapters. It does not
        # establish production transport admission or an isolated deadline.
        encoded = None if text_body is None else text_body.encode("utf-8")
        request = Request(
            url,
            data=encoded,
            method=method,
            headers={
                "Authorization": f"Bearer {token}",
                "Content-Type": "application/json",
                "Accept": "application/json",
            },
        )
        try:
            response = self.opener(request, timeout=self.timeout_seconds)
            try:
                status = int(response.getcode())
                raw = response.read(_MAX_BODY + 1)
            finally:
                response.close()
        except HTTPError as exc:
            status = int(getattr(exc, "code", 0) or 0)
            try:
                exc.read(_MAX_BODY)
            except Exception:
                pass
            raise ApprovalProtocolError(_status_message(status)) from None
        except Exception:
            raise ApprovalProtocolError("approval-engine unreachable") from None
        if status != 200 or len(raw) > _MAX_BODY:
            raise ApprovalProtocolError(_status_message(status if status != 200 else 502))
        try:
            return self._decode(raw.decode("utf-8"))
        except UnicodeDecodeError:
            raise ApprovalProtocolError("approval endpoint returned invalid JSON") from None

    def claim(self, approval_id: str) -> dict[str, Any]:
        if not _ID_RE.fullmatch(approval_id):
            raise ApprovalProtocolError("approval id is invalid")
        encoded_id = quote(approval_id, safe="")
        return self._request("GET", f"/v1/approvals/{encoded_id}/claim")

    def consume(
        self, approval_id: str, request_digest: str, decision_id: str | None = None
    ) -> dict[str, Any]:
        if not _ID_RE.fullmatch(approval_id):
            raise ApprovalProtocolError("approval id is invalid")
        if not DIGEST_RE.fullmatch(request_digest):
            raise ApprovalProtocolError("approval consume requires the canonical request digest")
        encoded_id = quote(approval_id, safe="")
        payload: dict[str, str] = {"request_digest": request_digest}
        if decision_id:
            payload["decision_id"] = decision_id
        result = self._request(
            "POST",
            f"/v1/approvals/{encoded_id}/consume",
            payload,
        )
        if result.get("status") != "consumed":
            raise ApprovalProtocolError("approval consumption was not confirmed")
        if result.get("request_digest") != request_digest:
            raise ApprovalProtocolError("approval consume digest does not match the request")
        return result


class ProtectedActionHarness:
    """Sequence a supplied PDP decision and dry-run/protected callback.

    The decision callback owns authorization. This class only enforces that a
    fresh approval claim precedes it and a successful CAS consume precedes the
    side effect. The consume response is mutation evidence, never a permission.
    """

    def __init__(self, client: ApprovalHTTPClient) -> None:
        self.client = client

    def execute(
        self,
        approval_id: str,
        request_digest: str,
        decide: Callable[[dict[str, Any]], dict[str, Any]],
        side_effect: Callable[[], Any],
    ) -> Any:
        claim = self.client.claim(approval_id)
        if claim.get("valid_now") is not True or claim.get("consumed") is not False:
            raise ApprovalProtocolError("approval claim is not valid for use")
        decision = decide(claim)
        if decision.get("effect") != "ALLOW":
            raise ApprovalProtocolError("authorization decision did not allow")
        decision_id = decision.get("decision_id")
        if not isinstance(decision_id, str) or not decision_id:
            raise ApprovalProtocolError("authorization decision lacks decision_id")
        if decision.get("request_digest") != request_digest:
            raise ApprovalProtocolError("authorization decision digest does not match")
        consumed = self.client.consume(approval_id, request_digest, decision_id)
        if "effect" in consumed or "allow" in consumed or "deny" in consumed:
            raise ApprovalProtocolError("consume response is not a permission")
        return side_effect()