diff --git a/dockerfiles/hermes-hux-foundation/hux/audit.py b/dockerfiles/hermes-hux-foundation/hux/audit.py new file mode 100644 index 00000000..6cf203bb --- /dev/null +++ b/dockerfiles/hermes-hux-foundation/hux/audit.py @@ -0,0 +1,40 @@ +"""Auditable outcome for every read and mutation. + +One JSONL ledger per tenant per day. The record shape is +``common.schema.json#/$defs/audit_outcome``; it never carries request bodies, +only the action name, the resource id and the decision. +""" + +from __future__ import annotations + +from hux.identity import Identity +from hux.store import TenantStore, now_iso + +OUTCOMES = ("allow", "deny", "not_found", "conflict", "flag_off") + + +def record(store: TenantStore, identity: Identity, action: str, resource: str, outcome: str, reason: str = "") -> dict: + """Append an audit outcome and return it.""" + if outcome not in OUTCOMES: + raise ValueError(f"unknown outcome {outcome!r}") + entry = { + "at": now_iso(), + "identity": identity.record(), + "action": action, + "resource": resource[:200], + "outcome": outcome, + } + if reason: + entry["reason"] = reason[:200] + store.append("audit", now_iso()[:10], entry) + return entry + + +def recent(store: TenantStore, limit: int = 200) -> list[dict]: + """Newest audit rows across day ledgers, newest last.""" + rows: list[dict] = [] + for name in reversed(store.ledgers("audit")): + rows = store.read("audit", name) + rows + if len(rows) >= limit: + break + return rows[-limit:] diff --git a/dockerfiles/hermes-hux-foundation/hux/errors.py b/dockerfiles/hermes-hux-foundation/hux/errors.py new file mode 100644 index 00000000..e6aa66e3 --- /dev/null +++ b/dockerfiles/hermes-hux-foundation/hux/errors.py @@ -0,0 +1,76 @@ +"""Error types that map one-to-one onto ``hux.error.v1`` records.""" + +from __future__ import annotations + + +class HuxError(Exception): + """Base class; ``status`` and ``code`` follow identity.schema.json#/$defs/error.""" + + status = 500 + code = "invalid" + + def __init__(self, message: str, details: list[str] | None = None) -> None: + super().__init__(message) + self.message = message + self.details = list(details or []) + + def record(self) -> dict: + """Serialise as a ``hux.error.v1`` record.""" + body = {"schema": "hux.error.v1", "status": self.status, "code": self.code, "message": self.message[:280]} + if self.details: + body["details"] = [d[:280] for d in self.details[:32]] + return body + + +class Unauthorized(HuxError): + """Identity headers missing, malformed or not trusted.""" + + status, code = 401, "unauthorized" + + +class Forbidden(HuxError): + """Identity is valid but does not own the resource.""" + + status, code = 403, "forbidden" + + +class NotFound(HuxError): + """Resource does not exist for this tenant.""" + + status, code = 404, "not_found" + + +class FlagOff(HuxError): + """Card (or one of its dependencies) is disabled; indistinguishable from not found on purpose.""" + + status, code = 404, "flag_off" + + +class Conflict(HuxError): + """If-Match revision mismatch or duplicate create.""" + + status, code = 409, "conflict" + + +class Invalid(HuxError): + """Body failed contract validation.""" + + status, code = 400, "invalid" + + +class TooLarge(HuxError): + """Body, record or family exceeds its bound.""" + + status, code = 413, "too_large" + + +class ApprovalRequired(HuxError): + """An action needs an approval record before it may proceed.""" + + status, code = 403, "approval_required" + + +class BudgetExhausted(HuxError): + """A run budget has been spent.""" + + status, code = 429, "budget_exhausted" diff --git a/dockerfiles/hermes-hux-foundation/hux/flags.py b/dockerfiles/hermes-hux-foundation/hux/flags.py new file mode 100644 index 00000000..2b151d35 --- /dev/null +++ b/dockerfiles/hermes-hux-foundation/hux/flags.py @@ -0,0 +1,71 @@ +"""Per-card feature flags and the capabilities record clients negotiate with. + +Flags come from the ``HUX_FLAGS`` comma list. A card counts as enabled only +when it and every card it depends on are enabled, so a half-configured +deployment fails closed. Route ownership per card is declared here so the +capabilities record can tell a client exactly what it may call. +""" + +from __future__ import annotations + +import os +from collections.abc import Mapping + +from hux.errors import FlagOff +from hux.identity import Identity +from hux.rules import flag_enabled, flag_registry + +CONTRACT_VERSION = "1.0.0" +CARD_ROUTES: dict[str, list[str]] = { + "HUX-11": ["/hux/v1/capabilities", "/hux/v1/manifest"], + "HUX-01": ["/hux/v1/conversations/{id}/events", "/hux/v1/conversations/{id}/events/stream"], + "HUX-02": ["/hux/v1/memory", "/hux/v1/memory/{id}", "/hux/v1/memory/{id}/{action}", "/hux/v1/memory/export"], + "HUX-03": ["/hux/v1/projects", "/hux/v1/projects/{id}", "/hux/v1/conversations", "/hux/v1/conversations/{id}", "/hux/v1/conversations/{id}/branch", "/hux/v1/conversations/{id}/lineage", "/hux/v1/search"], + "HUX-04": ["/hux/v1/artifacts", "/hux/v1/artifacts/{id}", "/hux/v1/artifacts/{id}/versions", "/hux/v1/artifacts/{id}/versions/{n}", "/hux/v1/artifacts/{id}/versions/{n}/diff", "/hux/v1/artifacts/{id}/promote"], + "HUX-05": ["/hux/v1/policy", "/hux/v1/approvals", "/hux/v1/approvals/{id}", "/hux/v1/runs/{id}/stop", "/hux/v1/runs/{id}/budget", "/hux/v1/runs/{id}/gate"], + "HUX-06": ["/hux/v1/modes", "/hux/v1/conversations/{id}/mode"], + "HUX-07": [], + "HUX-08": ["/hux/v1/sources", "/hux/v1/sources/{id}", "/hux/v1/passages", "/hux/v1/messages/{id}/citations", "/hux/v1/notebooks", "/hux/v1/notebooks/{id}"], + "HUX-09": ["/hux/v1/suggestions", "/hux/v1/suggestions/{id}/{action}"], + "HUX-10": ["/hux/v1/privacy/policy", "/hux/v1/privacy/notices", "/hux/v1/conversations/{id}/forget", "/hux/v1/privacy/audit"], + "HUX-12": ["/hux/v1/releases"], +} + + +class Flags: + """Snapshot of which cards are on for this process.""" + + def __init__(self, environ: Mapping[str, str] | None = None) -> None: + self._environ = dict(os.environ if environ is None else environ) + self._registry = flag_registry() + + def enabled(self, card: str) -> bool: + """True when the card and its whole dependency chain are on.""" + entry = self._registry.get(card) + return bool(entry) and flag_enabled(entry["flag"], self._environ) + + def require(self, card: str) -> None: + """Raise FlagOff unless the card is enabled.""" + if not self.enabled(card): + raise FlagOff(f"{card} is not enabled") + + def capabilities(self, identity: Identity, build: Mapping[str, str] | None = None) -> dict: + """Serialise ``hux.capabilities.v1`` for one caller.""" + cards = [ + {"card": card, "flag": entry["flag"], "enabled": self.enabled(card), "routes": CARD_ROUTES.get(card, [])} + for card, entry in sorted(self._registry.items()) + ] + server = {k: v for k, v in (build or {}).items() if k in {"commit", "image_digest"} and v} + return { + "schema": "hux.capabilities.v1", + "contract_version": CONTRACT_VERSION, + "identity": identity.record(), + "cards": cards, + "server": server, + } + + +def build_from_environ(environ: Mapping[str, str] | None = None) -> dict[str, str]: + """Commit and image digest the pod was started with, when the operator set them.""" + environ = os.environ if environ is None else environ + return {"commit": environ.get("HUX_BUILD_COMMIT", ""), "image_digest": environ.get("HUX_IMAGE_DIGEST", "")} diff --git a/dockerfiles/hermes-hux-foundation/hux/foundation.py b/dockerfiles/hermes-hux-foundation/hux/foundation.py new file mode 100644 index 00000000..5b26b739 --- /dev/null +++ b/dockerfiles/hermes-hux-foundation/hux/foundation.py @@ -0,0 +1,24 @@ +"""HUX-11 routes: capability negotiation and the data manifest.""" + +from __future__ import annotations + +from hux.flags import CONTRACT_VERSION +from hux.http import Request, Response, Router + + +def capabilities(request: Request) -> Response: + """``GET /hux/v1/capabilities``: what this tenant may call.""" + request.audit("foundation.capabilities", "capabilities") + return Response(200, request.flags.capabilities(request.identity, request.flags_build)) + + +def manifest(request: Request) -> Response: + """``GET /hux/v1/manifest``: the on-disk layout version for rollback readers.""" + request.audit("foundation.manifest", "manifest") + return Response(200, request.store.manifest(CONTRACT_VERSION)) + + +def register(router: Router) -> None: + """Attach HUX-11 routes.""" + router.add("GET", "/hux/v1/capabilities", "HUX-11", "foundation.capabilities", capabilities) + router.add("GET", "/hux/v1/manifest", "HUX-11", "foundation.manifest", manifest) diff --git a/dockerfiles/hermes-hux-foundation/hux/http.py b/dockerfiles/hermes-hux-foundation/hux/http.py new file mode 100644 index 00000000..cda03197 --- /dev/null +++ b/dockerfiles/hermes-hux-foundation/hux/http.py @@ -0,0 +1,213 @@ +"""Minimal HTTP layer for the per-tenant HUX service. + +Stdlib only. A ``Router`` maps method + path template to a handler; family +modules register their routes with it. Every request resolves identity from +trusted headers, checks the owning card's flag, and writes an audit outcome. +Errors always leave as ``hux.error.v1``. +""" + +from __future__ import annotations + +import json +import re +from dataclasses import dataclass, field +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from pathlib import Path +from typing import Any, Callable +from collections.abc import Mapping +from urllib.parse import parse_qs, urlsplit + +from hux import audit +from hux.errors import HuxError, Invalid, NotFound, TooLarge +from hux.flags import Flags, build_from_environ +from hux.identity import Identity, resolve +from hux.store import TenantStore + +MAX_BODY_BYTES = 1024 * 1024 +Handler = Callable[["Request"], "Response"] + + +@dataclass +class Request: + """Everything a handler needs; no raw socket access.""" + + method: str + path: str + params: dict[str, str] + query: dict[str, str] + headers: Mapping[str, str] + body: Any + identity: Identity + store: TenantStore + flags: Flags + flags_build: dict[str, str] = field(default_factory=dict) + + def if_match(self) -> int | None: + """Parsed If-Match revision, or None when absent.""" + raw = self.header("If-Match") + if raw == "": + return None + if not raw.isdigit(): + raise Invalid("If-Match must be a revision integer") + return int(raw) + + def idempotency_key(self) -> str: + """Client idempotency key, validated against the contract pattern.""" + raw = self.header("Idempotency-Key") + if raw and not re.match(r"^[A-Za-z0-9._:-]{8,120}$", raw): + raise Invalid("malformed Idempotency-Key") + return raw + + def header(self, name: str) -> str: + """Case-insensitive header lookup.""" + for key, value in self.headers.items(): + if key.lower() == name.lower(): + return value.strip() + return "" + + def audit(self, action: str, resource: str, outcome: str = "allow", reason: str = "") -> None: + """Write an audit outcome for this request.""" + audit.record(self.store, self.identity, action, resource, outcome, reason) + + +@dataclass +class Response: + """JSON (or SSE) response.""" + + status: int = 200 + body: Any = None + headers: dict[str, str] = field(default_factory=dict) + stream: Callable[[], Any] | None = None + + +@dataclass +class Route: + """One registered handler.""" + + method: str + template: str + card: str + action: str + handler: Handler + pattern: re.Pattern = field(init=False) + + def __post_init__(self) -> None: + regex = re.sub(r"\{(\w+)\}", r"(?P<\1>[A-Za-z0-9._:-]+)", self.template) + self.pattern = re.compile(f"^{regex}$") + + +class Router: + """Route table plus the request pipeline.""" + + def __init__(self, data_root: Path, environ: Mapping[str, str] | None = None) -> None: + self.data_root = Path(data_root) + self.environ = environ + self.flags = Flags(environ) + self.build = build_from_environ(environ) + self.routes: list[Route] = [] + + def add(self, method: str, template: str, card: str, action: str, handler: Handler) -> None: + """Register a handler; ``action`` is the audit action name (family.verb).""" + self.routes.append(Route(method, template, card, action, handler)) + + def match(self, method: str, path: str) -> tuple[Route | None, dict[str, str], bool]: + """Return (route, params, path_known).""" + known = False + for route in self.routes: + found = route.pattern.match(path) + if found: + known = True + if route.method == method: + return route, found.groupdict(), True + return None, {}, known + + def dispatch(self, method: str, raw_path: str, headers: Mapping[str, str], body: bytes) -> Response: + """Run the full pipeline and never raise.""" + parts = urlsplit(raw_path) + query = {k: v[-1] for k, v in parse_qs(parts.query).items()} + try: + identity = resolve(headers, self.environ) + store = TenantStore(self.data_root, identity) + except HuxError as error: + return Response(error.status, error.record()) + route, params, known = self.match(method, parts.path) + if route is None: + error = Invalid("method not allowed") if known else NotFound("no such route") + audit.record(store, identity, "http.route", parts.path, "not_found", error.message) + return Response(405 if known else 404, error.record()) + try: + self.flags.require(route.card) + payload = self._decode(body) + request = Request(method, parts.path, params, query, headers, payload, identity, store, self.flags, self.build) + response = route.handler(request) + except HuxError as error: + outcome = {"flag_off": "flag_off", "conflict": "conflict", "not_found": "not_found"}.get(error.code, "deny") + audit.record(store, identity, route.action, parts.path, outcome, error.message) + return Response(error.status, error.record()) + return response + + @staticmethod + def _decode(body: bytes) -> Any: + if not body: + return None + if len(body) > MAX_BODY_BYTES: + raise TooLarge("body exceeds 1 MiB") + try: + return json.loads(body) + except json.JSONDecodeError as error: + raise Invalid(f"body is not JSON: {error.msg}") from error + + +def page(items: list[Any], next_cursor: Any = None) -> Response: + """Standard list envelope.""" + return Response(200, {"items": items, "next": next_cursor}) + + +def make_handler(router: Router) -> type[BaseHTTPRequestHandler]: + """Bind a Router to a BaseHTTPRequestHandler subclass.""" + + class HuxHandler(BaseHTTPRequestHandler): + server_version = "hux-foundation/1.0" + + def log_message(self, fmt: str, *args: Any) -> None: # noqa: D102 - quiet by design; audit ledger is the log + return + + def _run(self) -> None: + if self.path == "/healthz": + self._send(Response(200, {"status": "ok", "contract_version": "1.0.0"})) + return + length = int(self.headers.get("Content-Length") or 0) + body = self.rfile.read(length) if length else b"" + self._send(router.dispatch(self.command, self.path, dict(self.headers.items()), body)) + + def _send(self, response: Response) -> None: + if response.stream is not None: + self.send_response(response.status) + self.send_header("Content-Type", "text/event-stream") + self.send_header("Cache-Control", "no-cache") + for key, value in response.headers.items(): + self.send_header(key, value) + self.end_headers() + for chunk in response.stream(): + self.wfile.write(chunk) + self.wfile.flush() + return + data = json.dumps(response.body, sort_keys=True).encode() + self.send_response(response.status) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(data))) + for key, value in response.headers.items(): + self.send_header(key, value) + self.end_headers() + self.wfile.write(data) + + do_GET = do_POST = do_PUT = do_PATCH = do_DELETE = _run + + return HuxHandler + + +def serve(router: Router, host: str = "127.0.0.1", port: int = 8790) -> ThreadingHTTPServer: + """Create (but do not start) the server; callers call serve_forever().""" + server = ThreadingHTTPServer((host, port), make_handler(router)) + server.daemon_threads = True + return server diff --git a/dockerfiles/hermes-hux-foundation/hux/identity.py b/dockerfiles/hermes-hux-foundation/hux/identity.py new file mode 100644 index 00000000..6012bea2 --- /dev/null +++ b/dockerfiles/hermes-hux-foundation/hux/identity.py @@ -0,0 +1,80 @@ +"""Caller identity resolved from the trusted hop's headers. + +The chat router, the Telegram relay and the Worker are the only callers. Each +asserts identity in headers; the request body is never trusted for identity. +Relay and worker callers must also present their shared key, compared in +constant time against the value the pod was started with. +""" + +from __future__ import annotations + +import hmac +import os +import re +from dataclasses import dataclass +from collections.abc import Mapping + +from hux.errors import Unauthorized + +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" + +SLOT_RE = re.compile(r"^slot-[0-9]{1,3}$") +SUBJECT_RE = re.compile(r"^usr_[0-9a-f]{16,64}$") +SURFACES = ("chat", "worker", "telegram", "voice", "api") +TRUSTS = ("router", "relay", "worker") +KEY_ENV = {"relay": "HUX_RELAY_KEY", "worker": "HUX_WORKER_KEY"} + + +@dataclass(frozen=True) +class Identity: + """Who is calling, on which surface, asserted by which trusted hop.""" + + tenant_slot: str + subject: str + surface: str + trust: str + + def record(self) -> dict[str, str]: + """Serialise as ``common.identity``.""" + return {"tenant_slot": self.tenant_slot, "subject": self.subject, "surface": self.surface, "trust": self.trust} + + +def _header(headers: Mapping[str, str], name: str) -> str: + for key, value in headers.items(): + if key.lower() == name.lower(): + return value.strip() + return "" + + +def resolve(headers: Mapping[str, str], environ: Mapping[str, str] | None = None) -> Identity: + """Build an Identity from request headers or raise Unauthorized. + + ``trust`` defaults to ``router``; relay and worker trust require the + matching shared key to be configured and presented. + """ + environ = os.environ if environ is None else environ + slot = _header(headers, HEADER_SLOT) + subject = _header(headers, HEADER_SUBJECT) + surface = _header(headers, HEADER_SURFACE) or "chat" + trust = _header(headers, HEADER_TRUST) or "router" + if not SLOT_RE.match(slot): + raise Unauthorized("missing or malformed tenant slot") + own_slot = environ.get("HUX_TENANT_SLOT", "") + if own_slot and slot != own_slot: + raise Unauthorized("tenant slot does not belong to this pod") + if not SUBJECT_RE.match(subject): + raise Unauthorized("missing or malformed subject") + if surface not in SURFACES or trust not in TRUSTS: + raise Unauthorized("unknown surface or trust") + if trust in KEY_ENV: + expected = environ.get(KEY_ENV[trust], "") + presented = _header(headers, HEADER_KEY) + if not expected or not presented or not hmac.compare_digest(expected, presented): + raise Unauthorized(f"{trust} key missing or wrong") + if trust == "router" and surface == "worker": + raise Unauthorized("worker surface needs worker trust") + return Identity(slot, subject, surface, trust) diff --git a/dockerfiles/hermes-hux-foundation/hux/server.py b/dockerfiles/hermes-hux-foundation/hux/server.py new file mode 100644 index 00000000..a22a4eff --- /dev/null +++ b/dockerfiles/hermes-hux-foundation/hux/server.py @@ -0,0 +1,48 @@ +"""Process entry point: ``python -m hux.server``. + +Environment: ``HUX_DATA_ROOT`` (tenant PVC path, default /opt/data), +``HUX_BIND`` (default 127.0.0.1), ``HUX_PORT`` (default 8790), ``HUX_FLAGS``, +``HUX_RELAY_KEY``, ``HUX_WORKER_KEY``, ``HUX_BUILD_COMMIT``, ``HUX_IMAGE_DIGEST``. +""" + +from __future__ import annotations + +import importlib +import os +from pathlib import Path + +from hux.http import Router, serve + +FAMILIES = ("foundation", "events", "memory", "privacy", "organization", "artifacts", "research", "policy") + + +def build_router(data_root: Path, environ: dict[str, str] | None = None) -> Router: + """Create a router with every family module that is present registered.""" + router = Router(data_root, environ) + for name in FAMILIES: + try: + module = importlib.import_module(f"hux.{name}") + except ModuleNotFoundError: + continue + module.register(router) + return router + + +def bind_address(env: dict[str, str]) -> str: + """Loopback only: the router is in the same pod, nothing else may reach the service.""" + bind = env.get("HUX_BIND", "127.0.0.1") + if bind not in {"127.0.0.1", "::1", "localhost"}: + raise SystemExit(f"refusing to bind {bind}: hux-foundation is loopback-only") + return bind + + +def main() -> None: + """Run the server until killed.""" + env = dict(os.environ) + router = build_router(Path(env.get("HUX_DATA_ROOT", "/opt/data")), env) + server = serve(router, bind_address(env), int(env.get("HUX_PORT", "8790"))) + server.serve_forever() + + +if __name__ == "__main__": # pragma: no cover + main() diff --git a/dockerfiles/hermes-hux-foundation/hux/store.py b/dockerfiles/hermes-hux-foundation/hux/store.py new file mode 100644 index 00000000..a453d312 --- /dev/null +++ b/dockerfiles/hermes-hux-foundation/hux/store.py @@ -0,0 +1,223 @@ +"""Per-tenant durable storage on the tenant PVC. + +Every path is derived from the caller's tenant slot and hashed subject, so +there is no shared cross-tenant store and no way to address another tenant's +data. Records are JSON documents with a ``revision`` for optimistic +concurrency; ledgers are append-only JSONL. Writes are atomic (temp file + +fsync + rename) and guarded by one lock per family per tenant. +""" + +from __future__ import annotations + +import json +import os +import re +import threading +import time +from datetime import datetime, timezone +from pathlib import Path +from typing import Any +from collections.abc import Iterator + +from hux.errors import Conflict, Invalid, NotFound, TooLarge +from hux.identity import Identity + +ID_RE = re.compile(r"^[a-z]{2,6}_[A-Za-z0-9._-]{4,80}$") +LAYOUT_VERSION = 1 +MAX_RECORD_BYTES = 256 * 1024 +MAX_FAMILY_RECORDS = 20000 + + +def now_iso() -> str: + """RFC 3339 UTC timestamp with second precision and a Z suffix.""" + return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") + + +def new_id(prefix: str) -> str: + """Opaque, prefix-typed identifier matching common.schema.json#/$defs/id.""" + return f"{prefix}_{os.urandom(8).hex()}{int(time.time() * 1000) % 100000:05d}" + + +def check_id(value: Any) -> str: + """Return ``value`` when it is a well-formed record id, else raise Invalid.""" + if not isinstance(value, str) or not ID_RE.match(value): + raise Invalid("malformed id") + return value + + +def _atomic_write(path: Path, data: bytes) -> None: + tmp = path.with_name(f".{path.name}.{os.getpid()}.{threading.get_ident()}.tmp") + with open(tmp, "wb") as handle: + handle.write(data) + handle.flush() + os.fsync(handle.fileno()) + os.replace(tmp, path) + + +class TenantStore: + """Files for exactly one (tenant slot, subject) pair.""" + + _locks: dict[str, threading.RLock] = {} + _locks_guard = threading.Lock() + + def __init__(self, root: Path, identity: Identity) -> None: + self.identity = identity + self.root = Path(root) / "hux" / f"v{LAYOUT_VERSION}" / identity.tenant_slot / identity.subject + self.root.mkdir(parents=True, exist_ok=True) + + # -- locking ----------------------------------------------------------- + def lock(self, family: str) -> threading.RLock: + """One re-entrant lock per family per tenant directory.""" + key = f"{self.root}:{family}" + with self._locks_guard: + return self._locks.setdefault(key, threading.RLock()) + + # -- documents --------------------------------------------------------- + def _doc_path(self, family: str, record_id: str) -> Path: + directory = self.root / family + directory.mkdir(parents=True, exist_ok=True) + return directory / f"{check_id(record_id)}.json" + + def get(self, family: str, record_id: str) -> dict[str, Any]: + """Read one document or raise NotFound.""" + path = self._doc_path(family, record_id) + if not path.exists(): + raise NotFound(f"{family} {record_id} not found") + return json.loads(path.read_bytes()) + + def exists(self, family: str, record_id: str) -> bool: + """True when the document is present.""" + return self._doc_path(family, record_id).exists() + + def put(self, family: str, record: dict[str, Any], expected_revision: int | None = None) -> dict[str, Any]: + """Create or replace a document, bumping ``revision``. + + ``expected_revision`` implements If-Match: it must equal the stored + revision (or be None for a create of a new id) or Conflict is raised. + """ + record_id = check_id(record.get("id")) + with self.lock(family): + path = self._doc_path(family, record_id) + current = json.loads(path.read_bytes()) if path.exists() else None + if current is None: + if expected_revision not in (None, 0): + raise Conflict("record does not exist yet") + if self.count(family) >= MAX_FAMILY_RECORDS: + raise TooLarge(f"{family} is full") + revision = 1 + else: + if expected_revision is not None and expected_revision != current.get("revision"): + raise Conflict(f"revision {expected_revision} does not match current revision {current.get('revision')}") + revision = int(current.get("revision", 0)) + 1 + stored = {**record, "revision": revision} + data = json.dumps(stored, sort_keys=True, separators=(",", ":")).encode() + if len(data) > MAX_RECORD_BYTES: + raise TooLarge("record exceeds size bound") + _atomic_write(path, data) + return stored + + def delete(self, family: str, record_id: str) -> None: + """Remove a document; missing is not an error.""" + path = self._doc_path(family, record_id) + with self.lock(family): + if path.exists(): + path.unlink() + + def count(self, family: str) -> int: + """Number of documents in a family.""" + directory = self.root / family + return sum(1 for p in directory.glob("*.json")) if directory.exists() else 0 + + def scan(self, family: str) -> Iterator[dict[str, Any]]: + """Yield every document in a family, oldest file first.""" + directory = self.root / family + if not directory.exists(): + return + for path in sorted(directory.glob("*.json"), key=lambda p: (p.stat().st_mtime_ns, p.name)): + yield json.loads(path.read_bytes()) + + # -- ledgers ----------------------------------------------------------- + def _ledger_path(self, family: str, name: str) -> Path: + directory = self.root / family + directory.mkdir(parents=True, exist_ok=True) + if not re.match(r"^[A-Za-z0-9._-]{1,120}$", name): + raise Invalid("malformed ledger name") + return directory / f"{name}.jsonl" + + def append(self, family: str, name: str, record: dict[str, Any]) -> None: + """Append one JSON line with fsync.""" + line = json.dumps(record, sort_keys=True, separators=(",", ":")).encode() + b"\n" + if len(line) > MAX_RECORD_BYTES: + raise TooLarge("ledger record exceeds size bound") + with self.lock(family): + path = self._ledger_path(family, name) + with open(path, "ab") as handle: + handle.write(line) + handle.flush() + os.fsync(handle.fileno()) + + def read(self, family: str, name: str) -> list[dict[str, Any]]: + """Read a whole ledger; a torn trailing line from a crash is dropped.""" + path = self._ledger_path(family, name) + if not path.exists(): + return [] + rows: list[dict[str, Any]] = [] + for line in path.read_bytes().split(b"\n"): + if not line: + continue + try: + rows.append(json.loads(line)) + except json.JSONDecodeError: + break + return rows + + def ledgers(self, family: str) -> list[str]: + """Names of the ledgers in a family.""" + directory = self.root / family + return sorted(p.stem for p in directory.glob("*.jsonl")) if directory.exists() else [] + + def rewrite(self, family: str, name: str, rows: list[dict[str, Any]]) -> None: + """Replace a ledger atomically (used by retention purges).""" + data = b"".join(json.dumps(r, sort_keys=True, separators=(",", ":")).encode() + b"\n" for r in rows) + with self.lock(family): + _atomic_write(self._ledger_path(family, name), data) + + # -- blobs ------------------------------------------------------------- + def put_blob(self, digest: str, data: bytes) -> Path: + """Store content-addressed bytes; the caller has already verified the digest.""" + if not re.match(r"^[0-9a-f]{64}$", digest): + raise Invalid("malformed digest") + directory = self.root / "blobs" / digest[:2] + directory.mkdir(parents=True, exist_ok=True) + path = directory / digest + if not path.exists(): + _atomic_write(path, data) + return path + + def get_blob(self, digest: str) -> bytes: + """Read content-addressed bytes or raise NotFound.""" + if not re.match(r"^[0-9a-f]{64}$", digest): + raise Invalid("malformed digest") + path = self.root / "blobs" / digest[:2] / digest + if not path.exists(): + raise NotFound("blob not found") + return path.read_bytes() + + # -- manifest ---------------------------------------------------------- + def manifest(self, contract_version: str) -> dict[str, Any]: + """Read or create the ``hux.manifest.v1`` for this tenant directory.""" + path = self.root / "MANIFEST.json" + with self.lock("manifest"): + if path.exists(): + return json.loads(path.read_bytes()) + stamp = now_iso() + record = { + "schema": "hux.manifest.v1", + "contract_version": contract_version, + "data_layout_version": LAYOUT_VERSION, + "min_reader_contract_version": "1.0.0", + "created_at": stamp, + "updated_at": stamp, + } + _atomic_write(path, json.dumps(record, sort_keys=True).encode()) + return record diff --git a/docs/hux/DATA-MODEL.md b/docs/hux/DATA-MODEL.md new file mode 100644 index 00000000..9d729061 --- /dev/null +++ b/docs/hux/DATA-MODEL.md @@ -0,0 +1,353 @@ +# HUX foundation: on-PVC data model + +`hux-foundation` is a Python stdlib service that runs inside every tenant pod +and owns the records defined in `services/hermes/contracts/hux/`. It has no +database and no storage shared between tenants. Everything it keeps lives on +the tenant's own `home` PVC (mounted at `/opt/data`, beside `webui/`, +`workspace/`, `home/` and the Telegram media roots `cache/images` and +`workspace`). The Go router in `services/hermes/router/` fronts it: the router +resolves Keycloak identity to a tenant slot and forwards the +`common.schema.json#/$defs/identity` tuple (`tenant_slot`, `subject`, +`surface`, `trust`) as headers, and enforces `hux.foundation` before any +request reaches this service. The service trusts nothing else about the +caller. + +The conventions below are lifted from what already works in this repo: +`hermes_model_routing._atomic_write` (temp file + `os.replace`), +`cli_lane_records.atomic_json` (journal first, act second), and +`execution_pool_store` (idempotent add by digest, terminal rows only are ever +garbage-collected). SQLite is deliberately not used: the WebUI already owns +`state.db` on this volume and a second writer with its own WAL is one more +thing to recover; JSON and JSONL are greppable during an incident. + +## 1. Directory layout + +Tenant root is `HUX_DATA_ROOT`, default `/opt/data/hux`. The layout version is +a directory (`v1`), so a v2 layout can be built beside v1 and swapped by +`MANIFEST.json`, never by rewriting v1 in place. + +``` +/opt/data/hux/ + MANIFEST.json hux.manifest.v1 (identity.schema.json): contract and layout versions + v1/ + .lock advisory lock file (flock) proving single writer per tenant + users// one subtree per hashed subject; nothing lives above it + profile.json {schema:"hux.user_profile.v1", memory_enabled, default_mode, revision} + events/ + / + events.jsonl append-only hux.event.v1, one per line, ordered by seq + seq.json {next_seq, last_event_id, bytes, checkpointed_at} + idempotency.jsonl {idempotency_key, event_id, seq, at}; last 10k keys kept + memory/ + ledger.jsonl append-only hux.memory.v1 snapshots; a status change appends a full record + tombstones.jsonl {memory_id, at, reason, purged:bool}; written on forget and on decay + index.json retrieval index (section 3), rebuilt from the ledger on demand + exports/.json GET /memory/export snapshots, audited, pruned after 7 days + projects/ + .json hux.project.v1 + revision + index.json {items:[{id,name,tags,pinned,archived,updated_at}], revision} + conversations/ + .json hux.conversation.v1 + revision + index.json {items:[{id,project_id,title,tags,pinned,archived,mode,branch,last_message_at}]} + search/ + .json per-conversation term postings for message_text and artifact_titles + artifacts/ + .json hux.artifact.v1 + revision (versions[] is the version list) + index.json {items:[{id,type,title,conversation_id,project_id,current_version}]} + blobs/sha256// content-addressed, immutable, 0400, fan-out on first two hex chars + blobs/refs.json {hash: [ "@", ...]} for safe purge + research/ + sources/.json hux.source.v1 + passages/.json hux.passage.v1 (text + locator; hash is the dedupe key) + citations/.jsonl hux.citation.v1 per message, append-only + notebooks/.json hux.research_notebook.v1 + revision + index.json {by_conversation:{conv_id:[nb_id]}, by_message:{msg_id:count}, passage_hashes:{hash:psg_id}} + policy/ + global.json hux.policy.v1 for scope level global + project/.json hux.policy.v1 per project + conversation/.json hux.policy.v1 per conversation + approvals/.json hux.approval.v1; terminal records never change again + approvals/pending.json {items:[apr_id], revision}; the queue the UI polls + receipts/.json hux.cancel_receipt.v1 + receipts/by_run.json {run_id: rcpt_id} + suggestions/ + state.json {schema:"hux.suggestion_states.v1", items:{sug_id: hux.suggestion_state.v1}, revision} + privacy/ + notices.jsonl hux.privacy_notice.v1 as shown, append-only + conversation_topics.json {conv_id:{topic, first_seen, decay_at, memory_disabled:bool}} + forgotten.jsonl {conv_id, requested_at, purged_at, counts} + audit/ + outcomes/.jsonl one hux.audit_outcome.v1 per read or mutation (section 6) + retention/.json hux.retention_audit.v1 per run + catalog/ + suggestions.json hux.suggestion.v1 catalog shipped with the image (read-only copy) + privacy_policy.json hux.privacy_policy.v1 as served, with the policy version +``` + +The pod is single-tenant, so the `users/` level is not multi-tenancy; it is +the guarantee that every path contains the hashed subject the router asserted +and that a second subject on the same pod (operator break-glass, future +household sharing) can never see another's tree without a distinct path. + +Record ids are minted by the service: `_<6 random base32>`, e.g. +`evt_m0k3xq9a2bcd7f`, so they sort by creation time and satisfy the +`common.schema.json` id pattern. `hux.audit_outcome.v1` and +`hux.user_profile.v1` are foundation-internal records; they follow the same +provenance and versioning rules but are not contracts the UI codes against. + +## 2. Write semantics + +**Atomic document write.** Serialise with `json.dumps(sort_keys=True, +ensure_ascii=False)` plus a trailing newline, write to +`....tmp` in the same directory, `flush()`, `os.fsync(fd)`, +`os.replace(tmp, final)`, then `os.fsync(dir_fd)` so the rename itself is +durable. Temp files that survive a crash are deleted on open. Blobs are +written the same way under their hash; an existing blob is never rewritten +(compare size and hash; on mismatch refuse and audit). + +**Append.** JSONL families open with `O_APPEND`, write the whole line in one +`os.write`, and `fsync` the file before the HTTP response is sent. The +controlling checkpoint (`seq.json`, `index.json`) is written atomically after +the append; a checkpoint may lag the log, never lead it. + +**Crash recovery.** On first access to a family the store validates the log: +each line must parse and, for events, `seq` must equal the previous `seq + 1`. +A trailing partial line (no newline, or JSON error on the last line only) is +truncated to the last good newline; a bad line anywhere else is a hard error +(the family is marked read-only, an audit outcome is written, `/healthz` +reports `degraded`). The checkpoint is then reconciled from the log: +`next_seq` = last good `seq + 1`, indexes rebuilt if `checkpointed_at` is +older than the log mtime. + +**Idempotency.** Every mutating request may carry `Idempotency-Key` +(`common.schema.json#/$defs/idempotency_key`, `^[A-Za-z0-9._:-]{8,120}$`). For +events the key is stored in `idempotency.jsonl`; a replay returns the original +event (same `id`, same `seq`) with `HTTP 200` and `HUX-Replayed: true`. For +documents the key is stored in the record's `_meta.idempotency_keys` (last 16) +and a replay returns the current record without bumping `revision`. Event +`id` is unique per tenant; an append whose `id` already exists in the last +checkpoint window is treated as a replay, not a duplicate. + +**Seq allocation.** The HTTP server is a single process, +`ThreadingHTTPServer`. Locks are per path, held in a process-wide +`dict[str, threading.RLock]` guarded by one `threading.Lock`. The unit of +locking is the family directory for JSONL (`events/`, `memory/`) and the +document path for JSON. Seq is allocated inside the conversation lock: +read `next_seq` from memory (loaded from `seq.json` once per open), assign, +append, fsync, write `seq.json`, release. Since one process owns the PVC, +`v1/.lock` is taken with `fcntl.flock(LOCK_EX|LOCK_NB)` at start-up so an +accidental second replica fails fast instead of interleaving appends. + +**Optimistic concurrency.** Every JSON document carries an integer +`revision` (starts at 1) beside the contract fields; it is served as the +`ETag`. `PATCH`/`PUT` require `If-Match: `; a mismatch returns +`409` with a `hux.error.v1` body (`code: conflict`, the current revision in +`details`) as `common.schema.json#/$defs/revision` specifies. A missing `If-Match` is accepted only when the +request carries an `Idempotency-Key`, and then the write is last-writer-wins +with the outcome audited as `why: "unconditional_write"`. Indexes carry their +own `revision` and are rewritten under the family lock after the document. + +**Caps.** Rejected with `hux.error.v1` `too_large` (413, size) or `conflict` +(409, count): +- event line 64 KiB, `detail` 32 KiB; 50 000 events per conversation, then + the conversation is `archived` and further appends need a branch +- memory content 2 000 chars (schema), 5 000 live entries, ledger 64 MiB + before compaction (section 4) +- conversation and project documents 256 KiB; 2 000 conversations, 200 + projects, `artifact_ids` 500 (schema) +- artifact blob 25 MiB (`tenantMediaLimit` is 50 MiB; half leaves room for + the Telegram path), 200 versions per artifact, 2 000 artifacts, blob store + 2 GiB per tenant +- passage text 4 000 chars, 10 000 sources, 50 000 passages +- audit outcomes are never capped by count; they rotate daily and age out + +## 3. Indexes + +Nothing is indexed that a scan cannot rebuild; every `index.json` is a cache +of its family and carries `built_from` (log bytes or document count) so a +stale index is detected and rebuilt rather than trusted. + +**Conversation search** (`GET /hux/v1/search?q=`) covers exactly +`project.schema.json#/$defs/search_index`: `title`, `tags`, `project_name` +come from `conversations/index.json` joined with `projects/index.json` in +memory (a few thousand rows, scanned per query); `message_text` and +`artifact_titles` come from `conversations/search/.json`, a per +conversation bag of lowercased, punctuation-stripped terms with positions, +updated when a `message.*` event or `artifact.*` event is appended. Query +terms are ANDed; ranking is title hit > tag hit > term frequency > recency. +Events with `sensitivity: restricted` or `redaction.level: full` never enter +the search bag, so a search result can never leak what the timeline hides. + +**Memory retrieval** (`memory/index.json`) is a term index over `content` of +`active` entries only, keyed by `scope.level`/`scope_id` and `topic`, plus +`expires` sorted by the effective expiry (`expires_at` or `created_at + +decay_days`) for the retention job. Entries are added on `approved` +(automatic or user), removed on `rejected`, `expired`, `forgotten`. Retrieval +reads the index, then loads the newest ledger snapshot for each id and drops +any whose status is no longer `active`, so a lagging index fails safe. + +**Tombstones remove from retrieval before content is gone.** `forget` is two +writes under the memory lock: append a ledger snapshot with +`status: forgotten`, `content: ""`, audit `forgotten`; then append +`tombstones.jsonl` `{memory_id, at, reason, purged:false}`. The index write +follows. Both the retrieval path and the export path consult the tombstone +set (loaded once, appended in memory) before returning anything, which is +what makes "do not remember" hold even if the index rebuild is interrupted. +The earlier ledger lines that still contain content are what the purge job +in section 4 rewrites. The same tombstone file records `disable_memory_here` +per conversation via `privacy/conversation_topics.json`, which retrieval also +checks: a conversation with `memory_disabled` contributes no entries and +receives none. + +**Research** indexes by hash: `passage_hashes` dedupes passages across +sources, `by_message` lets `GET /messages/{id}/citations` open one JSONL +without listing a directory, `by_conversation` backs the notebook drawer. +Approvals index only the pending queue; terminal approvals are found by id. + +## 4. Retention + +One thread runs the retention job every `retention_audit.interval_days` +(1 day) at a jittered hour, and on demand via `POST /hux/v1/admin/retention` +(worker surface only). Each run writes `audit/retention/.json` as +`hux.retention_audit.v1` with the counts the schema names, so a day without +the record is itself a finding. + +- `expire_memory`: entries whose effective expiry has passed get a new ledger + snapshot `status: expired` and leave the index. Decay means the entry + expires `decay_days` after `created_at` unless a later ledger snapshot + carries a newer `updated_at` from an `approved` audit action, which resets + the clock once. +- `decay_topic_context`: `privacy/conversation_topics.json` rows past + `decay_at` (`PRIVACY_TOPICS[topic].decay_days` after `first_seen`) cause + the conversation's events with that `sensitivity` to be rewritten with + `redaction.level: full` and `detail` removed; `summary` is replaced by the + topic notice text. Seq, ids and provenance are kept, so the timeline stays + contiguous. +- `purge_forgotten_content`: for every tombstone with `purged:false`, rewrite + `ledger.jsonl` (temp + replace, under the memory lock) replacing `content` + with `""` on every snapshot of that id, keeping `audit[]`; then set + `purged:true`. Forgotten conversations (`POST /conversations/{id}/forget`) + are handled the same way: `events.jsonl` is rewritten with `detail` + dropped and `redaction.level: full`, `search/.json` is deleted, + artifacts owned only by that conversation lose their blobs (via + `blobs/refs.json`), and `privacy/forgotten.jsonl` records the counts. The + conversation document stays with `archived: true` so branches still + resolve their parent. +- `report`: bounds check. Per-family bounds for a home cluster (10 Gi PVC + shared with the WebUI): events 1 GiB total and 180 days for archived + conversations, memory ledger compacted when over 64 MiB (rewrite keeping + only the newest snapshot per id plus every snapshot of ids with a + tombstone), blobs 2 GiB with unreferenced blobs deleted 7 days after their + last ref disappears, research 512 MiB and sources unreferenced by any + notebook or citation for 90 days deleted, audit outcomes 90 days, + retention audits 400 days, memory exports 7 days, idempotency keys 10 000 + per conversation. Terminal approvals and receipts are kept 180 days. + Nothing in `audit/` is ever removed by a forget or purge; only age. + +Private mode conversations (`retention: ephemeral`) are not written to +`events/` at all; the service returns `204` to appends and the router's +session memory is the only copy. + +## 5. Migrations + +`MANIFEST.json`: + +```json +{"schema":"hux.manifest.v1","contract_version":"1.0.0","data_layout_version":1, + "min_reader_contract_version":"1.0.0","created_at":"...","updated_at":"..."} +``` + +`contract_version` is `services/hermes/contracts/hux/VERSION`; the service +that last opened the tree writes its own version there and bumps +`updated_at`. `data_layout_version` names the `v/` directory in use. + +Rules: +- Additive only within a layout. A release may add optional fields to a + record, add a new file name, add a new directory, add an enum value that a + reader can ignore. It may not rename or remove a field, change a field's + type, change the meaning of an existing enum value, change the id pattern, + change `seq` semantics, or move a family to a different path. +- Every record keeps its contract `schema` value (`hux.event.v1`). Internal + bookkeeping is under `_meta` (`revision`, `idempotency_keys`, `built_from`), + which is stripped before a record is served and which readers must ignore. +- Readers ignore unknown fields and never fail a family because one record + has an unknown optional key. A reader from the previous release therefore + reads records from the next one; a rollback of the service leaves every + file readable because the older code sees the same required fields. +- A record whose `schema` is a version the running code does not know is + skipped on list and returned `409 hux.schema_unknown` on direct fetch, and + logged once per family per process. +- The service refuses to open a tree whose `data_layout_version` is + greater than the one it was built for, or whose + `min_reader_contract_version` is above its own `VERSION`, and it never + bumps either on its own; a v2 layout is a separate migration tool that + builds `v2/` beside `v1/` and flips the manifest last. A release that only + adds optional fields leaves `min_reader_contract_version` alone, which is + exactly what lets the previous release read after a rollback. +- Index files are never migrated; they are deleted and rebuilt. + +## 6. Authorization + +The router asserts the identity tuple: `tenant_slot` (`^slot-[0-9]{1,3}$`), +`subject` (`usr_`), `surface` and `trust` (`router|relay|worker`); the +service checks the slot equals its own `HERMES_TENANT_SLOT` and refuses +otherwise. Every filesystem path is then built by `store.path_for(user, +family, *ids)`, where `user` must match `^usr_[0-9a-f]{16,64}$` and each id +must match the `common.schema.json` id pattern +`^[a-z]{2,6}_[A-Za-z0-9._-]{4,80}$` (dates in `audit/` match +`^\d{4}-\d{2}-\d{2}$`, versions `^\d+$`, hashes `^[0-9a-f]{64}$`). The +pattern admits `.` but not `/`, so `..` alone is impossible in an id; the +resolved path is still checked with `os.path.commonpath` against the user's +subtree, the same belt-and-braces the router applies in +`normalizeTenantMediaPath`. No caller-supplied string is ever joined into a +path without going through `path_for`. + +Ownership: records with an `owner` field must equal the asserted user on +read and write; records without one (events, approvals, receipts, research) +are owned through their `conversation_id`, whose document is loaded and +checked first. A lookup that fails ownership returns `404`, not `403`, so +ids cannot be probed. + +Every read and mutation appends one line to `audit/outcomes/.jsonl` +shaped as `common.schema.json#/$defs/audit_outcome`, also for denials. The +line is the contract object plus an `_meta` envelope the store owns: + +```json +{"at":"...","identity":{"tenant_slot":"slot-3","subject":"usr_...","surface":"chat","trust":"router"}, + "action":"memory.forget","resource":"memory/mem_...","outcome":"allow","reason":"owner_match", + "_meta":{"schema":"hux.audit_outcome.v1","id":"aud_...","request_id":"...","idempotency_key":"...", + "revision_before":4,"revision_after":5,"build":{"commit":"...","image_digest":"sha256:..."}}} +``` + +`action` is `.`; `resource` is the family-relative record path; +`outcome` is `allow|deny|not_found|conflict|flag_off`; `reason` is a short +fixed vocabulary (`owner_match`, `owner_mismatch`, `invalid_id`, +`revision_conflict`, `cap_exceeded`, `policy_violation`, +`unconditional_write`, `replayed`, `family_readonly`). Audit lines never +contain record content. + +## 7. Module map + +All modules live in the `hux` package under `dockerfiles/hermes-hux-foundation/` +(where `contracts.py` and `rules.py`, the former `hux_contracts.py` and +`hux_policy.py`, already sit), stdlib only, each at most 500 lines, and each +tested in `testing/tests/test_hermes_hux_foundation_*.py`. + +| Module | Responsibility | +|---|---| +| `identity.py` | Parse and validate router headers (slot, `usr_` hash, request id); id and hash regexes from `common.schema.json`; id minting; ownership check helpers. | +| `flags.py` | Wrap `rules.flag_enabled` with per-request evaluation of `HUX_FLAGS`; map each route to its card flag; 404 when the chain is off. | +| `store.py` | `path_for`, `MANIFEST.json`, `.lock`, per-path lock registry, atomic document write, fsync'd append, tmp-file cleanup, crash validation and truncation, revision/`If-Match`, idempotency-key storage, size caps. No record semantics. | +| `events.py` | Per-conversation JSONL log: seq allocation, `seq.json`, `after_seq` reads, SSE cursor, search-bag updates, redaction on read, ephemeral-mode short circuit. | +| `memory.py` | Ledger append, state machine via `rules.MEMORY_TRANSITIONS`, `memory_policy_violations`, tombstones, retrieval index, export snapshots, compaction. | +| `privacy.py` | Topic detection hooks, `conversation_topics.json`, notices log, `forget` for a conversation, retention job (`expire_memory`, `decay_topic_context`, `purge_forgotten_content`, `report`) and `hux.retention_audit.v1`. | +| `artifacts.py` | Artifact documents, version list, content-addressed blobs and `refs.json`, diff between versions, promotion, lineage, blob GC. | +| `research.py` | Sources, passages (hash dedupe), per-message citation logs, notebooks, research index. | +| `policy.py` | Policy documents per scope, `effective_decision`, approvals queue and terminal transitions, cancellation receipts and `by_run.json`. | +| `organization.py` | Projects and conversations, indexes, branch lineage, search over the `search_index` fields, suggestion state and `suggestion_allowed` gating. | +| `audit.py` | `hux.audit_outcome.v1` writer with daily rotation, the `why` vocabulary, structured logging of degraded families, age-based pruning. | +| `http.py` | `ThreadingHTTPServer` on the tenant loopback port, route table for `/hux/v1`, JSON and SSE responses, `{"items":[],"next":cursor}` list wrapping, error mapping (`404/409/412/413`), `/healthz`. | + +Dependency direction is one way: `http` -> family modules -> `store`, with +`identity`, `flags` and `audit` used by everyone and importing only `store` +and `rules`/`contracts`. Family modules never touch the filesystem directly. diff --git a/docs/hux/THREAT-MODEL.md b/docs/hux/THREAT-MODEL.md new file mode 100644 index 00000000..6ea2b2a5 --- /dev/null +++ b/docs/hux/THREAT-MODEL.md @@ -0,0 +1,247 @@ +# hux-foundation threat model + +Scope: the per-tenant `hux-foundation` service (Python, stdlib +`ThreadingHTTPServer`) that runs as a sidecar in each `hermes-chat-tenant-N` +pod, keeps HUX records on that pod's `home` PVC under `/opt/data/hux/v1/` +(layout in `DATA-MODEL.md`), and is reached only through the chat router, +the Worker, and the Telegram relay. It implements the frozen 1.0.0 contracts +in `services/hermes/contracts/hux/` (ADR-0001) with the rules in +`dockerfiles/hermes-hux-foundation/hux/rules.py`. Where a NetworkPolicy or +pod setting is needed it is written as a requirement for Codex, not a +manifest. Every mitigation cites a contract field, a code obligation, or an +SO id from the checklist at the end. + +Assets, most important first: + +1. Tenant isolation: one Keycloak subject, one pod, one PVC + (`chat-statefulset.yaml`, `ai.bstein.dev/isolation`). Everything rests on it. +2. Secrets that transit the pod: the relay key at + `/runtime-access/chat-relay-key`, provider keys in `/opt/data/.env`, and + the OAuth headers the router already strips (`main.go` Director). +3. User content: memory, artifacts, sources, events. Sensitive and restricted + topics have hard rules in `PRIVACY_TOPICS` and `memory_policy_violations`. +4. Honesty of the record: `seq`, approval decisions, cancel receipts, budget + state, release evidence. If these can be forged the UI lies to the user. + +Attackers, in the order we expect to meet them: + +- A: another tenant's Hermes process or a tool it ran. Shell, Python, + Playwright, and a same-namespace address. +- B: this tenant's own agent gone wrong (prompt injection via web or files). + Tenant privileges, plus tool arguments and logs full of things the user + never meant to keep. +- C: the tenant's browser session, possibly a stolen `hermes_chat_session` + cookie, driving `/hux/v1` through the router. +- D: a linked Telegram peer, or a leaked relay key. +- E: operator mistakes: a flag on without its chain, an old binary reading + new records, a purge with a bad glob. + +## 1. Tenant and surface identity + +The router is the only identity authority. `slotFor` maps subject to slot, +the Director strips `Authorization`, `Cookie`, `X-Auth-Request-*` and +`X-Forwarded-*`, and sets `X-Hermes-Tenant-Identity: slot-N`. Nothing in the +pod ever sees a raw subject. ADR-0001 adds `X-Hux-Subject: usr_` +(`identityHash("keycloak", subject)`) and `X-Hux-Surface`. The service +turns those into `common.identity` (`tenant_slot`, `subject`, `surface`, +`trust`) and stamps it on every record a caller can create. + +| Threat | STRIDE | Who | Mitigation | +|---|---|---|---| +| Pod-network client sends `X-Hermes-Tenant-Identity: slot-2` straight to slot 2's hux port | S | A | Bind `127.0.0.1` only; the WebUI sidecar is the sole in-pod caller and forwards the router's headers verbatim. No cross-pod path exists to spoof. (SO-01) | +| Someone reaches 8787/8642/8788 from a pod that is not the router | S | A | `hermes-chat-tenant-isolation` already admits those ports only from `app: hermes-chat-router`. Requirement: no hux port is ever added to that ingress list; the sandbox pods keep their single-source ingress. (SO-02) | +| Browser sends `X-Hux-Subject`, router forwards it | S | C | Router deletes inbound `X-Hux-*` and `X-Hermes-Tenant-Identity` before setting its own, exactly as it deletes `Authorization`. Service refuses a slot header that differs from its own ordinal (`HERMES_TENANT_SLOT`). (SO-03) | +| Two header vocabularies (ADR: `X-Hux-Subject`; DATA-MODEL §6: `X-Hermes-User`) and one hop honours the wrong one | S, E | E | One header set, defined once in `identity.py`, with the other names rejected as `401 unauthorized`. (SO-04) | +| Telegram relay carries the shared relay key and no subject | S, E | D | `Authorization: Bearer ` compared with `hmac.compare_digest` (as `telegram_media_server.py` does); `X-Hux-Surface: telegram`, `trust: relay`. The owner is the slot's single assignment. A relay request that names `X-Hux-Subject` is `401`. (SO-05, SO-06) | +| Relay key ends up in a record or log | I | B | Key lives only under `/runtime-access` (read-only mount, not the PVC). It, and every value in `/opt/data/.env`, is a literal canary in the scrub of section 2. (SO-07) | +| Worker reads a tenant's user data through `trust: worker` | E, I | E | `trust: worker` may call `GET /hux/v1/releases`, `GET /hux/v1/capabilities` and `POST /hux/v1/admin/retention` only; any other route is `403 forbidden`. No worker bypass port. (SO-08) | +| Slot handed to a new subject; old subtree leaks | I | E | Paths are `users//...`; reads join on `identity.subject`. A different hash sees an empty tree, never another's. (SO-09) | +| Body claims `provenance.surface: worker` or `actor.type: user` | R, T | B, C | Server overwrites `identity`, `provenance.surface`, `provenance.actor` (when `type: user`) and `recorded_at` from the trusted hop and its own clock. (SO-10) | + +NetworkPolicy requirements for Codex: router-only ingress to every listening +port in the tenant pod; no egress from hux-foundation (it fetches nothing); +sandbox pods gain no route to the tenant's loopback services. + +## 2. Activity events + +`detail` is schema-open ("kind-specific payload"). Tool arguments carry file +paths, shell lines with tokens, URLs with query strings, pasted secrets. +The pipeline, in order, before any byte is written: drop keys outside the +per-kind allowlist; scrub every remaining string; cap sizes; raise +`redaction.level` if anything fired; then and only then allocate `seq`. + +| Threat | STRIDE | Who | Mitigation | +|---|---|---|---| +| `tool.call` detail persists raw arguments | I | B | Allowlist below; unknown keys dropped before write, never filtered on read only. (SO-11) | +| Secret inside an allowed string (`summary`, `detail.target_path`) | I | B | Pattern scrub: `sk-[A-Za-z0-9]{20,}`, `ghp_|gho_|glpat-`, `AKIA[0-9A-Z]{16}`, `xox[abp]-`, `Bearer [A-Za-z0-9._-]{16,}`, `-----BEGIN [A-Z ]*PRIVATE KEY`, JWT `eyJ[A-Za-z0-9_-]{20,}\.`, hex/base64 runs ≥ 40 chars, plus the literal canaries from SO-07. Match becomes `[redacted:]`, `redaction.level` ≥ `partial`, `redaction.reason` set. (SO-12) | +| Oversized detail fills the PVC or an SSE buffer | D | B | `detail` > 32 KiB (DATA-MODEL cap) is replaced by `{"truncated": true, "bytes": N}`; line cap 64 KiB; `evidence` ≤ 64 and `summary` ≤ 280 per schema. (SO-13) | +| Reader ignores `redaction.level` | I | C, D | Enforced at serve time too: `none` as stored; `partial` strips `detail`; `full` serves `kind`, `seq`, `ts`, `sensitivity`, `redaction` only. `surface: telegram` and `voice` never receive `none`. (SO-14) | +| Client supplies `seq` or `id` | T | B, C | Server-assigned under the conversation lock; a body carrying either is `400 invalid`. (SO-15) | +| Replay through `Idempotency-Key` returns someone else's event | I | C | Idempotency keys are scoped per conversation file; a replay is served only after the ownership check. (SO-16) | +| Reconnect storm, `after_seq=0` on a long log | D | C | Pages ≤ 200; `after_seq` before the retention window resumes from the oldest kept event; one SSE stream per (subject, conversation), a second open closes the first; idle timeout 15 min. (SO-17) | +| `GET /conversations/{guess}/events` | I | C | Ids are time-prefixed plus 30 random bits, so ownership is the real gate: `conversation_id` document loaded first, mismatch is `404 not_found`, not `403`. (SO-18) | +| `evidence[].uri` of `file:///opt/data/.env` expanded by a UI | I | B | Service never dereferences `evidence_ref.uri`; stored URIs must be `https://`, `hux://` or `artifact://`, others dropped. (SO-19) | +| Restricted event leaks via search | I | C | `sensitivity: restricted` or `redaction.level: full` events never enter `search/.json`. (SO-20) | + +Detail allowlist (anything else is dropped): + +- `message.*`: `message_id`, `chars`, `has_attachments` +- `decision.route`: `requested`, `resolved_target`, `provider`, `effort`, `reason` +- `decision.plan`: `steps` (≤ 20 strings ≤ 200 chars) +- `tool.call`: `tool`, `capability`, `argument_names`, `argument_hash`, + `target_path` (kept only under `/opt/data/workspace`, else dropped) +- `tool.result`: `tool`, `ok`, `duration_ms`, `bytes`, `exit_code` +- `approval.*`, `side_effect.*`: `approval_id`, `capability`, `choice`, `external` +- `memory.*`: `memory_id`, `kind`, `sensitivity`, `topic` +- `artifact.*`: `artifact_id`, `version`, `type`, `bytes`, `hash` +- `run.*`, `delegation.*`, `budget.exhausted`: `run_id`, `outcome`, + `receipt_id`, `spent`, `limits`, `child_run_id` +- `privacy.notice`: the `hux.privacy_notice.v1` fields only +- `citation.attached`, `mode.changed`, `suggestion.*`, `release.transition`: + ids and enum fields of their own schemas only + +## 3. Memory + +Defaults: `approval_mode: ask` for anything above `personal`; `no_store` is +a first-class status (`memory.status`, `memory.approval_mode`) so "do not +remember" is recorded as a decision rather than as absence. A proposal +unanswered for 7 days expires. + +| Threat | STRIDE | Who | Mitigation | +|---|---|---|---| +| Restricted content written because the topic detector missed it | I | B | `memory_policy_violations` on every write and transition; any problem is `422`/`invalid` with a `memory.suppressed` event. The service never lowers `sensitivity`; its own scrub may raise it. (SO-21) | +| "Do not remember" blocks the write but the entry surfaces later | I | B, C | `no_store` writes a content-free ledger line and a tombstone; `forget` writes `status: forgotten`, `content: ""`, `retrievable: false`, tombstone. Retrieval, `/search`, `/export` and the agent read hook check the tombstone set before the index and return only `status: active` with `retrievable: true`. (SO-22, SO-23) | +| Forgotten content lingers in earlier events or ledger lines | I | B | `memory.*` events carry ids only. `forget` re-redacts events referencing the id to `full` and queues `purge_forgotten_content` to rewrite older ledger snapshots. (SO-24) | +| Correction resurrects forgotten text | T | B | An entry whose `supersedes` or `source` points at a forgotten id needs `approval_mode: ask` answered by a `user` actor. (SO-25) | +| Export leaks another owner or forgotten entries | I | C | Export is owner-scoped, excludes `forgotten`, `rejected`, `expired`, `no_store` and `retrievable: false`, appends `exported` to `audit[]`, is served as an attachment, and the snapshot file is pruned after 7 days. (SO-26) | +| TTL never enforced because the audit job died | I | E | Expiry checked lazily at read (an expired entry is never served as active) and eagerly by the retention job; a missing `hux.retention_audit.v1` older than 48 h makes `/privacy/policy` report `audit_stale: true`. (SO-27) | +| Private mode writes memory | I | B | `MODE_CATALOG["private"].memory.write == False`: `POST /memory` is refused while the conversation's mode is `private`, and its events are not written at all (`204`). (SO-28) | +| `disable_memory_here` ignored on the read side | I | B | `privacy/conversation_topics.json` `memory_disabled` is checked by retrieval: that conversation contributes and receives nothing. (SO-23) | + +## 4. Artifacts and research + +Sources are records about where evidence came from. The service stores and +serves them; it never fetches a URI. + +| Threat | STRIDE | Who | Mitigation | +|---|---|---|---| +| SSRF via `source.uri` or `evidence_ref.uri` | S, I | B | No outbound HTTP client in the service; a test asserts `urllib.request`, `http.client` and `socket.create_connection` are absent from the module graph. (SO-29) | +| Wrong hash on upload; UI dedupes on it | T | B | `sha256` recomputed server-side; mismatch with `content_ref.hash` is `422`; blobs stored at `blobs/sha256//`, mode 0400, never rewritten. (SO-30) | +| Oversized artifact | D | B, C | 25 MiB per version, 200 versions, 2 000 artifacts, 2 GiB blob store per tenant (DATA-MODEL caps); over limit is `413 too_large` or `409 conflict`. (SO-31) | +| HTML/SVG served executable into the chat origin | I, E | B | Blobs served as `application/octet-stream`, `Content-Disposition: attachment`, `X-Content-Type-Options: nosniff`. `mime` is metadata for the UI's sandboxed preview, never a response header. (SO-32) | +| Lineage or promotion names a record the caller does not own | T, I | B, C | `lineage.artifact_id`, `promotion.project_id`, `passage.source_id`, `citation.passage_ids`, notebook id lists all resolved under the caller's subtree at write time; unknown or foreign is `422`. There is no cross-tenant store, and the check stays explicit so a future shared cache cannot widen it. (SO-33) | +| `access.mode: shared_readonly` used as a cross-tenant channel | I | C | Share tokens are minted by the owner, read-only, expire at `share_expires_at` ≤ 7 days, and resolve only inside the same pod; the router does not route them to another slot. (SO-34) | +| Passage `text` used to smuggle secrets into the citation strip | I | B | Passage text passes the SO-12 scrub; 4 000-char cap is the schema's. (SO-12) | + +## 5. Autonomy + +The service is the policy store and approval queue; the gateway executes +tools. The gate is only as strong as the gateway's call to +`effective_decision` and its presentation of the approval id at execution. + +| Threat | STRIDE | Who | Mitigation | +|---|---|---|---| +| Agent decides its own approval | S, E | B | `POST /approvals/{id}` needs `identity.surface` in `{chat, telegram, voice}` and `trust` in `{router, relay}`; the stored `decision.by` is the asserted user, not the body's actor. (SO-35) | +| `once` reused for two side effects | T | B | Approvals are terminal after one transition (`APPROVAL_TRANSITIONS`); consume marks `_meta.consumed_at` and a second consume of `once` is `409 conflict`. (SO-36) | +| TOCTOU: approved "write notes.md", executed "write ~/.ssh/authorized_keys" | T | B | `request.evidence` carries a `tool_call` ref whose `hash` is the sha256 of the canonical argument JSON; consume must present the same hash or it is `denied` with a `side_effect.blocked` event. (SO-37) | +| `always` grant never expires | E | C | `always` writes a grant with server-set `expires_at` ≤ 30 days and `granted_by`; the client cannot extend it. (SO-38) | +| `deploy`, `external_side_effect` or `network` slip through under `safe` | E | B, C | `effective_decision` is the only resolver; `_ALWAYS_ASK` ends `ask`, unexpired `deny` wins. `request.external: true` always requires an approval record. (SO-39) | +| Budget exhausted by delegation fan-out | D | B | `hux.budget_state.v1` is updated per `tool.result`/`delegation.*`; once `exhausted` the service refuses new approvals with `402`-style `budget_exhausted` and emits `budget.exhausted`. (SO-40) | +| Receipt says `cancelled` while a shell still runs | R | B | `outcome: cancelled` only after the gateway reports the run's process registry empty; else `failed_to_cancel`. `side_effects` must include every `tool.result` with `ok: true` after `requested_at`, each marked `reverted`. Stop is not done until the receipt exists. (SO-41) | +| Approval sits open forever | D | B | `expires_at` ≤ 24 h from `requested_at`; expiry is a terminal transition written by the retention thread. (SO-42) | + +## 6. Storage + +Layout per DATA-MODEL §1: `/opt/data/hux/v1/users///...`. +No table, index or cache is shared between tenants because no two tenants +share a PVC. + +| Threat | STRIDE | Who | Mitigation | +|---|---|---|---| +| Path traversal via id | I, T | B, C | `store.path_for` validates the user hash and every id against the `common.schema.json` patterns (`/` impossible), then `os.path.commonpath` against the user subtree, as `normalizeTenantMediaPath` does. No caller string reaches a path any other way. (SO-43) | +| Lost update between chat and Telegram | T | C, D | `revision` served as `ETag`; PATCH/PUT need `If-Match`, mismatch `409 conflict`; writes are temp + `os.replace` + fsync under the family lock. Unconditional writes are audited `why: unconditional_write`. (SO-44) | +| Agent edits the audit log; it has write on the PVC | R, T | B | hux-foundation runs as its own uid (not 10000); `/opt/data/hux` is 0700 to that uid; the agent reaches records only through the API. Requirement for Codex: separate `runAsUser`, `readOnlyRootFilesystem`, `drop: [ALL]`. (SO-45) | +| Audit line removed or altered | R | B, E | `audit/outcomes/*.jsonl` lines carry `prev` = sha256 of the previous line; a break is reported in the next retention audit as `report`. Audit lines never contain record content. (SO-46) | +| Purge deletes the wrong thing | D | E | `purge_forgotten_content` touches only tombstoned ids and blobs absent from `refs.json` for 7 days, writes the dry-run count into the retention record before the delete pass, and never touches `audit/`. (SO-47) | +| Second replica interleaves appends | T | E | `v1/.lock` taken with `flock(LOCK_EX\|LOCK_NB)` at start-up; failure is fatal. (SO-48) | +| PVC full stops the WebUI too | D | B | Per-family bounds from DATA-MODEL §4; a write over cap is `413 too_large`, and total `/opt/data/hux` above 4 GiB puts `/healthz` at `degraded`. (SO-49) | + +## 7. Feature flags + +| Threat | STRIDE | Who | Mitigation | +|---|---|---|---| +| `hux.memory_control` on, `hux.privacy` off | E | E | `flag_enabled` walks `depends_on`; a route serves only when the chain is on; `/hux/v1` is `404 flag_off` without `hux.foundation`. `GET /capabilities` reports the resolved chain, not the raw env. (SO-50) | +| Old binary reads new records | D | E | Additive-only within 1.x; `_meta` stripped on serve; unknown `schema` skipped on list, `409` on fetch; `MANIFEST.json` `min_reader_layout` refuses a tree the binary cannot read. (SO-51) | +| Rollback leaves data the disabled UI cannot show | I | E | Disabling hides routes, not data; TTL, forget and purge keep running. (SO-52) | + +## 8. Abuse and DoS bounds + +Sized for 4-8 tenants, one human each, on a 10 Gi PVC shared with the WebUI. + +- Bodies: 64 KiB JSON, 25 MiB blob, rejected before parsing. +- Rate: 30 writes/min and 300 reads/min per subject, `429 rate_limited` + with `Retry-After`. Relay and worker get the same limits per slot. +- Events: 50 000 per conversation then archived; 1 GiB total. +- Memory: 5 000 live entries; proposals expire in 7 days; ledger compacts at 64 MiB. +- Artifacts: 2 000, 200 versions each, 2 GiB blobs. +- Research: 10 000 sources, 50 000 passages, 512 MiB. +- Approvals: 50 pending per run; expiry ≤ 24 h. +- SSE: one per (subject, conversation); idle timeout 15 min; 8 per subject. +- Threads: `ThreadingHTTPServer` capped at 32 concurrent requests; beyond that `503`. + +## Security obligations checklist + +- SO-01 The service binds 127.0.0.1 only; any other bind address fails at startup. +- SO-02 Tenant pod ingress admits `app: hermes-chat-router` only; no hux port is added to any NetworkPolicy. +- SO-03 The router deletes inbound `X-Hux-*` and `X-Hermes-Tenant-Identity` before setting its own; the service rejects a slot header not equal to `HERMES_TENANT_SLOT`. +- SO-04 Exactly one identity header set is accepted (`X-Hermes-Tenant-Identity`, `X-Hux-Subject`, `X-Hux-Surface`); any other vocabulary is `401`. +- SO-05 Relay requests are checked with `hmac.compare_digest` against the relay key and carry `trust: relay`. +- SO-06 A relay request carrying `X-Hux-Subject` is `401`; the owner is derived from the slot. +- SO-07 The relay key and every `/opt/data/.env` value never appear in a record, event detail, audit line or log. +- SO-08 `trust: worker` may call only `/releases`, `/capabilities` and `/admin/retention`; everything else is `403`. +- SO-09 Every path contains the asserted `usr_` hash; a different subject on the same pod sees an empty tree. +- SO-10 `identity`, `provenance.surface`, user-type `provenance.actor` and `recorded_at` are server-set and client values ignored. +- SO-11 Event `detail` keys outside the per-kind allowlist are dropped before persistence. +- SO-12 Every stored string passes the secret scrub; a hit becomes `[redacted:]` and raises `redaction.level` to at least `partial`. +- SO-13 `detail` over 32 KiB is replaced by a truncation marker; a line over 64 KiB is rejected. +- SO-14 `redaction.level` is enforced at serve time; telegram and voice never receive `none`. +- SO-15 `seq` and `id` are server-assigned; a body containing either is `400`. +- SO-16 Idempotent replays are served only after the ownership check passes. +- SO-17 Event pages are ≤ 200; one SSE stream per (subject, conversation); idle timeout 15 min. +- SO-18 A conversation the caller does not own returns `404`, never `403`. +- SO-19 The service never dereferences `evidence_ref.uri` or `source.uri`; stored URIs use an allowlisted scheme. +- SO-20 Restricted or fully redacted events never enter the search index. +- SO-21 `memory_policy_violations` runs on every memory write and transition; a violation is `422` plus a `memory.suppressed` event. +- SO-22 `no_store` and `forget` write a content-free ledger line and a tombstone before returning. +- SO-23 Retrieval, search, export and the agent read hook consult tombstones and `memory_disabled` first and return only `active` entries with `retrievable: true`. +- SO-24 `forget` re-redacts events referencing the memory id to `full` and queues the ledger rewrite. +- SO-25 An entry that `supersedes` or is sourced from a forgotten id requires a user-actor approval. +- SO-26 Export is owner-scoped, excludes non-active entries, appends `exported` to `audit[]`, and its snapshot is pruned after 7 days. +- SO-27 Expired entries are never served as active; a retention audit older than 48 h is reported as `audit_stale`. +- SO-28 In `private` mode `POST /memory` is refused and event appends return `204` without writing. +- SO-29 A test asserts no outbound HTTP or raw socket client is importable from the service's module graph. +- SO-30 Blob hashes are recomputed on upload; a mismatch is `422`; blobs are 0400 and never rewritten. +- SO-31 Artifact caps: 25 MiB per version, 200 versions, 2 000 artifacts, 2 GiB blobs. +- SO-32 Blobs are served as `application/octet-stream` attachments with `nosniff`. +- SO-33 Every referenced id (lineage, promotion, source, passage, citation, notebook) must resolve under the caller's subtree. +- SO-34 Share tokens are owner-minted, read-only, expire within 7 days, and resolve only on the owning pod. +- SO-35 Approvals are decided only from a human surface with `trust` router or relay; `decision.by` is the asserted user. +- SO-36 A `once` approval can be consumed exactly once. +- SO-37 Consume must present the argument hash recorded in `request.evidence` at request time. +- SO-38 `always` grants carry a server-set `expires_at` of at most 30 days. +- SO-39 `effective_decision` is the sole resolver; `deploy` and `external_side_effect` always resolve to `ask`; `request.external: true` always needs an approval record. +- SO-40 An exhausted `budget_state` blocks new approvals with `budget_exhausted` and emits `budget.exhausted`. +- SO-41 A receipt says `cancelled` only after the run's process registry is empty, and lists every successful side effect after `requested_at`. +- SO-42 Pending approvals expire within 24 h of `requested_at`. +- SO-43 All paths go through `path_for`: pattern-validated ids plus `commonpath` containment in the user subtree. +- SO-44 Revisioned writes require `If-Match`; mismatch is `409`; unconditional writes are audited. +- SO-45 The service runs as a distinct uid with `/opt/data/hux` mode 0700; the agent has no filesystem path to records. +- SO-46 Audit outcome lines are hash-chained and a broken chain is reported in the next retention audit. +- SO-47 Purge touches only tombstoned ids and unreferenced blobs older than 7 days, records a dry-run count first, and never touches `audit/`. +- SO-48 A second service process against the same tree fails at startup on `v1/.lock`. +- SO-49 Writes beyond family caps are `413`; total store above 4 GiB marks `/healthz` degraded. +- SO-50 A route serves only when its flag and every dependency flag are on; `/capabilities` reports the resolved chain. +- SO-51 Responses strip `_meta`, unknown `schema` values are skipped on list and `409` on fetch, and a tree above `min_reader_layout` is refused. +- SO-52 Disabling a flag hides routes; TTL, forget and purge continue. +- SO-53 Rate limits (30 writes/min, 300 reads/min per subject) return `429` with `Retry-After`. +- SO-54 Bodies over 64 KiB (JSON) or 25 MiB (blob) are rejected with `413` before parsing. diff --git a/testing/tests/test_hermes_hux_contract_foundation.py b/testing/tests/test_hermes_hux_contract_foundation.py new file mode 100644 index 00000000..47be483e --- /dev/null +++ b/testing/tests/test_hermes_hux_contract_foundation.py @@ -0,0 +1,340 @@ +"""HUX-11 foundation: identity, flags, tenant store, audit and the HTTP pipeline. + +Security obligations exercised here: identity comes only from trusted headers +(relay/worker keys compared in constant time), every request lands in a +tenant-scoped directory, disabled cards are indistinguishable from unknown +routes, every read and denial leaves an audit outcome, and revisions guard +concurrent writers. +""" + +from __future__ import annotations + +import json +import sys +import threading +from http.client import HTTPConnection +from pathlib import Path + +import pytest + +ROOT = Path(__file__).resolve().parents[2] +FOUNDATION = ROOT / "dockerfiles" / "hermes-hux-foundation" +if str(FOUNDATION) not in sys.path: + sys.path.insert(0, str(FOUNDATION)) + +from hux import audit, contracts, errors, flags, identity, store # noqa: E402 +from hux.http import Router, Response, page, serve # noqa: E402 +from hux.server import build_router # noqa: E402 + +SCHEMAS = contracts.load_all() +HEADERS = {"X-Hermes-Tenant-Identity": "slot-3", "X-Hux-Subject": "usr_0123456789abcdef", "X-Hux-Surface": "chat"} +ALL_ON = ",".join(card["flag"] for card in contracts.load_flags()["cards"]) + + +def ident(**overrides) -> identity.Identity: + base = {"tenant_slot": "slot-3", "subject": "usr_0123456789abcdef", "surface": "chat", "trust": "router"} + return identity.Identity(**{**base, **overrides}) + + +# --- identity ------------------------------------------------------------------- + +def test_identity_from_router_headers(): + who = identity.resolve(HEADERS, {}) + assert who == ident() + assert contracts.validate("common.schema.json", who.record(), SCHEMAS, "/$defs/identity") == [] + + +@pytest.mark.parametrize("bad", [ + {"X-Hermes-Tenant-Identity": ""}, {"X-Hermes-Tenant-Identity": "slot-x"}, {"X-Hux-Subject": "brad@bstein.dev"}, + {"X-Hux-Subject": ""}, {"X-Hux-Surface": "admin"}, {"X-Hux-Trust": "god"}, {"X-Hux-Surface": "worker"}, +]) +def test_identity_rejects_bad_headers(bad): + with pytest.raises(errors.Unauthorized): + identity.resolve({**HEADERS, **bad}, {}) + + +def test_relay_and_worker_need_their_keys(): + relay = {**HEADERS, "X-Hux-Trust": "relay", "X-Hux-Surface": "telegram"} + with pytest.raises(errors.Unauthorized): + identity.resolve(relay, {}) + with pytest.raises(errors.Unauthorized): + identity.resolve({**relay, "X-Hux-Relay-Key": "nope"}, {"HUX_RELAY_KEY": "secret"}) + assert identity.resolve({**relay, "X-Hux-Relay-Key": "secret"}, {"HUX_RELAY_KEY": "secret"}).trust == "relay" + worker = {**HEADERS, "X-Hux-Trust": "worker", "X-Hux-Surface": "worker", "X-Hux-Relay-Key": "wk"} + assert identity.resolve(worker, {"HUX_WORKER_KEY": "wk"}).surface == "worker" + assert identity.resolve(HEADERS).trust == "router" + + +def test_slot_must_match_the_pod_it_reaches(): + with pytest.raises(errors.Unauthorized): + identity.resolve(HEADERS, {"HUX_TENANT_SLOT": "slot-4"}) + assert identity.resolve(HEADERS, {"HUX_TENANT_SLOT": "slot-3"}).tenant_slot == "slot-3" + + +def test_server_refuses_non_loopback_bind(): + from hux import server + assert server.bind_address({}) == "127.0.0.1" + with pytest.raises(SystemExit): + server.bind_address({"HUX_BIND": "0.0.0.0"}) + + +# --- flags --------------------------------------------------------------------- + +def test_flags_fail_closed_and_capabilities_validate(): + off = flags.Flags({}) + assert not off.enabled("HUX-11") + with pytest.raises(errors.FlagOff): + off.require("HUX-01") + partial = flags.Flags({"HUX_FLAGS": "hux.activity_timeline"}) + assert not partial.enabled("HUX-01") + on = flags.Flags({"HUX_FLAGS": ALL_ON}) + record = on.capabilities(ident(), {"commit": "d3cbeb06" * 5, "image_digest": "sha256:" + "4a" * 32, "junk": "x"}) + assert contracts.validate_record(record, SCHEMAS) == [] + assert all(card["enabled"] for card in record["cards"]) + assert {card["card"] for card in record["cards"]} == {f"HUX-{n:02d}" for n in range(1, 13)} + assert not on.enabled("HUX-99") + assert flags.build_from_environ({"HUX_BUILD_COMMIT": "abc"}) == {"commit": "abc", "image_digest": ""} + assert "commit" in flags.build_from_environ() + + +def test_every_declared_route_belongs_to_exactly_one_card(): + seen: dict[str, str] = {} + for card, routes in flags.CARD_ROUTES.items(): + for route in routes: + assert route.startswith("/hux/v1/") + assert route not in seen, f"{route} owned by {seen[route]} and {card}" + seen[route] = card + + +# --- store --------------------------------------------------------------------- + +def test_store_paths_are_tenant_scoped_and_ids_validated(tmp_path): + a = store.TenantStore(tmp_path, ident()) + b = store.TenantStore(tmp_path, ident(subject="usr_fedcba9876543210")) + assert a.root != b.root and a.root.parent == b.root.parent + a.put("things", {"id": "thg_0001", "v": 1}) + assert not b.exists("things", "thg_0001") + with pytest.raises(errors.Invalid): + a.get("things", "../../etc/passwd") + with pytest.raises(errors.Invalid): + a.put("things", {"id": "no-prefix"}) + with pytest.raises(errors.Invalid): + a.append("ledger", "../x", {}) + with pytest.raises(errors.NotFound): + a.get("things", "thg_9999") + assert store.ID_RE.match(store.new_id("evt")) + assert store.now_iso().endswith("Z") + + +def test_store_revisions_and_conflicts(tmp_path): + s = store.TenantStore(tmp_path, ident()) + first = s.put("docs", {"id": "doc_0001", "n": 1}) + assert first["revision"] == 1 + second = s.put("docs", {"id": "doc_0001", "n": 2}, expected_revision=1) + assert second["revision"] == 2 + with pytest.raises(errors.Conflict): + s.put("docs", {"id": "doc_0001", "n": 3}, expected_revision=1) + with pytest.raises(errors.Conflict): + s.put("docs", {"id": "doc_0002", "n": 1}, expected_revision=4) + assert s.put("docs", {"id": "doc_0003"}, expected_revision=0)["revision"] == 1 + assert s.count("docs") == 2 + assert [d["id"] for d in s.scan("docs")] == ["doc_0001", "doc_0003"] + assert list(s.scan("nothing")) == [] + s.delete("docs", "doc_0003") + s.delete("docs", "doc_0003") + assert s.count("docs") == 1 + + +def test_store_bounds(tmp_path, monkeypatch): + s = store.TenantStore(tmp_path, ident()) + with pytest.raises(errors.TooLarge): + s.put("docs", {"id": "doc_0001", "blob": "x" * store.MAX_RECORD_BYTES}) + with pytest.raises(errors.TooLarge): + s.append("ledger", "big", {"blob": "x" * store.MAX_RECORD_BYTES}) + monkeypatch.setattr(store, "MAX_FAMILY_RECORDS", 1) + s.put("docs", {"id": "doc_0001"}) + with pytest.raises(errors.TooLarge): + s.put("docs", {"id": "doc_0002"}) + s.put("docs", {"id": "doc_0001", "again": True}) + + +def test_store_ledgers_survive_torn_writes(tmp_path): + s = store.TenantStore(tmp_path, ident()) + s.append("ledger", "conv_1", {"seq": 1}) + s.append("ledger", "conv_1", {"seq": 2}) + path = s.root / "ledger" / "conv_1.jsonl" + with open(path, "ab") as handle: + handle.write(b'{"seq": 3, "tru') + assert [r["seq"] for r in s.read("ledger", "conv_1")] == [1, 2] + assert s.read("ledger", "missing") == [] + assert s.ledgers("ledger") == ["conv_1"] and s.ledgers("none") == [] + s.rewrite("ledger", "conv_1", [{"seq": 9}]) + assert s.read("ledger", "conv_1") == [{"seq": 9}] + + +def test_store_blobs_and_manifest(tmp_path): + s = store.TenantStore(tmp_path, ident()) + digest = "ab" * 32 + s.put_blob(digest, b"hello") + s.put_blob(digest, b"ignored") + assert s.get_blob(digest) == b"hello" + with pytest.raises(errors.NotFound): + s.get_blob("cd" * 32) + with pytest.raises(errors.Invalid): + s.put_blob("../x", b"") + with pytest.raises(errors.Invalid): + s.get_blob("zz") + manifest = s.manifest("1.0.0") + assert contracts.validate_record(manifest, SCHEMAS) == [] + assert s.manifest("1.9.0") == manifest + + +def test_store_locks_serialise_concurrent_writers(tmp_path): + s = store.TenantStore(tmp_path, ident()) + s.put("docs", {"id": "doc_0001", "n": 0}) + + def bump() -> None: + for _ in range(20): + with s.lock("docs"): + current = s.get("docs", "doc_0001") + s.put("docs", {**current, "n": current["n"] + 1}, expected_revision=current["revision"]) + + workers = [threading.Thread(target=bump) for _ in range(4)] + for worker in workers: + worker.start() + for worker in workers: + worker.join() + assert s.get("docs", "doc_0001")["n"] == 80 + + +# --- audit --------------------------------------------------------------------- + +def test_audit_rows_validate_and_never_carry_bodies(tmp_path): + s = store.TenantStore(tmp_path, ident()) + row = audit.record(s, ident(), "memory.read", "mem_0001", "deny", "not owner") + assert contracts.validate("common.schema.json", row, SCHEMAS, "/$defs/audit_outcome") == [] + with pytest.raises(ValueError): + audit.record(s, ident(), "x.y", "r", "maybe") + for _ in range(5): + audit.record(s, ident(), "memory.read", "mem_0002", "allow") + assert len(audit.recent(s, limit=3)) == 3 + assert audit.recent(s)[0]["outcome"] == "deny" + + +# --- http pipeline --------------------------------------------------------------- + +def _router(tmp_path, flags_value=ALL_ON, environ=None) -> Router: + env = {"HUX_FLAGS": flags_value, **(environ or {})} + return build_router(tmp_path, env) + + +def _call(router, method, path, headers=HEADERS, body=b"") -> tuple[int, dict]: + response = router.dispatch(method, path, headers, body) + return response.status, response.body + + +def test_capabilities_and_manifest_roundtrip(tmp_path): + router = _router(tmp_path, environ={"HUX_BUILD_COMMIT": "d3cbeb06" * 5}) + status, body = _call(router, "GET", "/hux/v1/capabilities") + assert status == 200 and contracts.validate_record(body, SCHEMAS) == [] + assert body["server"] == {"commit": "d3cbeb06" * 5} + status, body = _call(router, "GET", "/hux/v1/manifest") + assert status == 200 and body["schema"] == "hux.manifest.v1" + rows = audit.recent(store.TenantStore(tmp_path, ident())) + assert [r["action"] for r in rows] == ["foundation.capabilities", "foundation.manifest"] + + +def test_flag_off_and_unknown_route_look_the_same(tmp_path): + off = _router(tmp_path, flags_value="") + status, body = _call(off, "GET", "/hux/v1/capabilities") + assert (status, body["code"]) == (404, "flag_off") + status, body = _call(off, "GET", "/hux/v1/nothing") + assert (status, body["code"]) == (404, "not_found") + assert contracts.validate_record(body, SCHEMAS) == [] + status, body = _call(off, "POST", "/hux/v1/capabilities") + assert status == 405 + outcomes = [r["outcome"] for r in audit.recent(store.TenantStore(tmp_path, ident()))] + assert outcomes == ["flag_off", "not_found", "not_found"] + + +def test_unauthorized_requests_never_touch_storage(tmp_path): + router = _router(tmp_path) + status, body = _call(router, "GET", "/hux/v1/capabilities", headers={}) + assert (status, body["code"]) == (401, "unauthorized") + assert not (tmp_path / "hux").exists() + + +def test_body_decoding_and_request_helpers(tmp_path): + router = _router(tmp_path) + captured = {} + + def echo(request): + captured.update(body=request.body, if_match=request.if_match(), key=request.idempotency_key(), q=request.query) + return page([request.body], None) + + router.add("POST", "/hux/v1/echo/{id}", "HUX-11", "test.echo", echo) + status, body = _call(router, "POST", "/hux/v1/echo/abc?x=1&x=2", {**HEADERS, "If-Match": "3", "Idempotency-Key": "run:1234567"}, b'{"a": 1}') + assert status == 200 and body == {"items": [{"a": 1}], "next": None} + assert captured == {"body": {"a": 1}, "if_match": 3, "key": "run:1234567", "q": {"x": "2"}} + assert _call(router, "POST", "/hux/v1/echo/abc", HEADERS, b"{oops")[1]["code"] == "invalid" + assert _call(router, "POST", "/hux/v1/echo/abc", HEADERS, b"x" * (1024 * 1024 + 1))[1]["code"] == "too_large" + assert _call(router, "POST", "/hux/v1/echo/abc", {**HEADERS, "If-Match": "abc"})[1]["code"] == "invalid" + assert _call(router, "POST", "/hux/v1/echo/abc", {**HEADERS, "Idempotency-Key": "!"})[1]["code"] == "invalid" + _, body = _call(router, "POST", "/hux/v1/echo/abc", HEADERS) + assert body["items"] == [None] + + +def test_handler_errors_are_audited_with_outcome(tmp_path): + router = _router(tmp_path) + + def conflict(request): + raise errors.Conflict("stale") + + def denied(request): + raise errors.Forbidden("not yours") + + router.add("GET", "/hux/v1/c", "HUX-11", "test.conflict", conflict) + router.add("GET", "/hux/v1/d", "HUX-11", "test.denied", denied) + assert _call(router, "GET", "/hux/v1/c")[0] == 409 + assert _call(router, "GET", "/hux/v1/d")[0] == 403 + outcomes = [(r["action"], r["outcome"]) for r in audit.recent(store.TenantStore(tmp_path, ident()))] + assert outcomes == [("test.conflict", "conflict"), ("test.denied", "deny")] + + +def test_real_server_serves_json_and_sse(tmp_path): + router = _router(tmp_path) + + def stream(request): + return Response(200, stream=lambda: (b"id: 1\ndata: {}\n\n", b"id: 2\ndata: {}\n\n")) + + router.add("GET", "/hux/v1/s", "HUX-11", "test.stream", stream) + server = serve(router, "127.0.0.1", 0) + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + try: + conn = HTTPConnection("127.0.0.1", server.server_address[1], timeout=5) + conn.request("GET", "/healthz") + assert json.loads(conn.getresponse().read())["status"] == "ok" + conn.request("GET", "/hux/v1/capabilities", headers=HEADERS) + reply = conn.getresponse() + assert reply.status == 200 and json.loads(reply.read())["schema"] == "hux.capabilities.v1" + conn.request("POST", "/hux/v1/capabilities", body=b"{}", headers={**HEADERS, "Content-Length": "2"}) + assert conn.getresponse().status == 405 + conn.request("GET", "/hux/v1/s", headers=HEADERS) + reply = conn.getresponse() + assert reply.getheader("Content-Type") == "text/event-stream" + assert reply.read() == b"id: 1\ndata: {}\n\nid: 2\ndata: {}\n\n" + finally: + server.shutdown() + server.server_close() + + +def test_all_errors_serialise_to_contract(): + for cls in (errors.Unauthorized, errors.Forbidden, errors.NotFound, errors.FlagOff, errors.Conflict, errors.Invalid, errors.TooLarge, errors.ApprovalRequired, errors.BudgetExhausted): + record = cls("m" * 300, ["d" * 300] * 40).record() + assert contracts.validate_record(record, SCHEMAS) == [], cls + + +def test_foundation_sources_stay_under_500_lines(): + for path in sorted(FOUNDATION.rglob("*.py")): + assert len(path.read_text().splitlines()) <= 500, path