diff --git a/.gitignore b/.gitignore index 814946b..2c4e934 100644 --- a/.gitignore +++ b/.gitignore @@ -169,3 +169,4 @@ dist-mojo/ # Personal research notes (not shipped with the project) .tmp/ +uv.lock diff --git a/adr/003-mojo-default.md b/adr/003-mojo-default.md index 75aacad..6e02f5d 100644 --- a/adr/003-mojo-default.md +++ b/adr/003-mojo-default.md @@ -161,7 +161,7 @@ Each follow-up is a one-page issue. None blocks this ADR. 3. ~~**Rate limiting.**~~ **Shipped (per-process):** `RateLimit[Inner]` token bucket tuned by `RATE_LIMIT_RPS` / `RATE_LIMIT_BURST` (0 = disabled). Approximately global across workers via a shared atomic cell. Distributed rate limiting still needs Redis and stays open. -4. **TLS termination.** **Shipped via [PR #36](https://github.com/echohello-dev/opengateway/pull/36) on the flare#7 commit (`edf3f22`).** Bumped `pixi.toml` from `v0.9.0` to `edf3f22`; rebuilt `libflare_tls.so` against the new flare source (one-line `-Wl,-u,_flare_ssl_read_ex` linker flag because clang treats unreferenced extern C as dead-code; gcc includes them by default). `main.mojo` binds via `HttpServer.bind_tls(addr, cert, key, alpn=["http/1.1"])` + `serve_tls(stack)` when `TLS_CERT_FILE`/`TLS_KEY_FILE` are set. e2e: `openssl s_client` handshake completes, ALPN negotiates `http/1.1`, non-streaming + streaming `/v1/chat/completions` round-trip works over HTTPS, structured logger emits JSON per line. The flare reactor-side TLS reactor is real and working. Self-signed cert hostname gotcha documented in README ("Self-signed cert gotcha") — curl needs `--resolve localhost:port:127.0.0.1` when the cert is `CN=localhost`. Production terminates TLS at the edge LB; the in-binary path is for the edge-binary deployment story (Fly.io machines, bare metal, Lambda-ish). The stdlib-ssl proxy (`opengateway/mojo_bridge/tls_proxy.py` + `tests/test_tls_proxy.py` + `opengateway/mojo/repro_tls_thread.mojo`) stays in-tree as a vendored fallback for the v0.9.0 pin and as a reproduction of the early daemon-thread diagnostic; the wrapper usage in `main.mojo` was reverted. The flare issue I filed earlier (#8) was closed because the only "stall" I observed was curl's cert hostname verification failing silently under `-s`, not a reactor bug. +4. **TLS termination.** **Shipped via [PR #36](https://github.com/echohello-dev/opengateway/pull/36) on the flare#7 commit (`edf3f22`).** Bumped `pixi.toml` from `v0.9.0` to `edf3f22`; rebuilt `libflare_tls.so` against the new flare source (one-line `-Wl,-u,_flare_ssl_read_ex` linker flag because clang treats unreferenced extern C as dead-code; gcc includes them by default). `main.mojo` binds via `HttpServer.bind_tls(addr, cert, key, alpn=["http/1.1"])` + `serve_tls(stack)` when `TLS_CERT_FILE`/`TLS_KEY_FILE` are set. e2e: `openssl s_client` handshake completes, ALPN negotiates `http/1.1`, non-streaming + streaming `/v1/chat/completions` round-trip works over HTTPS, structured logger emits JSON per line. The flare reactor-side TLS reactor is real and working. Self-signed cert hostname gotcha documented in README ("Self-signed cert gotcha") — curl needs `--resolve localhost:port:127.0.0.1` when the cert is `CN=localhost`. Production terminates TLS at the edge LB; the in-binary path is for the edge-binary deployment story (Fly.io machines, bare metal, Lambda-ish). The flare issue I filed earlier (#8) was closed because the only "stall" I observed was curl's cert hostname verification failing silently under `-s`, not a reactor bug. 5. ~~**DB-backed virtual keys.**~~ **Shipped via the Python bridge** (no Mojo Postgres driver required): `opengateway/mojo_bridge/db.py` defines a `VirtualKeyStore` protocol with an asyncpg-backed implementation, and `authenticate_authorization` consults it after the root-key short-circuit with a 60 s in-process TTL cache. One connection per lookup — the bridge's one-shot `asyncio.run` model means a pool cannot outlive its event loop; at gateway latencies the connect cost is noise. `database_url` is now unset-by-default; when unset the store seam returns `None` and auth is root-key-only. Schema is created by `PostgresVirtualKeyStore.ensure_schema` (`virtual_keys` table with models / budget / tpm / rpm columns). **Spend recording shipped with it:** the bridge increments `budget_used` by `usage.total_tokens` after each unary completion, and the streaming pump parses the terminal usage chunk (the bridge injects `stream_options: {"include_usage": true}` so upstreams always send one). Budgets are token-denominated; a per-model dollar pricing table is a deliberate product decision left open. diff --git a/opengateway/mojo/repro_tls_thread.mojo b/opengateway/mojo/repro_tls_thread.mojo deleted file mode 100644 index 05c009e..0000000 --- a/opengateway/mojo/repro_tls_thread.mojo +++ /dev/null @@ -1,54 +0,0 @@ -"""Minimal reproducer for the Mojo-runtime × stdlib-ssl daemon-thread hang. - -Build: - pixi run -e mojo mojo build -I . opengateway/mojo/repro_tls_thread.mojo \\ - -o /tmp/repro-tls - -Run, comparing the two scenarios by toggling ``WITH_GILRELEASED`` env: - - MOJO_PYTHON_LIBRARY=$PWD/.pixi/envs/mojo/lib/libpython3.13.dylib \\ - PYTHONPATH=. \\ - /tmp/repro-tls # Mojo holds the GIL between calls - MOJO_PYTHON_LIBRARY=$PWD/.pixi/envs/mojo/lib/libpython3.13.dylib \\ - PYTHONPATH=. WITH_GILRELEASED=1 \\ - /tmp/repro-tls # main thread releases GIL after init - -The Python side (opengateway.mojo_bridge._repro_tls.repro) opens a TLS -listener on a random loopback port, spawns a daemon thread that does -accept + handshake + recv, and connects a TLS client to itself. The -output is a timeline of events so you can see where the hang is. - -A healthy run produces 10+ events ending with ``done``. A hung run -prints the first 5–6 events and then ``done`` from the timeout. -""" - -from std.python import Python, PythonObject -from std.python._cpython import GILReleased -from std.os import getenv - - -def main() raises: - var py = Python() - var mod = Python.import_module("opengateway.mojo_bridge._repro_tls") - var with_gil = getenv("WITH_GILRELEASED").byte_length() > 0 - print( - "repro-tls: " - + ("WITH GILReleased" if with_gil else "WITHOUT GILReleased"), - flush=True, - ) - - if with_gil: - with GILReleased(py): - _run_repro(mod) - else: - _run_repro(mod) - - -def _run_repro(mod: PythonObject) raises: - var result = mod.repro(5.0) - var events = result["events"] - print("events:", flush=True) - for ev in events: - var name = String(py=ev[0]) - var ts = Float64(py=ev[1]) - print(" ", ts, name, flush=True) \ No newline at end of file diff --git a/opengateway/mojo_bridge/__init__.py b/opengateway/mojo_bridge/__init__.py index 7a8e452..e2325c1 100644 --- a/opengateway/mojo_bridge/__init__.py +++ b/opengateway/mojo_bridge/__init__.py @@ -27,7 +27,6 @@ from opengateway.mojo_bridge.auth import AuthResult, authenticate_authorization from opengateway.mojo_bridge.chat import chat_completion, health_check from opengateway.mojo_bridge.stream import start_streaming_chat -from opengateway.mojo_bridge.tls_proxy import start_tls_proxy __all__ = [ "handle_chat", @@ -35,7 +34,6 @@ "health_check", "authenticate_authorization", "AuthResult", - "start_tls_proxy", ] logger = logging.getLogger("opengateway.mojo_bridge") diff --git a/opengateway/mojo_bridge/_repro_tls.py b/opengateway/mojo_bridge/_repro_tls.py deleted file mode 100644 index 2e10e1b..0000000 --- a/opengateway/mojo_bridge/_repro_tls.py +++ /dev/null @@ -1,141 +0,0 @@ -"""Minimal reproducer for the TLS proxy daemon-thread hang. - -Spawns a daemon thread that does TLS accept + recv on a loopback -socket, prints whether recv ever returns data, then exits. Used from -``opengateway/mojo/repro_tls_thread.mojo`` to compare with vs without -``GILReleased`` wrapping the reactor. -""" - -from __future__ import annotations - -import contextlib -import socket -import ssl -import threading -import time - - -def repro(timeout_s: float = 5.0) -> dict: - """Open a TLS listener, accept one connection, report what recv sees. - - Returns a dict with the timeline of events so the caller can see - exactly where the hang occurs. - """ - events: list[tuple[str, float]] = [] - - def record(name: str) -> None: - events.append((name, time.time())) - - record("start") - - context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER) - # Path used in tests/test_tls_proxy.py; the test that exercises this - # helper guarantees the cert exists before calling. - import os - - if not os.path.exists("/tmp/og-tls-certs/server.pem"): - os.makedirs("/tmp/og-tls-certs", exist_ok=True) - import subprocess - - subprocess.run( - [ - "openssl", - "req", - "-x509", - "-newkey", - "rsa:2048", - "-keyout", - "/tmp/og-tls-certs/server.key", - "-out", - "/tmp/og-tls-certs/server.pem", - "-days", - "1", - "-nodes", - "-subj", - "/CN=localhost", - ], - check=True, - capture_output=True, - ) - - context.load_cert_chain( - certfile="/tmp/og-tls-certs/server.pem", - keyfile="/tmp/og-tls-certs/server.key", - ) - - listener = socket.create_server(("127.0.0.1", 0)) - listener.listen() - port = listener.getsockname()[1] - record(f"bound 127.0.0.1:{port}") - - result: dict = {"port": port, "events": []} - - def accept_one() -> None: - record("accept: waiting for client") - try: - raw, _ = listener.accept() - record("accept: client connected") - tls = context.wrap_socket(raw, server_side=True) - record("accept: handshake complete") - tls.settimeout(timeout_s) - tls.setblocking(False) - record("accept: non-blocking set") - deadline = time.time() + timeout_s - try: - while time.time() < deadline: - record("accept: recv called") - try: - data = tls.recv(8192) - except ssl.SSLWantReadError: - record("accept: SSLWantReadError") - time.sleep(0.02) - continue - except ssl.SSLWantWriteError: - record("accept: SSLWantWriteError") - time.sleep(0.02) - continue - except BlockingIOError: - record("accept: BlockingIOError") - time.sleep(0.02) - continue - if not data: - record("accept: EOF") - break - record(f"accept: recv {len(data)} bytes") - finally: - with contextlib.suppress(Exception): - tls.close() - except Exception as exc: - record(f"accept: exception {type(exc).__name__}: {exc}") - finally: - with contextlib.suppress(Exception): - listener.close() - record("accept: closed") - - t = threading.Thread(target=accept_one, daemon=True) - t.start() - - record("client: connecting") - ctx_client = ssl._create_unverified_context() - with socket.create_connection(("127.0.0.1", port)) as raw: - record("client: tcp connected") - with ctx_client.wrap_socket(raw, server_hostname="localhost") as tls: - record("client: handshake complete") - tls.sendall(b"GET / HTTP/1.1\r\nHost: x\r\n\r\n") - record("client: sent GET") - # Hard cap on recv so the client cannot deadlock the Mojo - # call if the server thread is starved. - tls.settimeout(3.0) - try: - data = tls.recv(4096) - record(f"client: recv {len(data)} bytes") - except (TimeoutError, ssl.SSLWantReadError) as exc: - record(f"client: recv timeout: {type(exc).__name__}") - except Exception as exc: - record(f"client: recv exception {type(exc).__name__}: {exc}") - - t.join(timeout=2) - record("done") - - result["events"] = events - return result diff --git a/opengateway/mojo_bridge/tls_proxy.py b/opengateway/mojo_bridge/tls_proxy.py deleted file mode 100644 index 21f736f..0000000 --- a/opengateway/mojo_bridge/tls_proxy.py +++ /dev/null @@ -1,139 +0,0 @@ -"""In-binary TLS termination for the Mojo server. - -flare v0.9's server-side TLS surface is handshake-only (``TlsAcceptor`` -+ ``handshake_fd``); the data path (``SSL_read`` / ``SSL_write`` in the -reactor) is a deferred upstream follow-up, so the Mojo binary cannot -terminate TLS natively today. This module is the in-binary stand-in: -a stdlib ``ssl`` listener that accepts TLS on the public port and -proxies bytes to the cleartext flare listener on loopback. - -Design: thread-per-connection blocking pump, two threads per -connection (one per direction). The reactor-native replacement (flare -``STATE_TLS_HANDSHAKE`` + server-side read/write FFI) stays an upstream -follow-up; when it lands this module is deleted and ``main.mojo`` -binds TLS directly. - -Concurrency note: the pump threads need the GIL to call into the -``ssl`` module. ``main.mojo`` wraps the reactor in ``GILReleased`` so -these threads self-schedule while the Mojo reactor is blocked. -""" - -from __future__ import annotations - -import contextlib -import logging -import socket -import ssl -import threading - -logger = logging.getLogger("opengateway.mojo_bridge.tls_proxy") - -_BUF = 64 * 1024 - - -def start_tls_proxy( - listen_host: str, - listen_port: int, - target_host: str, - target_port: int, - cert_file: str, - key_file: str, -) -> threading.Thread: - """Start the TLS-terminating proxy on a daemon thread. - - Returns the acceptor thread. The thread (and every connection - thread it spawns) is a daemon, so process exit is owned by the - Mojo reactor, not by this proxy. - """ - context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER) - context.minimum_version = ssl.TLSVersion.TLSv1_2 - context.load_cert_chain(certfile=cert_file, keyfile=key_file) - - listener = socket.create_server((listen_host, listen_port), reuse_port=False) - listener.listen() - - thread = threading.Thread( - target=_accept_loop, - args=(listener, context, target_host, target_port), - name="opengateway-tls-proxy", - daemon=True, - ) - thread.start() - return thread - - -def _accept_loop( - listener: socket.socket, - context: ssl.SSLContext, - target_host: str, - target_port: int, -) -> None: - while True: - try: - client, _ = listener.accept() - except OSError: - logger.exception("tls proxy: accept failed") - continue - threading.Thread( - target=_handle, - args=(client, context, target_host, target_port), - name="opengateway-tls-conn", - daemon=True, - ).start() - - -def _handle( - client: socket.socket, - context: ssl.SSLContext, - target_host: str, - target_port: int, -) -> None: - """Hand a single TLS connection off to two blocking pump threads.""" - try: - tls_client = context.wrap_socket(client, server_side=True) - except (ssl.SSLError, OSError): - client.close() - return - try: - upstream = socket.create_connection((target_host, target_port)) - except OSError: - logger.exception("tls proxy: failed to reach cleartext target") - tls_client.close() - return - - # Client -> upstream runs on this thread; upstream -> client runs - # on the forwarder thread. A slow client cannot stall the upstream - # drain, which matters for SSE where the server writes far more - # than the client sends. - forwarder = threading.Thread( - target=_pump, - args=(tls_client, upstream), - name="opengateway-tls-fwd", - daemon=True, - ) - forwarder.start() - _pump(upstream, tls_client) - forwarder.join(timeout=5) - tls_client.close() - upstream.close() - - -def _pump(source: socket.socket, sink: socket.socket) -> None: - """Copy bytes from ``source`` to ``sink`` until EOF or error. - - On source EOF, half-closes the sink's write side so the peer sees - a clean FIN (important for SSE: the upstream's end-of-stream must - reach the client without tearing down the reverse direction - first). - """ - try: - while True: - data = source.recv(_BUF) - if not data: - break - sink.sendall(data) - except OSError: - pass - finally: - with contextlib.suppress(OSError): - sink.shutdown(socket.SHUT_WR) diff --git a/tests/test_tls_proxy.py b/tests/test_tls_proxy.py deleted file mode 100644 index a2e1fae..0000000 --- a/tests/test_tls_proxy.py +++ /dev/null @@ -1,152 +0,0 @@ -"""Tests for the in-binary TLS proxy (opengateway.mojo_bridge.tls_proxy). - -Integration-style: generates a self-signed cert with the openssl CLI, -serves a trivial cleartext HTTP responder as the upstream, and drives -real TLS clients through the proxy. Skips when openssl is unavailable. -""" - -from __future__ import annotations - -import shutil -import socket -import ssl -import subprocess -import threading -import time -from typing import Any - -import pytest - -from opengateway.mojo_bridge.tls_proxy import start_tls_proxy - -pytestmark = pytest.mark.skipif(shutil.which("openssl") is None, reason="openssl CLI not available") - - -@pytest.fixture(scope="module") -def self_signed_cert(tmp_path_factory: Any) -> Any: - import tempfile - - tmp = tempfile.mkdtemp() - key = f"{tmp}/server.key" - pem = f"{tmp}/server.pem" - subprocess.run( - [ - "openssl", - "req", - "-x509", - "-newkey", - "rsa:2048", - "-keyout", - key, - "-out", - pem, - "-days", - "1", - "-nodes", - "-subj", - "/CN=localhost", - ], - check=True, - capture_output=True, - ) - return pem, key - - -def _free_port() -> int: - with socket.socket() as s: - s.bind(("127.0.0.1", 0)) - return int(s.getsockname()[1]) - - -def _serve_cleartext(port: int, body: bytes, chunks: int = 1) -> threading.Thread: - """Minimal cleartext HTTP responder; writes the body in ``chunks`` - pieces to exercise the pump's incremental forwarding.""" - - def run() -> None: - srv = socket.create_server(("127.0.0.1", port)) - try: - conn, _ = srv.accept() - conn.recv(4096) - head = ( - b"HTTP/1.1 200 OK\r\nContent-Type: text/plain\r\n" - + f"Content-Length: {len(body)}\r\nConnection: close\r\n\r\n".encode() - ) - conn.sendall(head) - step = max(1, len(body) // chunks) - for i in range(0, len(body), step): - conn.sendall(body[i : i + step]) - time.sleep(0.01) - conn.close() - finally: - srv.close() - - t = threading.Thread(target=run, daemon=True) - t.start() - return t - - -def _tls_get(port: int, ca: str) -> bytes: - # Unverified client: these tests exercise the proxy byte path, not - # PKI. A self-signed server cert is not a valid CA for chain - # verification, and a full CA+leaf fixture adds nothing here. - context = ssl._create_unverified_context() # noqa: SLF001 - with ( - socket.create_connection(("127.0.0.1", port)) as raw, - context.wrap_socket(raw, server_hostname="localhost") as tls, - ): - tls.sendall(b"GET / HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n") - out = b"" - while True: - data = tls.recv(4096) - if not data: - return out - out += data - - -def test_tls_proxy_round_trip(self_signed_cert: Any) -> None: - pem, key = self_signed_cert - upstream_port = _free_port() - proxy_port = _free_port() - _serve_cleartext(upstream_port, b"hello over tls") - start_tls_proxy("127.0.0.1", proxy_port, "127.0.0.1", upstream_port, pem, key) - time.sleep(0.2) - - response = _tls_get(proxy_port, pem) - assert b"200 OK" in response - assert response.endswith(b"hello over tls") - - -def test_tls_proxy_rejects_cleartext(self_signed_cert: Any) -> None: - pem, key = self_signed_cert - upstream_port = _free_port() - proxy_port = _free_port() - _serve_cleartext(upstream_port, b"nope") - start_tls_proxy("127.0.0.1", proxy_port, "127.0.0.1", upstream_port, pem, key) - time.sleep(0.2) - - with socket.create_connection(("127.0.0.1", proxy_port)) as conn: - conn.sendall(b"GET / HTTP/1.1\r\n\r\n") - conn.settimeout(2.0) - try: - data = conn.recv(256) - except (ConnectionResetError, BrokenPipeError, TimeoutError): - # Server aborted the failed handshake without responding — - # also a valid rejection. - return - # TLS alert bytes, not an HTTP response: an HTTP response - # starts with 'H'. - assert not data.startswith(b"HTTP") - - -def test_tls_proxy_incremental_forwarding(self_signed_cert: Any) -> None: - """A multi-chunk upstream response arrives complete through the pump.""" - pem, key = self_signed_cert - upstream_port = _free_port() - proxy_port = _free_port() - payload = b"x" * 8192 - _serve_cleartext(upstream_port, payload, chunks=8) - start_tls_proxy("127.0.0.1", proxy_port, "127.0.0.1", upstream_port, pem, key) - time.sleep(0.2) - - response = _tls_get(proxy_port, pem) - assert response.endswith(payload)