"""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()