#!/usr/bin/env python3 """Serve the local Health Dashboard with strict routes and privacy headers.""" from __future__ import annotations import mimetypes import json import os import re import secrets import sqlite3 import stat import threading import time from datetime import date, datetime from http.cookies import CookieError, SimpleCookie from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path from urllib.parse import parse_qs, quote, unquote, urlparse from zoneinfo import ZoneInfo from dashboard_v5.read_api import ( APIError, connect_read_only, dispatch_api, _resolve_record_document, ) BASE = Path.home() / ".hermes" / "assets" / "Gesundheit" API_DB = ( Path(os.environ["HEALTH_DASHBOARD_DB"]).resolve() if os.environ.get("HEALTH_DASHBOARD_DB") else None ) DB = BASE / "health_data.db" REPORTS = BASE / "reports" DASHBOARD_FILE = Path(os.environ["HEALTH_DASHBOARD_FILE"]).resolve() DASHBOARD_V5_FILE = ( Path(os.environ["HEALTH_DASHBOARD_V5_FILE"]).resolve() if os.environ.get("HEALTH_DASHBOARD_V5_FILE") else None ) ASSET_DIR = Path( os.environ.get( "HEALTH_DASHBOARD_ASSET_DIR", str(BASE / "scripts" / "assets" / "health-assets"), ) ).resolve() BIND_HOST = os.environ.get("HEALTH_DASHBOARD_HOST", "127.0.0.1") BIND_PORT = int(os.environ.get("HEALTH_DASHBOARD_PORT", "8014")) ALLOWED_HOSTS = { value.strip().lower().rstrip(".") for value in os.environ.get( "HEALTH_DASHBOARD_ALLOWED_HOSTS", "127.0.0.1,localhost" ).split(",") if value.strip() } ROUTE = "/health-dashboard" V5_ROUTE = "/health-dashboard-v5" ASSET_ROUTES = { "/health-assets/chart.umd.min.js": ( "chart.umd.min.js", "text/javascript; charset=utf-8", ), "/health-assets/dashboard-v5.js": ( "dashboard-v5.js", "text/javascript; charset=utf-8", ), "/health-assets/dashboard-v5-range.js": ( "dashboard-v5-range.js", "text/javascript; charset=utf-8", ), "/health-assets/dashboard-v5-global-search.js": ( "dashboard-v5-global-search.js", "text/javascript; charset=utf-8", ), "/health-assets/dashboard-v5-api-explorer.js": ( "dashboard-v5-api-explorer.js", "text/javascript; charset=utf-8", ), "/health-assets/dashboard-v5-day-controller.js": ( "dashboard-v5-day-controller.js", "text/javascript; charset=utf-8", ), "/health-assets/dashboard-v5-record.js": ( "dashboard-v5-record.js", "text/javascript; charset=utf-8", ), "/health-assets/dashboard-v5.css": ("dashboard-v5.css", "text/css; charset=utf-8"), "/health-assets/echarts-6.1.0.min.js": ( "echarts-6.1.0.min.js", "text/javascript; charset=utf-8", ), "/health-assets/dashboard-v5-echarts-prototype.js": ( "dashboard-v5-echarts-prototype.js", "text/javascript; charset=utf-8", ), } CHECKIN_ROUTE = "/health-actions/symptom-checkin" ACTION_INBOX = Path( os.environ.get( "HEALTH_DASHBOARD_ACTION_INBOX", str(BASE / "runtime" / "dashboard-actions"), ) ).resolve() TEST_INSTANCE_ID = os.environ.get("HEALTH_DASHBOARD_TEST_INSTANCE_ID", "").strip() API_TOKEN_FILE = ( Path(os.environ["HEALTH_DASHBOARD_API_TOKEN_FILE"]).expanduser() if os.environ.get("HEALTH_DASHBOARD_API_TOKEN_FILE") else None ) BROWSER_SESSION_TTL_SECONDS = max( 60, min(int(os.environ.get("HEALTH_DASHBOARD_BROWSER_SESSION_TTL", "900")), 3600) ) BROWSER_SESSION_COOKIE_SECURE = ( os.environ.get("HEALTH_DASHBOARD_BROWSER_SESSION_COOKIE_SECURE") == "1" ) if TEST_INSTANCE_ID: if not re.fullmatch(r"[A-Za-z0-9_-]{8,64}", TEST_INSTANCE_ID): raise RuntimeError("invalid HEALTH_DASHBOARD_TEST_INSTANCE_ID") if not ACTION_INBOX.is_relative_to(Path("/tmp").resolve()): raise RuntimeError( "synthetic test instance requires an action inbox below /tmp" ) SYMPTOM_FIELDS = ("aphthen", "gi", "fatigue", "skin", "eyes", "joints", "vascular") CSRF_TTL_SECONDS = max( 60, min(int(os.environ.get("HEALTH_DASHBOARD_CSRF_TTL", "900")), 3600) ) QUEUE_RECEIPT_TTL_SECONDS = 120 MAX_API_RESPONSE_BYTES = 512_000 CAPTURE_TIMEZONE = ZoneInfo("Europe/Zurich") MAX_PENDING_ACTIONS = max( 1, min(int(os.environ.get("HEALTH_DASHBOARD_MAX_PENDING_ACTIONS", "64")), 256) ) MAX_ORIGINAL_BYTES = 20 * 1024 * 1024 ORIGINAL_MIME_MAGIC = ( (b"%PDF-", "application/pdf", ".pdf"), (b"\x89PNG\r\n\x1a\n", "image/png", ".png"), (b"\xff\xd8\xff", "image/jpeg", ".jpg"), ) _csrf_tokens: dict[str, float] = {} _csrf_lock = threading.Lock() _queue_receipts: dict[str, float] = {} _queue_receipt_lock = threading.Lock() _browser_sessions: dict[str, float] = {} _browser_session_lock = threading.Lock() _action_lock = threading.Lock() class QueueFullError(OSError): """Raised when the bounded private action queue cannot accept more work.""" def load_api_token() -> str: if API_TOKEN_FILE is None: raise OSError("read-only API authentication is not configured") descriptor = os.open( API_TOKEN_FILE, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK, ) try: metadata = os.fstat(descriptor) if ( not stat.S_ISREG(metadata.st_mode) or metadata.st_uid != os.getuid() or stat.S_IMODE(metadata.st_mode) != 0o600 or metadata.st_size > 256 ): raise OSError("read-only API token file is not private") raw = os.read(descriptor, 257) finally: os.close(descriptor) token = raw.decode("ascii").strip() if not re.fullmatch(r"[A-Za-z0-9_-]{32,128}", token): raise OSError("read-only API token is invalid") return token def api_request_is_authenticated(header: str, token: str) -> bool: if not header.startswith("Bearer ") or header.count(" ") != 1: return False supplied = header.removeprefix("Bearer ") return len(supplied) == len(token) and secrets.compare_digest(supplied, token) def issue_browser_session() -> str: now = time.monotonic() with _browser_session_lock: expired = [key for key, expiry in _browser_sessions.items() if expiry <= now] for key in expired: del _browser_sessions[key] session_id = secrets.token_urlsafe(32) _browser_sessions[session_id] = now + BROWSER_SESSION_TTL_SECONDS return session_id def browser_session_is_authenticated(cookie_header: str) -> bool: try: cookie = SimpleCookie(cookie_header) except (CookieError, ValueError): return False morsel = cookie.get("health_api_session") if morsel is None: return False session_id = morsel.value if not re.fullmatch(r"[A-Za-z0-9_-]{32,128}", session_id): return False now = time.monotonic() with _browser_session_lock: expiry = _browser_sessions.get(session_id) if expiry is None or expiry <= now: _browser_sessions.pop(session_id, None) return False return True def resolve_doc_path(*values: object) -> Path | None: for value in values: if not value: continue raw = str(value) candidates = [ Path(os.path.expanduser(raw)), BASE / raw, Path.home() / raw, Path(raw.replace(str(Path.home() / "Gesundheit"), str(BASE))), ] for candidate in candidates: try: resolved = candidate.resolve() if resolved.is_file() and ( resolved.is_relative_to(BASE.resolve()) or resolved.is_relative_to(REPORTS.resolve()) ): return resolved except OSError: continue return None def document_path(doc_id: int) -> tuple[Path, str] | None: connection = sqlite3.connect(f"{DB.resolve().as_uri()}?mode=ro", uri=True) connection.execute("PRAGMA query_only=ON") connection.row_factory = sqlite3.Row try: row = connection.execute( "SELECT datei_name,dateipfad,local_original_path FROM dokumente WHERE id=?", (doc_id,), ).fetchone() if not row: return None path = resolve_doc_path(row["local_original_path"], row["dateipfad"]) if not path: return None return path, str(row["datei_name"] or path.name) finally: connection.close() def _open_regular_beneath(root: Path, candidate: Path) -> int: """Open each path component without following symlinks.""" root = root.absolute() candidate = candidate.absolute() try: relative = candidate.relative_to(root) except ValueError as error: raise OSError("outside_root") from error if not relative.parts or any(part in {"", ".", ".."} for part in relative.parts): raise OSError("invalid_path") directory = os.open(root, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW) try: for part in relative.parts[:-1]: child = os.open( part, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW, dir_fd=directory, ) os.close(directory) directory = child return os.open( relative.parts[-1], os.O_RDONLY | os.O_NOFOLLOW | getattr(os, "O_NONBLOCK", 0), dir_fd=directory, ) finally: os.close(directory) def open_verified_original( database: Path, opaque: str ) -> tuple[int, int, str, str] | None: """Open a reviewed original beneath an allowlisted root using pinned descriptors.""" connection = connect_read_only(database) try: row = _resolve_record_document(connection, opaque, reviewed=True) except APIError: return None finally: connection.close() raw = row["local_original_path"] or row["dateipfad"] if not raw or "\x00" in str(raw): return None raw_path = Path(str(raw)).expanduser() roots = (BASE.absolute(), REPORTS.absolute()) candidates = ( (raw_path,) if raw_path.is_absolute() else tuple(root / raw_path for root in roots) ) descriptor = None for candidate in candidates: for root in roots: try: descriptor = _open_regular_beneath(root, candidate) break except OSError: continue if descriptor is not None: break if descriptor is None: return None try: metadata = os.fstat(descriptor) if not stat.S_ISREG(metadata.st_mode) or metadata.st_size > MAX_ORIGINAL_BYTES: raise OSError("original_not_available") magic = os.read(descriptor, 16) os.lseek(descriptor, 0, os.SEEK_SET) for prefix, mime, suffix in ORIGINAL_MIME_MAGIC: if magic.startswith(prefix): return descriptor, metadata.st_size, mime, f"document{suffix}" raise OSError("original_not_available") except OSError: os.close(descriptor) return None def safe_report_path(name: str) -> Path | None: clean = Path(unquote(name)).name path = (REPORTS / clean).resolve() try: if path.is_file() and path.is_relative_to(REPORTS.resolve()): return path except OSError: return None return None def canonical_request_host(raw_host: str) -> str | None: if not raw_host or any(char in raw_host for char in ("/", "@", "\\")): return None try: parsed = urlparse("//" + raw_host) host = parsed.hostname _ = parsed.port except ValueError: return None return host.lower().rstrip(".") if host else None def host_is_allowed(raw_host: str) -> bool: host = canonical_request_host(raw_host) return host is not None and host in ALLOWED_HOSTS def origin_matches_request(origin: str, raw_host: str) -> bool: try: request_authority = urlparse("//" + raw_host) supplied = urlparse(origin) request_port = request_authority.port supplied_port = supplied.port except ValueError: return False if ( supplied.scheme != "http" or not supplied.hostname or supplied.username is not None or supplied.password is not None or supplied.path or supplied.params or supplied.query or supplied.fragment ): return False request_host = request_authority.hostname return ( request_host is not None and supplied.hostname.lower().rstrip(".") == request_host.lower().rstrip(".") and supplied_port == request_port ) def issue_csrf_token(now: float | None = None) -> str: issued_at = time.monotonic() if now is None else now token = secrets.token_urlsafe(24) with _csrf_lock: expired = [ existing for existing, timestamp in _csrf_tokens.items() if issued_at - timestamp > CSRF_TTL_SECONDS ] for existing in expired: _csrf_tokens.pop(existing, None) _csrf_tokens[token] = issued_at while len(_csrf_tokens) > 256: oldest = min(_csrf_tokens, key=_csrf_tokens.get) # type: ignore[arg-type] _csrf_tokens.pop(oldest, None) return token def consume_csrf_token(token: str, now: float | None = None) -> bool: checked_at = time.monotonic() if now is None else now with _csrf_lock: issued_at = _csrf_tokens.pop(token, None) return issued_at is not None and 0 <= checked_at - issued_at <= CSRF_TTL_SECONDS def issue_queue_receipt(now: float | None = None) -> str: issued_at = time.monotonic() if now is None else now token = secrets.token_urlsafe(18) with _queue_receipt_lock: _queue_receipts[token] = issued_at expired = [ key for key, timestamp in _queue_receipts.items() if issued_at - timestamp > QUEUE_RECEIPT_TTL_SECONDS ] for key in expired: _queue_receipts.pop(key, None) while len(_queue_receipts) > 256: oldest = min(_queue_receipts, key=_queue_receipts.get) # type: ignore[arg-type] _queue_receipts.pop(oldest, None) return token def consume_queue_receipt(token: str, now: float | None = None) -> bool: checked_at = time.monotonic() if now is None else now with _queue_receipt_lock: issued_at = _queue_receipts.pop(token, None) return ( issued_at is not None and 0 <= checked_at - issued_at <= QUEUE_RECEIPT_TTL_SECONDS ) def dashboard_html_with_nonce( path: Path, *, capture_queued: bool = False, ) -> tuple[bytes, str, str, str]: nonce = secrets.token_urlsafe(18) csrf_token = issue_csrf_token() browser_session_csrf_token = issue_csrf_token() text = path.read_text(encoding="utf-8") capture_day = local_today() text = text.replace("__CSRF_TOKEN__", csrf_token) text = text.replace("__BROWSER_SESSION_CSRF__", browser_session_csrf_token) text = text.replace("__CAPTURE_QUEUED__", "true" if capture_queued else "false") text = text.replace("__CAPTURE_DATE__", capture_day.isoformat()) text = text.replace("__CAPTURE_DATE_LABEL__", capture_day.strftime("%d.%m.%Y")) text = text.replace(" date: return datetime.now(CAPTURE_TIMEZONE).date() def validate_symptom_form( form: dict[str, list[str]], ) -> tuple[str, dict[str, int], str]: day = (form.get("date") or [""])[0] try: parsed_day = date.fromisoformat(day) except ValueError as exc: raise ValueError("invalid date") from exc if parsed_day.year < 2000 or parsed_day > local_today(): raise ValueError("date outside allowed range") scores: dict[str, int] = {} for field in SYMPTOM_FIELDS: raw = (form.get(field) or [""])[0] if raw not in {"0", "1", "2", "3"}: raise ValueError(f"invalid or missing score: {field}") scores[field] = int(raw) notes = (form.get("notes") or [""])[0].strip() if len(notes) > 300: raise ValueError("notes too long") return parsed_day.isoformat(), scores, notes def validate_symptom_submission( form: dict[str, list[str]], ) -> tuple[str, dict[str, int], str, str]: required = {"csrf_token", "date", "notes", *SYMPTOM_FIELDS} allowed = required | {"return_to"} keys = frozenset(form) if keys not in {frozenset(required), frozenset(allowed)}: raise ValueError("invalid submission shape") if any(len(values) != 1 for values in form.values()): raise ValueError("every field must have a single value") return_to = form.get("return_to", ["v4"])[0] if return_to not in {"v4", "v5"}: raise ValueError("invalid return target") day, scores, notes = validate_symptom_form(form) return day, scores, notes, return_to def ensure_private_inbox() -> None: ACTION_INBOX.mkdir(mode=0o700, parents=True, exist_ok=True) metadata = ACTION_INBOX.lstat() if not stat.S_ISDIR(metadata.st_mode) or metadata.st_uid != os.getuid(): raise OSError("action inbox is not a private owned directory") os.chmod(ACTION_INBOX, 0o700) def write_symptom_checkin(day: str, scores: dict[str, int], notes: str) -> None: """Atomically enqueue a validated action; the network server never writes the DB.""" with _action_lock: ensure_private_inbox() if sum(1 for _ in ACTION_INBOX.glob("*.json")) >= MAX_PENDING_ACTIONS: raise QueueFullError("private action queue is full") token = secrets.token_hex(16) temporary = ACTION_INBOX / f".{token}.tmp" destination = ACTION_INBOX / f"{token}.json" payload = json.dumps( { "version": 1, "action": "symptom_checkin", "date": day, "scores": scores, "notes": notes, }, ensure_ascii=False, separators=(",", ":"), ).encode("utf-8") descriptor = os.open(temporary, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) try: with os.fdopen(descriptor, "wb") as handle: handle.write(payload) handle.flush() os.fsync(handle.fileno()) temporary.replace(destination) directory = os.open(ACTION_INBOX, os.O_RDONLY | os.O_DIRECTORY) try: os.fsync(directory) finally: os.close(directory) except Exception: temporary.unlink(missing_ok=True) raise class Handler(BaseHTTPRequestHandler): def send_error( self, code: int, message: str | None = None, explain: str | None = None, ) -> None: request_path = urlparse(getattr(self, "path", "")).path if request_path.startswith("/api/v1/"): send_body = getattr(self, "command", "") != "HEAD" if not host_is_allowed(self.headers.get("Host", "")): self._send_api_error(421, "invalid_host", send_body) return status = 405 if code == 501 else code error_code = "read_only_endpoint" if code == 501 else "request_rejected" self._send_api_error(status, error_code, send_body) return super().send_error(code, message, explain) def _reject_unsupported(self) -> None: if not host_is_allowed(self.headers.get("Host", "")): if urlparse(self.path).path.startswith("/api/v1/"): self._send_api_error(421, "host_not_allowed", self.command != "HEAD") else: self.send_error(421) return if urlparse(self.path).path.startswith("/api/v1/"): self._send_api_error(405, "read_only_endpoint", self.command != "HEAD") return self.send_error(405) do_OPTIONS = _reject_unsupported # noqa: N815 do_PUT = _reject_unsupported # noqa: N815 do_DELETE = _reject_unsupported # noqa: N815 do_PATCH = _reject_unsupported # noqa: N815 do_TRACE = _reject_unsupported # noqa: N815 do_CONNECT = _reject_unsupported # noqa: N815 def do_HEAD(self) -> None: # noqa: N802 if not host_is_allowed(self.headers.get("Host", "")): if urlparse(self.path).path.startswith("/api/v1/"): self._send_api_error(421, "host_not_allowed", False) else: self.send_error(421) return self._handle(send_body=False) def do_GET(self) -> None: # noqa: N802 if not host_is_allowed(self.headers.get("Host", "")): if urlparse(self.path).path.startswith("/api/v1/"): self._send_api_error(421, "host_not_allowed", True) else: self.send_error(421) return self._handle(send_body=True) def do_POST(self) -> None: # noqa: N802 if not host_is_allowed(self.headers.get("Host", "")): if urlparse(self.path).path.startswith("/api/v1/"): self._send_api_error(421, "host_not_allowed", True) else: self.send_error(421) return api_path = urlparse(self.path).path if api_path.startswith("/api/v1/"): if api_path == "/api/v1/browser-session": self._handle_browser_session(True) else: self._send_api_error(405, "read_only_endpoint", True) return if urlparse(self.path).path != CHECKIN_ROUTE: self.send_error(404) return origin = self.headers.get("Origin", "") host = self.headers.get("Host", "") if not origin_matches_request(origin, host): self.send_error(403) return if self.headers.get_content_type() != "application/x-www-form-urlencoded": self.send_error(415) return try: length = int(self.headers.get("Content-Length", "0")) except ValueError: self.send_error(400) return if length < 1 or length > 4096: self.send_error(413) return try: form = parse_qs( self.rfile.read(length).decode("utf-8"), keep_blank_values=True, strict_parsing=True, ) except (UnicodeError, ValueError): self.send_error(400) return token = (form.get("csrf_token") or [""])[0] cookie = SimpleCookie(self.headers.get("Cookie", "")) cookie_token = cookie.get("health_csrf") if ( not token or not cookie_token or not secrets.compare_digest(token, cookie_token.value) ): self.send_error(403) return if not consume_csrf_token(token): self.send_error(403) return try: day, scores, notes, return_to = validate_symptom_submission(form) write_symptom_checkin(day, scores, notes) except ValueError: self.send_error(400) return except QueueFullError: self.send_error(503) return except OSError: self.send_error(500) return if return_to == "v5": receipt = quote(issue_queue_receipt(), safe="") target = V5_ROUTE + f"?queued={receipt}#capture-status" else: target = ROUTE + "?queued=1#symptome" self.send_response(303) self.send_header("Location", target) self.send_header("Cache-Control", "no-store") self.end_headers() def _handle(self, send_body: bool) -> None: request = urlparse(self.path) path = request.path if path.startswith("/api/v1/"): self._handle_api(path, request.query, send_body) return if path == "/": self.send_response(302) self.send_header("Location", ROUTE) self.end_headers() return if path in {ROUTE, V5_ROUTE}: selected = DASHBOARD_FILE if path == ROUTE else DASHBOARD_V5_FILE if selected is None: self.send_error(404) return try: queued_values = parse_qs(request.query, keep_blank_values=True).get( "queued", [] ) capture_queued = ( send_body and path == V5_ROUTE and len(queued_values) == 1 and consume_queue_receipt(queued_values[0]) ) data, nonce, csrf_token, browser_session_csrf_token = ( dashboard_html_with_nonce( selected, capture_queued=capture_queued, ) ) except (OSError, UnicodeError): self.send_error(404) return self._send_bytes( data, selected.name, "text/html; charset=utf-8", send_body, nonce=nonce, csrf_token=csrf_token, browser_session_csrf_token=browser_session_csrf_token, connect_self=path == V5_ROUTE, ) return if path in ASSET_ROUTES: filename, content_type = ASSET_ROUTES[path] self._serve_file( ASSET_DIR / filename, filename, content_type, send_body, ) return if path.startswith("/health-doc/"): try: doc_id = int(path.rsplit("/", 1)[-1]) except ValueError: self.send_error(404) return found = document_path(doc_id) if not found: self.send_error(404) return self._serve_file(found[0], found[1], None, send_body) return if path.startswith("/health-report/"): report = safe_report_path(path.rsplit("/", 1)[-1]) if not report: self.send_error(404) return self._serve_file(report, report.name, None, send_body) return self.send_error(404) def _handle_browser_session(self, send_body: bool) -> None: request = urlparse(self.path) if request.query or API_DB is None: self._send_api_error( 503 if API_DB is None else 400, "api_unavailable" if API_DB is None else "unknown_parameter", send_body, ) return origin = self.headers.get("Origin", "") fetch_site = self.headers.get("Sec-Fetch-Site", "") if ( origin and not origin_matches_request(origin, self.headers.get("Host", "")) ) or (fetch_site and fetch_site not in {"same-origin", "none"}): self._send_api_error(403, "origin_not_allowed", send_body) return if self.headers.get("Content-Length", "0") not in {"", "0"}: self._send_api_error(400, "request_rejected", send_body) return try: cookie = SimpleCookie(self.headers.get("Cookie", "")) except (CookieError, ValueError): self._send_api_error(403, "csrf_rejected", send_body) return csrf_header = self.headers.get("X-Health-Browser-CSRF", "") csrf_cookie = cookie.get("health_browser_csrf") if ( not csrf_header or csrf_cookie is None or len(csrf_header) != len(csrf_cookie.value) or not secrets.compare_digest(csrf_header, csrf_cookie.value) or not consume_csrf_token(csrf_header) ): self._send_api_error(403, "csrf_rejected", send_body) return session_id = issue_browser_session() cookie = ( f"health_api_session={session_id}; Path=/api/v1; HttpOnly; SameSite=Strict; " f"Max-Age={BROWSER_SESSION_TTL_SECONDS}" ) if BROWSER_SESSION_COOKIE_SECURE: cookie += "; Secure" self.send_response(204) self.send_header("Set-Cookie", cookie) self.send_header("Cache-Control", "no-store") self.send_header("Pragma", "no-cache") self.send_header("X-Content-Type-Options", "nosniff") self.send_header("X-Frame-Options", "DENY") self.send_header("Referrer-Policy", "no-referrer") self.send_header( "Content-Security-Policy", "default-src 'none'; frame-ancestors 'none'" ) self.send_header("Content-Length", "0") self.end_headers() def _handle_api(self, path: str, query: str, send_body: bool) -> None: if API_DB is None: self._send_api_error(503, "api_unavailable", send_body) return try: token = load_api_token() except (OSError, UnicodeError): token = None bearer_authenticated = token is not None and api_request_is_authenticated( self.headers.get("Authorization", ""), token ) session_authenticated = browser_session_is_authenticated( self.headers.get("Cookie", "") ) if token is None: self._send_api_error(503, "api_unavailable", send_body) return if not bearer_authenticated and not session_authenticated: self._send_api_error(401, "authentication_required", send_body) return origin = self.headers.get("Origin", "") fetch_site = self.headers.get("Sec-Fetch-Site", "") if ( origin and not origin_matches_request(origin, self.headers.get("Host", "")) ) or (fetch_site and fetch_site not in {"same-origin", "none"}): self._send_api_error(403, "origin_not_allowed", send_body) return original = re.fullmatch( r"/api/v1/documents/(api-document-[a-f0-9]{24})/original", path ) if original: if query: self._send_api_error(400, "unknown_parameter", send_body) return opened = open_verified_original(API_DB, original.group(1)) if not opened: self._send_api_error(404, "original_not_available", send_body) return descriptor, size, mime, filename = opened try: self.send_response(200) self.send_header("Content-Type", mime) self.send_header("Content-Length", str(size)) self.send_header("Content-Disposition", f"inline; filename={filename}") self.send_header("Cache-Control", "no-store") self.send_header("Pragma", "no-cache") self.send_header("X-Content-Type-Options", "nosniff") self.send_header("X-Frame-Options", "DENY") self.send_header("Referrer-Policy", "no-referrer") self.send_header( "Content-Security-Policy", "default-src 'none'; frame-ancestors 'none'", ) self.end_headers() if send_body: with os.fdopen(descriptor, "rb", closefd=False) as handle: while data := handle.read(64 * 1024): self.wfile.write(data) except OSError: return finally: os.close(descriptor) return try: payload = dispatch_api(API_DB, path, query) except APIError as error: self._send_api_error(error.status, error.code, send_body) return except (OSError, UnicodeError): self._send_api_error(503, "api_unavailable", send_body) return except Exception: self._send_api_error(500, "internal_error", send_body) return self._send_api_json(200, payload, send_body) def _send_api_error(self, status: int, code: str, send_body: bool) -> None: self._send_api_json(status, {"error": {"code": code}}, send_body) def _send_api_json( self, status: int, payload: dict[str, object], send_body: bool ) -> None: try: data = json.dumps( payload, ensure_ascii=False, allow_nan=False, sort_keys=True, separators=(",", ":"), ).encode("utf-8") except (TypeError, ValueError, OverflowError): status = 500 data = b'{"error":{"code":"internal_error"}}' if len(data) > MAX_API_RESPONSE_BYTES: status = 422 data = b'{"error":{"code":"response_too_large"}}' self.send_response(status) self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Content-Length", str(len(data))) self.send_header("Cache-Control", "no-store") self.send_header("Pragma", "no-cache") self.send_header("X-Content-Type-Options", "nosniff") self.send_header("X-Frame-Options", "DENY") self.send_header("Referrer-Policy", "no-referrer") self.send_header( "Permissions-Policy", "camera=(), microphone=(), geolocation=()" ) self.send_header( "Content-Security-Policy", "default-src 'none'; frame-ancestors 'none'" ) if status == 401: self.send_header("WWW-Authenticate", 'Bearer realm="health-dashboard-api"') if TEST_INSTANCE_ID: self.send_header("X-Health-Synthetic-Instance", TEST_INSTANCE_ID) self.end_headers() if send_body: self.wfile.write(data) def _serve_file( self, path: Path, filename: str, content_type: str | None, send_body: bool, ) -> None: try: data = path.read_bytes() except OSError: self.send_error(404) return mime = ( content_type or mimetypes.guess_type(str(path))[0] or "application/octet-stream" ) self._send_bytes(data, filename, mime, send_body) def _send_bytes( self, data: bytes, filename: str, content_type: str, send_body: bool, *, nonce: str | None = None, csrf_token: str | None = None, browser_session_csrf_token: str | None = None, connect_self: bool = False, ) -> None: self.send_response(200) self.send_header("Content-Type", content_type) self.send_header("Content-Length", str(len(data))) self.send_header("Cache-Control", "no-store") self.send_header("X-Content-Type-Options", "nosniff") self.send_header("X-Frame-Options", "DENY") self.send_header("Referrer-Policy", "strict-origin") self.send_header( "Permissions-Policy", "camera=(), microphone=(), geolocation=()" ) if TEST_INSTANCE_ID: self.send_header("X-Health-Synthetic-Instance", TEST_INSTANCE_ID) if nonce: connect_source = "'self'" if connect_self else "'none'" policy = ( "default-src 'none'; " f"script-src 'self' 'nonce-{nonce}'; " f"style-src 'self' 'nonce-{nonce}'; " "img-src 'self' data:; font-src 'self'; " f"connect-src {connect_source}; " "object-src 'none'; base-uri 'none'; frame-ancestors 'none'; form-action 'self'" ) self.send_header("Content-Security-Policy", policy) if csrf_token: self.send_header( "Set-Cookie", f"health_csrf={csrf_token}; Path=/; HttpOnly; SameSite=Strict", ) if browser_session_csrf_token: self.send_header( "Set-Cookie", f"health_browser_csrf={browser_session_csrf_token}; Path=/api/v1; HttpOnly; SameSite=Strict", ) self.send_header( "Content-Disposition", f"inline; filename*=UTF-8''{quote(filename)}" ) self.end_headers() if send_body: self.wfile.write(data) def log_message(self, format: str, *args: object) -> None: return if __name__ == "__main__": if not DASHBOARD_FILE.is_file(): raise SystemExit("health dashboard file missing") if not (ASSET_DIR / "chart.umd.min.js").is_file(): raise SystemExit("local Chart.js asset missing") ThreadingHTTPServer((BIND_HOST, BIND_PORT), Handler).serve_forever()