Add reverse proxy: origin mapping, HTML/redirect rewriting, frame-option stripping

src/uwudash_mk3/proxy.py

@@ -0,0 +1,187 @@
+"""Reverse proxy.
+
+Every framed service is reached through ``/proxy/<service-id>/...``.  The proxy
+is the *only* way an upstream service is reachable, so the services themselves
+never need to be exposed.
+
+Mapping rule: the upstream **origin** (``scheme://host[:port]``) maps onto the
+proxy prefix, so ``/proxy/grafana/d/abc`` -> ``http://internal:3000/d/abc``.
+Absolute URLs inside HTML and ``Location`` headers are rewritten back onto the
+proxy prefix so the framed app keeps talking to the dashboard instead of
+escaping to its own (private) origin.
+"""
+
+from __future__ import annotations
+
+import re
+from typing import Dict, Optional
+
+import httpx
+from fastapi import HTTPException, Request
+from fastapi.responses import Response, StreamingResponse
+
+from .config import Config
+from .models import Service
+from .store import ServiceStore
+
+# Headers that must not be forwarded verbatim between hops.
+REQUEST_BLOCKED = {
+    "connection",
+    "keep-alive",
+    "transfer-encoding",
+    "upgrade",
+    "content-length",
+    "host",
+    "accept-encoding",  # let httpx negotiate its own encoding
+}
+RESPONSE_BLOCKED = {
+    "connection",
+    "keep-alive",
+    "transfer-encoding",
+    "upgrade",
+    "content-length",  # recomputed by the response class
+}
+
+MAX_BODY_BYTES = 10 * 1024 * 1024
+
+_FRAME_ANCESTORS = re.compile(r"frame-ancestors\s*=?[^;]*;?")
+
+
+class Proxy:
+    def __init__(self, store: ServiceStore, config: Config) -> None:
+        self.store = store
+        self.config = config
+        self.client = httpx.AsyncClient(
+            timeout=config.proxy_timeout,
+            follow_redirects=False,
+            headers={"user-agent": "uwudash-mk3/0.1"},
+        )
+
+    def close(self) -> None:
+        self.client.close()
+
+    # --------------------------------------------------------------- helpers
+    @staticmethod
+    def _prefix(service: Service) -> str:
+        return f"/proxy/{service.id}"
+
+    @staticmethod
+    def _netloc(service: Service) -> str:
+        return service.origin.split("://", 1)[1]
+
+    def _upstream_url(self, service: Service, path: str, query: str) -> str:
+        url = f"{service.origin}/{path.lstrip('/')}"
+        return f"{url}?{query}" if query else url
+
+    def _request_headers(self, request: Request, service: Service) -> Dict[str, str]:
+        headers = {
+            key: value
+            for key, value in request.headers.items()
+            if key.lower() not in REQUEST_BLOCKED
+        }
+        headers.update(service.headers)
+        return headers
+
+    def _response_headers(self, upstream: httpx.Response, service: Service) -> Dict[str, str]:
+        headers: Dict[str, str] = {}
+        for key, value in upstream.headers.items():
+            lowered = key.lower()
+            if lowered in RESPONSE_BLOCKED:
+                continue
+            if lowered == "location":
+                headers[key] = self._rewrite_location(value, service)
+            elif self.config.strip_frame_options and lowered == "x-frame-options":
+                continue
+            elif self.config.strip_frame_options and lowered == "content-security-policy":
+                headers[key] = _FRAME_ANCESTORS.sub("", value).strip("; ")
+            else:
+                headers[key] = value
+        return headers
+
+    def _rewrite_location(self, location: str, service: Service) -> str:
+        prefix = self._prefix(service)
+        if location.startswith(service.origin):
+            return prefix + location[len(service.origin) :]
+        if location.startswith("//" + self._netloc(service)):
+            return prefix + location[2 + len(self._netloc(service)) :]
+        # Root-relative redirects resolve against the proxy prefix.
+        if location.startswith("/"):
+            return prefix + location
+        return location
+
+    def _rewrite_html(self, text: str, service: Service) -> str:
+        prefix = self._prefix(service)
+        return text.replace(service.origin, prefix).replace(
+            f"//{self._netloc(service)}", prefix
+        )
+
+    # --------------------------------------------------------------- handler
+    async def handle(self, service_id: str, path: str, request: Request) -> Response:
+        service = self.store.get(service_id)
+        if service is None:
+            raise HTTPException(status_code=404, detail=f"unknown service: {service_id}")
+        if not service.enabled:
+            raise HTTPException(status_code=403, detail=f"service {service_id} is disabled")
+
+        upgrade = request.headers.get("upgrade", "")
+        if upgrade.lower() == "websocket":
+            raise HTTPException(
+                status_code=400,
+                detail="websocket proxying is not supported by this dashboard",
+            )
+
+        body: Optional[bytes] = None
+        if request.headers.get("content-length"):
+            body = await request.body()
+            if len(body) > MAX_BODY_BYTES:
+                raise HTTPException(status_code=413, detail="request body too large")
+
+        url = self._upstream_url(service, path, request.url.query)
+        upstream_request = self.client.build_request(
+            request.method, url, headers=self._request_headers(request, service), content=body
+        )
+        try:
+            upstream = await self.client.send(upstream_request, stream=True)
+        except httpx.HTTPError as exc:
+            raise HTTPException(
+                status_code=502, detail=f"upstream {service.origin} unreachable: {exc}"
+            ) from exc
+
+        content_type = upstream.headers.get("content-type", "")
+        headers = self._response_headers(upstream, service)
+
+        if content_type.startswith("text/html"):
+            await upstream.aread()
+            return Response(
+                content=self._rewrite_html(upstream.text, service),
+                status_code=upstream.status_code,
+                headers=headers,
+                media_type="text/html; charset=utf-8",
+            )
+
+        async def stream():
+            try:
+                async for chunk in upstream.aiter_bytes():
+                    yield chunk
+            finally:
+                await upstream.aclose()
+
+        return StreamingResponse(
+            stream(), status_code=upstream.status_code, headers=headers
+        )
+
+    # ----------------------------------------------------------- health check
+    async def check(self, service: Service) -> Dict[str, object]:
+        try:
+            response = await self.client.get(
+                service.url,
+                headers=service.headers,
+                timeout=min(10.0, self.config.proxy_timeout),
+            )
+            return {
+                "reachable": response.status_code < 400,
+                "status": response.status_code,
+                "error": None,
+            }
+        except httpx.HTTPError as exc:
+            return {"reachable": False, "status": None, "error": str(exc)}