#!/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 SimpleCookie from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path from urllib.parse import parse_qs, quote, unquote, urlparse from zoneinfo import ZoneInfo BASE = Path.home() / ".hermes" / "assets" / "Gesundheit" 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.css": ("dashboard-v5.css", "text/css; 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() 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 CAPTURE_TIMEZONE = ZoneInfo("Europe/Zurich") MAX_PENDING_ACTIONS = max(1, min(int(os.environ.get("HEALTH_DASHBOARD_MAX_PENDING_ACTIONS", "64")), 256)) _csrf_tokens: dict[str, float] = {} _csrf_lock = threading.Lock() _queue_receipts: dict[str, float] = {} _queue_receipt_lock = threading.Lock() _action_lock = threading.Lock() class QueueFullError(OSError): """Raised when the bounded private action queue cannot accept more work.""" 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(DB) 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 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]: nonce = secrets.token_urlsafe(18) 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("__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 _reject_unsupported(self) -> None: if not host_is_allowed(self.headers.get("Host", "")): self.send_error(421) 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", "")): 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", "")): 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", "")): self.send_error(421) 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 == "/": 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 = 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, ) 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 _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, ) -> 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: policy = ( "default-src 'none'; " f"script-src 'self' 'nonce-{nonce}'; " f"style-src 'self' 'nonce-{nonce}'; " "img-src 'self' data:; font-src 'self'; connect-src 'none'; " "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", ) 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()