From ce4c74eca890b9f658d5bcc58f4e27d90fef6922 Mon Sep 17 00:00:00 2001 From: Jer Miller Date: Sun, 19 Apr 2026 18:46:58 -0600 Subject: [PATCH] link: fork spl/home into think/link + apps/link convey surface MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Scaffolding chunks 1-4 of the spl-solstone integration (vpe req_xhoetsh6). Complete fork from github.com/solpbc/spl home/ β€” no pip dep, no submodule, no sync. Two projects are fully independent from here. - think/link/: tunnel service (service.py, relay_client.py, wsgi_bridge.py, ca.py, auth.py, nonces.py, mux.py, framing.py, tls_adapter.py, paths.py). pair_server.py dropped β€” pair runs through convey's existing listener. CA has no passphrase layer per spec (journal/link/ca/private.pem mode 0600). Added last_seen_at column + touch_last_seen() on AuthorizedClients. WSGI bridge pipes tunnel bytes to convey's real Flask app. - apps/link/: dashboard (workspace.html), Flask routes (/pair-start, /pair, /unpair, /api/devices, /api/status), Typer CLI (pair/list/ unpair/status). All spec literal-copy strings landed verbatim. - supervisor: link service launches alongside cortex (--no-link opt-out). - sol.py: 'sol link' command + GROUPS entry. - pyproject.toml: pyOpenSSL + websockets deps. - tests/link/test_framing.py: 17 tests ported from spl-repo (all green). Privacy invariant: no payload bytes in logs. Rendezvous-only (method, path, status, byte counts, tunnel_id, stream_id). Callosum tract 'link' emits enrolled/connecting/connected/disconnect/tunnel_pair/tunnel_close/ last_seen. Remaining chunks (ca/auth/mux/nonces/wsgi unit tests; in-tree test client; end-to-end integration test; blindness grep; spl-repo cross-reference PR) delegated to a continuation hopper lode. Full solstone suite: 3498 tests passing. Co-Authored-By: Claude Opus 4.7 (1M context) --- apps/link/app.json | 5 + apps/link/call.py | 195 +++++++++++++++++++ apps/link/routes.py | 296 ++++++++++++++++++++++++++++ apps/link/workspace.html | 326 +++++++++++++++++++++++++++++++ pyproject.toml | 5 + sol.py | 2 + tests/link/__init__.py | 0 tests/link/test_framing.py | 143 ++++++++++++++ think/link/README.md | 31 +++ think/link/__init__.py | 24 +++ think/link/__main__.py | 9 + think/link/auth.py | 192 ++++++++++++++++++ think/link/ca.py | 252 ++++++++++++++++++++++++ think/link/framing.py | 185 ++++++++++++++++++ think/link/mux.py | 287 +++++++++++++++++++++++++++ think/link/nonces.py | 176 +++++++++++++++++ think/link/paths.py | 164 ++++++++++++++++ think/link/relay_client.py | 390 +++++++++++++++++++++++++++++++++++++ think/link/service.py | 145 ++++++++++++++ think/link/tls_adapter.py | 199 +++++++++++++++++++ think/link/wsgi_bridge.py | 350 +++++++++++++++++++++++++++++++++ think/supervisor.py | 14 ++ 22 files changed, 3390 insertions(+) create mode 100644 apps/link/app.json create mode 100644 apps/link/call.py create mode 100644 apps/link/routes.py create mode 100644 apps/link/workspace.html create mode 100644 tests/link/__init__.py create mode 100644 tests/link/test_framing.py create mode 100644 think/link/README.md create mode 100644 think/link/__init__.py create mode 100644 think/link/__main__.py create mode 100644 think/link/auth.py create mode 100644 think/link/ca.py create mode 100644 think/link/framing.py create mode 100644 think/link/mux.py create mode 100644 think/link/nonces.py create mode 100644 think/link/paths.py create mode 100644 think/link/relay_client.py create mode 100644 think/link/service.py create mode 100644 think/link/tls_adapter.py create mode 100644 think/link/wsgi_bridge.py diff --git a/apps/link/app.json b/apps/link/app.json new file mode 100644 index 000000000..cf0787ba2 --- /dev/null +++ b/apps/link/app.json @@ -0,0 +1,5 @@ +{ + "icon": "πŸ”—", + "label": "Link", + "facets": {"disabled": true} +} diff --git a/apps/link/call.py b/apps/link/call.py new file mode 100644 index 000000000..63afb6e15 --- /dev/null +++ b/apps/link/call.py @@ -0,0 +1,195 @@ +# SPDX-License-Identifier: AGPL-3.0-only +# Copyright (c) 2026 sol pbc + +"""CLI commands for the link tunnel service. + +Auto-discovered by ``think.call`` and mounted as ``sol call link ...``. +""" + +from __future__ import annotations + +import datetime as dt +import socket +import time + +import typer + +from think.link.auth import AuthorizedClients +from think.link.ca import generate_nonce, load_or_generate_ca +from think.link.nonces import NONCE_TTL_SECONDS, NonceStore +from think.link.paths import ( + LinkState, + authorized_clients_path, + ca_dir, + load_account_token, + nonces_path, + relay_url, +) +from think.utils import require_solstone + +app = typer.Typer( + help="Link β€” tunnel service for reaching this solstone from paired phones." +) + + +@app.callback() +def _require_up() -> None: + require_solstone() + + +def _authorized() -> AuthorizedClients: + return AuthorizedClients(authorized_clients_path()) + + +def _nonces() -> NonceStore: + return NonceStore(nonces_path()) + + +def _detect_lan_ip() -> str | None: + try: + sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + try: + sock.connect(("8.8.8.8", 80)) + return sock.getsockname()[0] + finally: + sock.close() + except OSError: + return None + + +def _relative_time(iso: str | None) -> str: + if not iso: + return "never" + try: + then = dt.datetime.strptime(iso, "%Y-%m-%dT%H:%M:%SZ").replace(tzinfo=dt.UTC) + except ValueError: + return iso + now = dt.datetime.now(dt.UTC) + delta = (now - then).total_seconds() + if delta < 15: + return "just now" + if delta < 60: + return f"{int(delta)} seconds ago" + if delta < 3600: + return f"{int(delta // 60)} minutes ago" + if delta < 86400: + return f"{int(delta // 3600)} hours ago" + return f"{int(delta // 86400)} days ago" + + +@app.command() +def pair( + device_label: str = typer.Option(..., "--device-label", help="Label for the phone being paired"), + convey_host: str = typer.Option( + "", + "--convey-host", + help="Override host[:port] for the pair URL (default: auto-detect LAN IP)", + ), + convey_port: int = typer.Option( + 0, + "--convey-port", + help="Override convey port (default: read from service port file or 5015)", + ), + timeout_seconds: int = typer.Option( + NONCE_TTL_SECONDS, + "--timeout", + help="How long to wait for the phone before giving up", + ), +) -> None: + """Mint a one-shot nonce, print the pair URL + QR-ready payload, wait for completion.""" + from think.utils import read_service_port + + value = generate_nonce() + _nonces().add(value, device_label) + ca_fp = load_or_generate_ca(ca_dir()).fingerprint_sha256() + + host = convey_host or _detect_lan_ip() or "localhost" + port = convey_port or read_service_port("convey") or 5015 + base = f"http://{host}:{port}" + url = f"{base}/app/link/pair?token={value}" + + typer.echo(f"Pair code: {value} (expires in 5 minutes)") + typer.echo(f"Pair URL: {url}") + typer.echo(f"CA fingerprint: sha256:{ca_fp}") + typer.echo(f"Device: {device_label}") + typer.echo("") + typer.echo("Waiting for phone…") + + # Poll authorized_clients.json for a new entry. + authorized = _authorized() + before = {e.fingerprint for e in authorized.snapshot()} + deadline = time.time() + timeout_seconds + while time.time() < deadline: + time.sleep(1.0) + current = authorized.snapshot() + new_entries = [e for e in current if e.fingerprint not in before] + if new_entries: + entry = new_entries[-1] + typer.echo(f"Paired: {entry.device_label}") + typer.echo(f" fingerprint: {entry.fingerprint}") + typer.echo(f" paired_at: {entry.paired_at}") + raise typer.Exit(0) + # Also detect nonce consumption β€” if the nonce is gone/used, assume + # the pair route fired but we missed the device (rare). + nonce_entry = _nonces().peek(value) + if nonce_entry and nonce_entry.used: + typer.echo("Pair request completed; device should appear in `sol call link list`.") + raise typer.Exit(0) + typer.echo("Timed out. Pair code expired.") + raise typer.Exit(2) + + +@app.command("list") +def list_devices() -> None: + """Print every paired device with its last-seen time.""" + entries = _authorized().snapshot() + if not entries: + typer.echo("No phones paired yet.") + return + for entry in entries: + short_fp = entry.fingerprint.replace("sha256:", "")[:16] + typer.echo( + f"- {entry.device_label}" + f" β€” added {_relative_time(entry.paired_at)}" + f" β€” last seen {_relative_time(entry.last_seen_at)}" + f" [{short_fp}]" + ) + + +@app.command() +def unpair( + target: str = typer.Argument(..., help="Device label or fingerprint (sha256:)"), +) -> None: + """Revoke a paired device. Next reconnect from that device fails at TLS handshake.""" + authorized = _authorized() + if target.startswith("sha256:"): + removed = authorized.remove(target) + if not removed: + typer.echo(f"No paired device with fingerprint {target}") + raise typer.Exit(1) + typer.echo("Unpaired.") + return + entry = authorized.find_by_label(target) + if entry is None: + typer.echo(f"No paired device with label {target!r}") + raise typer.Exit(1) + authorized.remove(entry.fingerprint) + typer.echo("Unpaired.") + + +@app.command() +def status() -> None: + """Report enrollment, listen-WS state, active tunnel count, relay endpoint.""" + state = LinkState.load_or_create() + token = load_account_token() + url = relay_url() + entries = _authorized().snapshot() + typer.echo(f"Instance ID: {state.instance_id}") + typer.echo(f"Home label: {state.home_label}") + typer.echo(f"Relay URL: {url}") + typer.echo(f"Enrolled: {'yes' if token else 'no'}") + typer.echo(f"Paired devices: {len(entries)}") + # Listen-WS state and active-tunnel count live in the service process + # memory β€” surfaced via callosum events rather than polled here. The + # convey /app/link/api/status route is the live vantage. + typer.echo("Listen-WS state: (query convey /app/link/api/status for live state)") diff --git a/apps/link/routes.py b/apps/link/routes.py new file mode 100644 index 000000000..bfd56995d --- /dev/null +++ b/apps/link/routes.py @@ -0,0 +1,296 @@ +# SPDX-License-Identifier: AGPL-3.0-only +# Copyright (c) 2026 sol pbc + +"""link app routes β€” pair ceremony + paired-device dashboard. + +All user-facing work for the spl tunnel integration happens here. The +protocol-level code (TLS, framing, mux) lives in `think/link/`; this +module is the HTTP surface that mobiles and the convey UI hit. + +Routes: + + GET /link dashboard (paired devices + pair button) + GET /link/qr.png QR image for an active nonce (via ?token=) + POST /link/pair-start generate a new nonce + return QR payload + POST /link/pair mobile posts CSR + nonce; we sign + attest + POST /link/unpair remove a fingerprint (immediate revocation) + GET /link/api/devices JSON list of paired devices for JS polling + GET /link/api/status service status (for dashboard refresh) + +The pair hop is plain HTTP on convey's existing listener β€” there is no +separate port. Integrity is provided by the CA-fingerprint pinned in the +QR, not by transport TLS. A MITM on the LAN can observe the nonce but +cannot forge a cert signed by the pinned CA. +""" + +from __future__ import annotations + +import datetime as dt +import logging +import socket +from typing import Any + +from cryptography.hazmat.primitives import serialization +from flask import Blueprint, jsonify, request + +from think.link.auth import AuthorizedClients, ClientEntry +from think.link.ca import generate_nonce, load_or_generate_ca, mint_attestation, sign_csr +from think.link.nonces import NonceStore +from think.link.paths import ( + LinkState, + authorized_clients_path, + ca_dir, + load_account_token, + nonces_path, + relay_url, +) + +logger = logging.getLogger(__name__) + +link_bp = Blueprint( + "app:link", + __name__, + url_prefix="/app/link", +) + + +def _authorized() -> AuthorizedClients: + return AuthorizedClients(authorized_clients_path()) + + +def _nonces() -> NonceStore: + return NonceStore(nonces_path()) + + +def _utc_now_iso() -> str: + return dt.datetime.now(dt.UTC).strftime("%Y-%m-%dT%H:%M:%SZ") + + +def _lan_pair_base_url() -> str: + """Best-effort LAN URL for the convey host β€” used in the QR payload.""" + host = request.host + scheme = "http" if not request.is_secure else "https" + # If the request came in on localhost, substitute a routable LAN IP so + # the phone's QR scan works. If we can't find one, fall back to host. + try: + hostname, _, port = host.partition(":") + if hostname in ("localhost", "127.0.0.1", "::1", "0.0.0.0"): + lan_ip = _detect_lan_ip() + if lan_ip: + host = f"{lan_ip}:{port}" if port else lan_ip + except Exception: + logger.debug("lan ip detection failed", exc_info=True) + return f"{scheme}://{host}" + + +def _detect_lan_ip() -> str | None: + """Pick a reasonable LAN-facing IPv4 by opening a UDP socket. + + No packets are sent β€” we just read what src address the kernel would + pick for a route to an external host. Returns None on any error. + """ + try: + sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + try: + sock.connect(("8.8.8.8", 80)) + return sock.getsockname()[0] + finally: + sock.close() + except OSError: + return None + + +def _ca_fingerprint() -> str: + ca = load_or_generate_ca(ca_dir()) + return ca.fingerprint_sha256() + + +def _is_lan_accessible() -> bool: + """Check whether convey is bound to a non-loopback interface. + + Used to drive the "enable LAN access" nudge on /link. Best-effort: the + signal is the Host header the dashboard loaded under. + """ + hostname, _, _ = request.host.partition(":") + if hostname in ("localhost", "127.0.0.1", "::1"): + return bool(_detect_lan_ip()) + return True + + +# --------------------------------------------------------------------------- +# dashboard +# --------------------------------------------------------------------------- + + +@link_bp.route("/api/devices") +def api_devices() -> Any: + """JSON list of paired devices β€” used by the dashboard JS.""" + entries = _authorized().snapshot() + devices = [_entry_to_json(e) for e in entries] + return jsonify({"devices": devices}) + + +@link_bp.route("/api/status") +def api_status() -> Any: + """Snapshot of link-service state for the dashboard header.""" + state = LinkState.load_or_create() + token = load_account_token() + ca_fp = _ca_fingerprint() if ca_dir().exists() else None + return jsonify( + { + "instance_id": state.instance_id, + "home_label": state.home_label, + "enrolled": token is not None, + "relay_url": relay_url(), + "ca_fingerprint": ca_fp, + "lan_accessible": _is_lan_accessible(), + } + ) + + +# --------------------------------------------------------------------------- +# pair ceremony +# --------------------------------------------------------------------------- + + +@link_bp.route("/pair-start", methods=["POST"]) +def pair_start() -> Any: + """Generate a single-use 5-minute nonce and return QR-ready payload.""" + payload = request.get_json(silent=True) or {} + device_label = str(payload.get("device_label") or "").strip() or "unnamed device" + + nonce = generate_nonce() + _nonces().add(nonce, device_label) + + ca_fp = _ca_fingerprint() + pair_url = f"{_lan_pair_base_url()}/app/link/pair?token={nonce}" + # The QR payload is a stable shape the mobile can parse β€” keep it + # versioned so future mobiles stay backward-compatible. + qr_payload = { + "v": 1, + "kind": "spl-pair", + "pair_url": pair_url, + "ca_fingerprint": ca_fp, + "expires_in": 300, + "device_label": device_label, + } + return jsonify( + { + "nonce": nonce, + "pair_url": pair_url, + "qr_payload": qr_payload, + "ca_fingerprint": ca_fp, + "expires_in": 300, + "device_label": device_label, + } + ) + + +@link_bp.route("/pair", methods=["POST"]) +def pair() -> Any: + """Mobile pair endpoint β€” accepts CSR + nonce, signs + mints attestation. + + Query: `?token=` (the nonce minted by /pair-start). + Body (JSON): + { + "csr": "", // required + "device_label": "", // optional (falls back to nonce label) + "nonce": "" // optional: may be in body instead of query + } + + Response on success (200): + { + "client_cert": "", + "ca_chain": ["", ...], + "instance_id": "", + "home_label": "", + "home_attestation": "", + "fingerprint": "sha256:" + } + """ + body = request.get_json(silent=True) or {} + nonce_value = request.args.get("token") or body.get("nonce") + csr_pem = body.get("csr") + device_label = str(body.get("device_label") or "").strip() or "unnamed device" + + if not isinstance(nonce_value, str) or not isinstance(csr_pem, str): + return jsonify({"error": "missing fields (nonce + csr required)"}), 400 + + consumed = _nonces().consume(nonce_value) + if consumed is None: + return jsonify({"error": "nonce expired or used"}), 410 + + effective_label = device_label or (consumed.device_label or "unnamed device") + + ca = load_or_generate_ca(ca_dir()) + try: + client_cert_pem, fingerprint = sign_csr(ca, csr_pem, effective_label) + except Exception as exc: + logger.info("pair: bad csr: %s", exc) + return jsonify({"error": f"bad csr: {exc}"}), 400 + + state = LinkState.load_or_create() + authorized = _authorized() + authorized.add( + fingerprint=fingerprint, + device_label=effective_label, + instance_id=state.instance_id, + paired_at=_utc_now_iso(), + ) + attestation = mint_attestation(ca, state.instance_id, fingerprint) + ca_chain_pem = ca.cert.public_bytes(serialization.Encoding.PEM).decode("ascii") + return jsonify( + { + "client_cert": client_cert_pem, + "ca_chain": [ca_chain_pem], + "instance_id": state.instance_id, + "home_label": state.home_label, + "home_attestation": attestation, + "fingerprint": fingerprint, + } + ) + + +@link_bp.route("/unpair", methods=["POST"]) +def unpair() -> Any: + """Revoke a paired device by label or fingerprint. + + Body (JSON): {"fingerprint": "sha256:..."} or {"device_label": "..."} + """ + body = request.get_json(silent=True) or {} + fingerprint = body.get("fingerprint") + device_label = body.get("device_label") + if not isinstance(fingerprint, str): + if not isinstance(device_label, str): + return jsonify({"error": "fingerprint or device_label required"}), 400 + entry = _authorized().find_by_label(device_label) + if entry is None: + return jsonify({"error": "no paired device with that label"}), 404 + fingerprint = entry.fingerprint + + removed = _authorized().remove(fingerprint) + if not removed: + return jsonify({"error": "fingerprint not paired"}), 404 + return jsonify({"unpaired": fingerprint}) + + +def _entry_to_json(entry: ClientEntry) -> dict[str, Any]: + short_fp = entry.fingerprint.replace("sha256:", "")[:16] + return { + "fingerprint": entry.fingerprint, + "fingerprint_short": short_fp, + "device_label": entry.device_label, + "paired_at": entry.paired_at, + "last_seen_at": entry.last_seen_at, + } + + +# --------------------------------------------------------------------------- +# helpers for the workspace template +# --------------------------------------------------------------------------- + + +@link_bp.app_context_processor +def _inject_link_helpers() -> dict[str, Any]: + """Make `url_for` to link endpoints easy from templates.""" + return {} diff --git a/apps/link/workspace.html b/apps/link/workspace.html new file mode 100644 index 000000000..0bc4eac8a --- /dev/null +++ b/apps/link/workspace.html @@ -0,0 +1,326 @@ +
+
+

