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 == ""