"""Sprint-6B read-only API contracts and negative HTTP boundaries.""" from __future__ import annotations import importlib.util import json import socket import sqlite3 import stat import sys import threading import urllib.error import urllib.request from datetime import date, datetime, time, timedelta, timezone from pathlib import Path from urllib.parse import urlencode import pytest ROOT = Path(__file__).resolve().parents[1] HEALTH = ROOT / "scripts" / "health" SERVER = HEALTH / "health_dashboard_server.py" if str(HEALTH) not in sys.path: sys.path.insert(0, str(HEALTH)) import dashboard_v5.read_api as read_api # noqa: E402 from dashboard_v5.read_api import ( # noqa: E402 APIError, MAX_LAB_ROWS, MAX_RANGE_DAYS, MAX_SERIES_ROWS, dispatch_api, ) from tests.fixtures.dashboard_v5_fixture import build_dashboard_v5_fixture # noqa: E402 ANCHOR = date(2026, 6, 15) FORBIDDEN_RESPONSE_KEYS = { "dateipfad", "local_original_path", "drive_file_id", "drive_web_url", "raw_json", "file_name", "file_hash", "record_hash", "quelle", } def iso_noon(day: date) -> str: return datetime.combine(day, time(12), tzinfo=timezone.utc).isoformat() def api_database(tmp_path: Path) -> Path: database = tmp_path / "sprint6b.db" build_dashboard_v5_fixture(database, anchor_date=ANCHOR) connection = sqlite3.connect(database) try: zero_day = ANCHOR - timedelta(days=4) connection.execute( "UPDATE apple_health_records SET value=0 WHERE metric='step_count' AND substr(start_date,1,10)=?", (zero_day.isoformat(),), ) extra = [] for offset in range(1, 41): day = ANCHOR - timedelta(days=offset) for metric, value, unit in ( ("walking_running_distance", 3.0 + offset / 100, "km"), ("active_energy", 418.4 + offset, "kJ"), ("blood_oxygen_saturation", 97.0 + (offset % 2), "%"), ("respiratory_rate", 14.0 + (offset % 3), "count/min"), ("physical_effort", 2.0 + offset / 100, "kcal/hr·kg"), ): key = f"s6b-{metric}-{offset}" extra.append(( "HKQuantityType", metric, iso_noon(day), iso_noon(day), value, None, unit, "Synthetic Sprint6B Source", "1", "Synthetic Device", f"{key}.json", key, f"{key}-record", "{}", iso_noon(day), )) for metric, value, unit in ( ("weight_body_mass", 70.5, "kg"), ("body_mass_index", 22.4, "count"), ): key = f"s6b-{metric}" extra.append(( "HKQuantityType", metric, iso_noon(ANCHOR - timedelta(days=2)), iso_noon(ANCHOR - timedelta(days=2)), value, None, unit, "Synthetic Sprint6B Source", "1", "Synthetic Device", f"{key}.json", key, f"{key}-record", "{}", iso_noon(ANCHOR - timedelta(days=2)), )) timezone_instant = datetime.combine( ANCHOR - timedelta(days=6), time(23, 30), tzinfo=timezone.utc ).isoformat() extra.append(( "HKQuantityType", "weight_body_mass", timezone_instant, timezone_instant, 71.0, None, "kg", "Synthetic Sprint6B Source", "1", "Synthetic Device", "s6b-zurich-boundary.json", "s6b-zurich-boundary", "s6b-zurich-boundary-record", "{}", timezone_instant, )) for offset, systolic, diastolic in ((3, 121, 79), (10, 118, 76)): day = ANCHOR - timedelta(days=offset) key = f"s6b-bp-{offset}" raw = json.dumps({ "type": "blood_pressure", "metric": "blood_pressure", "date": iso_noon(day), "systolic": systolic, "diastolic": diastolic, "unit": "mmHg", "source": "Synthetic", }) extra.append(( "HKCorrelationType", "blood_pressure", iso_noon(day), iso_noon(day), None, None, "mmHg", "Synthetic Sprint6B Source", "1", "Synthetic Device", f"{key}.json", key, f"{key}-record", raw, iso_noon(day), )) connection.executemany( """INSERT INTO apple_health_records (record_type,metric,start_date,end_date,value,value_text,unit,source_name, source_version,device,file_name,file_hash,record_hash,raw_json,imported_at) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)""", extra, ) cursor = connection.execute( """INSERT INTO dokumente (datei_name,dateipfad,daten_typ,status,kategorie,review_status, processing_quality,drive_file_id,drive_web_url,local_original_path, document_date,institution) VALUES(?,?,?,?,?,?,?,?,?,?,?,?)""", ( "synthetic-lab.pdf", "/tmp/private/synthetic-lab.pdf", "pdf", "eingearbeitet", "Labor", "geprueft", "synthetic_contract", "drive-secret-id", "https://drive.invalid/private", "/tmp/private/original.pdf", (ANCHOR - timedelta(days=45)).isoformat(), "Synthetisches Labor", ), ) document_id = int(cursor.lastrowid) connection.execute( "UPDATE laborwerte SET dokument_id=?, canonical_document_id=? WHERE parameter_name='C-Reaktives Protein (CRP)'", (document_id, document_id), ) connection.execute( """INSERT INTO laborwerte (dokument_id,canonical_document_id,parameter_name,wert,einheit, reference_min,reference_max,abnahme_datum,befund_datum,wert_original, validierungsstatus,source_type,reference_range_source, verified_against_original,provenance_note) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)""", ( document_id, document_id, "D-Dimer", "0", "µg/L", "0", "500", (ANCHOR - timedelta(days=2)).isoformat(), (ANCHOR - timedelta(days=2)).isoformat(), "0", "validiert", "synthetic", "scanned_original", 1, "Synthetic observation-specific reference contract", ), ) connection.execute( "INSERT INTO health_events(date,category,parameter,value,unit,source,notes) VALUES(?,?,?,?,?,?,?)", ((ANCHOR - timedelta(days=2)).isoformat(), "appointment", "Kontrolle", None, None, "fixture", "/private/note must never leave"), ) connection.execute( "INSERT INTO health_events(date,category,parameter,value,unit,source,notes) VALUES(?,?,?,?,?,?,?)", ((ANCHOR - timedelta(days=2)).isoformat(), "malformed", "prefix /home/agent/private", None, None, "fixture", None), ) connection.executemany( "INSERT INTO health_events(date,category,parameter,value,unit,source,notes) VALUES(?,?,?,?,?,?,?)", [ ((ANCHOR - timedelta(days=2)).isoformat(), "malformed", value, None, None, "fixture", None) for value in ( "prefix:/home/agent/private.db", "see:https://drive.invalid/private", "x|/tmp/private/report.pdf", "meta:1A2B3C4D5E6F7G8H9I0J1K2L3M4N5O6P7Q8R9S0T", "prefix /var/lib/secret.db", "prefix=/var/lib/secret.db", "relative ../../secret.pdf", ) ], ) connection.execute( """INSERT INTO health_event_periods (start_date,end_date,event_type,label,severity,location,source,notes,confidence) VALUES(?,?,?,?,?,?,?,?,?)""", ((ANCHOR - timedelta(days=12)).isoformat(), (ANCHOR - timedelta(days=8)).isoformat(), "symptom_phase", "Synthetische Phase", None, None, "fixture", "private", 1.0), ) connection.commit() finally: connection.close() database.chmod(0o600) return database def assert_no_forbidden_fields(value): if isinstance(value, dict): assert not (FORBIDDEN_RESPONSE_KEYS & set(value)) for item in value.values(): assert_no_forbidden_fields(item) elif isinstance(value, list): for item in value: assert_no_forbidden_fields(item) elif isinstance(value, str): assert "/tmp/private" not in value assert "/home/agent" not in value assert "drive-secret-id" not in value assert "drive.invalid" not in value assert "1A2B3C4D5E6F7G8H9I0J1K2L3M4N5O6P7Q8R9S0T" not in value assert "/var/lib/secret.db" not in value assert "prefix=/var" not in value assert "../../secret.pdf" not in value def test_catalog_and_series_contracts_preserve_zero_gaps_units_and_complete_weeks(tmp_path): database = api_database(tmp_path) catalog = dispatch_api(database, "/api/v1/metric-catalog", "q=Blutdruck") assert [item["id"] for item in catalog["groups"]["metrics"][:3]] == [ "panel.blood_pressure", "apple.blood_pressure.systolic", "apple.blood_pressure.diastolic", ] start = (ANCHOR - timedelta(days=22)).isoformat() end = (ANCHOR - timedelta(days=1)).isoformat() daily = dispatch_api( database, "/api/v1/series", urlencode({"metric": "apple.steps", "from": start, "to": end, "resolution": "day"}), ) assert daily["timezone"] == "Europe/Zurich" assert daily["unit"] == "Schritte" and daily["aggregation"] == "sum" assert daily["source"] == {"type": "apple_health", "identifier": "step_count", "parser": "apple_health_analytics_v2"} zero_day = (ANCHOR - timedelta(days=4)).isoformat() assert {point["date"]: point["value"] for point in daily["points"]}[zero_day] == 0 assert all(point["value"] is not None for point in daily["points"]) assert daily["coverage"]["observed_days"] == 22 assert daily["coverage"]["missing_days"] == 0 weekly = dispatch_api( database, "/api/v1/series", urlencode({"metric": "apple.steps", "from": start, "to": end, "resolution": "week"}), ) assert weekly["aggregation_rule"] == { "id": "complete_calendar_week_daily_metric", "minimum_observations": 7, "requires_complete_calendar_week": True, } assert all(point["observation_count"] == 7 for point in weekly["points"]) assert all(point["observed_days"] == 7 for point in weekly["points"]) assert all(point["quality"] == "complete_calendar_week" for point in weekly["points"]) assert all(point["week"].endswith(tuple(f"W{i:02d}" for i in range(1, 54))) for point in weekly["points"]) gap_day = (ANCHOR - timedelta(days=4)).isoformat() day = dispatch_api(database, f"/api/v1/day/{gap_day}", "") assert day["metrics"]["apple.steps"]["value"] == 0 assert "apple.sleep" not in day["metrics"] assert all(value is not None for item in day["metrics"].values() for key, value in item.items() if key == "value") def test_blood_pressure_and_free_ranges_are_deterministic_and_bounded(tmp_path): database = api_database(tmp_path) query = urlencode({ "metric": "apple.blood_pressure.systolic", "from": (ANCHOR - timedelta(days=30)).isoformat(), "to": ANCHOR.isoformat(), "resolution": "day", }) first = dispatch_api(database, "/api/v1/series", query) second = dispatch_api(database, "/api/v1/series", query) assert first == second assert [point["value"] for point in first["points"]] == [118, 121] assert first["unit"] == "mmHg" with pytest.raises(APIError) as panel: dispatch_api(database, "/api/v1/series", query.replace("apple.blood_pressure.systolic", "panel.blood_pressure")) assert (panel.value.status, panel.value.code) == (400, "metric_not_series") with pytest.raises(APIError) as too_wide: dispatch_api( database, "/api/v1/series", urlencode({"metric": "apple.steps", "from": "2000-01-01", "to": "2026-01-01", "resolution": "day"}), ) assert too_wide.value.code == "range_too_large" assert MAX_RANGE_DAYS < 26 * 366 def test_seven_thirty_ninety_180_all_units_and_zurich_boundary_are_deterministic(tmp_path): database = api_database(tmp_path) for days in (7, 30, 90, 180): query = urlencode({ "metric": "apple.distance", "from": (ANCHOR - timedelta(days=days)).isoformat(), "to": (ANCHOR - timedelta(days=1)).isoformat(), "resolution": "day", }) assert dispatch_api(database, "/api/v1/series", query) == dispatch_api( database, "/api/v1/series", query ) all_query = "metric=apple.distance&resolution=day" all_series = dispatch_api(database, "/api/v1/series", all_query) assert all_series == dispatch_api(database, "/api/v1/series", all_query) assert all_series["unit"] == "km" energy = dispatch_api( database, "/api/v1/series", urlencode({ "metric": "apple.active_energy", "from": (ANCHOR - timedelta(days=1)).isoformat(), "to": (ANCHOR - timedelta(days=1)).isoformat(), "resolution": "day", }), ) assert energy["unit"] == "kcal" assert energy["points"][0]["value"] != 419.4 local_day = (ANCHOR - timedelta(days=5)).isoformat() weight = dispatch_api( database, "/api/v1/series", urlencode({"metric": "apple.weight", "from": local_day, "to": local_day, "resolution": "day"}), ) assert weight["points"] == [{ "date": local_day, "value": 71.0, "quality": "direct", "source_class": "app", "baseline": None, }] assert weight["timezone"] == "Europe/Zurich" def test_laboratory_api_keeps_observation_reference_document_and_redaction(tmp_path): database = api_database(tmp_path) labs = dispatch_api(database, "/api/v1/labs", "") d_dimer = next(item for item in labs["observations"] if item["parameter"] == "D-Dimer") assert d_dimer["value"] == 0 assert d_dimer["reference"] == {"min": "0", "max": "500", "source": "scanned_original"} assert d_dimer["document"] == { "id": d_dimer["document"]["id"], "date": (ANCHOR - timedelta(days=45)).isoformat(), "category": "Labor", "institution": "Synthetisches Labor", } assert d_dimer["document"]["id"].startswith("api-document-") assert_no_forbidden_fields(labs) def test_event_and_search_apis_are_grouped_bounded_and_redacted(tmp_path): database = api_database(tmp_path) events = dispatch_api( database, "/api/v1/events", urlencode({"from": (ANCHOR - timedelta(days=20)).isoformat(), "to": ANCHOR.isoformat(), "types": "medication_administered,health_event,health_period"}), ) assert {item["type"] for item in events["events"]} <= {"medication_administered", "health_event", "health_period"} assert_no_forbidden_fields(events) for term, group in (("CRP", "laboratory"), ("Aphthen", "symptoms_events"), ("Medikamente", "symptoms_events")): result = dispatch_api(database, "/api/v1/search", urlencode({"q": term})) assert result["groups"][group] assert set(result["groups"]) == {"metrics", "laboratory", "symptoms_events", "documents", "days"} assert_no_forbidden_fields(result) documents = dispatch_api(database, "/api/v1/search", "q=Labor") assert documents["groups"]["documents"] assert documents["groups"]["documents"][0]["id"].startswith("api-document-") assert documents["groups"]["documents"][0]["drill_down_target"] == "document_metadata" date_result = dispatch_api(database, "/api/v1/search", f"q={(ANCHOR - timedelta(days=2)).isoformat()}") assert date_result["groups"]["days"] == [{"date": (ANCHOR - timedelta(days=2)).isoformat(), "drill_down_target": "day"}] def test_dispatch_rejects_unknown_metrics_columns_paths_queries_dates_and_limits(tmp_path): assert MAX_RANGE_DAYS == 3660 assert MAX_SERIES_ROWS == 3660 assert MAX_LAB_ROWS == 100 database = api_database(tmp_path) rejected = [ ("/api/v1/series", "metric=sqlite_master&from=2026-01-01&to=2026-01-02&resolution=day"), ("/api/v1/series", "metric=apple.steps&from=2026-01-01&to=2026-01-02&resolution=raw"), ("/api/v1/series", "metric=apple.steps&from=2026-01-01&to=2026-01-02&resolution=day&column=raw_json"), ("/api/v1/metric-catalog", "q=HRV&q=CRP"), ("/api/v1/labs", "path=/tmp/private"), ("/api/v1/events", "from=2026-01-01&to=2026-01-02&types=notes"), ("/api/v1/day/../../etc/passwd", ""), ("/api/v1/day/2026-02-30", ""), ("/api/v1/search", "q=x"), ] for path, query in rejected: with pytest.raises(APIError) as error: dispatch_api(database, path, query) assert error.value.status in {400, 404, 422} assert error.value.code not in {"internal_error", "sql_error"} def test_event_and_laboratory_row_limits_fail_closed_without_truncation(tmp_path): database = api_database(tmp_path) connection = sqlite3.connect(database) try: connection.executemany( """INSERT INTO health_events(date,category,parameter,value,unit,source,notes) VALUES(?,?,?,?,?,?,?)""", [ (ANCHOR.isoformat(), "synthetic_event", f"event-{index}", None, None, "synthetic", None) for index in range(1001) ], ) connection.executemany( """INSERT INTO laborwerte( dokument_id,parameter_name,wert,einheit,reference_min,reference_max, abnahme_datum,befund_datum,wert_original,quelle,validierungsstatus, source_type,reference_range_source,verified_against_original, canonical_document_id,provenance_note ) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)""", [ ( None, "CRP", str(index), "mg/L", "0", "5", (ANCHOR - timedelta(days=index + 100)).isoformat(), None, str(index), "synthetic_contract", "validiert", "scanned_original", "scanned_original", 1, None, "synthetic_contract", ) for index in range(101) ], ) connection.commit() finally: connection.close() with pytest.raises(APIError, match="row_limit_exceeded"): dispatch_api( database, "/api/v1/events", urlencode({"from": ANCHOR.isoformat(), "to": ANCHOR.isoformat(), "types": "health_event"}), ) with pytest.raises(APIError, match="row_limit_exceeded"): dispatch_api(database, "/api/v1/labs", "") def test_source_inventory_identifier_limit_fails_closed(tmp_path): database = api_database(tmp_path) connection = sqlite3.connect(database) try: connection.executemany( "INSERT INTO apple_health_records(metric,value,unit,raw_json) VALUES(?,?,?,?)", [(f"unreleased-synthetic-{index}", 1.0, "widget", "{}") for index in range(1001)], ) connection.commit() finally: connection.close() with pytest.raises(APIError, match="source_inventory_limit_exceeded"): dispatch_api(database, "/api/v1/metric-catalog", "") def test_future_observations_reversed_periods_and_iso_boundaries_fail_closed(tmp_path): database = api_database(tmp_path) connection = sqlite3.connect(database) try: connection.execute( """INSERT INTO apple_health_records(metric,value,unit,raw_json,start_date) VALUES('step_count',999,'count','{}','2099-01-02T12:00:00+01:00')""" ) connection.execute( """INSERT INTO laborwerte( parameter_name,wert,einheit,reference_min,reference_max,abnahme_datum, wert_original,validierungsstatus,source_type,reference_range_source, verified_against_original,provenance_note ) VALUES('CRP','1','mg/L','0','5','2099-01-03','1','validiert', 'scanned_original','scanned_original',1,'synthetic_future')""" ) connection.execute( """INSERT INTO health_event_periods(start_date,end_date,event_type,label,source,confidence) VALUES('2026-06-10','2026-06-01','synthetic_reversed','Reversed','fixture',1.0)""" ) connection.commit() finally: connection.close() all_steps = dispatch_api(database, "/api/v1/series", "metric=apple.steps&resolution=day") assert "2099-01-02" not in {point["date"] for point in all_steps["points"]} labs = dispatch_api(database, "/api/v1/labs", "") assert "2099-01-03" not in {item["date"] for item in labs["observations"]} periods = dispatch_api( database, "/api/v1/events", "from=2026-06-01&to=2026-06-15&types=health_period", ) assert all(item["category"] != "synthetic_reversed" for item in periods["events"]) earliest = dispatch_api( database, "/api/v1/series", "metric=apple.steps&from=0001-01-01&to=0001-01-01&resolution=day", ) assert earliest["points"] == [] connection = sqlite3.connect(database) try: connection.execute( """INSERT INTO apple_health_records(metric,value,unit,raw_json,start_date) VALUES('step_count',1,'count','{}','2010-01-01T12:00:00+01:00')""" ) connection.commit() finally: connection.close() with pytest.raises(APIError, match="range_too_large"): dispatch_api(database, "/api/v1/series", "metric=apple.steps&resolution=day") with pytest.raises(APIError, match="future_date_not_allowed"): dispatch_api(database, "/api/v1/day/9999-12-31", "") with pytest.raises(APIError, match="future_range_not_allowed"): dispatch_api( database, "/api/v1/series", "metric=apple.steps&from=2099-01-01&to=2099-01-02&resolution=day", ) def test_planned_medication_rows_cannot_hide_a_later_actual_administration(tmp_path): database = api_database(tmp_path) connection = sqlite3.connect(database) try: connection.executemany( """INSERT INTO medication_administrations(datum,medication_name,event_type) VALUES(?,?,?)""", [(ANCHOR.isoformat(), f"Plan-{index:05d}", "planned") for index in range(10001)], ) connection.execute( """INSERT INTO medication_administrations(datum,medication_name,event_type) VALUES(?,?,?)""", (ANCHOR.isoformat(), "Synthetic actual", "administered"), ) connection.commit() finally: connection.close() response = dispatch_api( database, "/api/v1/events", f"from={ANCHOR.isoformat()}&to={ANCHOR.isoformat()}&types=medication_administered", ) assert any(item["label"] == "Synthetic actual" for item in response["events"]) def test_sqlite_query_deadline_is_enforced_as_a_controlled_api_error(tmp_path, monkeypatch): database = api_database(tmp_path) monkeypatch.setattr(read_api, "QUERY_TIMEOUT_SECONDS", -1.0) with pytest.raises(APIError, match="data_unavailable"): dispatch_api(database, "/api/v1/metric-catalog", "") def test_availability_matches_incomplete_symptom_and_ambiguous_or_unreferenced_labs(tmp_path): database = api_database(tmp_path) incomplete_day = (ANCHOR - timedelta(days=9)).isoformat() connection = sqlite3.connect(database) lab_columns = """parameter_name,wert,einheit,reference_min,reference_max, abnahme_datum,wert_original,validierungsstatus,source_type, reference_range_source,verified_against_original,provenance_note""" try: connection.execute("DELETE FROM symptom_log WHERE datum<>?", (incomplete_day,)) connection.execute("DELETE FROM laborwerte") connection.executemany( f"INSERT INTO laborwerte({lab_columns}) VALUES(?,?,?,?,?,?,?,?,?,?,?,?)", [ ("CRP", value, "mg/L", "0", "5", "2026-05-01", value, "validiert", "scanned_original", "scanned_original", 1, "synthetic_duplicate") for value in ("1", "2") ], ) connection.commit() finally: connection.close() symptom_catalog = dispatch_api(database, "/api/v1/metric-catalog", "q=Symptome") symptom_metric = next( item for item in symptom_catalog["groups"]["metrics"] if item["id"] == "symptom.total" ) assert symptom_metric["availability"] == {"status": "supported_no_data", "observations": 0} assert dispatch_api(database, "/api/v1/series", "metric=symptom.total&resolution=day")["points"] == [] crp = dispatch_api(database, "/api/v1/metric-catalog", "q=CRP")["groups"]["metrics"][0] assert crp["availability"] == {"status": "supported_no_data", "observations": 0} assert dispatch_api(database, "/api/v1/labs", "")["observations"] == [] connection = sqlite3.connect(database) try: connection.execute("DELETE FROM laborwerte") connection.execute( f"INSERT INTO laborwerte({lab_columns}) VALUES(?,?,?,?,?,?,?,?,?,?,?,?)", ("CRP", "1", "mg/L", "9" * 400, "<5", "2026-05-01", "1", "validiert", "scanned_original", "scanned_original", 1, "synthetic_bad_reference"), ) connection.commit() finally: connection.close() crp = dispatch_api(database, "/api/v1/metric-catalog", "q=CRP")["groups"]["metrics"][0] assert crp["availability"] == {"status": "supported_no_data", "observations": 0} assert dispatch_api(database, "/api/v1/labs", "")["observations"] == [] def load_server(name: str, dashboard: Path, database: Path, token_file: Path, monkeypatch): monkeypatch.delenv("HEALTH_DASHBOARD_TEST_INSTANCE_ID", raising=False) monkeypatch.delenv("HEALTH_DASHBOARD_ACTION_INBOX", raising=False) monkeypatch.delenv("HEALTH_DASHBOARD_BASIC_PASSWORD_FILE", raising=False) monkeypatch.setenv("HEALTH_DASHBOARD_FILE", str(dashboard)) monkeypatch.setenv("HEALTH_DASHBOARD_V5_FILE", str(dashboard)) monkeypatch.setenv("HEALTH_DASHBOARD_DB", str(database)) monkeypatch.setenv("HEALTH_DASHBOARD_API_TOKEN_FILE", str(token_file)) spec = importlib.util.spec_from_file_location(name, SERVER) assert spec and spec.loader module = importlib.util.module_from_spec(spec) sys.modules[name] = module spec.loader.exec_module(module) return module def test_http_api_requires_private_bearer_token_and_never_caches(tmp_path, monkeypatch): database = api_database(tmp_path) dashboard = tmp_path / "dashboard.html" dashboard.write_text("", encoding="utf-8") token = "synthetic_api_token_0123456789abcdef" token_file = tmp_path / "api-token" token_file.write_text(token, encoding="ascii") token_file.chmod(0o600) module = load_server("health_dashboard_server_sprint6b", dashboard, database, token_file, monkeypatch) assert module.API_DB == database.resolve() assert module.DB != database.resolve() server = module.ThreadingHTTPServer(("127.0.0.1", 0), module.Handler) thread = threading.Thread(target=server.serve_forever, daemon=True) thread.start() base = f"http://127.0.0.1:{server.server_port}" try: bad_host = urllib.request.Request( base + "/api/v1/labs", headers={"Host": "attacker.invalid"}, ) with pytest.raises(urllib.error.HTTPError) as rejected_host: urllib.request.urlopen(bad_host, timeout=5) assert rejected_host.value.code == 421 assert rejected_host.value.headers["Cache-Control"] == "no-store" with pytest.raises(urllib.error.HTTPError) as missing: urllib.request.urlopen(base + "/api/v1/metric-catalog?q=HRV", timeout=5) assert missing.value.code == 401 assert missing.value.headers["Cache-Control"] == "no-store" bad = urllib.request.Request( base + "/api/v1/metric-catalog?q=HRV", headers={"Authorization": "Bearer wrong-token-0123456789abcdef"}, ) with pytest.raises(urllib.error.HTTPError) as denied: urllib.request.urlopen(bad, timeout=5) assert denied.value.code == 401 request = urllib.request.Request( base + "/api/v1/metric-catalog?q=HRV", headers={"Authorization": f"Bearer {token}"}, ) response = urllib.request.urlopen(request, timeout=5) payload = json.loads(response.read()) assert response.status == 200 assert response.headers["Cache-Control"] == "no-store" assert response.headers["Content-Type"] == "application/json; charset=utf-8" assert response.headers["X-Content-Type-Options"] == "nosniff" foreign_origin = urllib.request.Request( base + "/api/v1/search?q=HRV", headers={ "Authorization": f"Bearer {token}", "Origin": "https://attacker.invalid", }, ) with pytest.raises(urllib.error.HTTPError) as cross_origin: urllib.request.urlopen(foreign_origin, timeout=5) assert cross_origin.value.code == 403 assert cross_origin.value.headers["Cache-Control"] == "no-store" cross_site = urllib.request.Request( base + "/api/v1/search?q=HRV", headers={ "Authorization": f"Bearer {token}", "Sec-Fetch-Site": "cross-site", }, ) with pytest.raises(urllib.error.HTTPError) as cross_site_error: urllib.request.urlopen(cross_site, timeout=5) assert cross_site_error.value.code == 403 assert payload["groups"]["metrics"][0]["id"] == "apple.hrv" assert_no_forbidden_fields(payload) document_request = urllib.request.Request( base + "/api/v1/search?q=Labor", headers={"Authorization": f"Bearer {token}"}, ) document_payload = json.load(urllib.request.urlopen(document_request, timeout=5)) opaque_document_id = document_payload["groups"]["documents"][0]["id"] assert opaque_document_id.startswith("api-document-") assert not opaque_document_id.removeprefix("api-document-").isdigit() legacy_url = base + f"/health-doc/{opaque_document_id}" with pytest.raises(urllib.error.HTTPError) as unauthenticated_legacy: urllib.request.urlopen(legacy_url, timeout=5) assert unauthenticated_legacy.value.code == 401 legacy_cross_link = urllib.request.Request( legacy_url, headers={"Authorization": f"Bearer {token}"}, ) with pytest.raises(urllib.error.HTTPError) as no_legacy_cross_link: urllib.request.urlopen(legacy_cross_link, timeout=5) assert no_legacy_cross_link.value.code == 404 token_in_query = urllib.request.Request( base + f"/api/v1/metric-catalog?q=HRV&token={token}", headers={"Authorization": f"Bearer {token}"}, ) with pytest.raises(urllib.error.HTTPError) as query_secret: urllib.request.urlopen(token_in_query, timeout=5) assert query_secret.value.code == 400 post = urllib.request.Request( base + "/api/v1/search?q=HRV", data=b"{}", method="POST", headers={"Authorization": f"Bearer {token}"}, ) with pytest.raises(urllib.error.HTTPError) as read_only: urllib.request.urlopen(post, timeout=5) assert read_only.value.code == 405 assert read_only.value.headers["Cache-Control"] == "no-store" propfind = urllib.request.Request( base + "/api/v1/labs", method="PROPFIND", headers={"Authorization": f"Bearer {token}"}, ) with pytest.raises(urllib.error.HTTPError) as unknown_method: urllib.request.urlopen(propfind, timeout=5) assert unknown_method.value.code == 405 assert unknown_method.value.headers["Cache-Control"] == "no-store" assert unknown_method.value.headers["Content-Type"] == "application/json; charset=utf-8" def raw_method(target: str, host: str) -> bytes: with socket.create_connection(("127.0.0.1", server.server_port), timeout=5) as client: client.sendall( f"PROPFIND {target} HTTP/1.1\r\nHost: {host}\r\nConnection: close\r\n\r\n".encode("ascii") ) chunks = [] while chunk := client.recv(65536): chunks.append(chunk) return b"".join(chunks) absolute = raw_method(f"http://127.0.0.1:{server.server_port}/api/v1/labs", f"127.0.0.1:{server.server_port}") assert b" 405 " in absolute.split(b"\r\n", 1)[0] assert b"Content-Type: application/json; charset=utf-8" in absolute assert b"Cache-Control: no-store" in absolute invalid_absolute_host = raw_method( f"http://127.0.0.1:{server.server_port}/api/v1/labs", "attacker.invalid" ) assert b" 421 " in invalid_absolute_host.split(b"\r\n", 1)[0] assert b"Cache-Control: no-store" in invalid_absolute_host head = urllib.request.Request( base + "/api/v1/labs", method="HEAD", headers={"Authorization": f"Bearer {token}"}, ) assert urllib.request.urlopen(head, timeout=5).read() == b"" original_dispatch = getattr(module, "dispatch_api") setattr(module, "dispatch_api", lambda *_args: {"data": "x" * 512_001}) try: with pytest.raises(urllib.error.HTTPError) as oversized: urllib.request.urlopen(request, timeout=5) assert oversized.value.code == 422 assert json.load(oversized.value) == {"error": {"code": "response_too_large"}} assert oversized.value.headers["Cache-Control"] == "no-store" finally: setattr(module, "dispatch_api", original_dispatch) setattr(module, "dispatch_api", lambda *_args: (_ for _ in ()).throw(RuntimeError("synthetic"))) try: with pytest.raises(urllib.error.HTTPError) as internal: urllib.request.urlopen(request, timeout=5) assert internal.value.code == 500 assert json.load(internal.value) == {"error": {"code": "internal_error"}} assert internal.value.headers["Cache-Control"] == "no-store" finally: setattr(module, "dispatch_api", original_dispatch) setattr(module, "dispatch_api", lambda *_args: {"not_serializable": {1, 2}}) try: with pytest.raises(urllib.error.HTTPError) as serialization: urllib.request.urlopen(request, timeout=5) assert serialization.value.code == 500 assert json.load(serialization.value) == {"error": {"code": "internal_error"}} assert serialization.value.headers["Cache-Control"] == "no-store" finally: setattr(module, "dispatch_api", original_dispatch) token_file.chmod(0o644) with pytest.raises(urllib.error.HTTPError) as insecure: urllib.request.urlopen(request, timeout=5) assert insecure.value.code == 503 token_file.chmod(0o600) token_alias = tmp_path / "api-token-link" token_alias.symlink_to(token_file) setattr(module, "API_TOKEN_FILE", token_alias) with pytest.raises(urllib.error.HTTPError) as symlinked: urllib.request.urlopen(request, timeout=5) assert symlinked.value.code == 503 finally: server.shutdown() server.server_close() thread.join(timeout=5) assert stat.S_IMODE(token_file.stat().st_mode) == 0o600 def test_http_api_is_fail_closed_without_explicit_database_and_token_paths(tmp_path, monkeypatch): dashboard = tmp_path / "dashboard.html" dashboard.write_text("", encoding="utf-8") monkeypatch.delenv("HEALTH_DASHBOARD_TEST_INSTANCE_ID", raising=False) monkeypatch.delenv("HEALTH_DASHBOARD_ACTION_INBOX", raising=False) monkeypatch.delenv("HEALTH_DASHBOARD_BASIC_PASSWORD_FILE", raising=False) monkeypatch.setenv("HEALTH_DASHBOARD_FILE", str(dashboard)) monkeypatch.setenv("HEALTH_DASHBOARD_V5_FILE", str(dashboard)) monkeypatch.delenv("HEALTH_DASHBOARD_DB", raising=False) monkeypatch.delenv("HEALTH_DASHBOARD_API_TOKEN_FILE", raising=False) spec = importlib.util.spec_from_file_location("health_dashboard_server_sprint6b_disabled", SERVER) assert spec and spec.loader module = importlib.util.module_from_spec(spec) sys.modules[spec.name] = module spec.loader.exec_module(module) server = module.ThreadingHTTPServer(("127.0.0.1", 0), module.Handler) thread = threading.Thread(target=server.serve_forever, daemon=True) thread.start() try: with urllib.request.urlopen( f"http://127.0.0.1:{server.server_port}/health-dashboard-v5", timeout=5 ) as response: assert response.status == 200 with pytest.raises(urllib.error.HTTPError) as disabled: urllib.request.urlopen( f"http://127.0.0.1:{server.server_port}/api/v1/labs", timeout=5 ) assert disabled.value.code == 503 assert disabled.value.headers["Cache-Control"] == "no-store" finally: server.shutdown() server.server_close() thread.join(timeout=5)