From c0742994cd58328571598e076e004982387b3d9b Mon Sep 17 00:00:00 2001 From: retoor Date: Sun, 9 Aug 2026 11:26:17 +0200 Subject: [PATCH] Make the workspace editor reachable through the sub-path proxy code-server runs authenticateOrigin on every websocket and resolves the request host as Forwarded, then X-Forwarded-Host, then Host. The forward core put the public host into additional_headers, but the websockets client already writes its own Host for the real TCP target and Headers appends, so the handshake carried two Host lines; Node keeps the first (the internal gateway:port), the origin check failed, and code-server answered 403. Because the browser socket was accepted before the upstream was dialled, that surfaced as a 101 followed by 1011 and the editor died on "the workbench failed to connect to the server". Dialling first and carrying the public host in the connect URI fixes both planes. The two header builders that had drifted apart are now one core, so a websocket carries the same client and forwarded headers as an HTTP request. Responses stream instead of buffering whole, which is what makes a large tunnel download cost constant memory and lets SSE work; byte accounting moved onto the completion callback. Subprotocols negotiate, the upstream client is reused across requests, and the path and query are forwarded byte-exactly. Co-Authored-By: Claude Opus 5 (1M context) --- CLAUDE.md | 2 +- devplacepy/main.py | 17 +- .../routers/projects/containers/workspace.py | 3 +- devplacepy/routers/proxy.py | 3 +- devplacepy/routers/tunnel.py | 17 +- devplacepy/services/containers/CLAUDE.md | 57 +++- devplacepy/services/containers/forward.py | 252 +++++++++++++----- tests/unit/services/containers.py | 108 +++++++- 8 files changed, 380 insertions(+), 79 deletions(-) diff --git a/CLAUDE.md b/CLAUDE.md index 869093d5..7552d8b2 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -269,7 +269,7 @@ Devii is also reachable over **Telegram** (one supervised long-poller subprocess - **No comments, no docstrings in source.** Code is self-documenting. - **Forbidden variable name patterns:** `_new`, `_old`, `_temp`, `_v2`, `better_`, `my_`, `the_`. -- **Outbound HTTP goes through the stealth client only.** Never instantiate a bare `httpx.AsyncClient(...)`/`httpx.Client(...)` for outbound traffic - build it via `devplacepy.stealth.stealth_async_client(...)`/`stealth_sync_client(...)`. The async factory routes bytes through **`curl_cffi` (curl-impersonate, BoringSSL)** behind an `httpx` transport adapter, so TLS JA3/JA4 and the HTTP/2 fingerprint match Chrome 146 byte-for-byte; falls back to a pure-httpx Chrome-aligned transport if `curl_cffi` is missing. Applies to news, gateway upstream, Devii/DeepSearch/SEO fetch, gitea, push, attachments. SSRF-guarded fetches via `net_guard.guarded_async_client(...)` get the fingerprint automatically. Sole exception: `routers/proxy.py` relays the user's own headers verbatim. **Load-bearing gotcha:** cleartext `http://` is forced to HTTP/1.1 (`curl_transport.http_version_for`) - the Chrome-impersonation profile defaults to HTTP/2, but a plaintext origin has no TLS/ALPN to negotiate h2 and uvicorn (HTTP/1.1 only) returns `400 "Invalid HTTP request received."` once a cleartext HTTP/2 body crosses ~128KB. Every internal AI consumer calls `INTERNAL_GATEWAY_URL` = `http://localhost:/openai/v1/...`, so without this override a large Devii turn (big system prompt + tool list) fails on every request. `https://` keeps the HTTP/2 default. `stealth_sync_client` stays pure-httpx Chrome-aligned (no curl_cffi sync adapter). **Never bolt a hand-written browser-header dict (UA/`sec-ch-ua`/`Sec-Fetch-*`) onto a `stealth_async_client`/`stealth_sync_client` call** - curl_cffi's impersonation already emits an internally-consistent Chrome 146 identity across TLS, UA, and Client Hints; a caller-supplied header set claiming a different Chrome version or platform creates a UA-vs-TLS-vs-Client-Hints mismatch that is a textbook anti-bot tripwire (this was a real bug in Devii's `fetch_url`/`http_request` tool, fixed by deleting its hardcoded Chrome-131 header block). `stealth.detect_consent_gate(html)` recognizes the common JS-redirect cookie-consent wall pattern (DPG Media's `nu.nl` and sibling sites, and generically any page gated behind a `decodeURIComponent(...)` callback referencing a "privacy gate"/`consent.js`) - reuse it wherever a fetched page's real content sits behind that gate rather than special-casing a domain. `stealth_transport`/`stealth_async_client`/`stealth_sync_client` all accept a `proxy=` override, defaulting to `stealth.configured_proxy_url()` (the admin-editable `outbound_proxy_url` site setting, falling back to `DEVPLACE_OUTBOUND_PROXY_URL` when the setting is unset) - this is the only real fix for a destination that blocks by IP/ASN reputation (e.g. reddit.com returns an identical "blocked by network security" page across every TLS/header fingerprint tried, confirmed by testing curl_cffi's full impersonation list - it is blocking the datacenter ASN outright, not fingerprinting the client). +- **Outbound HTTP goes through the stealth client only.** Never instantiate a bare `httpx.AsyncClient(...)`/`httpx.Client(...)` for outbound traffic - build it via `devplacepy.stealth.stealth_async_client(...)`/`stealth_sync_client(...)`. The async factory routes bytes through **`curl_cffi` (curl-impersonate, BoringSSL)** behind an `httpx` transport adapter, so TLS JA3/JA4 and the HTTP/2 fingerprint match Chrome 146 byte-for-byte; falls back to a pure-httpx Chrome-aligned transport if `curl_cffi` is missing. Applies to news, gateway upstream, Devii/DeepSearch/SEO fetch, gitea, push, attachments. SSRF-guarded fetches via `net_guard.guarded_async_client(...)` get the fingerprint automatically. Sole exception: the reverse-proxy forwarding core `services/containers/forward.py` (and the three routers that call it - `routers/proxy.py`, `routers/tunnel.py`, the workspace editor route), which relays the user's own headers verbatim and therefore keeps a plain `httpx.AsyncClient`; bolting the Chrome identity onto it would overwrite the very headers it exists to forward. **Load-bearing gotcha:** cleartext `http://` is forced to HTTP/1.1 (`curl_transport.http_version_for`) - the Chrome-impersonation profile defaults to HTTP/2, but a plaintext origin has no TLS/ALPN to negotiate h2 and uvicorn (HTTP/1.1 only) returns `400 "Invalid HTTP request received."` once a cleartext HTTP/2 body crosses ~128KB. Every internal AI consumer calls `INTERNAL_GATEWAY_URL` = `http://localhost:/openai/v1/...`, so without this override a large Devii turn (big system prompt + tool list) fails on every request. `https://` keeps the HTTP/2 default. `stealth_sync_client` stays pure-httpx Chrome-aligned (no curl_cffi sync adapter). **Never bolt a hand-written browser-header dict (UA/`sec-ch-ua`/`Sec-Fetch-*`) onto a `stealth_async_client`/`stealth_sync_client` call** - curl_cffi's impersonation already emits an internally-consistent Chrome 146 identity across TLS, UA, and Client Hints; a caller-supplied header set claiming a different Chrome version or platform creates a UA-vs-TLS-vs-Client-Hints mismatch that is a textbook anti-bot tripwire (this was a real bug in Devii's `fetch_url`/`http_request` tool, fixed by deleting its hardcoded Chrome-131 header block). `stealth.detect_consent_gate(html)` recognizes the common JS-redirect cookie-consent wall pattern (DPG Media's `nu.nl` and sibling sites, and generically any page gated behind a `decodeURIComponent(...)` callback referencing a "privacy gate"/`consent.js`) - reuse it wherever a fetched page's real content sits behind that gate rather than special-casing a domain. `stealth_transport`/`stealth_async_client`/`stealth_sync_client` all accept a `proxy=` override, defaulting to `stealth.configured_proxy_url()` (the admin-editable `outbound_proxy_url` site setting, falling back to `DEVPLACE_OUTBOUND_PROXY_URL` when the setting is unset) - this is the only real fix for a destination that blocks by IP/ASN reputation (e.g. reddit.com returns an identical "blocked by network security" page across every TLS/header fingerprint tried, confirmed by testing curl_cffi's full impersonation list - it is blocking the datacenter ASN outright, not fingerprinting the client). - **Form validation is Pydantic-native.** Routers declare `data: Annotated[SomeForm, Form()]` (models in `models.py`); the global `RequestValidationError` handler re-renders auth pages with messages (400) or redirects other routes. Read raw `await request.form()` only when also reading an uploaded file alongside a model. - **`RedirectResponse(url=..., status_code=302)`** - always pass `status_code` as an integer literal, never `status.HTTP_302_FOUND`. - **Ownership checks:** compare `resource["user_uid"] == user["uid"]` before edit/delete (use `content.is_owner`). **Deletes additionally allow any admin:** `is_owner(...) or is_admin(user)` on every content delete endpoint, so an administrator may delete any member's content while members are limited to their own. Edits stay owner-only. Deletes are **soft** and cascade under one shared stamp - see `devplacepy/database/CLAUDE.md`. diff --git a/devplacepy/main.py b/devplacepy/main.py index 8f11555d..c25f6a7c 100644 --- a/devplacepy/main.py +++ b/devplacepy/main.py @@ -307,6 +307,9 @@ async def lifespan(app: FastAPI): flush_visits() await service_manager.shutdown_all() await background.stop() + from devplacepy.services.containers import forward + + await forward.close_client() app = FastAPI( @@ -646,14 +649,24 @@ def _terms_exempt(path: str) -> bool: async def terms_acceptance_gate(request: Request, call_next): if request.method not in _TERMS_GATED_METHODS or _terms_exempt(request.url.path): return await call_next(request) - from devplacepy.routers.auth.terms import needs_acceptance + from devplacepy.routers.auth.terms import ( + TERMS_ACCEPTANCE_CODE, + current_terms_version, + needs_acceptance, + ) user = get_current_user(request) if not needs_acceptance(user): return await call_next(request) message = "Accept the updated Terms of Service to continue." if wants_json(request): - return json_error(403, message, redirect="/auth/accept-terms") + return json_error( + 403, + message, + code=TERMS_ACCEPTANCE_CODE, + redirect="/auth/accept-terms", + terms_version=current_terms_version(), + ) return RedirectResponse(url="/auth/accept-terms", status_code=303) diff --git a/devplacepy/routers/projects/containers/workspace.py b/devplacepy/routers/projects/containers/workspace.py index fcbd61a7..88c68ebb 100644 --- a/devplacepy/routers/projects/containers/workspace.py +++ b/devplacepy/routers/projects/containers/workspace.py @@ -281,4 +281,5 @@ async def editor_proxy_ws( await websocket.close(code=1011) return activity.touch(instance["uid"]) - await forward.proxy_ws(websocket, host, port, path) + prefix = f"/projects/{slug}/containers/instances/{uid}/code" + await forward.proxy_ws(websocket, host, port, path, prefix=prefix) diff --git a/devplacepy/routers/proxy.py b/devplacepy/routers/proxy.py index 1cd94b2e..4ee82a75 100644 --- a/devplacepy/routers/proxy.py +++ b/devplacepy/routers/proxy.py @@ -55,7 +55,6 @@ async def proxy_ws(websocket: WebSocket, slug: str, path: str = ""): if instance is None or not host or not port: await websocket.close(code=1011) return - await websocket.accept() audit.record( websocket, "proxy.access", @@ -67,4 +66,4 @@ async def proxy_ws(websocket: WebSocket, slug: str, path: str = ""): summary=f"websocket proxied to instance {instance.get('name')} via ingress {slug}", links=[audit.instance(instance["uid"], instance.get("name"))], ) - await forward.proxy_ws(websocket, host, port, path, accepted=True) + await forward.proxy_ws(websocket, host, port, path, prefix=f"/p/{slug}") diff --git a/devplacepy/routers/tunnel.py b/devplacepy/routers/tunnel.py index 4d9cae3d..1fd8cfd2 100644 --- a/devplacepy/routers/tunnel.py +++ b/devplacepy/routers/tunnel.py @@ -50,11 +50,18 @@ async def handle_http(request: Request, path: str) -> Response: return Response("this workspace is suspended", status_code=403) if not gateway or not port: return Response("the tunnel has no reachable port", status_code=502) - response = await forward.proxy_http(request, gateway, port, path) - size = len(response.body) if hasattr(response, "body") and response.body else 0 - activity.touch(instance["uid"], egress_bytes=size) - tunnels.record_hit(row["uid"], size) - return response + return await forward.proxy_http( + request, + gateway, + port, + path, + on_complete=lambda sent: record_traffic(instance["uid"], row["uid"], sent), + ) + + +def record_traffic(instance_uid: str, tunnel_uid: str, sent: int) -> None: + activity.touch(instance_uid, egress_bytes=sent) + tunnels.record_hit(tunnel_uid, sent) async def handle_ws(websocket: WebSocket, path: str) -> None: diff --git a/devplacepy/services/containers/CLAUDE.md b/devplacepy/services/containers/CLAUDE.md index 46b9df64..8222e4b7 100644 --- a/devplacepy/services/containers/CLAUDE.md +++ b/devplacepy/services/containers/CLAUDE.md @@ -222,8 +222,61 @@ WebSocket, so a tunnel host can never render the application. molohttp's `HostIn **exactly one label**, which is why the pattern is `{port}-{name}`, never `{port}.{name}`. **One forwarding core.** `forward.py` owns header filtering, prefix stripping, `Location` rewriting, -`` injection and the bidirectional WS pump. `/p/{slug}`, the editor route and the tunnel route -all call it. Never write a second proxy. +`` injection, response streaming and the bidirectional WS pump. `/p/{slug}`, the editor route +and the tunnel route all call it. Never write a second proxy, and never let the two planes drift: +**both build their headers from the single `base_headers` core**, so anything a tunnelled app sees +over HTTP (`X-Real-IP`, `X-Forwarded-For`, `Accept-Language`, custom auth headers) it also sees over +a websocket. `forward_headers` adds only what is HTTP-specific (`Host`, `Accept-Encoding: identity` +so the internal hop is never compressed); `ws_headers` drops only `WS_HANDSHAKE_HEADERS`, which the +`websockets` client regenerates itself. Both planes therefore send the same `Host` (the public one), +the same `X-Forwarded-Host`/`-Proto`/`-Prefix`, and the same `X-Script-Name`. + +**The websocket handshake carries exactly ONE `Host`, and it is the public one (load-bearing).** +Never put `Host` in `additional_headers`: the `websockets` client already writes its own for the real +TCP target and `Headers.update` *appends*, so a second one produces a handshake with two `Host` +lines. Node keeps the FIRST (the internal `gateway:port`) and Go's `net/http` rejects the request +outright. The fix is `websockets.connect(uri, host=..., port=...)` - the public host goes in the URI +(so it becomes the single `Host` header) while the connection still dials the container. code-server +resolves the request host as `Forwarded` -> `X-Forwarded-Host` -> `Host` and runs +`authenticateOrigin` on EVERY websocket (`wsRouter.ws(/.*/, ensureOrigin, ...)`), so with only the +losing duplicate it compared the browser's `Origin` (`devplace.net`) against the internal host and +answered **403**, and the editor died on *"The workbench failed to connect to the server (Error: Time +limit reached)"* - a blank page under the whole +`/projects/{slug}/containers/instances/{uid}/code/` sub-path while its HTTP assets loaded fine +(`forward_headers` already sent `X-Forwarded-Host`, which is why only the websocket plane broke). +This hit the tunnel plane identically. Reproduce with the real image, not a mock: `docker run ppy +code-server --auth none` and replay the header set - the origin check runs before authentication. + +**Dial upstream BEFORE accepting the browser socket.** `proxy_ws` connects first and only then calls +`websocket.accept(subprotocol=upstream.subprotocol)`. Accepting first turns every upstream refusal +into a phantom `101` followed by an immediate `1011`, which is exactly what made the 403 above +present as an opaque client-side timeout instead of a handshake failure; it also makes it impossible +to echo the negotiated subprotocol, because the answer is not known yet. Subprotocols travel through +`subprotocols=` (never as a forwarded header) and come back on the `accept`. + +**Responses stream; only the ``-injected HTML is buffered.** `proxy_http` sends with +`stream=True` and returns a `StreamingResponse` over `aiter_raw()`, passing `content-length` and +`content-encoding` through untouched (`RESPONSE_HOP_HEADERS` strips hop-by-hop only), so a large +download through a tunnel costs a constant few hundred KiB of worker memory instead of its full size, +and SSE works. The HTML branch must buffer because injection needs the whole body, and it therefore +drops `content-length`/`content-encoding` and lets Starlette recompute. Byte accounting rides the +`on_complete(sent)` callback fired when the stream ends - `routers/tunnel.py` uses it for +`activity.touch` + `tunnels.record_hit`, which is ONE `touch` per request (`activity` accumulates +while throttled, so a second call would double the request counter). **Request bodies stay +buffered on purpose**: they are bounded by nginx `client_max_body_size`, and streaming them would +force chunked encoding onto arbitrary upstream apps. + +**One keep-alive client, closed on shutdown.** `forward.client()` is a lazily-created module-level +`httpx.AsyncClient` with `httpx.Limits` (the `ChromeStealthClient` pattern), closed by +`forward.close_client()` in the `main.py` lifespan. A client per request meant a fresh pool and TCP +handshake for every one of the workbench's hundreds of assets. + +**Path and query are forwarded byte-exactly.** The route-matched path is `quote()`d before it is +interpolated (a filename containing `#` or a space otherwise truncates or corrupts the upstream URL), +and the query comes from `raw_query(connection)` = the raw ASGI `scope["query_string"]`, NEVER +`request.url.query`/`request.query_params`. Starlette rebuilds `.url` by string-joining path and +query, so a literal `#` in the path makes the reconstructed URL treat the query as a fragment and +`.url.query` silently returns empty - which would drop code-server's `?reconnectionToken=`. **Editor persistence.** code-server's user-data and extensions live in `config.WORKSPACE_STATE_DIR/`, bind-mounted at `WORKSPACE_STATE_MOUNT`, so extensions diff --git a/devplacepy/services/containers/forward.py b/devplacepy/services/containers/forward.py index 2c85f935..695e838e 100644 --- a/devplacepy/services/containers/forward.py +++ b/devplacepy/services/containers/forward.py @@ -4,11 +4,13 @@ from __future__ import annotations import asyncio import logging +from collections.abc import AsyncIterator, Callable +from urllib.parse import quote import httpx import websockets from fastapi import Request, WebSocket -from starlette.responses import Response +from starlette.responses import Response, StreamingResponse logger = logging.getLogger(__name__) @@ -26,40 +28,110 @@ HOP_HEADERS = { "content-encoding", } +RESPONSE_HOP_HEADERS = { + "connection", + "keep-alive", + "proxy-authenticate", + "proxy-authorization", + "te", + "trailers", + "transfer-encoding", + "upgrade", +} + +WS_HANDSHAKE_HEADERS = { + "sec-websocket-key", + "sec-websocket-version", + "sec-websocket-extensions", + "sec-websocket-protocol", +} + METHODS = ["GET", "POST", "PUT", "DELETE", "PATCH", "OPTIONS", "HEAD"] +SCHEME_PROTO = {"http": "http", "https": "https", "ws": "http", "wss": "https"} + DEFAULT_TIMEOUT = 300.0 +OPEN_TIMEOUT = 10.0 +MAX_CONNECTIONS = 200 +MAX_KEEPALIVE_CONNECTIONS = 50 + +_client: httpx.AsyncClient | None = None -def forward_headers(request: Request, prefix: str = "") -> dict: - headers = {k: v for k, v in request.headers.items() if k.lower() not in HOP_HEADERS} +def client() -> httpx.AsyncClient: + global _client + if _client is None or _client.is_closed: + _client = httpx.AsyncClient( + follow_redirects=False, + timeout=DEFAULT_TIMEOUT, + limits=httpx.Limits( + max_connections=MAX_CONNECTIONS, + max_keepalive_connections=MAX_KEEPALIVE_CONNECTIONS, + ), + ) + return _client + + +async def close_client() -> None: + global _client + if _client is not None and not _client.is_closed: + await _client.aclose() + _client = None + + +def public_host(connection) -> str: + return connection.headers.get("host", "") or connection.url.hostname or "" + + +def raw_query(connection) -> str: + return connection.scope.get("query_string", b"").decode("latin-1") + + +def base_headers(connection, prefix: str, drop: set[str]) -> dict: + headers = { + name: value + for name, value in connection.headers.items() + if name.lower() not in HOP_HEADERS and name.lower() not in drop + } if prefix: headers["X-Forwarded-Prefix"] = prefix headers["X-Script-Name"] = prefix - origin_host = request.headers.get("host", request.url.hostname or "") - headers["X-Forwarded-Host"] = origin_host + origin_host = public_host(connection) + if origin_host: + headers["X-Forwarded-Host"] = origin_host + headers["X-Forwarded-Proto"] = connection.headers.get( + "x-forwarded-proto", SCHEME_PROTO.get(connection.url.scheme, "http") + ) + return headers + + +def forward_headers(request: Request, prefix: str = "") -> dict: + headers = base_headers(request, prefix, set()) + origin_host = public_host(request) if origin_host: headers["Host"] = origin_host - headers["X-Forwarded-Proto"] = request.headers.get( - "x-forwarded-proto", request.url.scheme - ) headers["Accept-Encoding"] = "identity" return headers -WS_FORWARD_HEADERS = ("cookie", "authorization", "user-agent", "origin") +def ws_headers(websocket: WebSocket, prefix: str = "") -> dict: + return base_headers(websocket, prefix, WS_HANDSHAKE_HEADERS) -def ws_headers(websocket) -> dict: - headers = { - name: websocket.headers[name] - for name in WS_FORWARD_HEADERS - if name in websocket.headers - } - origin_host = websocket.headers.get("host", "") - if origin_host: - headers["Host"] = origin_host - return headers +def ws_subprotocols(websocket: WebSocket) -> list[str] | None: + offered = [ + value.strip() + for value in websocket.headers.get("sec-websocket-protocol", "").split(",") + if value.strip() + ] + return offered or None + + +def upstream_url(scheme: str, authority: str, path: str, query: str) -> str: + url = f"{scheme}://{authority}/{quote(path)}" + if query: + url = f"{url}?{query}" + return url def inject_base(body: bytes, prefix: str) -> bytes: @@ -86,6 +158,37 @@ def rewrite_location(value: str, prefix: str) -> str: return value +def response_headers(upstream: httpx.Response, prefix: str) -> dict: + headers = { + name: value + for name, value in upstream.headers.items() + if name.lower() not in RESPONSE_HOP_HEADERS and name.lower() != "set-cookie" + } + if "location" in headers: + headers["location"] = rewrite_location(headers["location"], prefix) + return headers + + +def apply_cookies(response: Response, upstream: httpx.Response) -> Response: + for cookie in upstream.headers.get_list("set-cookie"): + response.headers.append("set-cookie", cookie) + return response + + +async def stream_body( + upstream: httpx.Response, on_complete: Callable[[int], None] | None +) -> AsyncIterator[bytes]: + sent = 0 + try: + async for chunk in upstream.aiter_raw(): + sent += len(chunk) + yield chunk + finally: + await upstream.aclose() + if on_complete is not None: + on_complete(sent) + + async def proxy_http( request: Request, host: str, @@ -95,67 +198,86 @@ async def proxy_http( prefix: str = "", timeout: float = DEFAULT_TIMEOUT, rewrite_html: bool = True, + on_complete: Callable[[int], None] | None = None, ) -> Response: - url = f"http://{host}:{port}/{path}" - headers = forward_headers(request, prefix) - body = await request.body() + upstream_request = client().build_request( + request.method, + upstream_url("http", f"{host}:{port}", path, raw_query(request)), + headers=forward_headers(request, prefix), + content=await request.body(), + timeout=timeout, + ) try: - async with httpx.AsyncClient( - timeout=timeout, follow_redirects=False - ) as client: - upstream = await client.request( - request.method, - url, - params=request.query_params, - headers=headers, - content=body, - ) + upstream = await client().send(upstream_request, stream=True) except httpx.HTTPError as error: return Response(f"upstream error: {error}", status_code=502) - out_headers = { - k: v - for k, v in upstream.headers.items() - if k.lower() not in HOP_HEADERS and k.lower() != "set-cookie" - } - if "location" in out_headers: - out_headers["location"] = rewrite_location(out_headers["location"], prefix) content_type = upstream.headers.get("content-type", "") - content = upstream.content + headers = response_headers(upstream, prefix) if prefix and rewrite_html and "text/html" in content_type.lower(): - content = inject_base(content, prefix) - response = Response( - content=content, - status_code=upstream.status_code, - headers=out_headers, - media_type=content_type or None, + body = await upstream.aread() + await upstream.aclose() + content = inject_base(body, prefix) + headers.pop("content-length", None) + headers.pop("content-encoding", None) + if on_complete is not None: + on_complete(len(content)) + return apply_cookies( + Response( + content=content, + status_code=upstream.status_code, + headers=headers, + media_type=content_type or None, + ), + upstream, + ) + return apply_cookies( + StreamingResponse( + stream_body(upstream, on_complete), + status_code=upstream.status_code, + headers=headers, + media_type=content_type or None, + ), + upstream, ) - for cookie in upstream.headers.get_list("set-cookie"): - response.headers.append("set-cookie", cookie) - return response + + +async def close_quietly(websocket: WebSocket) -> None: + try: + await websocket.close(code=1011) + except Exception: + pass async def proxy_ws( - websocket: WebSocket, host: str, port: int, path: str, *, accepted: bool = False + websocket: WebSocket, + host: str, + port: int, + path: str, + *, + prefix: str = "", + timeout: float = OPEN_TIMEOUT, ) -> None: - upstream_url = f"ws://{host}:{port}/{path}" - if websocket.url.query: - upstream_url += f"?{websocket.url.query}" - if not accepted: - await websocket.accept() + authority = public_host(websocket) or f"{host}:{port}" + target = upstream_url("ws", authority, path, raw_query(websocket)) try: - async with websockets.connect( - upstream_url, - open_timeout=10, + upstream = await websockets.connect( + target, + host=host, + port=port, + open_timeout=timeout, max_size=None, - additional_headers=ws_headers(websocket), - ) as upstream: - await pump(websocket, upstream) + additional_headers=ws_headers(websocket, prefix), + subprotocols=ws_subprotocols(websocket), + ) except Exception as error: logger.debug("ws proxy to %s:%s failed: %s", host, port, error) - try: - await websocket.close(code=1011) - except Exception: - pass + await close_quietly(websocket) + return + try: + await websocket.accept(subprotocol=upstream.subprotocol) + await pump(websocket, upstream) + finally: + await upstream.close() async def pump(client_ws: WebSocket, upstream) -> None: diff --git a/tests/unit/services/containers.py b/tests/unit/services/containers.py index f19ebf71..45aee097 100644 --- a/tests/unit/services/containers.py +++ b/tests/unit/services/containers.py @@ -7,7 +7,9 @@ from tests.conftest import BASE_URL, run_async from devplacepy.database import db, get_table, init_db, refresh_snapshot from devplacepy.utils import generate_uid from devplacepy import config, project_files -from devplacepy.services.containers import api, store, runtime +from starlette.requests import Request +from starlette.websockets import WebSocket +from devplacepy.services.containers import api, forward, store, runtime from devplacepy.services.containers.backend.base import Mount, PortMapping, RunSpec from devplacepy.services.containers.backend.docker_cli import build_run_argv, parse_size from devplacepy.services.containers.backend.fake import FakeBackend @@ -401,3 +403,107 @@ def test_bidirectional_sync_newer_wins(env, tmp_path): assert (workspace / "shared.txt").read_text() == "from project\n" imported = project_files.read_file(pid, "fromfs.txt") assert imported["content"] == "from fs\n" + + + +def _proxy_scope(kind: str, headers: dict, scheme: str, query: str = "") -> dict: + return { + "type": kind, + "scheme": scheme, + "server": ("devplace.net", 443), + "path": "/projects/demo/containers/instances/abc/code/stable-1", + "query_string": query.encode(), + "headers": [ + (name.encode(), value.encode()) for name, value in headers.items() + ], + } + + +def _proxy_websocket(headers: dict, scheme: str = "wss", query: str = "") -> WebSocket: + return WebSocket(_proxy_scope("websocket", headers, scheme, query), None, None) + + +def _proxy_request(headers: dict, scheme: str = "https", query: str = "") -> Request: + return Request(_proxy_scope("http", headers, scheme, query)) + + +def test_ws_headers_send_the_public_host_only_as_x_forwarded_host(): + headers = forward.ws_headers( + _proxy_websocket( + { + "host": "devplace.net", + "origin": "https://devplace.net", + "cookie": "session=abc", + } + ) + ) + assert headers["X-Forwarded-Host"] == "devplace.net" + assert headers["origin"] == "https://devplace.net" + assert headers["cookie"] == "session=abc" + assert not [name for name in headers if name.lower() == "host"] + + +def test_ws_headers_drop_only_the_handshake_headers_the_client_regenerates(): + headers = forward.ws_headers( + _proxy_websocket( + { + "host": "devplace.net", + "x-real-ip": "203.0.113.7", + "accept-language": "nl-NL", + "sec-websocket-key": "should-not-survive", + "sec-websocket-version": "13", + "sec-websocket-extensions": "permessage-deflate", + "sec-websocket-protocol": "v1", + } + ) + ) + assert headers["x-real-ip"] == "203.0.113.7" + assert headers["accept-language"] == "nl-NL" + for name in forward.WS_HANDSHAKE_HEADERS: + assert name not in headers + + +def test_ws_headers_derive_the_forwarded_proto_from_the_socket_scheme(): + assert forward.ws_headers(_proxy_websocket({"host": "d.net"}))[ + "X-Forwarded-Proto" + ] == "https" + assert forward.ws_headers(_proxy_websocket({"host": "d.net"}, scheme="ws"))[ + "X-Forwarded-Proto" + ] == "http" + assert forward.ws_headers( + _proxy_websocket({"host": "d.net", "x-forwarded-proto": "https"}, scheme="ws") + )["X-Forwarded-Proto"] == "https" + + +def test_forward_headers_send_the_public_host_and_the_prefix(): + headers = forward.forward_headers( + _proxy_request({"host": "devplace.net", "x-real-ip": "203.0.113.7"}), + prefix="/p/demo", + ) + assert headers["Host"] == "devplace.net" + assert headers["X-Forwarded-Host"] == "devplace.net" + assert headers["X-Forwarded-Prefix"] == "/p/demo" + assert headers["X-Script-Name"] == "/p/demo" + assert headers["x-real-ip"] == "203.0.113.7" + + +def test_ws_subprotocols_are_parsed_for_negotiation(): + assert forward.ws_subprotocols(_proxy_websocket({"host": "d.net"})) is None + assert forward.ws_subprotocols( + _proxy_websocket({"host": "d.net", "sec-websocket-protocol": "v2, v1"}) + ) == ["v2", "v1"] + + +def test_upstream_url_encodes_the_path_and_keeps_the_query_verbatim(): + assert ( + forward.upstream_url("http", "h:1", "dir/a b#c", "keep=1") + == "http://h:1/dir/a%20b%23c?keep=1" + ) + assert forward.upstream_url("ws", "h:1", "", "") == "ws://h:1/" + + +def test_raw_query_survives_a_hash_in_the_path(): + request = _proxy_request({"host": "d.net"}, query="keep=1") + request.scope["path"] = "/weird/a b#c" + assert forward.raw_query(request) == "keep=1" + assert request.url.query == ""