"""Read-only HTTP service for health, status, evidence, and dashboard views.""" from __future__ import annotations import json import os from collections.abc import Mapping from dataclasses import dataclass from http import HTTPStatus from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from types import MappingProxyType from . import __version__ from .dashboard import build_dashboard_payload, render_dashboard from .errors import InvalidConfiguration _TRUE_VALUES = frozenset({"1", "true", "yes", "on"}) _FALSE_VALUES = frozenset({"0", "false", "no", "off"}) @dataclass(frozen=True, slots=True) class RouteResponse: """A deterministic HTTP response independent of socket handling.""" status: HTTPStatus content_type: str body: bytes def _enabled(environ: Mapping[str, str], name: str, *, default: bool = False) -> bool: raw = environ.get(name) if raw is None: return default normalized = raw.strip().lower() if normalized in _TRUE_VALUES: return True if normalized in _FALSE_VALUES: return False return default def _bounded_runtime_text( environ: Mapping[str, str], name: str, *, default: str, max_length: int, ) -> str: value = environ.get(name, default).strip() if not value or len(value) > max_length: return default return value def service_status(environ: Mapping[str, str] | None = None) -> Mapping[str, object]: """Return a deterministic, secret-free operational status document.""" source = os.environ if environ is None else environ writer_enabled = _enabled(source, "OSM_LEAD_SOURCE_WRITER_ENABLED") legacy_disabled = _enabled(source, "OSM_LEAD_SOURCE_LEGACY_WRITER_DISABLED") scheduler_enabled = _enabled( source, "OSM_LEAD_SOURCE_SCHEDULER_ENABLED", ) and _enabled(source, "OSM_LEAD_SOURCE_SCHEDULER_APPROVED") noco_write_enabled = _enabled(source, "OSM_LEAD_SOURCE_NOCO_WRITE_ENABLED") stale_deletion_enabled = _enabled(source, "OSM_LEAD_SOURCE_STALE_DELETION_ENABLED") mcp_enabled = _enabled(source, "OSM_LEAD_SOURCE_MCP_ENABLED") read_only = _enabled(source, "OSM_LEAD_SOURCE_READ_ONLY", default=True) write_ready = writer_enabled and legacy_disabled return MappingProxyType( { "dashboard_snapshot_configured": bool( source.get("OSM_LEAD_SOURCE_DASHBOARD_SNAPSHOT_PATH", "").strip() ), "environment": _bounded_runtime_text( source, "OSM_LEAD_SOURCE_ENVIRONMENT", default="local", max_length=80, ), "legacy_writer_disabled": legacy_disabled, "mcp_enabled": mcp_enabled, "mode": "controlled-writer" if write_ready else "shadow-read-only", "noco_write_enabled": noco_write_enabled, "phase": "2B", "read_only": read_only, "release_sha": _bounded_runtime_text( source, "OSM_LEAD_SOURCE_RELEASE_SHA", default="unknown", max_length=64, ), "scheduler_enabled": scheduler_enabled, "service": "osm-lead-source-service", "sidecar_configured": bool( source.get("OSM_LEAD_SOURCE_SIDECAR_DATABASE_URL", "").strip() ), "stale_deletion_enabled": stale_deletion_enabled, "version": __version__, "write_runtime_armed": write_ready, "writer_enabled": writer_enabled, } ) def _json_bytes(value: Mapping[str, object]) -> bytes: return (json.dumps(dict(value), sort_keys=True, separators=(",", ":")) + "\n").encode() def route_response( path: str, environ: Mapping[str, str] | None = None, ) -> RouteResponse: """Resolve one fixed read-only route without opening a socket.""" normalized_path = path.split("?", 1)[0] status = service_status(environ) if normalized_path == "/": payload = build_dashboard_payload(status, environ) return RouteResponse( HTTPStatus.OK, "text/html; charset=utf-8", render_dashboard(payload), ) if normalized_path == "/healthz": return RouteResponse( HTTPStatus.OK, "application/json; charset=utf-8", _json_bytes({"ok": True}), ) if normalized_path == "/readyz": return RouteResponse( HTTPStatus.OK, "application/json; charset=utf-8", _json_bytes({"ok": True, "mode": status["mode"]}), ) if normalized_path in {"/status", "/api/status"}: return RouteResponse( HTTPStatus.OK, "application/json; charset=utf-8", _json_bytes(status), ) if normalized_path == "/api/dashboard": payload = build_dashboard_payload(status, environ) return RouteResponse( HTTPStatus.OK, "application/json; charset=utf-8", _json_bytes(payload), ) if normalized_path == "/favicon.ico": return RouteResponse(HTTPStatus.NO_CONTENT, "image/x-icon", b"") return RouteResponse( HTTPStatus.NOT_FOUND, "application/json; charset=utf-8", _json_bytes({"error": "not_found"}), ) class StatusRequestHandler(BaseHTTPRequestHandler): """Serve only fixed GET/HEAD endpoints; there is no mutation surface.""" server_version = "osm-lead-source-service" sys_version = "" def _respond(self, *, include_body: bool) -> None: response = route_response(self.path) self.send_response(response.status) self.send_header("Content-Type", response.content_type) self.send_header("Cache-Control", "no-store") self.send_header("Content-Length", str(len(response.body))) self.send_header("Content-Security-Policy", "default-src 'none'; style-src 'unsafe-inline'") self.send_header("Cross-Origin-Resource-Policy", "same-origin") self.send_header("Permissions-Policy", "camera=(), geolocation=(), microphone=()") self.send_header("Referrer-Policy", "no-referrer") self.send_header("X-Content-Type-Options", "nosniff") self.send_header("X-Frame-Options", "DENY") self.end_headers() if include_body and response.body: self.wfile.write(response.body) def do_GET(self) -> None: # noqa: N802 self._respond(include_body=True) def do_HEAD(self) -> None: # noqa: N802 self._respond(include_body=False) def do_POST(self) -> None: # noqa: N802 self.send_error(HTTPStatus.METHOD_NOT_ALLOWED) def do_PUT(self) -> None: # noqa: N802 self.send_error(HTTPStatus.METHOD_NOT_ALLOWED) def do_PATCH(self) -> None: # noqa: N802 self.send_error(HTTPStatus.METHOD_NOT_ALLOWED) def do_DELETE(self) -> None: # noqa: N802 self.send_error(HTTPStatus.METHOD_NOT_ALLOWED) def log_message(self, format: str, *args: object) -> None: return def serve(*, host: str, port: int) -> None: """Run the HTTP server and optional read-only monthly worker.""" bounded_host = host.strip() if not bounded_host or len(bounded_host) > 255: raise InvalidConfiguration("HTTP host must be nonblank and bounded") if isinstance(port, bool) or not 1 <= port <= 65535: raise InvalidConfiguration("HTTP port must be between 1 and 65535") server = ThreadingHTTPServer((bounded_host, port), StatusRequestHandler) server.daemon_threads = True try: from .worker import start_scheduler_from_env start_scheduler_from_env() server.serve_forever(poll_interval=0.5) finally: server.server_close() __all__ = [ "RouteResponse", "StatusRequestHandler", "route_response", "serve", "service_status", ]