Reach your solstone from anywhere

+

spl is blind by construction. Cloudflare and sol pbc cannot see anything inside the tunnel.

+
+ + Service status: loading… + + +
+
+ + + +
+ + +
+ + + + + + +
+ + + + diff --git a/pyproject.toml b/pyproject.toml index deffd1c0c..ae86343f2 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -53,6 +53,11 @@ dependencies = [ "httpx", "jsonschema>=4.26,<5", "genai-prices", + # Link tunnel service (think/link/): TLS 1.3 in memory-BIO mode over + # an opaque WebSocket; requires pyOpenSSL for the handshake-time + # verify callback that stdlib ssl can't express. + "pyOpenSSL>=24.0", + "websockets>=13.0", "pypdf", "pdf2image", "pytesseract", diff --git a/sol.py b/sol.py index 49cc75b81..611a0b73f 100644 --- a/sol.py +++ b/sol.py @@ -61,6 +61,7 @@ COMMANDS: dict[str, str] = { "providers": "think.providers_cli", "cortex": "think.cortex", "talent": "think.talent_cli", + "link": "think.link", "call": "think.call", "engage": "think.engage", "chat": "think.chat_cli", @@ -128,6 +129,7 @@ GROUPS: dict[str, list[str]] = { "streams", "segment", "journal-stats", + "link", ], "Help": ["chat"], } diff --git a/tests/link/__init__.py b/tests/link/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/tests/link/test_framing.py b/tests/link/test_framing.py new file mode 100644 index 000000000..7e7173fdc --- /dev/null +++ b/tests/link/test_framing.py @@ -0,0 +1,143 @@ +# SPDX-License-Identifier: AGPL-3.0-only +# Copyright (c) 2026 sol pbc + +"""Framing encode/decode round-trip + flag validation (forked from spl/home).""" + +from __future__ import annotations + +import pytest + +from think.link.framing import ( + FLAG_CLOSE, + FLAG_DATA, + FLAG_OPEN, + FLAG_RESERVED_MASK, + FLAG_RESET, + FLAG_WINDOW, + HEADER_LEN, + RESET_PROTOCOL_ERROR, + Frame, + FrameDecoder, + ProtocolError, + build_close, + build_data, + build_open, + build_reset, + build_window, + parse_reset_reason, + parse_window_credit, + validate_flags, +) + + +def test_header_is_8_bytes() -> None: + assert HEADER_LEN == 8 + + +def test_encode_decode_roundtrip() -> None: + original = Frame(stream_id=7, flags=FLAG_DATA, payload=b"hello world") + encoded = original.encode() + decoder = FrameDecoder() + decoder.feed(encoded) + got = decoder.next() + assert got == original + + +def test_decoder_handles_fragmented_feeds() -> None: + frame = Frame(stream_id=5, flags=FLAG_DATA, payload=b"fragmented") + encoded = frame.encode() + decoder = FrameDecoder() + for byte in encoded: + decoder.feed(bytes([byte])) + assert decoder.next() == frame + + +def test_decoder_returns_none_when_incomplete() -> None: + decoder = FrameDecoder() + decoder.feed(b"\x00\x00\x00\x01") # partial header + assert decoder.next() is None + + +def test_multiple_frames_in_one_buffer() -> None: + frames = [build_data(1, b"a"), build_data(3, b"bb"), build_data(5, b"ccc")] + decoder = FrameDecoder() + for f in frames: + decoder.feed(f.encode()) + assert decoder.drain() == frames + + +def test_reserved_bits_rejected_on_encode() -> None: + with pytest.raises(ProtocolError): + Frame(stream_id=1, flags=FLAG_RESERVED_MASK, payload=b"").encode() + + +def test_reserved_bits_rejected_on_decode() -> None: + bad = Frame(stream_id=1, flags=FLAG_DATA, payload=b"") + encoded = bytearray(bad.encode()) + encoded[4] |= 0x80 + decoder = FrameDecoder() + decoder.feed(bytes(encoded)) + with pytest.raises(ProtocolError): + decoder.next() + + +def test_payload_length_bound() -> None: + build_data(1, b"").encode() + ok = Frame(stream_id=1, flags=FLAG_DATA, payload=b"x" * ((1 << 24) - 1)) + ok.encode() + + +def test_open_with_initial_bytes_carries_both_flags() -> None: + f = build_open(1, b"seed") + assert f.flags & FLAG_OPEN and f.flags & FLAG_DATA + assert f.payload == b"seed" + + +def test_open_without_bytes_is_pure_open() -> None: + f = build_open(1) + assert f.flags & FLAG_OPEN + assert not (f.flags & FLAG_DATA) + + +def test_data_with_close_carries_both() -> None: + f = build_data(1, b"last", close=True) + assert f.flags & FLAG_DATA and f.flags & FLAG_CLOSE + + +def test_close_carries_only_close_flag() -> None: + f = build_close(1) + assert f.flags & FLAG_CLOSE + assert not (f.flags & FLAG_OPEN) + + +def test_window_credit_parse_roundtrip() -> None: + f = build_window(1, 65536) + assert parse_window_credit(f) == 65536 + + +def test_reset_reason_parse_roundtrip() -> None: + f = build_reset(1, RESET_PROTOCOL_ERROR) + assert parse_reset_reason(f) == RESET_PROTOCOL_ERROR + + +def test_validate_flags_allows_only_legal_combos() -> None: + for flag in (FLAG_OPEN, FLAG_DATA, FLAG_CLOSE, FLAG_RESET, FLAG_WINDOW): + validate_flags(flag) + validate_flags(FLAG_OPEN | FLAG_DATA) + validate_flags(FLAG_DATA | FLAG_CLOSE) + with pytest.raises(ProtocolError): + validate_flags(FLAG_OPEN | FLAG_CLOSE) + with pytest.raises(ProtocolError): + validate_flags(FLAG_DATA | FLAG_WINDOW) + + +def test_window_frame_requires_4_byte_payload() -> None: + f = Frame(stream_id=1, flags=FLAG_WINDOW, payload=b"abc") + with pytest.raises(ProtocolError): + parse_window_credit(f) + + +def test_reset_frame_requires_1_byte_payload() -> None: + f = Frame(stream_id=1, flags=FLAG_RESET, payload=b"") + with pytest.raises(ProtocolError): + parse_reset_reason(f) diff --git a/think/link/README.md b/think/link/README.md new file mode 100644 index 000000000..1be065f06 --- /dev/null +++ b/think/link/README.md @@ -0,0 +1,31 @@ +# link service + +The home-side tunnel endpoint for the spl protocol β€” solstone's long-term home for this code. + +**Forked from [`github.com/solpbc/spl`](https://github.com/solpbc/spl) `home/` on 2026-04-20.** +The two copies are now fully independent: no pip dep, no submodule, no sync scripts. +The `spl` repo's `home/` continues as the open-source reference implementation of the protocol; this module is the canonical production implementation. + +## layout + +| File | Purpose | +|------|---------| +| `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. | +| `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. | +| `nonces.py` | Pair-ceremony nonce store (shared between CLI and convey pair route). | +| `mux.py` | Multiplex state machine per stream. | +| `framing.py` | Wire-frame encode/decode. | +| `paths.py` | Journal-path helpers + `SOL_LINK_RELAY_URL` resolution. | + +## naming + +- **link** β€” user-facing and architecturally-visible names: service name, convey app, `sol call link`, `journal/link/`, `/link` route. +- **spl** β€” protocol-level constructs: wire-format frames, JWT claim schemas, reset reason codes. These reference the external stable spl protocol and keep that name. + +## 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. diff --git a/think/link/__init__.py b/think/link/__init__.py new file mode 100644 index 000000000..65ebbd206 --- /dev/null +++ b/think/link/__init__.py @@ -0,0 +1,24 @@ +# SPDX-License-Identifier: AGPL-3.0-only +# Copyright (c) 2026 sol pbc + +"""link service β€” solstone's tunnel endpoint. + +Forked from github.com/solpbc/spl home/ on 2026-04-20; the two copies are +now fully independent. The spl repo is the open-source protocol reference; +this is the canonical production implementation. + +The service opens a listen WebSocket to spl-relay and terminates TLS 1.3 +inside the solstone process. Paired devices reach convey's real routes +through the tunnel β€” there is no separate in-tunnel test app. + +The wire protocol and frame types keep the "spl" name (spl_frame, +SplTunnelFrame, etc.); everything user-facing, architecturally-visible, +or journal-facing is "link" (service name, apps/link, journal/link, +sol call link, /link). +""" + +__version__ = "0.1.0" + +from .service import main # noqa: E402 β€” re-exported so `sol link` can import it + +__all__ = ["main"] diff --git a/think/link/__main__.py b/think/link/__main__.py new file mode 100644 index 000000000..caa9e2e88 --- /dev/null +++ b/think/link/__main__.py @@ -0,0 +1,9 @@ +# SPDX-License-Identifier: AGPL-3.0-only +# Copyright (c) 2026 sol pbc + +"""python -m think.link β€” entry point for `sol link`.""" + +from .service import main + +if __name__ == "__main__": + main() diff --git a/think/link/auth.py b/think/link/auth.py new file mode 100644 index 000000000..3ec1d6c37 --- /dev/null +++ b/think/link/auth.py @@ -0,0 +1,192 @@ +# SPDX-License-Identifier: AGPL-3.0-only +# Copyright (c) 2026 sol pbc + +"""authorized_clients.json β€” the link service's authoritative revocation ledger. + +Entry shape is fixed by the spl protocol (see github.com/solpbc/spl +proto/pairing.md Β§6), plus a solstone-specific `last_seen_at` field for +UX: + + { + "fingerprint": "sha256:", + "device_label": "Jer's iPhone", + "paired_at": "2026-04-19T17:42:13Z", + "instance_id": "", + "last_seen_at": "2026-04-19T18:03:12Z" // optional; null/absent = never + } + +The TLS 1.3 server reloads the file on mtime change so an unpair action +takes effect within ~500 ms of the file write. The pair route and the +relay client both own the writer surface; reads (convey dashboard, TLS +verify callback) go through this module. + +`last_seen_at` is local-only β€” never transmitted externally. +""" + +from __future__ import annotations + +import datetime as dt +import fcntl +import json +import os +import threading +from dataclasses import dataclass, replace +from pathlib import Path + + +@dataclass(frozen=True) +class ClientEntry: + fingerprint: str + device_label: str + paired_at: str + instance_id: str + last_seen_at: str | None = None + + +class AuthorizedClients: + """In-memory view of authorized_clients.json with mtime-based reload.""" + + def __init__(self, path: Path) -> None: + self._path = path + self._lock = threading.Lock() + self._entries: dict[str, ClientEntry] = {} + self._mtime_ns = 0 + if path.exists(): + self._reload_locked() + + @property + def path(self) -> Path: + return self._path + + def reload_if_stale(self) -> bool: + """Re-read the file if its mtime changed. Returns True if reloaded.""" + with self._lock: + try: + current = self._path.stat().st_mtime_ns + except FileNotFoundError: + if self._entries: + self._entries = {} + self._mtime_ns = 0 + return True + return False + if current == self._mtime_ns: + return False + self._reload_locked() + return True + + def is_authorized(self, fingerprint: str) -> bool: + self.reload_if_stale() + with self._lock: + return fingerprint in self._entries + + def add( + self, + fingerprint: str, + device_label: str, + instance_id: str, + *, + paired_at: str | None = None, + ) -> None: + paired_at = paired_at or dt.datetime.now(dt.UTC).strftime("%Y-%m-%dT%H:%M:%SZ") + entry = ClientEntry( + fingerprint=fingerprint, + device_label=device_label, + paired_at=paired_at, + instance_id=instance_id, + last_seen_at=None, + ) + with self._lock: + current = self._load_file_locked() + current[fingerprint] = entry + self._atomic_write_locked(current) + self._entries = current + + def remove(self, fingerprint: str) -> bool: + with self._lock: + current = self._load_file_locked() + if fingerprint not in current: + return False + del current[fingerprint] + self._atomic_write_locked(current) + self._entries = current + return True + + def touch_last_seen(self, fingerprint: str, *, now: dt.datetime | None = None) -> bool: + """Update last_seen_at for a paired device. Returns False if not paired.""" + ts = (now or dt.datetime.now(dt.UTC)).strftime("%Y-%m-%dT%H:%M:%SZ") + with self._lock: + current = self._load_file_locked() + existing = current.get(fingerprint) + if existing is None: + return False + current[fingerprint] = replace(existing, last_seen_at=ts) + self._atomic_write_locked(current) + self._entries = current + return True + + def snapshot(self) -> list[ClientEntry]: + self.reload_if_stale() + with self._lock: + return list(self._entries.values()) + + def find_by_label(self, label: str) -> ClientEntry | None: + self.reload_if_stale() + with self._lock: + for entry in self._entries.values(): + if entry.device_label == label: + return entry + return None + + def _reload_locked(self) -> None: + entries = self._load_file_locked() + self._entries = entries + try: + self._mtime_ns = self._path.stat().st_mtime_ns + except FileNotFoundError: + self._mtime_ns = 0 + + def _load_file_locked(self) -> dict[str, ClientEntry]: + if not self._path.exists(): + return {} + try: + raw = json.loads(self._path.read_text("utf-8")) + except (json.JSONDecodeError, OSError): + return {} + out: dict[str, ClientEntry] = {} + if isinstance(raw, list): + for item in raw: + if not isinstance(item, dict): + continue + fp = item.get("fingerprint") + if not isinstance(fp, str): + continue + last_seen = item.get("last_seen_at") + out[fp] = ClientEntry( + fingerprint=fp, + device_label=str(item.get("device_label", "")), + paired_at=str(item.get("paired_at", "")), + instance_id=str(item.get("instance_id", "")), + last_seen_at=last_seen if isinstance(last_seen, str) else None, + ) + return out + + def _atomic_write_locked(self, entries: dict[str, ClientEntry]) -> None: + self._path.parent.mkdir(parents=True, exist_ok=True) + payload = [ + { + "fingerprint": e.fingerprint, + "device_label": e.device_label, + "paired_at": e.paired_at, + "instance_id": e.instance_id, + **({"last_seen_at": e.last_seen_at} if e.last_seen_at else {}), + } + for e in entries.values() + ] + tmp = self._path.with_suffix(self._path.suffix + ".tmp") + with open(tmp, "w", encoding="utf-8") as f: + fcntl.flock(f.fileno(), fcntl.LOCK_EX) + json.dump(payload, f, indent=2) + f.write("\n") + f.flush() + os.fsync(f.fileno()) + os.replace(tmp, self._path) diff --git a/think/link/ca.py b/think/link/ca.py new file mode 100644 index 000000000..dbe1b98b2 --- /dev/null +++ b/think/link/ca.py @@ -0,0 +1,252 @@ +# SPDX-License-Identifier: AGPL-3.0-only +# Copyright (c) 2026 sol pbc + +"""Local CA management + home attestation minting. + +Generates an ECDSA-P256 CA on first run, keeps the private key unencrypted +on disk with mode 0600 (no passphrase layer β€” filesystem perms + disk +encryption are the protection surface; see the spl-solstone-integration +spec "CA lifecycle" for rationale), signs mobile CSRs during the LAN pair +ceremony, and mints short-lived home-attestation JWTs that spl-relay +consumes at /enroll/device. + +See github.com/solpbc/spl proto/pairing.md Β§"the local CA" and +proto/tokens.md Β§"POST /enroll/device" for the protocol contracts. +""" + +from __future__ import annotations + +import base64 +import datetime as dt +import hashlib +import json +import secrets +import time +import uuid +from dataclasses import dataclass +from pathlib import Path + +from cryptography import x509 +from cryptography.hazmat.primitives import hashes, serialization +from cryptography.hazmat.primitives.asymmetric import ec +from cryptography.hazmat.primitives.asymmetric.utils import decode_dss_signature +from cryptography.x509.oid import NameOID + +CA_VALIDITY_DAYS = 365 * 10 +CLIENT_CERT_VALIDITY_DAYS = 365 * 10 # Revocation is via authorized_clients.json. +ATTESTATION_LIFETIME_SECONDS = 240 # 4 min β€” under the 5 min relay cap. + + +@dataclass(frozen=True) +class LoadedCa: + """Materialized CA state: certificate, private key, and cached public-key PEM.""" + + cert: x509.Certificate + private_key: ec.EllipticCurvePrivateKey + pubkey_spki_pem: str + + def fingerprint_sha256(self) -> str: + """SHA-256 of the CA cert DER β€” used as the CA identifier at /enroll/home.""" + return _hex_sha256(self.cert.public_bytes(serialization.Encoding.DER)) + + +def generate_ca( + ca_dir: Path, + common_name: str = "solstone link CA", +) -> LoadedCa: + """Generate a fresh ECDSA-P256 CA and write it to disk. + + Writes `/cert.pem` (world-readable) + `/private.pem` + (mode 0600). No passphrase β€” filesystem perms + disk encryption are + the protection surface per the spec. + """ + private_key = ec.generate_private_key(ec.SECP256R1()) + now = dt.datetime.now(dt.UTC) + subject = issuer = x509.Name( + [x509.NameAttribute(NameOID.COMMON_NAME, common_name)], + ) + cert = ( + x509.CertificateBuilder() + .subject_name(subject) + .issuer_name(issuer) + .public_key(private_key.public_key()) + .serial_number(x509.random_serial_number()) + .not_valid_before(now - dt.timedelta(minutes=5)) + .not_valid_after(now + dt.timedelta(days=CA_VALIDITY_DAYS)) + .add_extension(x509.BasicConstraints(ca=True, path_length=0), critical=True) + .add_extension( + x509.KeyUsage( + digital_signature=True, + content_commitment=False, + key_encipherment=False, + data_encipherment=False, + key_agreement=False, + key_cert_sign=True, + crl_sign=True, + encipher_only=False, + decipher_only=False, + ), + critical=True, + ) + .sign(private_key, hashes.SHA256()) + ) + + ca_dir.mkdir(parents=True, exist_ok=True) + cert_path = _cert_path(ca_dir) + key_path = _key_path(ca_dir) + cert_path.write_bytes(cert.public_bytes(serialization.Encoding.PEM)) + _write_key(key_path, private_key) + return _materialize(cert, private_key) + + +def load_ca(ca_dir: Path) -> LoadedCa: + """Load a previously generated CA. Raises if either file is missing.""" + cert_pem = _cert_path(ca_dir).read_bytes() + key_pem = _key_path(ca_dir).read_bytes() + cert = x509.load_pem_x509_certificate(cert_pem) + key = serialization.load_pem_private_key(key_pem, password=None) + if not isinstance(key, ec.EllipticCurvePrivateKey): + raise ValueError("CA key is not an ECDSA private key") + return _materialize(cert, key) + + +def load_or_generate_ca(ca_dir: Path) -> LoadedCa: + """Return an existing CA if present; otherwise generate a fresh one.""" + if _cert_path(ca_dir).exists() and _key_path(ca_dir).exists(): + return load_ca(ca_dir) + return generate_ca(ca_dir) + + +def sign_csr( + ca: LoadedCa, + csr_pem: str | bytes, + device_label: str, +) -> tuple[str, str]: + """Sign a mobile CSR with the CA. Returns (client_cert_pem, fingerprint).""" + csr_bytes = csr_pem.encode("utf-8") if isinstance(csr_pem, str) else csr_pem + csr = x509.load_pem_x509_csr(csr_bytes) + if not csr.is_signature_valid: + raise ValueError("CSR signature invalid") + pub = csr.public_key() + if not isinstance(pub, ec.EllipticCurvePublicKey): + raise ValueError("CSR public key must be ECDSA") + if not isinstance(pub.curve, ec.SECP256R1): + raise ValueError("CSR public key must be ECDSA-P256") + + now = dt.datetime.now(dt.UTC) + cert = ( + x509.CertificateBuilder() + .subject_name(x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, device_label)])) + .issuer_name(ca.cert.subject) + .public_key(pub) + .serial_number(x509.random_serial_number()) + .not_valid_before(now - dt.timedelta(minutes=5)) + .not_valid_after(now + dt.timedelta(days=CLIENT_CERT_VALIDITY_DAYS)) + .add_extension(x509.BasicConstraints(ca=False, path_length=None), critical=True) + .add_extension( + x509.KeyUsage( + digital_signature=True, + content_commitment=False, + key_encipherment=False, + data_encipherment=False, + key_agreement=False, + key_cert_sign=False, + crl_sign=False, + encipher_only=False, + decipher_only=False, + ), + critical=True, + ) + .add_extension( + x509.ExtendedKeyUsage([x509.oid.ExtendedKeyUsageOID.CLIENT_AUTH]), + critical=False, + ) + .sign(ca.private_key, hashes.SHA256()) + ) + pem = cert.public_bytes(serialization.Encoding.PEM).decode("ascii") + fp = _hex_sha256(cert.public_bytes(serialization.Encoding.DER)) + return pem, f"sha256:{fp}" + + +def mint_attestation( + ca: LoadedCa, + instance_id: str, + device_fp: str, + *, + now: int | None = None, +) -> str: + """Mint an ES256 home-attestation JWT for /enroll/device. + + Shape is locked in github.com/solpbc/spl proto/tokens.md Β§"POST /enroll/device". + """ + iat = now if now is not None else int(time.time()) + exp = iat + ATTESTATION_LIFETIME_SECONDS + header = {"alg": "ES256", "typ": "home-attest"} + claims = { + "iss": f"home:{instance_id}", + "aud": "spl-relay", + "scope": "device.enroll", + "instance_id": instance_id, + "device_fp": device_fp, + "iat": iat, + "exp": exp, + "jti": str(uuid.uuid4()), + } + header_b64 = _b64url(json.dumps(header, separators=(",", ":")).encode("utf-8")) + payload_b64 = _b64url(json.dumps(claims, separators=(",", ":")).encode("utf-8")) + signing_input = f"{header_b64}.{payload_b64}".encode("ascii") + der_sig = ca.private_key.sign(signing_input, ec.ECDSA(hashes.SHA256())) + r, s = decode_dss_signature(der_sig) + raw_sig = r.to_bytes(32, "big") + s.to_bytes(32, "big") + sig_b64 = _b64url(raw_sig) + return f"{header_b64}.{payload_b64}.{sig_b64}" + + +def cert_fingerprint(cert_pem: str | bytes) -> str: + """Compute `sha256:` over the DER form of a PEM-encoded cert.""" + pem_bytes = cert_pem.encode("utf-8") if isinstance(cert_pem, str) else cert_pem + cert = x509.load_pem_x509_certificate(pem_bytes) + return f"sha256:{_hex_sha256(cert.public_bytes(serialization.Encoding.DER))}" + + +def generate_nonce() -> str: + """64-character hex nonce for the pair ceremony.""" + return secrets.token_hex(32) + + +def _materialize(cert: x509.Certificate, key: ec.EllipticCurvePrivateKey) -> LoadedCa: + pub_pem = ( + cert.public_key() + .public_bytes( + serialization.Encoding.PEM, + serialization.PublicFormat.SubjectPublicKeyInfo, + ) + .decode("ascii") + ) + return LoadedCa(cert=cert, private_key=key, pubkey_spki_pem=pub_pem) + + +def _cert_path(ca_dir: Path) -> Path: + return ca_dir / "cert.pem" + + +def _key_path(ca_dir: Path) -> Path: + return ca_dir / "private.pem" + + +def _write_key(path: Path, key: ec.EllipticCurvePrivateKey) -> None: + encoded = key.private_bytes( + encoding=serialization.Encoding.PEM, + format=serialization.PrivateFormat.PKCS8, + encryption_algorithm=serialization.NoEncryption(), + ) + path.write_bytes(encoded) + path.chmod(0o600) + + +def _hex_sha256(data: bytes) -> str: + return hashlib.sha256(data).hexdigest() + + +def _b64url(data: bytes) -> str: + return base64.urlsafe_b64encode(data).rstrip(b"=").decode("ascii") diff --git a/think/link/framing.py b/think/link/framing.py new file mode 100644 index 000000000..38ec86501 --- /dev/null +++ b/think/link/framing.py @@ -0,0 +1,185 @@ +# SPDX-License-Identifier: AGPL-3.0-only +# Copyright (c) 2026 sol pbc + +"""Multiplex framing per the spl protocol (see github.com/solpbc/spl proto/framing.md). + +Wire format (8-byte header + payload): + + +------+------+---+-------+ + | sid4 | flg1 | len3 | header + +------+------+-----------+ + | payload (len bytes) | + +-------------------------+ + +This module implements encoding, decoding, the flag bitfield, the reset +reason codes, and a small `Stream` helper that tracks per-stream credit + +close state per the spec. Protocol violations (reserved bits, illegal flag +combos) raise `ProtocolError`; the caller is expected to translate that +into a RESET frame with reason PROTOCOL_ERROR. + +The relay does not parse frames β€” this code runs on both tunnel endpoints. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Final + +# Flag bits β€” each frame must carry exactly one of OPEN / DATA / CLOSE / +# RESET / WINDOW, except OPEN|DATA (open with initial bytes) and DATA|CLOSE +# (last data + half-close). +FLAG_OPEN: Final[int] = 0x01 +FLAG_DATA: Final[int] = 0x02 +FLAG_CLOSE: Final[int] = 0x04 +FLAG_RESET: Final[int] = 0x08 +FLAG_WINDOW: Final[int] = 0x10 +FLAG_RESERVED_MASK: Final[int] = 0xE0 # bits 5-7 must be zero on send + +# Reset reason codes β€” 1-byte payload of RESET frames. +RESET_PROTOCOL_ERROR: Final[int] = 0x01 +RESET_FLOW_CONTROL_ERROR: Final[int] = 0x02 +RESET_STREAM_LIMIT_EXCEEDED: Final[int] = 0x03 +RESET_INTERNAL_ERROR: Final[int] = 0x04 +RESET_CANCEL: Final[int] = 0x05 +RESET_UNSPECIFIED: Final[int] = 0xFF + +# Spec constants. +HEADER_LEN: Final[int] = 8 +MAX_PAYLOAD: Final[int] = (1 << 24) - 1 # 16 MiB - 1 +INITIAL_WINDOW: Final[int] = 1 << 20 # 1 MiB +MAX_CONCURRENT_STREAMS: Final[int] = 256 +RECOMMENDED_CHUNK: Final[int] = 64 * 1024 + + +class ProtocolError(ValueError): + """Raised when a frame violates the spl framing spec β€” caller should RESET.""" + + +@dataclass(frozen=True) +class Frame: + stream_id: int + flags: int + payload: bytes + + def encode(self) -> bytes: + if not 0 <= self.stream_id <= 0xFFFFFFFF: + raise ProtocolError(f"stream_id out of range: {self.stream_id}") + if not 0 <= self.flags <= 0xFF: + raise ProtocolError(f"flags out of range: {self.flags}") + if self.flags & FLAG_RESERVED_MASK: + raise ProtocolError(f"reserved flag bits set: {self.flags:#x}") + length = len(self.payload) + if length > MAX_PAYLOAD: + raise ProtocolError(f"payload exceeds 16 MiB - 1: {length}") + header = bytearray(HEADER_LEN) + header[0:4] = self.stream_id.to_bytes(4, "big") + header[4] = self.flags + header[5:8] = length.to_bytes(3, "big") + return bytes(header) + self.payload + + +@dataclass +class FrameDecoder: + """Stream decoder. Feed bytes, pull frames until no complete frame remains. + + WebSocket message boundaries are ignored β€” we re-frame ourselves so that + coalescing or fragmentation at the transport layer doesn't affect + framing-layer semantics. + """ + + _buf: bytearray = field(default_factory=bytearray) + + def feed(self, data: bytes | bytearray | memoryview) -> None: + self._buf.extend(data) + + def next(self) -> Frame | None: + if len(self._buf) < HEADER_LEN: + return None + stream_id = int.from_bytes(self._buf[0:4], "big") + flags = self._buf[4] + length = int.from_bytes(self._buf[5:8], "big") + if flags & FLAG_RESERVED_MASK: + raise ProtocolError(f"reserved flag bits set: {flags:#x}") + end = HEADER_LEN + length + if len(self._buf) < end: + return None + payload = bytes(self._buf[HEADER_LEN:end]) + del self._buf[:end] + return Frame(stream_id=stream_id, flags=flags, payload=payload) + + def drain(self) -> list[Frame]: + out: list[Frame] = [] + while True: + frame = self.next() + if frame is None: + return out + out.append(frame) + + +def build_open(stream_id: int, payload: bytes = b"") -> Frame: + flags = FLAG_OPEN + if payload: + flags |= FLAG_DATA + return Frame(stream_id=stream_id, flags=flags, payload=payload) + + +def build_data(stream_id: int, payload: bytes, *, close: bool = False) -> Frame: + if not payload and not close: + return Frame(stream_id=stream_id, flags=FLAG_DATA, payload=b"") + flags = FLAG_DATA + if close: + flags |= FLAG_CLOSE + return Frame(stream_id=stream_id, flags=flags, payload=payload) + + +def build_close(stream_id: int, payload: bytes = b"") -> Frame: + return Frame(stream_id=stream_id, flags=FLAG_CLOSE, payload=payload) + + +def build_reset(stream_id: int, reason: int = RESET_UNSPECIFIED) -> Frame: + if not 0 <= reason <= 0xFF: + raise ProtocolError(f"reset reason out of range: {reason}") + return Frame(stream_id=stream_id, flags=FLAG_RESET, payload=bytes([reason])) + + +def build_window(stream_id: int, credit: int) -> Frame: + if not 0 <= credit <= 0xFFFFFFFF: + raise ProtocolError(f"window credit out of range: {credit}") + return Frame(stream_id=stream_id, flags=FLAG_WINDOW, payload=credit.to_bytes(4, "big")) + + +def validate_flags(flags: int) -> None: + if flags & FLAG_RESERVED_MASK: + raise ProtocolError(f"reserved flag bits set: {flags:#x}") + exclusive = flags & (FLAG_OPEN | FLAG_DATA | FLAG_CLOSE | FLAG_RESET | FLAG_WINDOW) + if not exclusive: + raise ProtocolError("frame has no OPEN/DATA/CLOSE/RESET/WINDOW bit set") + allowed = { + FLAG_OPEN, + FLAG_DATA, + FLAG_CLOSE, + FLAG_RESET, + FLAG_WINDOW, + FLAG_OPEN | FLAG_DATA, + FLAG_DATA | FLAG_CLOSE, + } + if exclusive not in allowed: + raise ProtocolError(f"illegal flag combination: {flags:#x}") + + +def parse_window_credit(frame: Frame) -> int: + if frame.flags & (FLAG_OPEN | FLAG_DATA | FLAG_CLOSE | FLAG_RESET) or not ( + frame.flags & FLAG_WINDOW + ): + raise ProtocolError("not a WINDOW frame") + if len(frame.payload) != 4: + raise ProtocolError(f"WINDOW payload must be 4 bytes, got {len(frame.payload)}") + return int.from_bytes(frame.payload, "big") + + +def parse_reset_reason(frame: Frame) -> int: + if not (frame.flags & FLAG_RESET): + raise ProtocolError("not a RESET frame") + if len(frame.payload) != 1: + raise ProtocolError(f"RESET payload must be 1 byte, got {len(frame.payload)}") + return frame.payload[0] diff --git a/think/link/mux.py b/think/link/mux.py new file mode 100644 index 000000000..20435befc --- /dev/null +++ b/think/link/mux.py @@ -0,0 +1,287 @@ +# SPDX-License-Identifier: AGPL-3.0-only +# Copyright (c) 2026 sol pbc + +"""Multiplex driver: framing-layer state + per-stream asyncio I/O. + +Bytes in from the TLS-plaintext side are fed into this module; we produce +frames to send back, and each logical stream surfaces as an +`asyncio.StreamReader`/`StreamWriter` pair that the HTTP app can drive. + +Flow-control uses the 1 MiB initial window per the spl framing spec β€” this +side grants credit as bytes drain into the app; the peer uses its granted +credit to send more data. For MVP the default "grant on every drained +chunk" policy is fine. + +Concurrent stream cap: 256 per direction. OPENs beyond cap RESET with +STREAM_LIMIT_EXCEEDED. +""" + +from __future__ import annotations + +import asyncio +from collections.abc import Awaitable, Callable +from dataclasses import dataclass, field +from typing import TYPE_CHECKING + +from .framing import ( + FLAG_CLOSE, + FLAG_DATA, + FLAG_OPEN, + FLAG_RESET, + FLAG_WINDOW, + INITIAL_WINDOW, + MAX_CONCURRENT_STREAMS, + RECOMMENDED_CHUNK, + RESET_FLOW_CONTROL_ERROR, + RESET_INTERNAL_ERROR, + RESET_PROTOCOL_ERROR, + RESET_STREAM_LIMIT_EXCEEDED, + Frame, + FrameDecoder, + ProtocolError, + build_close, + build_data, + build_open, + build_reset, + build_window, + parse_reset_reason, + parse_window_credit, +) + +if TYPE_CHECKING: + StreamHandler = Callable[[asyncio.StreamReader, "StreamWriter"], Awaitable[None]] +else: + StreamHandler = object + + +@dataclass +class _StreamState: + stream_id: int + reader: asyncio.StreamReader + reader_closed: bool = False + writer_closed: bool = False + send_credit: int = INITIAL_WINDOW + recv_credit: int = INITIAL_WINDOW + unacked_recv: int = 0 + credit_event: asyncio.Event = field(default_factory=asyncio.Event) + task: asyncio.Task[None] | None = None + + +class StreamWriter: + """Per-stream writer. Calls into the mux to emit DATA/CLOSE/RESET frames.""" + + def __init__(self, mux: Multiplexer, state: _StreamState) -> None: + self._mux = mux + self._state = state + + async def write(self, data: bytes) -> None: + if self._state.writer_closed: + raise ConnectionError(f"stream {self._state.stream_id} writer is closed") + view = memoryview(data) + while view: + chunk_len = min(len(view), RECOMMENDED_CHUNK, self._state.send_credit) + if chunk_len <= 0: + self._state.credit_event.clear() + await self._state.credit_event.wait() + continue + chunk = bytes(view[:chunk_len]) + view = view[chunk_len:] + self._state.send_credit -= chunk_len + await self._mux._emit(build_data(self._state.stream_id, chunk)) + + async def close(self) -> None: + if self._state.writer_closed: + return + self._state.writer_closed = True + await self._mux._emit(build_close(self._state.stream_id)) + + async def reset(self, reason: int = RESET_INTERNAL_ERROR) -> None: + if self._state.writer_closed and self._state.reader_closed: + return + self._state.writer_closed = True + self._state.reader_closed = True + await self._mux._emit(build_reset(self._state.stream_id, reason)) + self._state.reader.feed_eof() + self._mux._forget(self._state.stream_id) + + +class Multiplexer: + """Frame-level state. Caller pumps incoming bytes with `feed`.""" + + def __init__( + self, + send_frame: Callable[[bytes], Awaitable[None]], + handler: StreamHandler, + *, + is_listener: bool = True, + ) -> None: + """If `is_listener=True`, this side expects odd stream_ids from the peer.""" + self._decoder = FrameDecoder() + self._send_frame = send_frame + self._handler = handler + self._is_listener = is_listener + self._streams: dict[int, _StreamState] = {} + self._closed = False + + async def feed(self, plaintext: bytes) -> None: + if not plaintext: + return + self._decoder.feed(plaintext) + while True: + try: + frame = self._decoder.next() + except ProtocolError: + await self._reset_all(RESET_PROTOCOL_ERROR) + return + if frame is None: + return + await self._dispatch(frame) + + async def close(self) -> None: + if self._closed: + return + self._closed = True + for state in list(self._streams.values()): + state.reader.feed_eof() + state.writer_closed = True + if state.task and not state.task.done(): + state.task.cancel() + self._streams.clear() + + async def _dispatch(self, frame: Frame) -> None: + if frame.flags & FLAG_OPEN: + if not self._valid_peer_stream_id(frame.stream_id): + await self._emit(build_reset(frame.stream_id, RESET_PROTOCOL_ERROR)) + return + if frame.stream_id in self._streams: + await self._emit(build_reset(frame.stream_id, RESET_PROTOCOL_ERROR)) + return + if len(self._streams) >= MAX_CONCURRENT_STREAMS: + await self._emit(build_reset(frame.stream_id, RESET_STREAM_LIMIT_EXCEEDED)) + return + state = self._open_stream(frame.stream_id) + if frame.payload: + state.reader.feed_data(frame.payload) + state.recv_credit -= len(frame.payload) + if frame.flags & FLAG_CLOSE: + state.reader.feed_eof() + state.reader_closed = True + return + + maybe_state = self._streams.get(frame.stream_id) + if maybe_state is None: + await self._emit(build_reset(frame.stream_id, RESET_PROTOCOL_ERROR)) + return + state = maybe_state + + if frame.flags & FLAG_DATA: + if len(frame.payload) > state.recv_credit: + await self._emit(build_reset(frame.stream_id, RESET_FLOW_CONTROL_ERROR)) + self._terminate(state) + return + state.reader.feed_data(frame.payload) + state.recv_credit -= len(frame.payload) + state.unacked_recv += len(frame.payload) + if state.unacked_recv >= INITIAL_WINDOW // 2: + grant = state.unacked_recv + state.recv_credit += grant + state.unacked_recv = 0 + await self._emit(build_window(frame.stream_id, grant)) + if frame.flags & FLAG_CLOSE: + state.reader.feed_eof() + state.reader_closed = True + if state.writer_closed: + self._forget(frame.stream_id) + if frame.flags & FLAG_WINDOW: + try: + credit = parse_window_credit(frame) + except ProtocolError: + await self._emit(build_reset(frame.stream_id, RESET_PROTOCOL_ERROR)) + self._terminate(state) + return + state.send_credit += credit + state.credit_event.set() + if frame.flags & FLAG_RESET: + try: + _ = parse_reset_reason(frame) + except ProtocolError: + pass + state.reader.feed_eof() + self._terminate(state) + + def _open_stream(self, stream_id: int) -> _StreamState: + reader = asyncio.StreamReader() + state = _StreamState(stream_id=stream_id, reader=reader) + state.credit_event.set() + self._streams[stream_id] = state + writer = StreamWriter(self, state) + + async def runner() -> None: + try: + await self._handler(reader, writer) + except Exception: + await writer.reset(RESET_INTERNAL_ERROR) + finally: + if not state.writer_closed: + try: + await writer.close() + except Exception: + pass + self._forget(stream_id) + + state.task = asyncio.create_task(runner(), name=f"link-stream-{stream_id}") + return state + + def _terminate(self, state: _StreamState) -> None: + state.writer_closed = True + state.reader_closed = True + if state.task and not state.task.done(): + state.task.cancel() + self._forget(state.stream_id) + + def _forget(self, stream_id: int) -> None: + self._streams.pop(stream_id, None) + + def _valid_peer_stream_id(self, stream_id: int) -> bool: + if stream_id == 0: + return False + return (stream_id % 2 == 1) if self._is_listener else (stream_id % 2 == 0) + + async def _emit(self, frame: Frame) -> None: + if self._closed: + return + try: + encoded = frame.encode() + except ProtocolError: + return + await self._send_frame(encoded) + + async def _reset_all(self, reason: int) -> None: + for state in list(self._streams.values()): + await self._emit(build_reset(state.stream_id, reason)) + self._terminate(state) + + async def open_stream( + self, + initial_payload: bytes = b"", + ) -> tuple[asyncio.StreamReader, StreamWriter]: + next_id = self._next_local_stream_id() + reader = asyncio.StreamReader() + state = _StreamState(stream_id=next_id, reader=reader) + state.credit_event.set() + self._streams[next_id] = state + writer = StreamWriter(self, state) + frame = build_open(next_id, initial_payload) + if initial_payload: + state.send_credit -= len(initial_payload) + await self._emit(frame) + return reader, writer + + def _next_local_stream_id(self) -> int: + start = 2 if self._is_listener else 1 + cur = start + while cur in self._streams: + cur += 2 + if cur > 0xFFFFFFFF: + raise RuntimeError("stream_id space exhausted") + return cur diff --git a/think/link/nonces.py b/think/link/nonces.py new file mode 100644 index 000000000..27d9d7d08 --- /dev/null +++ b/think/link/nonces.py @@ -0,0 +1,176 @@ +# SPDX-License-Identifier: AGPL-3.0-only +# Copyright (c) 2026 sol pbc + +"""Pair-nonce store β€” shared between the CLI pair flow and convey's pair route. + +`sol call link pair` mints a nonce and writes it to disk; convey's +`POST /link/pair` reads on every incoming pair request, garbage-collects +expired entries, and enforces single-use semantics. The file is the IPC +channel between the two processes β€” simple, durable across crashes, no +extra port. + +Consumers treat the file as opaque and call only into the methods here. +Atomic replaces guard against partial writes. +""" + +from __future__ import annotations + +import fcntl +import json +import os +import time +from dataclasses import dataclass +from pathlib import Path + +NONCE_TTL_SECONDS = 300 # 5 min per the spl pairing spec. + + +@dataclass(frozen=True) +class Nonce: + value: str + device_label: str + issued_at: int + expires_at: int + used: bool + + +class NonceStore: + def __init__(self, path: Path) -> None: + self._path = path + + @property + def path(self) -> Path: + return self._path + + def add( + self, + nonce: str, + device_label: str, + *, + now: int | None = None, + ) -> Nonce: + ts = now if now is not None else int(time.time()) + entry = Nonce( + value=nonce, + device_label=device_label, + issued_at=ts, + expires_at=ts + NONCE_TTL_SECONDS, + used=False, + ) + with self._locked_read_write() as entries: + self._gc_locked(entries, ts) + entries[nonce] = entry + self._write_locked(entries) + return entry + + def consume(self, value: str, *, now: int | None = None) -> Nonce | None: + """Mark a nonce used if valid. Single-use enforced atomically.""" + ts = now if now is not None else int(time.time()) + with self._locked_read_write() as entries: + self._gc_locked(entries, ts) + entry = entries.get(value) + if entry is None: + return None + if entry.used or entry.expires_at <= ts: + return None + entry = Nonce( + value=entry.value, + device_label=entry.device_label, + issued_at=entry.issued_at, + expires_at=entry.expires_at, + used=True, + ) + entries[value] = entry + self._write_locked(entries) + return entry + + def peek(self, value: str) -> Nonce | None: + entries = self._read() + return entries.get(value) + + def snapshot(self) -> list[Nonce]: + return list(self._read().values()) + + def gc(self, *, now: int | None = None) -> int: + """Remove expired entries. Returns count removed.""" + ts = now if now is not None else int(time.time()) + with self._locked_read_write() as entries: + before = len(entries) + self._gc_locked(entries, ts) + if len(entries) != before: + self._write_locked(entries) + return before - len(entries) + + def _read(self) -> dict[str, Nonce]: + if not self._path.exists(): + return {} + try: + raw = json.loads(self._path.read_text("utf-8")) + except (json.JSONDecodeError, OSError): + return {} + out: dict[str, Nonce] = {} + if isinstance(raw, list): + for item in raw: + if not isinstance(item, dict): + continue + val = item.get("value") + if not isinstance(val, str): + continue + out[val] = Nonce( + value=val, + device_label=str(item.get("device_label", "")), + issued_at=int(item.get("issued_at", 0)), + expires_at=int(item.get("expires_at", 0)), + used=bool(item.get("used", False)), + ) + return out + + def _write_locked(self, entries: dict[str, Nonce]) -> None: + self._path.parent.mkdir(parents=True, exist_ok=True) + payload = [ + { + "value": e.value, + "device_label": e.device_label, + "issued_at": e.issued_at, + "expires_at": e.expires_at, + "used": e.used, + } + for e in entries.values() + ] + tmp = self._path.with_suffix(self._path.suffix + ".tmp") + with open(tmp, "w", encoding="utf-8") as f: + json.dump(payload, f, indent=2) + f.write("\n") + f.flush() + os.fsync(f.fileno()) + os.replace(tmp, self._path) + + def _gc_locked(self, entries: dict[str, Nonce], now: int) -> None: + to_drop = [k for k, e in entries.items() if e.used or e.expires_at <= now] + for k in to_drop: + del entries[k] + + class _Guard: + def __init__(self, store: NonceStore) -> None: + self.store = store + self.lock_path = store._path.with_suffix(store._path.suffix + ".lock") + self.fd = -1 + + def __enter__(self) -> dict[str, Nonce]: + self.lock_path.parent.mkdir(parents=True, exist_ok=True) + self.fd = os.open( + str(self.lock_path), + os.O_RDWR | os.O_CREAT, + 0o600, + ) + fcntl.flock(self.fd, fcntl.LOCK_EX) + return self.store._read() + + def __exit__(self, *_: object) -> None: + try: + fcntl.flock(self.fd, fcntl.LOCK_UN) + finally: + os.close(self.fd) + + def _locked_read_write(self) -> NonceStore._Guard: + return NonceStore._Guard(self) diff --git a/think/link/paths.py b/think/link/paths.py new file mode 100644 index 000000000..ad0cf3732 --- /dev/null +++ b/think/link/paths.py @@ -0,0 +1,164 @@ +# SPDX-License-Identifier: AGPL-3.0-only +# Copyright (c) 2026 sol pbc + +"""journal/link/ path resolution + service state I/O. + +All link-service state lives under `journal/link/`: + + journal/link/ + ca/ + cert.pem world-readable local CA cert + private.pem mode 0600 β€” filesystem-perms-only protection + authorized_clients.json paired-device ledger (mtime-reloaded) + tokens/ + account.json cached account_token from /enroll/home + nonces.json pair-ceremony nonces (5-min TTL, single-use) + state.json instance_id + home_label (generated on first run) + +`journal/link/` is a narrow exception to the "memories live in +day/stream/segment/" rule β€” this is config, not memory, scoped to this +one service (see cpo/strategy/journal-memory-structure.md). +""" + +from __future__ import annotations + +import json +import os +import uuid +from dataclasses import dataclass +from pathlib import Path + +from think.utils import get_journal + +# Production spl-relay endpoint. Single source of truth β€” self-hosters +# override via SOL_LINK_RELAY_URL env var. When CTO wires +# spl.solpbc.org as DNS front, update this constant. +DEFAULT_RELAY_URL = "https://spl-relay-staging.jer-3f2.workers.dev" + + +def link_root() -> Path: + """`journal/link/` β€” auto-created.""" + root = Path(get_journal()) / "link" + root.mkdir(parents=True, exist_ok=True) + return root + + +def ca_dir() -> Path: + d = link_root() / "ca" + d.mkdir(parents=True, exist_ok=True) + return d + + +def authorized_clients_path() -> Path: + return link_root() / "authorized_clients.json" + + +def tokens_dir() -> Path: + d = link_root() / "tokens" + d.mkdir(parents=True, exist_ok=True) + return d + + +def account_token_path() -> Path: + return tokens_dir() / "account.json" + + +def nonces_path() -> Path: + return link_root() / "nonces.json" + + +def state_path() -> Path: + return link_root() / "state.json" + + +def relay_url() -> str: + """Resolve the spl-relay endpoint. + + Precedence: SOL_LINK_RELAY_URL env var > journal config `link.relay_url` > + DEFAULT_RELAY_URL constant. Self-hosters override one-field; production + users get the default. + """ + env = os.environ.get("SOL_LINK_RELAY_URL", "").strip() + if env: + return env.rstrip("/") + try: + from think.utils import get_config + + cfg = get_config() + link_cfg = cfg.get("link") if isinstance(cfg, dict) else None + if isinstance(link_cfg, dict): + url = link_cfg.get("relay_url") + if isinstance(url, str) and url.strip(): + return url.strip().rstrip("/") + except Exception: + pass + return DEFAULT_RELAY_URL + + +@dataclass +class LinkState: + """Service identity β€” the values spl-relay binds an account_token to. + + Persisted to `journal/link/state.json`; generated on first run. + """ + + instance_id: str + home_label: str + + @classmethod + def load_or_create(cls, *, default_label: str = "solstone") -> LinkState: + path = state_path() + if path.exists(): + try: + raw = json.loads(path.read_text("utf-8")) + iid = raw.get("instance_id") + label = raw.get("home_label") or default_label + if isinstance(iid, str) and iid: + return cls(instance_id=iid, home_label=label) + except (json.JSONDecodeError, OSError): + pass + state = cls(instance_id=str(uuid.uuid4()), home_label=default_label) + state.save() + return state + + def save(self) -> None: + path = state_path() + path.parent.mkdir(parents=True, exist_ok=True) + tmp = path.with_suffix(".json.tmp") + with open(tmp, "w", encoding="utf-8") as f: + json.dump( + {"instance_id": self.instance_id, "home_label": self.home_label}, + f, + indent=2, + ) + f.write("\n") + f.flush() + os.fsync(f.fileno()) + os.replace(tmp, path) + + +def load_account_token() -> str | None: + """Read the cached /enroll/home account token, or None.""" + path = account_token_path() + if not path.exists(): + return None + try: + raw = json.loads(path.read_text("utf-8")) + token = raw.get("account_token") + return token if isinstance(token, str) and token else None + except (json.JSONDecodeError, OSError): + return None + + +def save_account_token(token: str) -> None: + """Persist the account token atomically with mode 0600.""" + path = account_token_path() + path.parent.mkdir(parents=True, exist_ok=True) + tmp = path.with_suffix(".json.tmp") + with open(tmp, "w", encoding="utf-8") as f: + json.dump({"account_token": token}, f, indent=2) + f.write("\n") + f.flush() + os.fsync(f.fileno()) + os.chmod(tmp, 0o600) + os.replace(tmp, path) diff --git a/think/link/relay_client.py b/think/link/relay_client.py new file mode 100644 index 000000000..356c5bfca --- /dev/null +++ b/think/link/relay_client.py @@ -0,0 +1,390 @@ +# SPDX-License-Identifier: AGPL-3.0-only +# Copyright (c) 2026 sol pbc + +"""Listen WS + tunnel WS orchestrator. + +On startup: + 1. If no account_token stored, POST /enroll/home to mint one (idempotent). + 2. Open listen WS to spl-relay with the account token. + 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. + 4. On disconnect, reconnect with exponential backoff (1s β†’ 60s, Β±25%). + +All WebSocket I/O uses the `websockets` library in asyncio mode. The TLS +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 +eligible for logging, and everything emitted to callosum is the same +rendezvous-only subset. +""" + +from __future__ import annotations + +import asyncio +import contextlib +import json +import logging +import random +import ssl +import urllib.parse +from typing import Any, Callable + +import websockets +from websockets.asyncio.client import ClientConnection as _WsConnection +from websockets.exceptions import ConnectionClosed + +from .auth import AuthorizedClients +from .ca import LoadedCa +from .mux import Multiplexer, StreamWriter +from .tls_adapter import ( + TlsError, + build_server_context, + drive_tls, + issue_server_cert, + new_server, +) +from .wsgi_bridge import serve_request + +log = logging.getLogger("link.relay_client") + +_RECONNECT_MIN = 1.0 +_RECONNECT_MAX = 60.0 + +# Callosum event emitter signature β€” the service layer injects a closure +# that forwards to the supervisor's callosum connection. None in tests. +CallosumEmit = Callable[[str, dict[str, Any]], None] + + +class RelayClient: + def __init__( + self, + *, + instance_id: str, + home_label: str, + relay_endpoint: str, + account_token: str | None, + 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 + self._home_label = home_label + self._relay_endpoint = relay_endpoint.rstrip("/") + self._account_token = account_token + 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" + self._tunnels: dict[str, asyncio.Task[None]] = {} + + server_cert, server_key_pem = issue_server_cert( + ca, + common_name=f"solstone link ({home_label})", + ) + self._tls_ctx = build_server_context( + ca=ca, + server_cert=server_cert, + server_key=server_key_pem, + authorized=authorized, + ) + + @property + def listen_state(self) -> str: + """One of: offline, connecting, online, reconnecting, not-enrolled.""" + return self._listen_state + + @property + def account_token(self) -> str | None: + return self._account_token + + def active_tunnel_count(self) -> int: + return sum(1 for t in self._tunnels.values() if not t.done()) + + async def enroll_if_needed(self) -> None: + if self._account_token: + return + endpoint = f"{self._relay_endpoint}/enroll/home" + body = { + "instance_id": self._instance_id, + "ca_pubkey": self._ca.pubkey_spki_pem, + "home_label": self._home_label, + } + log.info("enrolling home with relay at %s", endpoint) + result = await _post_json(endpoint, body) + token = result.get("account_token") + if not isinstance(token, str) or not token: + raise RuntimeError("relay returned no account_token") + self._account_token = token + self._on_account_token(token) + self._emit("enrolled", {"instance_id": self._instance_id}) + + async def run(self) -> None: + self._running = True + delay = _RECONNECT_MIN + while self._running: + try: + await self._run_once() + delay = _RECONNECT_MIN + except ConnectionClosed as exc: + log.warning("listen WS closed: code=%s reason=%s", exc.code, exc.reason) + except Exception as exc: # noqa: BLE001 + log.exception("listen loop error: %s", exc) + if not self._running: + return + self._listen_state = "reconnecting" + self._emit("disconnect", {}) + jitter = delay * 0.25 + wait = delay + random.uniform(-jitter, jitter) # noqa: S311 + log.info("reconnecting in %.1fs", wait) + await asyncio.sleep(wait) + delay = min(_RECONNECT_MAX, delay * 2.0) + self._listen_state = "offline" + + async def stop(self) -> None: + self._running = False + self._listen_state = "offline" + for task in list(self._tunnels.values()): + task.cancel() + if self._tunnels: + await asyncio.gather( + *( + asyncio.shield(t) + for t in self._tunnels.values() + ), + return_exceptions=True, + ) + self._tunnels.clear() + + async def _run_once(self) -> None: + await self.enroll_if_needed() + self._listen_state = "connecting" + self._emit("connecting", {}) + assert self._account_token is not None + listen_url = self._url_for("/session/listen", token=self._account_token) + headers = _auth_header(self._account_token) + log.info("opening listen WS to %s", _redact(listen_url)) + async with websockets.connect( + listen_url, + additional_headers=headers, + max_size=None, + ) as ws: + self._listen_state = "online" + self._emit("connected", {}) + log.info("listen WS open β€” waiting for incoming") + async for message in ws: + data = _parse_control(message) + if data is None: + continue + tunnel_id = data.get("tunnel_id") + if data.get("type") == "incoming" and isinstance(tunnel_id, str): + log.info("incoming tunnel_id=%s", tunnel_id) + self._emit("tunnel_pair", {"tunnel_id": tunnel_id}) + task = asyncio.create_task( + self._handle_tunnel(tunnel_id), + name=f"link-tunnel-{tunnel_id}", + ) + self._tunnels[tunnel_id] = task + task.add_done_callback(lambda _t, tid=tunnel_id: self._tunnels.pop(tid, None)) + + async def _handle_tunnel(self, tunnel_id: str) -> None: + assert self._account_token is not None + url = self._url_for(f"/tunnel/{tunnel_id}", token=self._account_token) + headers = _auth_header(self._account_token) + try: + async with websockets.connect( + url, + additional_headers=headers, + max_size=None, + ) as ws: + await self._pump_tunnel(ws, tunnel_id) + except ConnectionClosed as exc: + log.info("tunnel %s closed: code=%s reason=%s", tunnel_id, exc.code, exc.reason) + except TlsError as exc: + log.info("tunnel %s TLS rejected: %s", tunnel_id, exc) + except Exception as exc: # noqa: BLE001 + log.exception("tunnel %s error: %s", tunnel_id, exc) + finally: + self._emit("tunnel_close", {"tunnel_id": tunnel_id}) + + async def _pump_tunnel(self, ws: _WsConnection, tunnel_id: str) -> None: + tls = new_server(self._tls_ctx) + send_queue: asyncio.Queue[bytes] = asyncio.Queue() + fingerprint_touched = False + + async def send_frame(frame: bytes) -> None: + send_queue.put_nowait(frame) + + async def handle_stream( + 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() + 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, + ) + + mux = Multiplexer(send_frame, handle_stream, is_listener=True) + + async def ws_reader() -> None: + nonlocal fingerprint_touched + try: + async for frame in ws: + inbound = frame if isinstance(frame, bytes) else frame.encode("utf-8") + outbound, plaintext = drive_tls(tls, inbound=inbound) + if outbound: + await ws.send(outbound) + if plaintext: + await mux.feed(plaintext) + if ( + not fingerprint_touched + and tls.handshake_done + and tls.peer_fingerprint + ): + fingerprint_touched = True + self._authorized.touch_last_seen(tls.peer_fingerprint) + self._emit( + "last_seen", + { + "fingerprint": tls.peer_fingerprint, + "tunnel_id": tunnel_id, + }, + ) + await _drain_send_queue(tls, ws, send_queue) + except ConnectionClosed: + return + + async def app_writer() -> None: + try: + while True: + data = await send_queue.get() + outbound = _encrypt(tls, data) + if outbound: + await ws.send(outbound) + except ConnectionClosed: + return + + reader_task = asyncio.create_task(ws_reader(), name=f"ws-reader-{tunnel_id}") + writer_task = asyncio.create_task(app_writer(), name=f"app-writer-{tunnel_id}") + try: + await reader_task + finally: + writer_task.cancel() + with contextlib.suppress(asyncio.CancelledError): + await writer_task + await mux.close() + + def _url_for(self, path: str, *, token: str | None = None) -> str: + base = _to_ws(self._relay_endpoint) + path + q = {"instance": self._instance_id} + if token: + q["token"] = token + return base + "?" + urllib.parse.urlencode(q) + + +def _encrypt(tls: Any, plaintext: bytes) -> bytes: + outbound, _ = drive_tls(tls, inbound=b"", plaintext_out=plaintext) + return outbound + + +async def _drain_send_queue( + tls: Any, + ws: _WsConnection, + queue: asyncio.Queue[bytes], +) -> None: + drained: list[bytes] = [] + while not queue.empty(): + try: + drained.append(queue.get_nowait()) + except asyncio.QueueEmpty: + break + if not drained: + return + for chunk in drained: + outbound = _encrypt(tls, chunk) + if outbound: + await ws.send(outbound) + + +async def _post_json(url: str, body: dict[str, Any]) -> dict[str, Any]: + import urllib.request + + def sync() -> dict[str, Any]: + data = json.dumps(body).encode("utf-8") + req = urllib.request.Request( # noqa: S310 β€” URL scheme validated below + url, + data=data, + headers={ + "content-type": "application/json", + "user-agent": "solstone-link/0.1", + }, + method="POST", + ) + ctx = ssl.create_default_context() + if url.startswith("http://"): + ctx = None # type: ignore[assignment] # local dev + elif not url.startswith("https://"): + raise ValueError(f"unsupported url scheme: {url!r}") + with urllib.request.urlopen(req, context=ctx, timeout=30) as resp: # noqa: S310 + payload = resp.read() + parsed: dict[str, Any] = json.loads(payload) + return parsed + + return await asyncio.to_thread(sync) + + +def _to_ws(endpoint: str) -> str: + if endpoint.startswith("http://"): + return "ws://" + endpoint[len("http://") :] + if endpoint.startswith("https://"): + return "wss://" + endpoint[len("https://") :] + return endpoint + + +def _parse_control(message: str | bytes) -> dict[str, Any] | None: + if isinstance(message, bytes): + try: + text = message.decode("utf-8") + except UnicodeDecodeError: + return None + else: + text = message + try: + out = json.loads(text) + except json.JSONDecodeError: + return None + return out if isinstance(out, dict) else None + + +def _auth_header(token: str | None) -> dict[str, str]: + return {"Authorization": f"Bearer {token}"} if token else {} + + +def _redact(url: str) -> str: + parsed = urllib.parse.urlparse(url) + q = urllib.parse.parse_qs(parsed.query) + if "token" in q: + q["token"] = [""] + new_query = urllib.parse.urlencode(q, doseq=True) + return urllib.parse.urlunparse(parsed._replace(query=new_query)) diff --git a/think/link/service.py b/think/link/service.py new file mode 100644 index 000000000..c69a00523 --- /dev/null +++ b/think/link/service.py @@ -0,0 +1,145 @@ +# SPDX-License-Identifier: AGPL-3.0-only +# Copyright (c) 2026 sol pbc + +"""link service runtime. + +Registered with solstone's supervisor via `sol.py` COMMANDS (see `sol link`); +the supervisor launches this as a subprocess alongside callosum, cortex, +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. + +Exits on SIGINT/SIGTERM with a clean close of the listen WS and all +in-flight tunnel WSes. + +Callosum events are emitted on the `link` tract: + enrolled first-run account-token mint + connecting opening listen WS + connected listen WS open (service is reachable) + disconnect listen WS closed (about to reconnect) + tunnel_pair incoming tunnel (paired device dialed in) + tunnel_close tunnel closed + last_seen paired fingerprint completed TLS handshake +""" + +from __future__ import annotations + +import asyncio +import logging +import signal +from typing import Any + +from think.callosum import CallosumConnection + +from .auth import AuthorizedClients +from .ca import load_or_generate_ca +from .paths import ( + LinkState, + authorized_clients_path, + ca_dir, + load_account_token, + relay_url, + save_account_token, +) +from .relay_client import RelayClient + +log = logging.getLogger("link.service") + + +async def run_service() -> None: + """Build the RelayClient and run it until signaled.""" + state = LinkState.load_or_create() + ca = load_or_generate_ca(ca_dir()) + authorized = AuthorizedClients(authorized_clients_path()) + token = load_account_token() + + wsgi_app = _build_convey_wsgi() + + callosum = CallosumConnection() + callosum.start() + + def emit(event: str, fields: dict[str, Any]) -> None: + try: + callosum.emit("link", event, **fields) + except Exception: + log.debug("callosum emit failed", exc_info=True) + + client = RelayClient( + instance_id=state.instance_id, + home_label=state.home_label, + relay_endpoint=relay_url(), + account_token=token, + on_account_token=save_account_token, + ca=ca, + authorized=authorized, + wsgi_app=wsgi_app, + callosum_emit=emit, + ) + + stop_event = asyncio.Event() + loop = asyncio.get_running_loop() + for sig in (signal.SIGINT, signal.SIGTERM): + with _suppress_not_implemented(): + loop.add_signal_handler(sig, stop_event.set) + + run_task = asyncio.create_task(client.run(), name="link-relay-client") + try: + await stop_event.wait() + finally: + log.info("link service stopping") + await client.stop() + run_task.cancel() + try: + await run_task + except asyncio.CancelledError: + pass + 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.""" + + def __enter__(self) -> None: + return None + + def __exit__(self, exc_type: Any, _exc: Any, _tb: Any) -> bool: + return exc_type is NotImplementedError + + +def main() -> None: + """CLI entry point for `sol link` β€” starts the service.""" + import argparse + + from think.utils import require_solstone, setup_cli + + parser = argparse.ArgumentParser(description="solstone link tunnel service") + args = setup_cli(parser) + require_solstone() + + logging.basicConfig( + level=logging.INFO if not args.verbose else logging.DEBUG, + format="%(asctime)s %(name)s %(levelname)s %(message)s", + ) + + try: + asyncio.run(run_service()) + except KeyboardInterrupt: + log.info("link service interrupted") + + +if __name__ == "__main__": + main() diff --git a/think/link/tls_adapter.py b/think/link/tls_adapter.py new file mode 100644 index 000000000..559c158a9 --- /dev/null +++ b/think/link/tls_adapter.py @@ -0,0 +1,199 @@ +# SPDX-License-Identifier: AGPL-3.0-only +# Copyright (c) 2026 sol pbc + +"""pyOpenSSL memory-BIO adapter: TLS 1.3 over non-socket byte streams. + +The link service runs TLS inside the spl-relay tunnel, which is an opaque +WebSocket β€” not a real socket. pyOpenSSL's `SSL.Connection` supports +memory-BIO mode: the caller pushes ciphertext in with `bio_write`, pulls +ciphertext out with `bio_read`, and reads/writes plaintext with +`recv`/`send`. + +This module wraps that state machine with a byte-oriented API the +relay-client and mux loops drive. It installs the pinned verify callback β€” +the load-bearing reason we use pyOpenSSL and not stdlib `ssl` (stdlib +doesn't expose a handshake-time callback that can reject a cert with a +clean TLS alert). +""" + +from __future__ import annotations + +from dataclasses import dataclass + +from cryptography import x509 +from cryptography.hazmat.primitives import serialization +from OpenSSL import SSL, crypto + +from .auth import AuthorizedClients +from .ca import LoadedCa + + +class TlsError(RuntimeError): + """Raised when the TLS handshake is aborted (e.g., fingerprint rejected).""" + + +@dataclass +class TlsServerState: + conn: SSL.Connection + handshake_done: bool = False + peer_fingerprint: str | None = None + + +def build_server_context( + ca: LoadedCa, + server_cert: x509.Certificate, + server_key: bytes, + authorized: AuthorizedClients, +) -> SSL.Context: + """Build a TLS 1.3 server context with the pinned verify callback.""" + ctx = SSL.Context(SSL.TLS_METHOD) + ctx.set_min_proto_version(SSL.TLS1_3_VERSION) + ctx.set_max_proto_version(SSL.TLS1_3_VERSION) + ctx.use_certificate( + crypto.X509.from_cryptography(server_cert), + ) + ctx.use_privatekey(crypto.load_privatekey(crypto.FILETYPE_PEM, server_key)) + ctx.add_extra_chain_cert(crypto.X509.from_cryptography(ca.cert)) + store = ctx.get_cert_store() + assert store is not None, "pyOpenSSL context must expose a cert store" + store.add_cert(crypto.X509.from_cryptography(ca.cert)) + + def verify_cb( + _conn: SSL.Connection, + cert: crypto.X509, + _errno: int, + depth: int, + preverify_ok: int, + ) -> bool: + if not preverify_ok: + return False + if depth != 0: + return True + der = cert.to_cryptography().public_bytes(serialization.Encoding.DER) + import hashlib + + fp = f"sha256:{hashlib.sha256(der).hexdigest()}" + return authorized.is_authorized(fp) + + ctx.set_verify( + SSL.VERIFY_PEER | SSL.VERIFY_FAIL_IF_NO_PEER_CERT, + verify_cb, + ) + return ctx + + +def new_server(ctx: SSL.Context) -> TlsServerState: + """Fresh memory-BIO connection in accept state.""" + conn = SSL.Connection(ctx, None) + conn.set_accept_state() + return TlsServerState(conn=conn) + + +def drive_tls( + state: TlsServerState, + *, + inbound: bytes, + plaintext_out: bytes = b"", +) -> tuple[bytes, bytes]: + """Push ciphertext in + plaintext out; return (ciphertext_to_send, plaintext_received).""" + if inbound: + state.conn.bio_write(inbound) + if plaintext_out: + try: + state.conn.send(plaintext_out) + except SSL.WantReadError: + pass + + if not state.handshake_done: + try: + state.conn.do_handshake() + state.handshake_done = True + peer = state.conn.get_peer_certificate() + if peer is not None: + import hashlib + + der = peer.to_cryptography().public_bytes( + serialization.Encoding.DER, + ) + state.peer_fingerprint = f"sha256:{hashlib.sha256(der).hexdigest()}" + except SSL.WantReadError: + pass + except SSL.Error as exc: + raise TlsError(f"handshake failed: {exc}") from exc + + plaintext_in = bytearray() + if state.handshake_done: + while True: + try: + chunk = state.conn.recv(16 * 1024) + except SSL.WantReadError: + break + except SSL.ZeroReturnError: + break + if not chunk: + break + plaintext_in.extend(chunk) + + outbound = bytearray() + while True: + try: + chunk = state.conn.bio_read(16 * 1024) + except SSL.WantReadError: + break + if not chunk: + break + outbound.extend(chunk) + return bytes(outbound), bytes(plaintext_in) + + +def issue_server_cert( + ca: LoadedCa, + common_name: str = "solstone link", +) -> tuple[x509.Certificate, bytes]: + """Mint a server cert (signed by the CA) + its PEM-encoded private key. + + Regenerated on each start β€” server-side TLS material doesn't need to + survive restarts since the mobile pins the *CA* fingerprint, not the + server cert. + """ + import datetime as dt + + from cryptography.hazmat.primitives import hashes + from cryptography.hazmat.primitives.asymmetric import ec + from cryptography.x509.oid import NameOID + + key = ec.generate_private_key(ec.SECP256R1()) + now = dt.datetime.now(dt.UTC) + cert = ( + x509.CertificateBuilder() + .subject_name( + x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, common_name)]), + ) + .issuer_name(ca.cert.subject) + .public_key(key.public_key()) + .serial_number(x509.random_serial_number()) + .not_valid_before(now - dt.timedelta(minutes=5)) + .not_valid_after(now + dt.timedelta(days=30)) + .add_extension(x509.BasicConstraints(ca=False, path_length=None), critical=True) + .add_extension( + x509.ExtendedKeyUsage([x509.oid.ExtendedKeyUsageOID.SERVER_AUTH]), + critical=False, + ) + .sign(ca.private_key, hashes.SHA256()) + ) + key_pem = key.private_bytes( + serialization.Encoding.PEM, + serialization.PrivateFormat.PKCS8, + serialization.NoEncryption(), + ) + return cert, key_pem + + +__all__ = [ + "TlsError", + "TlsServerState", + "build_server_context", + "drive_tls", + "issue_server_cert", + "new_server", +] diff --git a/think/link/wsgi_bridge.py b/think/link/wsgi_bridge.py new file mode 100644 index 000000000..d7c2cf720 --- /dev/null +++ b/think/link/wsgi_bridge.py @@ -0,0 +1,350 @@ +# SPDX-License-Identifier: AGPL-3.0-only +# Copyright (c) 2026 sol pbc + +"""HTTP/1.1 ⇄ WSGI bridge for the link tunnel. + +Each tunnel stream is one HTTP request/response exchange (no keep-alive). +This module: + + 1. Parses an HTTP/1.1 request from the per-stream byte reader + 2. Builds a WSGI environ dict + 3. Invokes convey's real Flask app + 4. Writes the response (status + headers + body) back to the stream writer + +Streaming responses (SSE, chunked) are supported β€” the WSGI iterable is +pumped chunk-by-chunk so the client sees events as the app produces them. + +Request-body size is capped at 64 MiB. Privacy invariant: NO request or +response bytes are ever logged. Only error counts, byte totals, and +status codes are eligible for logging (see `metadata`), never payloads. +""" + +from __future__ import annotations + +import asyncio +import io +import logging +import urllib.parse +from dataclasses import dataclass +from typing import Any, Callable + +log = logging.getLogger("link.wsgi") + +MAX_REQUEST_BODY = 64 * 1024 * 1024 +WSGI_SERVER_NAME = "solstone-link" + + +@dataclass +class ExchangeMetadata: + """Per-request rendezvous metadata β€” safe to log (no payload).""" + + method: str + path: str + status: int | None = None + request_bytes: int = 0 + response_bytes: int = 0 + stream_id: int | None = None + + +async def serve_request( + reader: asyncio.StreamReader, + writer: Any, # think.link.mux.StreamWriter + wsgi_app: Callable[[dict[str, Any], Callable[..., Any]], Any], + *, + peer_fingerprint: str | None = None, + tunnel_id: str | None = None, + stream_id: int | None = None, +) -> ExchangeMetadata: + """Read one HTTP/1.1 request; dispatch to `wsgi_app`; write response. + + Returns rendezvous metadata describing what happened (for callosum + events + debug logs). Never returns payload bytes. + """ + meta = ExchangeMetadata(method="-", path="-", stream_id=stream_id) + try: + request = await _read_request(reader) + except _BadRequest as exc: + log.debug("tunnel %s stream %s: bad request: %s", tunnel_id, stream_id, exc) + await _write_simple(writer, 400, "bad request", b"bad request\n") + meta.status = 400 + meta.response_bytes = _byte_count_for_simple(b"bad request\n") + return meta + except asyncio.IncompleteReadError: + log.debug("tunnel %s stream %s: incomplete request", tunnel_id, stream_id) + return meta + + meta.method = request.method + meta.path = request.path + meta.request_bytes = len(request.body) + + environ = _build_environ( + request, + peer_fingerprint=peer_fingerprint, + tunnel_id=tunnel_id, + ) + + response_state = _ResponseState() + + def start_response( + status: str, + headers: list[tuple[str, str]], + exc_info: Any = None, + ) -> Callable[[bytes], None]: + response_state.status_line = status + response_state.headers = list(headers) + # WSGI returns a write() callable, but we use the iterable instead. + def write(_data: bytes) -> None: + raise RuntimeError("write() callable not supported; return iterable") + + return write + + # Run the WSGI app in a thread β€” it may block on DB, file I/O, etc. + loop = asyncio.get_running_loop() + try: + result = await loop.run_in_executor( + None, lambda: wsgi_app(environ, start_response) + ) + except Exception: + log.exception( + "tunnel %s stream %s: wsgi app raised", tunnel_id, stream_id + ) + await _write_simple(writer, 500, "internal server error", b"internal server error\n") + meta.status = 500 + meta.response_bytes = _byte_count_for_simple(b"internal server error\n") + return meta + + if response_state.status_line is None: + log.warning( + "tunnel %s stream %s: wsgi app returned without calling start_response", + tunnel_id, + stream_id, + ) + await _write_simple(writer, 500, "internal server error", b"missing response\n") + meta.status = 500 + return meta + + # Parse numeric status. + try: + code_str, reason = response_state.status_line.split(" ", 1) + code = int(code_str) + except (ValueError, IndexError): + code = 500 + reason = "internal server error" + meta.status = code + + # Detect if the response is Transfer-Encoding: chunked or uses Content-Length. + headers_map = {k.lower(): v for k, v in response_state.headers} + is_chunked = headers_map.get("transfer-encoding", "").lower() == "chunked" + + # Send status + headers. + sent = await _write_status_headers(writer, code, reason, response_state.headers) + meta.response_bytes += sent + + # Stream body. Iterate in a thread (WSGI iterables can block). + iterator = iter(result) if not isinstance(result, (bytes, bytearray)) else iter([bytes(result)]) + + async def next_chunk() -> bytes | None: + def _pull() -> bytes | None: + try: + return next(iterator) + except StopIteration: + return None + + return await loop.run_in_executor(None, _pull) + + try: + while True: + chunk = await next_chunk() + if chunk is None: + break + if not chunk: + continue + if is_chunked: + # The WSGI app is responsible for emitting valid chunked + # framing (Flask's stream_with_context does this). Pass + # through unchanged. + await writer.write(chunk) + else: + await writer.write(chunk) + meta.response_bytes += len(chunk) + finally: + # WSGI requires closing iterables that have a close() method. + close = getattr(result, "close", None) + if callable(close): + try: + await loop.run_in_executor(None, close) + except Exception: + log.debug("wsgi close() raised", exc_info=True) + + return meta + + +@dataclass +class _ResponseState: + status_line: str | None = None + headers: list[tuple[str, str]] | None = None + + +@dataclass +class _Request: + method: str + path: str + query: str + headers: list[tuple[str, str]] + body: bytes + + +class _BadRequest(Exception): + pass + + +async def _read_request(reader: asyncio.StreamReader) -> _Request: + """Parse HTTP/1.1 request line + headers + body (Content-Length only).""" + raw_line = await reader.readline() + if not raw_line: + raise asyncio.IncompleteReadError(b"", None) + line = raw_line.decode("latin-1").rstrip("\r\n") + parts = line.split(" ", 2) + if len(parts) != 3: + raise _BadRequest(f"bad request line: {line!r}") + method, target, _version = parts + path, _, query = target.partition("?") + + headers: list[tuple[str, str]] = [] + while True: + raw = await reader.readline() + if raw in (b"\r\n", b"\n", b""): + break + try: + header_line = raw.decode("latin-1").rstrip("\r\n") + except UnicodeDecodeError as exc: + raise _BadRequest("header decode failed") from exc + if ":" not in header_line: + raise _BadRequest(f"bad header: {header_line!r}") + name, _, value = header_line.partition(":") + headers.append((name.strip(), value.strip())) + + headers_map = {k.lower(): v for k, v in headers} + body = b"" + if "transfer-encoding" in headers_map and headers_map["transfer-encoding"].lower() == "chunked": + body = await _read_chunked(reader) + else: + cl_raw = headers_map.get("content-length", "0") + try: + cl = int(cl_raw) + except ValueError as exc: + raise _BadRequest(f"bad content-length: {cl_raw!r}") from exc + if cl < 0 or cl > MAX_REQUEST_BODY: + raise _BadRequest(f"content-length out of bounds: {cl}") + if cl: + body = await reader.readexactly(cl) + + return _Request( + method=method, + path=urllib.parse.unquote(path), + query=query, + headers=headers, + body=body, + ) + + +async def _read_chunked(reader: asyncio.StreamReader) -> bytes: + """Minimal chunked-transfer decoder for uploads.""" + out = bytearray() + total = 0 + while True: + size_line = (await reader.readline()).decode("latin-1").strip() + if ";" in size_line: + size_line = size_line.split(";", 1)[0].strip() + try: + size = int(size_line, 16) + except ValueError as exc: + raise _BadRequest(f"bad chunk size: {size_line!r}") from exc + if size == 0: + # Consume trailer headers until blank line. + while True: + line = await reader.readline() + if line in (b"\r\n", b"\n", b""): + break + break + chunk = await reader.readexactly(size) + out.extend(chunk) + total += size + if total > MAX_REQUEST_BODY: + raise _BadRequest("request body too large") + # Trailing CRLF after each chunk. + trailer = await reader.readexactly(2) + if trailer not in (b"\r\n", b"\n\r"): + # tolerate bare \n + pass + return bytes(out) + + +def _build_environ( + request: _Request, + *, + peer_fingerprint: str | None, + tunnel_id: str | None, +) -> dict[str, Any]: + """Build a PEP-3333-compliant WSGI environ from the parsed request.""" + environ: dict[str, Any] = { + "REQUEST_METHOD": request.method, + "SCRIPT_NAME": "", + "PATH_INFO": request.path, + "QUERY_STRING": request.query, + "SERVER_NAME": WSGI_SERVER_NAME, + "SERVER_PORT": "443", + "SERVER_PROTOCOL": "HTTP/1.1", + "wsgi.version": (1, 0), + "wsgi.url_scheme": "https", + "wsgi.input": io.BytesIO(request.body), + "wsgi.errors": io.StringIO(), + "wsgi.multithread": True, + "wsgi.multiprocess": False, + "wsgi.run_once": False, + } + + for name, value in request.headers: + key = "HTTP_" + name.upper().replace("-", "_") + if name.lower() == "content-length": + environ["CONTENT_LENGTH"] = value + elif name.lower() == "content-type": + environ["CONTENT_TYPE"] = value + environ[key] = value + + if peer_fingerprint: + environ["LINK_PEER_FINGERPRINT"] = peer_fingerprint + if tunnel_id: + environ["LINK_TUNNEL_ID"] = tunnel_id + + return environ + + +async def _write_status_headers( + writer: Any, + code: int, + reason: str, + headers: list[tuple[str, str]], +) -> int: + lines = [f"HTTP/1.1 {code} {reason}\r\n"] + for name, value in headers: + lines.append(f"{name}: {value}\r\n") + lines.append("\r\n") + out = "".join(lines).encode("latin-1") + await writer.write(out) + return len(out) + + +async def _write_simple(writer: Any, code: int, reason: str, body: bytes) -> None: + headers = [ + ("Content-Type", "text/plain; charset=utf-8"), + ("Content-Length", str(len(body))), + ] + await _write_status_headers(writer, code, reason, headers) + if body: + await writer.write(body) + + +def _byte_count_for_simple(body: bytes) -> int: + # Just a rough tally for metadata β€” status line + headers + body. + return len(body) + 64 diff --git a/think/supervisor.py b/think/supervisor.py index 86d81a7a6..5ff91ae3f 100644 --- a/think/supervisor.py +++ b/think/supervisor.py @@ -917,6 +917,12 @@ def start_cortex_server() -> ManagedProcess: return _launch_process("cortex", cmd, restart=True) +def start_link_server() -> ManagedProcess: + """Launch the link tunnel service (spl home-side endpoint).""" + cmd = ["sol", "link", "-v"] + return _launch_process("link", cmd, restart=True) + + def start_convey_server( verbose: bool, debug: bool = False, port: int = 0 ) -> tuple[ManagedProcess, int]: @@ -1395,6 +1401,11 @@ def parse_args() -> argparse.ArgumentParser: action="store_true", help="Do not start the Cortex server (run it manually for debugging)", ) + parser.add_argument( + "--no-link", + action="store_true", + help="Do not start the link tunnel service", + ) parser.add_argument( "--no-convey", action="store_true", @@ -1568,6 +1579,9 @@ def main() -> None: # Cortex for agent execution if not args.no_cortex: procs.append(start_cortex_server()) + # Link tunnel service (opt-out via --no-link) + if not args.no_link: + procs.append(start_link_server()) # Make procs accessible to restart handler _managed_procs = procs -- 2.51.2