"""admin-canvas — the write side of the sys-admin-llm's display surface. Companion to digest-engine/digest-web (docs/project-plan.md Phase 12), but for ad hoc content instead of a scheduled synthesis run: statistics, power-consumption readings, charts, snapshots — anything Home Assistant's tool-calling LLM wants to put on a thin client's admin canvas (Phase 13), on demand rather than 4x/day. SECURITY BOUNDARY. This is a real network listener — unlike digest-engine, which is a oneshot batch job with no listening socket at all. It must stay reachable only from the container host's internal Docker network (HA and nothing else — no `ports:` entry in the compose block that starts this service), and every request additionally requires the bearer token below, as defense in depth in case that network boundary is ever loosened by mistake. This module knows nothing about Home Assistant, tool calls, or the LLM that ultimately triggers a write — it only ever validates and stores what it is given. Reads are somebody else's job entirely: admin-web (nginx, read-only, LAN-published) serves output/, exactly mirroring how digest-web serves digest-engine's output/ — this process never serves a GET. Two endpoints, both POST, both token-gated: - /show JSON {"windows": [...]}, overwrites output/latest.json - /media/ raw bytes, written to output/media/ See admin-canvas/README.md for the window schema (stat/image/video/chart kinds) and a worked example of the intended HA-side `rest_command` -> here call. """ from __future__ import annotations import json import logging import os import re import sys import threading from datetime import datetime, timezone from http import HTTPStatus from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path from urllib.parse import urlsplit LOG = logging.getLogger("admin-canvas") OUTPUT_DIR = Path(os.environ.get("ADMIN_CANVAS_OUTPUT_DIR", "/output")) MEDIA_DIR = OUTPUT_DIR / "media" # Fails closed: an unset token means every request is rejected, never "auth is # optional". See admin-canvas.env.example. TOKEN = os.environ.get("ADMIN_CANVAS_TOKEN", "") MAX_SHOW_BYTES = int(os.environ.get("ADMIN_CANVAS_MAX_SHOW_KB", "256")) * 1024 MAX_MEDIA_BYTES = int(os.environ.get("ADMIN_CANVAS_MAX_MEDIA_MB", "25")) * 1024 * 1024 # `kind` may also be omitted entirely for plain markdown-ish content — handled # client-side by DigestWindow.open exactly as a digest window is today. KNOWN_KINDS = {"stat", "image", "video", "chart"} # No leading "/", no "..", no scheme — a relative path under the media/ directory # this service itself writes to, and nothing else. Mirrors the "a payload never # becomes a URL host" invariant thinclient_agent/mqtt_discovery.py documents for the # thin client's own control surface; the trust boundary just lives here instead, # since this is the component that now decides what a browser is told to fetch. SRC_RE = re.compile(r"^media/[A-Za-z0-9_.-]+$") MEDIA_NAME_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_.-]{0,63}$") MEDIA_EXTENSIONS = {".png", ".jpg", ".jpeg", ".gif", ".webp", ".mp4", ".webm"} _write_lock = threading.Lock() def validate_show_payload(payload) -> str | None: """Returns an error string, or None if payload is acceptable.""" if not isinstance(payload, dict): return "body must be a JSON object" windows = payload.get("windows") if not isinstance(windows, list): return "'windows' must be a list" for i, win in enumerate(windows): if not isinstance(win, dict): return f"windows[{i}] must be an object" kind = win.get("kind") if kind is not None and kind not in KNOWN_KINDS: return ( f"windows[{i}].kind {kind!r} is not one of {sorted(KNOWN_KINDS)} " "(omit 'kind' entirely for plain text/markdown-ish content)" ) if kind in ("image", "video"): content = win.get("content") src = content.get("src") if isinstance(content, dict) else None if not isinstance(src, str) or not SRC_RE.match(src): return ( f"windows[{i}].content.src must be a 'media/' path " "already uploaded via POST /media/" ) return None class Handler(BaseHTTPRequestHandler): server_version = "admin-canvas/1" def log_message(self, format, *args): # noqa: A002 - must match BaseHTTPRequestHandler's signature LOG.info("%s - %s", self.address_string(), format % args) def _authorized(self) -> bool: if not TOKEN: return False return self.headers.get("Authorization", "") == f"Bearer {TOKEN}" def _respond(self, status: HTTPStatus, payload: dict) -> None: body = json.dumps(payload).encode("utf-8") self.send_response(status) self.send_header("Content-Type", "application/json") self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) def _read_body(self, max_bytes: int) -> bytes: try: length = int(self.headers.get("Content-Length") or 0) except ValueError: raise ValueError("missing or invalid Content-Length") from None if length <= 0: return b"" if length > max_bytes: raise ValueError(f"body too large ({length} > {max_bytes} bytes)") return self.rfile.read(length) def do_GET(self): # noqa: N802 - stdlib method name self._respond( HTTPStatus.NOT_FOUND, {"error": "admin-canvas only accepts writes; reads are served by admin-web"}, ) def do_POST(self): # noqa: N802 - stdlib method name if not self._authorized(): self._respond(HTTPStatus.UNAUTHORIZED, {"error": "missing or invalid bearer token"}) return path = urlsplit(self.path).path if path == "/show": self._handle_show() elif path.startswith("/media/"): self._handle_media(path[len("/media/"):]) else: self._respond(HTTPStatus.NOT_FOUND, {"error": "no such endpoint"}) def _handle_show(self) -> None: try: raw = self._read_body(MAX_SHOW_BYTES) except ValueError as exc: self._respond(HTTPStatus.REQUEST_ENTITY_TOO_LARGE, {"error": str(exc)}) return try: payload = json.loads(raw or b"{}") except ValueError: self._respond(HTTPStatus.BAD_REQUEST, {"error": "body is not valid JSON"}) return error = validate_show_payload(payload) if error: self._respond(HTTPStatus.BAD_REQUEST, {"error": error}) return payload["generated_at"] = datetime.now(timezone.utc).isoformat().replace("+00:00", "Z") with _write_lock: OUTPUT_DIR.mkdir(parents=True, exist_ok=True) tmp = OUTPUT_DIR / "latest.json.tmp" tmp.write_text(json.dumps(payload, indent=2, ensure_ascii=False), encoding="utf-8") tmp.replace(OUTPUT_DIR / "latest.json") window_count = len(payload.get("windows", [])) LOG.info("wrote %d window(s) to output/latest.json", window_count) self._respond(HTTPStatus.OK, {"ok": True, "windows": window_count}) def _handle_media(self, name: str) -> None: if not MEDIA_NAME_RE.match(name) or Path(name).suffix.lower() not in MEDIA_EXTENSIONS: self._respond( HTTPStatus.BAD_REQUEST, {"error": "invalid media filename — see admin-canvas/README.md for the allowed pattern"}, ) return try: data = self._read_body(MAX_MEDIA_BYTES) except ValueError as exc: self._respond(HTTPStatus.REQUEST_ENTITY_TOO_LARGE, {"error": str(exc)}) return if not data: self._respond(HTTPStatus.BAD_REQUEST, {"error": "empty body"}) return with _write_lock: MEDIA_DIR.mkdir(parents=True, exist_ok=True) (MEDIA_DIR / name).write_bytes(data) LOG.info("wrote media/%s (%d bytes)", name, len(data)) self._respond(HTTPStatus.OK, {"ok": True, "path": f"media/{name}"}) def main() -> int: logging.basicConfig( level=os.environ.get("LOG_LEVEL", "INFO").upper(), format="%(asctime)s %(levelname)-7s %(name)s: %(message)s", ) if not TOKEN: LOG.error( "ADMIN_CANVAS_TOKEN is not set — every request will be rejected until it is. " "See admin-canvas.env.example." ) port = int(os.environ.get("ADMIN_CANVAS_PORT", "8092")) server = ThreadingHTTPServer(("0.0.0.0", port), Handler) LOG.info("admin-canvas listening on :%d (output dir: %s)", port, OUTPUT_DIR) try: server.serve_forever() except KeyboardInterrupt: pass return 0 if __name__ == "__main__": sys.exit(main())