301 lines
13 KiB
Python
301 lines
13 KiB
Python
"""Thin urllib client for the HUX service on the pod loopback.
|
|
|
|
The client owns three things: the identity headers ``hux.identity`` expects,
|
|
the mapping of ``hux.error.v1`` bodies to ``HuxServiceError``, and the
|
|
per-process capabilities cache. It never logs or embeds request bodies in
|
|
exceptions (SO-07, SO-11): an error carries status, code and the service's
|
|
own message only.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import http.client
|
|
import hmac
|
|
import json
|
|
import os
|
|
import re
|
|
import stat
|
|
import threading
|
|
import urllib.error
|
|
import urllib.parse
|
|
import urllib.request
|
|
from collections.abc import Mapping
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
HEADER_SLOT = "X-Hermes-Tenant-Identity"
|
|
HEADER_SUBJECT = "X-Hux-Subject"
|
|
HEADER_SURFACE = "X-Hux-Surface"
|
|
HEADER_TRUST = "X-Hux-Trust"
|
|
HEADER_KEY = "X-Hux-Relay-Key"
|
|
DEFAULT_BASE_URL = "http://127.0.0.1:8790"
|
|
MESSAGE_MAX = 280
|
|
MAX_RESPONSE_BYTES = 4 * 1024 * 1024
|
|
MAX_KEY_BYTES = 4096
|
|
MAX_SUBJECT_BYTES = 128
|
|
LOOPBACK_HOSTS = frozenset({"127.0.0.1", "::1"})
|
|
SUBJECT_RE = re.compile(r"^usr_[0-9a-f]{16,64}$")
|
|
|
|
|
|
class _RejectRedirect(urllib.request.HTTPRedirectHandler):
|
|
"""Never replay tenant identity or worker credentials to a redirect target."""
|
|
|
|
def redirect_request(self, req, fp, code, msg, headers, newurl): # noqa: ANN001, ANN201, ARG002
|
|
"""Refuse every redirect rather than replaying the original headers."""
|
|
return None
|
|
|
|
|
|
def _validated_base_url(raw: str) -> str:
|
|
"""Return a canonical literal-loopback HTTP origin or reject it."""
|
|
parts = urllib.parse.urlsplit(raw)
|
|
if (
|
|
parts.scheme != "http"
|
|
or parts.hostname not in LOOPBACK_HOSTS
|
|
or parts.username is not None
|
|
or parts.password is not None
|
|
or parts.query
|
|
or parts.fragment
|
|
or parts.path not in {"", "/"}
|
|
or parts.port is None
|
|
):
|
|
raise ValueError("HUX base URL must be a literal loopback HTTP origin with an explicit port")
|
|
host = f"[{parts.hostname}]" if parts.hostname == "::1" else parts.hostname
|
|
return f"http://{host}:{parts.port}"
|
|
|
|
|
|
def _validated_path(path: str) -> str:
|
|
"""Accept only canonical relative HUX API paths owned by this client."""
|
|
if not isinstance(path, str):
|
|
raise HuxServiceError(400, "invalid", "malformed request path")
|
|
decoded = urllib.parse.unquote(path)
|
|
if (
|
|
not path.startswith("/hux/v1/")
|
|
or path.startswith("//")
|
|
or any(char in path for char in "?#\\\r\n")
|
|
or any(segment in {".", ".."} for segment in decoded.split("/"))
|
|
):
|
|
raise HuxServiceError(400, "invalid", "malformed request path")
|
|
return path
|
|
|
|
|
|
def _key_from_file(key_file: str | Path) -> str:
|
|
"""Read one 0400 regular-file credential without accepting weak permissions or unbounded data."""
|
|
path = Path(key_file)
|
|
try:
|
|
mode = path.stat().st_mode
|
|
if not stat.S_ISREG(mode) or stat.S_IMODE(mode) != 0o400:
|
|
raise ValueError("HUX key file must be a 0400 regular file")
|
|
data = path.read_bytes()
|
|
if not data or len(data) > MAX_KEY_BYTES:
|
|
raise ValueError("HUX key file is empty or oversized")
|
|
value = data.decode("utf-8", errors="strict").strip()
|
|
if not value:
|
|
raise ValueError("HUX key file is empty")
|
|
return value
|
|
except OSError as error:
|
|
raise ValueError("HUX key file is unavailable") from error
|
|
except UnicodeDecodeError as error:
|
|
raise ValueError("HUX key file is not UTF-8") from error
|
|
|
|
|
|
def _subject_from_file(subject_file: str | Path) -> str:
|
|
"""Read one router-published subject binding from a read-only shared file."""
|
|
path = Path(subject_file)
|
|
flags = os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0)
|
|
try:
|
|
descriptor = os.open(path, flags)
|
|
except OSError as error:
|
|
raise ValueError("HUX subject file is unavailable") from error
|
|
try:
|
|
mode = os.fstat(descriptor).st_mode
|
|
if not stat.S_ISREG(mode) or stat.S_IMODE(mode) not in {0o400, 0o440}:
|
|
raise ValueError("HUX subject file must be a 0400 or 0440 regular file")
|
|
with os.fdopen(descriptor, "rb", closefd=False) as subject_stream:
|
|
data = subject_stream.read(MAX_SUBJECT_BYTES + 1)
|
|
except OSError as error:
|
|
raise ValueError("HUX subject file is unavailable") from error
|
|
finally:
|
|
os.close(descriptor)
|
|
if not data or len(data) > MAX_SUBJECT_BYTES:
|
|
raise ValueError("HUX subject file is empty or oversized")
|
|
try:
|
|
value = data.decode("utf-8", errors="strict").strip()
|
|
except UnicodeDecodeError as error:
|
|
raise ValueError("HUX subject file is not UTF-8") from error
|
|
if not SUBJECT_RE.fullmatch(value):
|
|
raise ValueError("HUX subject file is malformed")
|
|
return value
|
|
|
|
|
|
class HuxServiceError(Exception):
|
|
"""The service answered with a ``hux.error.v1`` body (or a non-JSON failure)."""
|
|
|
|
def __init__(self, status: int, code: str, message: str) -> None:
|
|
super().__init__(f"{status} {code}: {message}")
|
|
self.status = int(status)
|
|
self.code = str(code)
|
|
self.message = str(message)[:MESSAGE_MAX]
|
|
|
|
|
|
class HuxUnavailable(HuxServiceError):
|
|
"""The service could not be reached at all; callers decide open or closed."""
|
|
|
|
def __init__(self, message: str = "hux service unreachable") -> None:
|
|
super().__init__(0, "unavailable", message)
|
|
|
|
|
|
class HuxResponse:
|
|
"""Status, parsed JSON body and headers of one answer."""
|
|
|
|
def __init__(self, status: int, body: Any, headers: Mapping[str, str]) -> None:
|
|
self.status = status
|
|
self.body = body
|
|
self.headers = {k.lower(): v for k, v in headers.items()}
|
|
|
|
def header(self, name: str) -> str:
|
|
"""Case-insensitive header lookup; empty when absent."""
|
|
return self.headers.get(name.lower(), "")
|
|
|
|
|
|
def _error_from(status: int, body: Any) -> HuxServiceError:
|
|
if isinstance(body, dict) and body.get("schema") == "hux.error.v1":
|
|
return HuxServiceError(int(body.get("status", status)), str(body.get("code", "invalid")), str(body.get("message", "")))
|
|
return HuxServiceError(status, "invalid", f"unexpected response {status}")
|
|
|
|
|
|
class HuxClient:
|
|
"""One tenant identity talking to one HUX base URL.
|
|
|
|
``identity`` carries ``tenant_slot``, ``subject``, ``surface`` and
|
|
``trust``; ``key`` is the relay or worker shared key when ``trust`` needs
|
|
one. The agent hook normally runs as ``surface=worker, trust=worker``.
|
|
"""
|
|
|
|
def __init__(self, base_url: str = DEFAULT_BASE_URL, identity: Mapping[str, str] | None = None,
|
|
key: str | None = None, timeout: float = 5, *, key_file: str | Path | None = None,
|
|
subject_file: str | Path | None = None) -> None:
|
|
identity = dict(identity or {})
|
|
configured_subject_file = subject_file if subject_file is not None else os.environ.get("HUX_SUBJECT_FILE")
|
|
if configured_subject_file:
|
|
bound_subject = _subject_from_file(configured_subject_file)
|
|
asserted_subject = str(identity.get("subject", ""))
|
|
if asserted_subject and not hmac.compare_digest(asserted_subject, bound_subject):
|
|
raise ValueError("HUX identity subject conflicts with the trusted binding")
|
|
identity["subject"] = bound_subject
|
|
self.base_url = _validated_base_url(base_url)
|
|
self.identity = {
|
|
"tenant_slot": str(identity.get("tenant_slot", "")), "subject": str(identity.get("subject", "")),
|
|
"surface": str(identity.get("surface", "worker")), "trust": str(identity.get("trust", "worker")),
|
|
}
|
|
if key is not None and key_file is not None:
|
|
raise ValueError("set either key or key_file, not both")
|
|
self._key = _key_from_file(key_file) if key_file is not None else key
|
|
if isinstance(timeout, bool) or not isinstance(timeout, int | float) or not 0.1 <= float(timeout) <= 30:
|
|
raise ValueError("timeout must be between 0.1 and 30 seconds")
|
|
self.timeout = float(timeout)
|
|
self._opener = urllib.request.build_opener(urllib.request.ProxyHandler({}), _RejectRedirect())
|
|
self._capabilities: dict[str, Any] | None = None
|
|
self._guard = threading.Lock()
|
|
|
|
# -- transport -------------------------------------------------------------
|
|
|
|
def headers(self, extra: Mapping[str, str] | None = None) -> dict[str, str]:
|
|
"""Identity headers exactly as ``hux.identity.resolve`` reads them, plus ``extra``."""
|
|
out = {
|
|
HEADER_SLOT: self.identity["tenant_slot"], HEADER_SUBJECT: self.identity["subject"],
|
|
HEADER_SURFACE: self.identity["surface"], HEADER_TRUST: self.identity["trust"],
|
|
"Accept": "application/json",
|
|
}
|
|
if self._key:
|
|
out[HEADER_KEY] = self._key
|
|
for name, value in (extra or {}).items():
|
|
if value:
|
|
out[name] = str(value)
|
|
return out
|
|
|
|
def request(self, method: str, path: str, body: Any = None, *, idempotency_key: str | None = None,
|
|
if_match: int | str | None = None, query: Mapping[str, str] | None = None) -> HuxResponse:
|
|
"""Send one request; raise ``HuxServiceError`` on 4xx/5xx and ``HuxUnavailable`` on transport failure."""
|
|
extra: dict[str, str] = {}
|
|
if idempotency_key:
|
|
extra["Idempotency-Key"] = idempotency_key
|
|
if if_match is not None:
|
|
extra["If-Match"] = str(if_match)
|
|
data = None
|
|
if body is not None:
|
|
data = json.dumps(body, sort_keys=True, separators=(",", ":")).encode()
|
|
extra["Content-Type"] = "application/json"
|
|
url = self.base_url + _validated_path(path)
|
|
if query:
|
|
url += "?" + urllib.parse.urlencode({str(k): str(v) for k, v in query.items()})
|
|
req = urllib.request.Request(url, data=data, method=method, headers=self.headers(extra))
|
|
try:
|
|
with self._opener.open(req, timeout=self.timeout) as raw: # noqa: S310 - validated literal loopback only
|
|
response = HuxResponse(raw.status, _decode(_bounded_read(raw)), dict(raw.headers.items()))
|
|
except urllib.error.HTTPError as error:
|
|
payload = _decode(_bounded_read(error))
|
|
raise _error_from(error.code, payload) from None
|
|
except (urllib.error.URLError, OSError, TimeoutError) as error:
|
|
raise HuxUnavailable(f"hux service unreachable: {type(error).__name__}") from None
|
|
except http.client.InvalidURL:
|
|
# http.client refuses control characters and spaces in the path; treat it as our own bad request.
|
|
raise HuxServiceError(400, "invalid", "malformed request path") from None
|
|
except http.client.HTTPException as error:
|
|
raise HuxUnavailable(f"hux service unreachable: {type(error).__name__}") from None
|
|
if response.status >= 400:
|
|
raise _error_from(response.status, response.body)
|
|
return response
|
|
|
|
def get(self, path: str, query: Mapping[str, str] | None = None) -> HuxResponse:
|
|
"""``GET path``."""
|
|
return self.request("GET", path, query=query)
|
|
|
|
def post(self, path: str, body: Any, idempotency_key: str | None = None) -> HuxResponse:
|
|
"""``POST path`` with an optional ``Idempotency-Key``."""
|
|
return self.request("POST", path, body, idempotency_key=idempotency_key)
|
|
|
|
def put(self, path: str, body: Any, if_match: int | str | None = None) -> HuxResponse:
|
|
"""``PUT path`` with an optional ``If-Match`` revision."""
|
|
return self.request("PUT", path, body, if_match=if_match)
|
|
|
|
# -- capabilities ------------------------------------------------------------
|
|
|
|
def capabilities(self, refresh: bool = False) -> dict[str, Any]:
|
|
"""Which cards are on, cached per process; unreachable or off means every card reads as off."""
|
|
with self._guard:
|
|
if self._capabilities is not None and not refresh:
|
|
return self._capabilities
|
|
try:
|
|
body = self.get("/hux/v1/capabilities").body
|
|
except HuxServiceError as error:
|
|
return {"reachable": error.code != "unavailable", "cards": {}, "contract_version": ""}
|
|
cards = {c["card"]: bool(c.get("enabled")) for c in body.get("cards", []) if isinstance(c, dict) and "card" in c}
|
|
self._capabilities = {"reachable": True, "cards": cards, "contract_version": str(body.get("contract_version", ""))}
|
|
return self._capabilities
|
|
|
|
def card_enabled(self, card: str) -> bool:
|
|
"""True only when the service answered and reports ``card`` on."""
|
|
return bool(self.capabilities()["cards"].get(card))
|
|
|
|
def forget_capabilities(self) -> None:
|
|
"""Drop the cache so the next call re-reads flags (used after a reload signal)."""
|
|
with self._guard:
|
|
self._capabilities = None
|
|
|
|
|
|
def _decode(raw: bytes) -> Any:
|
|
if not raw:
|
|
return None
|
|
try:
|
|
return json.loads(raw)
|
|
except ValueError:
|
|
return None
|
|
|
|
|
|
def _bounded_read(raw: Any) -> bytes:
|
|
"""Read one bounded response so a compromised sidecar cannot exhaust the worker."""
|
|
data = raw.read(MAX_RESPONSE_BYTES + 1)
|
|
if len(data) > MAX_RESPONSE_BYTES:
|
|
raise HuxUnavailable("hux service returned an oversized response")
|
|
return data
|