From 003b953cc02edd77cfd86f69fa49b33067b5bd56 Mon Sep 17 00:00:00 2001 From: Jer Miller Date: Sun, 19 Apr 2026 21:18:25 -0600 Subject: [PATCH] link: pipe tunnel streams to convey Add a loopback TCP pipe for relay streams, switch the link service off the in-process WSGI app path, and log stream-close metadata around the new conduit. Co-Authored-By: Claude Opus 4.7 (1M context) --- think/link/README.md | 4 +- think/link/mux.py | 1 + think/link/relay_client.py | 38 +++++----- think/link/service.py | 17 +---- think/link/tcp_pipe.py | 137 +++++++++++++++++++++++++++++++++++++ 5 files changed, 159 insertions(+), 38 deletions(-) create mode 100644 think/link/tcp_pipe.py diff --git a/think/link/README.md b/think/link/README.md index 1be065f06..e670c4fbc 100644 --- a/think/link/README.md +++ b/think/link/README.md @@ -12,7 +12,7 @@ The `spl` repo's `home/` continues as the open-source reference implementation o |------|---------| | `service.py` | Entry point + runtime. `sol link` runs `main()` here. | | `relay_client.py` | Listen-WS + per-tunnel TLS pump. Spawns a task per incoming tunnel. | -| `wsgi_bridge.py` | HTTP/1.1 ⇄ WSGI adapter that pipes tunnel bytes to convey's real Flask app. | +| `tcp_pipe.py` | Byte pump that opens a loopback TCP connection to convey and forwards stream bytes both ways. | | `tls_adapter.py` | pyOpenSSL memory-BIO adapter. Runs TLS 1.3 over opaque byte streams. | | `ca.py` | Local CA lifecycle + CSR signing + home-attestation minting. | | `auth.py` | `authorized_clients.json` reader/writer with mtime-reload and last-seen tracking. | @@ -28,4 +28,4 @@ The `spl` repo's `home/` continues as the open-source reference implementation o ## privacy -No payload bytes are ever logged. The tunnel pump only emits rendezvous metadata (tunnel_id, stream_id, method, path, status, byte counts) to logs and callosum. The CA private key never leaves `journal/link/ca/private.pem`; account tokens live in `journal/link/tokens/` and device tokens live on the phone. +No payload bytes are ever logged. The tunnel pump only emits rendezvous metadata (tunnel_id, stream_id, bytes_in, bytes_out, closed_reason) to logs and callosum. The CA private key never leaves `journal/link/ca/private.pem`; account tokens live in `journal/link/tokens/` and device tokens live on the phone. diff --git a/think/link/mux.py b/think/link/mux.py index 7cdf197a7..fc9e1f768 100644 --- a/think/link/mux.py +++ b/think/link/mux.py @@ -73,6 +73,7 @@ class StreamWriter: def __init__(self, mux: Multiplexer, state: _StreamState) -> None: self._mux = mux self._state = state + self.stream_id = state.stream_id async def write(self, data: bytes) -> None: if self._state.writer_closed: diff --git a/think/link/relay_client.py b/think/link/relay_client.py index f11d6e2d1..3b347d7d6 100644 --- a/think/link/relay_client.py +++ b/think/link/relay_client.py @@ -9,7 +9,7 @@ On startup: 3. Loop: wait for {"type":"incoming","tunnel_id":...} control messages. On each signal, spawn a tunnel task that opens /tunnel/, drives pyOpenSSL TLS 1.3 in memory-BIO mode, and hands the plaintext byte - stream to the multiplexer + WSGI bridge. + stream to the multiplexer + loopback TCP pipe into convey. 4. On disconnect, reconnect with exponential backoff (1s → 60s, ±25%). All WebSocket I/O uses the `websockets` library in asyncio mode. The TLS @@ -17,7 +17,7 @@ state machine runs inline on the event loop — each tunnel is a dedicated task pumping bytes between the WS and the TLS engine. Privacy invariant: NO payload bytes ever appear in logs. Only rendezvous -metadata (tunnel_id, stream_id, byte_count, status code, duration) is +metadata (tunnel_id, stream_id, bytes_in, bytes_out, closed_reason) is eligible for logging, and everything emitted to callosum is the same rendezvous-only subset. """ @@ -40,6 +40,7 @@ from websockets.exceptions import ConnectionClosed from .auth import AuthorizedClients from .ca import LoadedCa from .mux import Multiplexer, StreamWriter +from .tcp_pipe import ConveyUnreachable, PipeMetadata, pump_stream from .tls_adapter import ( TlsError, build_server_context, @@ -47,7 +48,6 @@ from .tls_adapter import ( issue_server_cert, new_server, ) -from .wsgi_bridge import serve_request log = logging.getLogger("link.relay_client") @@ -70,7 +70,6 @@ class RelayClient: on_account_token: Callable[[str], None], ca: LoadedCa, authorized: AuthorizedClients, - wsgi_app: Callable[..., Any], callosum_emit: CallosumEmit | None = None, ) -> None: self._instance_id = instance_id @@ -80,7 +79,6 @@ class RelayClient: self._on_account_token = on_account_token self._ca = ca self._authorized = authorized - self._wsgi_app = wsgi_app self._emit = callosum_emit or (lambda _event, _fields: None) self._running = False self._listen_state = "offline" @@ -228,22 +226,22 @@ class RelayClient: reader: asyncio.StreamReader, writer: StreamWriter, ) -> None: - meta = await serve_request( - reader, - writer, - self._wsgi_app, - peer_fingerprint=tls.peer_fingerprint, - tunnel_id=tunnel_id, - ) - await writer.close() + try: + meta: PipeMetadata = await pump_stream( + reader, + writer, + tunnel_id=tunnel_id, + stream_id=writer.stream_id, + ) + except ConveyUnreachable: + raise log.debug( - "tunnel %s exchange: method=%s path=%s status=%s in=%s out=%s", - tunnel_id, - meta.method, - meta.path, - meta.status, - meta.request_bytes, - meta.response_bytes, + "link stream closed tunnel=%s stream_id=%d bytes_in=%d bytes_out=%d reason=%s", + meta.tunnel_id, + meta.stream_id, + meta.bytes_in, + meta.bytes_out, + meta.closed_reason, ) mux = Multiplexer(send_frame, handle_stream, is_listener=True) diff --git a/think/link/service.py b/think/link/service.py index c69a00523..bf8a9a2c0 100644 --- a/think/link/service.py +++ b/think/link/service.py @@ -9,7 +9,7 @@ convey, etc. Service lifecycle: start → load state + CA → ensure account_token (enroll once) → open listen WS to spl-relay → accept tunnel pairs → pump bytes through - TLS → convey WSGI. On disconnect, reconnect with exponential backoff. + TLS → convey (TCP pipe). On disconnect, reconnect with exponential backoff. Exits on SIGINT/SIGTERM with a clean close of the listen WS and all in-flight tunnel WSes. @@ -55,8 +55,6 @@ async def run_service() -> None: authorized = AuthorizedClients(authorized_clients_path()) token = load_account_token() - wsgi_app = _build_convey_wsgi() - callosum = CallosumConnection() callosum.start() @@ -74,7 +72,6 @@ async def run_service() -> None: on_account_token=save_account_token, ca=ca, authorized=authorized, - wsgi_app=wsgi_app, callosum_emit=emit, ) @@ -98,18 +95,6 @@ async def run_service() -> None: callosum.stop() -def _build_convey_wsgi() -> Any: - """Return convey's Flask app as a WSGI callable. - - Imported lazily so `sol call link status` (and other dry reads) don't - pay the convey import cost. The returned object is the Flask app — - calling it as `app(environ, start_response)` invokes its WSGI entry. - """ - from convey import create_app - - return create_app() - - class _suppress_not_implemented: """Context manager that swallows NotImplementedError for Windows/TTYs.""" diff --git a/think/link/tcp_pipe.py b/think/link/tcp_pipe.py new file mode 100644 index 000000000..664e3c2ca --- /dev/null +++ b/think/link/tcp_pipe.py @@ -0,0 +1,137 @@ +# SPDX-License-Identifier: AGPL-3.0-only +# Copyright (c) 2026 sol pbc + +"""Byte pump for tunnel→convey. + +Opens a loopback TCP connection to the running convey server and pumps +bytes bidirectionally with half-close handling. Does not parse HTTP or +inspect payloads — the pipe carries whatever the tunnel stream carries. +""" + +from __future__ import annotations + +import asyncio +import logging +from dataclasses import dataclass + +from think.utils import read_service_port + +from .mux import StreamWriter + +log = logging.getLogger("link.pipe") + +_BUF = 65536 + + +@dataclass +class PipeMetadata: + tunnel_id: str + stream_id: int + bytes_in: int + bytes_out: int + closed_reason: str + + +class ConveyUnreachable(RuntimeError): ... + + +async def _pump_up(src: asyncio.StreamReader, dst: asyncio.StreamWriter) -> int: + total = 0 + try: + while True: + data = await src.read(_BUF) + if not data: + try: + dst.write_eof() + except (OSError, RuntimeError): + pass + return total + dst.write(data) + await dst.drain() + total += len(data) + except (BrokenPipeError, ConnectionResetError): + return total + + +async def _pump_down(src: asyncio.StreamReader, dst: StreamWriter) -> int: + total = 0 + try: + while True: + data = await src.read(_BUF) + if not data: + await dst.close() + return total + await dst.write(data) + total += len(data) + except (BrokenPipeError, ConnectionResetError): + return total + + +async def pump_stream( + reader: asyncio.StreamReader, + writer: StreamWriter, + *, + tunnel_id: str, + stream_id: int, +) -> PipeMetadata: + port = read_service_port("convey") + if port is None: + log.debug( + "convey unreachable for stream tunnel=%s stream_id=%d: no convey port", + tunnel_id, + stream_id, + ) + raise ConveyUnreachable("convey port not published") + + try: + tcp_reader, tcp_writer = await asyncio.open_connection("127.0.0.1", port) + except (ConnectionRefusedError, OSError) as err: + log.debug( + "convey unreachable for stream tunnel=%s stream_id=%d: %s", + tunnel_id, + stream_id, + err, + ) + raise ConveyUnreachable(str(err)) from err + + bytes_in = 0 + bytes_out = 0 + closed_reason = "both_eof" + try: + if hasattr(asyncio, "TaskGroup"): + async with asyncio.TaskGroup() as tg: + up = tg.create_task(_pump_up(reader, tcp_writer)) + down = tg.create_task(_pump_down(tcp_reader, writer)) + bytes_in, bytes_out = up.result(), down.result() + else: + tasks = [ + asyncio.create_task(_pump_up(reader, tcp_writer)), + asyncio.create_task(_pump_down(tcp_reader, writer)), + ] + try: + bytes_in, bytes_out = await asyncio.gather(*tasks) + except BaseException: + for task in tasks: + task.cancel() + await asyncio.gather(*tasks, return_exceptions=True) + raise + if reader.at_eof() and tcp_reader.at_eof(): + closed_reason = "both_eof" + elif reader.at_eof(): + closed_reason = "client_eof" + elif tcp_reader.at_eof(): + closed_reason = "server_eof" + except asyncio.CancelledError: + closed_reason = "cancelled" + raise + except Exception: + closed_reason = "error" + raise + finally: + try: + tcp_writer.close() + await tcp_writer.wait_closed() + except (OSError, RuntimeError): + pass + + return PipeMetadata(tunnel_id, stream_id, bytes_in, bytes_out, closed_reason) -- 2.51.2