#!/usr/bin/env python3 """Serve the local Health Dashboard with strict routes and privacy headers.""" from __future__ import annotations import base64 import binascii import json import mimetypes import os import re import secrets import sqlite3 import stat import threading import time from datetime import date, datetime, timedelta 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, ) from dashboard_v5.document_originals import configured_original_roots, probe_original 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-nutrition.js": ( "dashboard-v5-nutrition.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-echarts.js": ( "dashboard-v5-echarts.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/fullcalendar-6.1.21.min.js": ( "fullcalendar-6.1.21.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" NUTRITION_MAPPING_ROUTE = "/health-actions/nutrition-mapping" SYMPTOM_EVENT_ROUTE = "/health-actions/symptom-event" MEDICATION_EVENT_ROUTE = "/health-actions/medication-event" GENERAL_EVENT_ROUTE = "/health-actions/general-event" 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 ) BASIC_PASSWORD_FILE = ( Path(os.environ["HEALTH_DASHBOARD_BASIC_PASSWORD_FILE"]).expanduser() if os.environ.get("HEALTH_DASHBOARD_BASIC_PASSWORD_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") MAPPING_DECISIONS = {"assign", "composite", "ignore"} MAPPING_SCORES = {"0", "1", "2", "3", "unknown"} MAPPING_CONFIDENCE = {"low", "medium", "high"} MAPPING_METHODS = {"sighi_reference", "ingredient_label", "manual_review", "local_alias"} PERSONAL_TOLERANCE_STATUSES = { "unknown", "documented_tolerated", "documented_not_tolerated", "unclear", } 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"), ) ORIGINAL_ROOTS = configured_original_roots() _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 load_dashboard_basic_password() -> str: if BASIC_PASSWORD_FILE is None: return load_api_token() descriptor = os.open( BASIC_PASSWORD_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 > 129 ): raise OSError("dashboard password file is not private") raw = os.read(descriptor, 130) finally: os.close(descriptor) password = raw.decode("ascii").strip() if not re.fullmatch(r"[A-Za-z0-9_-]{5,128}", password): raise OSError("dashboard password is invalid") return password 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 dashboard_principal_is_authenticated( header: str, token: str, basic_password: str | None = None ) -> bool: if api_request_is_authenticated(header, token): return True if not header.startswith("Basic ") or header.count(" ") != 1: return False try: decoded = base64.b64decode( header.removeprefix("Basic "), validate=True ).decode("utf-8") except (binascii.Error, UnicodeError): return False username, separator, supplied = decoded.partition(":") expected = token if basic_password is None else basic_password return ( separator == ":" and username == "health" and len(supplied) == len(expected) and secrets.compare_digest(supplied, expected) ) 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_safe_original( database: Path, opaque: str ) -> tuple[int, int, str, str] | None: """Resolve any known document and pin the safely opened descriptor for HEAD/GET.""" connection = connect_read_only(database) try: row = _resolve_record_document(connection, opaque, reviewed=False) except APIError: return None finally: connection.close() roots = ( ORIGINAL_ROOTS if os.environ.get("HEALTH_DASHBOARD_ORIGINAL_ROOTS") else (BASE.absolute(), REPORTS.absolute()) ) probe = probe_original( row["local_original_path"] or row["dateipfad"], roots=roots, max_bytes=MAX_ORIGINAL_BYTES, opener=_open_regular_beneath, ) if probe.status != "available" or probe.descriptor is None: probe.close() return None return probe.descriptor, probe.size, probe.mime, probe.download_name 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("__CAPTURE_DATETIME__", datetime.now(CAPTURE_TIMEZONE).strftime("%Y-%m-%dT%H:%M")) 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 _safe_action_text(value: str, maximum: int, *, required: bool = False) -> str: cleaned = " ".join(value.split()) if (required and not cleaned) or len(cleaned) > maximum or re.search(r"[\x00-\x1f\x7f]|https?://|/|\\\\|\.hermes", cleaned, re.I): raise ValueError("invalid action text") return cleaned def _local_action_datetime(value: str) -> str: try: parsed = datetime.fromisoformat(value) except ValueError as exc: raise ValueError("invalid action datetime") from exc if parsed.tzinfo is not None or parsed.year < 2000: raise ValueError("invalid action datetime") aware = parsed.replace(tzinfo=CAPTURE_TIMEZONE) if aware > datetime.now(CAPTURE_TIMEZONE) + timedelta(minutes=5): raise ValueError("future action datetime") return parsed.isoformat(timespec="minutes") def validate_patient_action(form: dict[str, list[str]], action_path: str) -> tuple[dict[str, object], str]: common = {"csrf_token", "occurred_at", "notes", "return_to"} if action_path == SYMPTOM_EVENT_ROUTE: required = common | {"symptom_type", "severity", "onset_at", "duration_minutes", "label"} elif action_path == MEDICATION_EVENT_ROUTE: required = common | {"event_type", "medication_name", "dose", "unit", "route"} elif action_path == GENERAL_EVENT_ROUTE: required = common | {"category", "label", "intensity"} else: raise ValueError("unsupported patient action") if set(form) != required or any(len(values) != 1 for values in form.values()): raise ValueError("invalid patient action shape") if form["return_to"][0] != "v5": raise ValueError("invalid return target") occurred_at = _local_action_datetime(form["occurred_at"][0]) notes = _safe_action_text(form["notes"][0], 300) if action_path == SYMPTOM_EVENT_ROUTE: symptom_type = form["symptom_type"][0] if symptom_type not in {"headache", "aphthae", "gi", "joints", "skin", "eyes", "fatigue", "other"} or form["severity"][0] not in {"0", "1", "2", "3"}: raise ValueError("invalid symptom event") onset = form["onset_at"][0] onset_at = _local_action_datetime(onset) if onset else "" duration = form["duration_minutes"][0] if duration and (not duration.isdigit() or int(duration) > 10080): raise ValueError("invalid symptom duration") label = _safe_action_text(form["label"][0], 80, required=symptom_type == "other") return {"version": 1, "action": "symptom_event", "occurred_at": occurred_at, "symptom_type": symptom_type, "severity": int(form["severity"][0]), "onset_at": onset_at, "duration_minutes": int(duration) if duration else None, "label": label, "notes": notes}, "v5" if action_path == MEDICATION_EVENT_ROUTE: event_type = form["event_type"][0] if event_type not in {"administered", "missed", "corrected"}: raise ValueError("invalid medication event") medication = _safe_action_text(form["medication_name"][0], 120, required=True) if API_DB is None: raise ValueError("medication catalog unavailable") connection = connect_read_only(API_DB) try: known = connection.execute( "SELECT 1 FROM medication_administrations WHERE medication_name=? LIMIT 1", (medication,), ).fetchone() finally: connection.close() if known is None: raise ValueError("medication is not in exact catalog") dose = _safe_action_text(form["dose"][0], 40) unit = _safe_action_text(form["unit"][0], 30) route = _safe_action_text(form["route"][0], 60) return {"version": 1, "action": "medication_event", "occurred_at": occurred_at, "event_type": event_type, "medication_name": medication, "dose": dose, "unit": unit, "route": route, "notes": notes}, "v5" category = form["category"][0] if category not in {"stress", "infection", "appointment", "physical_load", "sleep_disruption", "heat", "travel", "other"}: raise ValueError("invalid event category") intensity = form["intensity"][0] if intensity and intensity not in {"0", "1", "2", "3"}: raise ValueError("invalid event intensity") label = _safe_action_text(form["label"][0], 100, required=True) return {"version": 1, "action": "general_event", "occurred_at": occurred_at, "category": category, "label": label, "intensity": int(intensity) if intensity else None, "notes": notes}, "v5" def validate_mapping_submission(form: dict[str, list[str]]) -> tuple[dict[str, object], str]: required = { "csrf_token", "decision", "queue_key", "alias", "canonical_food", "sighi_score", "confidence", "mapping_method", "source_label", "source_version", "note", "ingredient_review_required", "personal_tolerance_status", "personal_tolerance_note", "return_to", } if set(form) != required or any(len(values) != 1 for values in form.values()): raise ValueError("invalid mapping shape") return_to = form["return_to"][0] if return_to != "v5": raise ValueError("invalid return target") decision = form["decision"][0] queue_key = form["queue_key"][0].strip() alias = " ".join(form["alias"][0].split()) canonical_food = " ".join(form["canonical_food"][0].split()) score = form["sighi_score"][0] confidence = form["confidence"][0] method = form["mapping_method"][0] source_label = " ".join(form["source_label"][0].split()) source_version = " ".join(form["source_version"][0].split()) note = " ".join(form["note"][0].split()) review = form["ingredient_review_required"][0] tolerance_status = form["personal_tolerance_status"][0] tolerance_note = " ".join(form["personal_tolerance_note"][0].split()) if not re.fullmatch(r"[0-9a-f]{32}", queue_key): raise ValueError("invalid queue key") for value, maximum in ( (alias, 160), (canonical_food, 120), (source_label, 80), (source_version, 40), (note, 300), (tolerance_note, 160), ): if len(value) > maximum or re.search(r"[\x00-\x1f\x7f]|https?://|/|\\\\|\.hermes", value, re.I): raise ValueError("invalid mapping text") if not alias or not source_label or not source_version: raise ValueError("mapping documentation required") if ( decision not in MAPPING_DECISIONS or score not in MAPPING_SCORES or confidence not in MAPPING_CONFIDENCE or method not in MAPPING_METHODS or review not in {"true", "false"} or tolerance_status not in PERSONAL_TOLERANCE_STATUSES ): raise ValueError("invalid mapping value") ingredient_review_required = review == "true" if decision == "assign" and (score == "unknown" or not canonical_food or ingredient_review_required): raise ValueError("assign requires canonical score") if decision == "composite" and (score != "unknown" or not ingredient_review_required): raise ValueError("composite remains unclassified") if decision == "ignore" and (score != "unknown" or canonical_food or ingredient_review_required): raise ValueError("ignore must not classify") return { "version": 1, "action": "nutrition_mapping", "decision": decision, "queue_key": queue_key, "alias": alias, "canonical_food": canonical_food, "sighi_score": "unknown" if score == "unknown" else int(score), "confidence": confidence, "mapping_method": method, "source_label": source_label, "source_version": source_version, "note": note, "ingredient_review_required": ingredient_review_required, "personal_tolerance_status": tolerance_status, "personal_tolerance_note": tolerance_note, }, 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_action_payload(payload: dict[str, object]) -> 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" packed = json.dumps( payload, ensure_ascii=False, separators=(",", ":"), ).encode("utf-8") descriptor = os.open( temporary, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600, ) try: with os.fdopen(descriptor, "wb") as handle: handle.write(packed) 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 def write_symptom_checkin(day: str, scores: dict[str, int], notes: str) -> None: write_action_payload( { "version": 1, "action": "symptom_checkin", "date": day, "scores": scores, "notes": notes, } ) 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 action_path = urlparse(self.path).path if action_path not in {CHECKIN_ROUTE, NUTRITION_MAPPING_ROUTE, SYMPTOM_EVENT_ROUTE, MEDICATION_EVENT_ROUTE, GENERAL_EVENT_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, max_num_fields=20, ) 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 mapping_queue_key_value = "" try: if action_path == CHECKIN_ROUTE: day, scores, notes, return_to = validate_symptom_submission(form) write_symptom_checkin(day, scores, notes) elif action_path == NUTRITION_MAPPING_ROUTE: payload, return_to = validate_mapping_submission(form) mapping_queue_key_value = str(payload["queue_key"]) write_action_payload(payload) else: payload, return_to = validate_patient_action(form, action_path) write_action_payload(payload) 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="") if action_path == NUTRITION_MAPPING_ROUTE: queue_key = quote(mapping_queue_key_value, safe="") target = ( V5_ROUTE + "?view=nutrition&workspace=mapping&map_status=open" + f"&map_sort=frequency&mapping={queue_key}&queued={receipt}" ) else: 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 _v5_principal_authenticated(self) -> bool: if API_DB is None: return True try: token = load_api_token() basic_password = load_dashboard_basic_password() except (OSError, UnicodeError): return False return dashboard_principal_is_authenticated( self.headers.get("Authorization", ""), token, basic_password ) def _send_dashboard_auth_required(self) -> None: self.send_response(401) self.send_header("WWW-Authenticate", 'Basic realm="Health Dashboard V5"') self.send_header("Cache-Control", "no-store") self.send_header("Pragma", "no-cache") self.send_header("X-Content-Type-Options", "nosniff") self.send_header("Content-Length", "0") 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}: if path == V5_ROUTE and not self._v5_principal_authenticated(): self._send_dashboard_auth_required() return 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 if not self._v5_principal_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 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_safe_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()