sync-transport.ts

1// WebSocket sync transport — v3 sync Phase 2.2//3// The extension is always the WebSocket client; Android (or future iOS /4// desktop sync hosts) is always the server. Each binary WebSocket message5// carries exactly one Noise frame — no length prefix, no fragmentation.6//7// On connect:8//   1. Open ws://<ip>:<port>/9//   2. Run Noise XX as the initiator, using our long-term sync identity10//      static keypair.11//   3. Verify the responder's static public key matches the paired-device12//      registry entry we expect. Mismatch is a hard fail — no implicit13//      pubkey trust on the LAN.14//   4. Hand the caller a `SyncSession` for the SyncMessage protocol.15//16// Subnet probe falls back to scanning the /24 returned by `getLocalSubnet`17// when the cached last-known IP fails. We use a short connection timeout18// (~500ms per IP) and parallel batches of 16 to keep the worst-case scan19// under ~10 seconds.2021import { noiseXXInitiator, type NoiseXXSession } from "./noise-xx";22import { SYNC_LIMITS } from "./sync-limits";23import { hexToPubkey, pubkeyToHex } from "./sync-identity";2425const DEFAULT_PROBE_TIMEOUT_MS = 500;26const DEFAULT_PROBE_BATCH = 16;27const DEFAULT_CONNECT_TIMEOUT_MS = 4_000;28/** v2 brief §5.2 — subnet probe is rate-limited to once per 60 seconds. */29export const SUBNET_PROBE_COOLDOWN_MS = 60_000;3031export interface SyncSession {32  send(plaintext: Uint8Array): void;33  receive(): Promise<Uint8Array>;34  close(): Promise<void>;35  remoteStaticPublicKey: Uint8Array;36}3738export interface SyncEndpoint {39  ip: string;40  port: number;41}4243/**44 * Open a WebSocket to (ip, port), run Noise XX, verify the responder's45 * long-term static key matches `expectedRemotePubkeyHex`, and return a46 * ready-to-use session.47 *48 * On a successful return, `setLastKnownIp` in the caller can persist49 * (ip, port) for the next session's fast path.50 */51export async function openSyncSession(52  endpoint: SyncEndpoint,53  ourLongTermPrivateKey: Uint8Array,54  ourLongTermPublicKey: Uint8Array,55  expectedRemotePubkeyHex: string,56  options: { connectTimeoutMs?: number } = {},57): Promise<SyncSession> {58  const ws = await connectWebSocket(endpoint, options.connectTimeoutMs ?? DEFAULT_CONNECT_TIMEOUT_MS);59  const inbox: Uint8Array[] = [];60  const readers: Array<(b: Uint8Array) => void> = [];61  let closed = false;6263  ws.binaryType = "arraybuffer";64  ws.onmessage = (ev: MessageEvent) => {65    const data = ev.data;66    if (data instanceof ArrayBuffer) {67      // MV-3: refuse oversized frames before they are buffered or decrypted.68      // A peer that sends one is misbehaving — tear the session down rather69      // than letting the inbox grow toward OOM.70      if (data.byteLength > SYNC_LIMITS.MAX_FRAME_BYTES) {71        // MV-3: oversized frame — tear the session down. close() routes through72        // onclose, which rejects any parked receive(), so no error array is73        // needed (PCH-01-026 #9).74        console.warn(75          `[Peach] sync frame exceeds max size (${SYNC_LIMITS.MAX_FRAME_BYTES} bytes); closing`,76        );77        try { ws.close(); } catch { /* ignore */ }78        return;79      }80      const bytes = new Uint8Array(data);81      const r = readers.shift();82      if (r) r(bytes);83      else inbox.push(bytes);84    }85  };86  ws.onerror = () => {87    // PCH-01-026 #9: surface the error and unblock any parked receive() by88    // closing. Browsers usually follow error with close, but not always; an89    // explicit close guarantees the handshake fails fast instead of hanging.90    console.warn("[Peach] sync WebSocket error; closing transport");91    try { ws.close(); } catch { /* ignore */ }92  };93  ws.onclose = () => {94    closed = true;95    while (readers.length > 0) {96      const r = readers.shift()!;97      r(new Uint8Array(0));98    }99  };100101  function send(bytes: Uint8Array): Promise<void> {102    return new Promise((resolve, reject) => {103      try {104        ws.send(bytes as Uint8Array<ArrayBuffer>);105        resolve();106      } catch (err) {107        reject(err as Error);108      }109    });110  }111  function receive(): Promise<Uint8Array> {112    if (closed) return Promise.reject(new Error("WebSocket closed before frame"));113    const v = inbox.shift();114    if (v) return Promise.resolve(v);115    return new Promise<Uint8Array>((resolve, reject) => {116      readers.push((b) => {117        if (b.length === 0 && closed) reject(new Error("WebSocket closed"));118        else resolve(b);119      });120    });121  }122123  let session: NoiseXXSession;124  try {125    session = await noiseXXInitiator(126      ourLongTermPrivateKey,127      ourLongTermPublicKey,128      send,129      receive,130      // PCH-01-003: abort BEFORE msg3 if the responder is not the peer we131      // expect, so a probed-but-unexpected LAN host never receives our static.132      (remoteStatic) => {133        const seenHex = pubkeyToHex(remoteStatic);134        if (seenHex !== expectedRemotePubkeyHex) {135          throw new Error(136            `paired-device static key mismatch (got ${seenHex.slice(0, 16)}…, ` +137            `expected ${expectedRemotePubkeyHex.slice(0, 16)}…)`,138          );139        }140      },141    );142  } catch (err) {143    try { ws.close(); } catch { /* ignore */ }144    throw err;145  }146147  const remoteHex = pubkeyToHex(session.remoteStaticPublicKey);148  if (remoteHex !== expectedRemotePubkeyHex) {149    try { ws.close(); } catch { /* ignore */ }150    throw new Error(151      `paired-device static key mismatch (got ${remoteHex.slice(0, 16)}…, ` +152      `expected ${expectedRemotePubkeyHex.slice(0, 16)}…)`,153    );154  }155  // Bind so the unused expectation-source check helper stays honest.156  void hexToPubkey(expectedRemotePubkeyHex);157158  return {159    send(plaintext: Uint8Array) {160      const ciphertext = session.encrypt(plaintext);161      ws.send(ciphertext as Uint8Array<ArrayBuffer>);162    },163    async receive() {164      const ct = await receive();165      return session.decrypt(ct);166    },167    async close() {168      try { ws.close(); } catch { /* ignore */ }169    },170    remoteStaticPublicKey: session.remoteStaticPublicKey,171  };172}173174function connectWebSocket(endpoint: SyncEndpoint, timeoutMs: number): Promise<WebSocket> {175  return new Promise((resolve, reject) => {176    let ws: WebSocket;177    try {178      ws = new WebSocket(`ws://${endpoint.ip}:${endpoint.port}/`);179    } catch (err) {180      reject(err as Error);181      return;182    }183    const t = setTimeout(() => {184      try { ws.close(); } catch { /* ignore */ }185      reject(new Error(`WebSocket connect to ${endpoint.ip}:${endpoint.port} timed out`));186    }, timeoutMs);187    ws.onopen = () => {188      clearTimeout(t);189      resolve(ws);190    };191    ws.onerror = () => {192      clearTimeout(t);193      reject(new Error(`WebSocket connect to ${endpoint.ip}:${endpoint.port} errored`));194    };195  });196}197198/**199 * Per-pubkey cooldown tracker for the subnet probe. v2 brief §5.2 caps200 * full /24 scans at once per minute so an extension whose cached IP has201 * gone stale doesn't hammer the LAN with handshake attempts every time202 * the popup opens.203 *204 * Keyed by the expected remote pubkey because each paired peer is probed205 * independently; the rate limit applies to "scan looking for THIS peer",206 * not "scan the subnet at all".207 */208const lastProbeAt = new Map<string, number>();209210export function subnetProbeOnCooldown(211  expectedRemotePubkeyHex: string,212  now: number = Date.now(),213  cooldownMs: number = SUBNET_PROBE_COOLDOWN_MS,214): boolean {215  const last = lastProbeAt.get(expectedRemotePubkeyHex);216  return last !== undefined && now - last < cooldownMs;217}218219/** Test-only — clear the per-pubkey probe cooldown ledger. */220export function _resetSubnetProbeCooldown(): void {221  lastProbeAt.clear();222}223224/**225 * Subnet probe — scan a /24 looking for a host that completes Noise XX226 * with the expected long-term static key. Returns the first match, or227 * null if the cooldown is still active or no host responded.228 *229 * Rate-limited to once per `SUBNET_PROBE_COOLDOWN_MS` per peer pubkey230 * (v2 brief §5.2). Callers should still attempt the cached231 * `last_known_ip`/`last_known_port` directly first — the subnet probe is232 * fallback only.233 *234 * For each IP we attempt the full Noise XX handshake, but with a tight235 * timeout: dead hosts fail at the WebSocket connect step in <1ms (browser236 * surfaces ECONNREFUSED immediately) or by the connectTimeout otherwise.237 */238export async function probeSubnetForPeer(239  subnet: string, // e.g. "192.168.1"240  port: number,241  ourLongTermPrivateKey: Uint8Array,242  ourLongTermPublicKey: Uint8Array,243  expectedRemotePubkeyHex: string,244  options: { batch?: number; timeoutMs?: number; cooldownMs?: number; now?: number } = {},245): Promise<SyncEndpoint | null> {246  const now = options.now ?? Date.now();247  const cooldownMs = options.cooldownMs ?? SUBNET_PROBE_COOLDOWN_MS;248  if (subnetProbeOnCooldown(expectedRemotePubkeyHex, now, cooldownMs)) {249    return null;250  }251  lastProbeAt.set(expectedRemotePubkeyHex, now);252  const batchSize = options.batch ?? DEFAULT_PROBE_BATCH;253  const timeoutMs = options.timeoutMs ?? DEFAULT_PROBE_TIMEOUT_MS;254  for (let start = 1; start <= 254; start += batchSize) {255    const end = Math.min(start + batchSize - 1, 254);256    const ips: string[] = [];257    for (let i = start; i <= end; i++) ips.push(`${subnet}.${i}`);258259    const results = await Promise.all(ips.map((ip) => probeOne(260      { ip, port },261      ourLongTermPrivateKey,262      ourLongTermPublicKey,263      expectedRemotePubkeyHex,264      timeoutMs,265    )));266    const hit = results.find((r): r is SyncEndpoint => r !== null);267    if (hit) return hit;268  }269  return null;270}271272async function probeOne(273  endpoint: SyncEndpoint,274  ourLongTermPrivateKey: Uint8Array,275  ourLongTermPublicKey: Uint8Array,276  expectedRemotePubkeyHex: string,277  timeoutMs: number,278): Promise<SyncEndpoint | null> {279  try {280    const session = await openSyncSession(281      endpoint,282      ourLongTermPrivateKey,283      ourLongTermPublicKey,284      expectedRemotePubkeyHex,285      { connectTimeoutMs: timeoutMs },286    );287    await session.close();288    return endpoint;289  } catch {290    return null;291  }292}