sync-transport.ts

1// WebSocket reconnect transport — Noise KK with both paired static keys.2// First pairing remains Noise NK; reconnect never negotiates or falls back.3//4// The extension is always the WebSocket client; Android (or future iOS /5// native sync hosts) is always the server. Each binary WebSocket message6// carries exactly one Noise frame — no length prefix, no fragmentation.7//8// On connect:9//   1. Open ws://<ip>:<port>/10//   2. Send PEACH_RECONNECT with an opaque rotating rendezvous capability so11//      the listener can select paired local state without exposing identities.12//   3. Run standard Noise KK using both locally-known static keys.13//   4. Exchange PAIR_IDENTITY frames over the encrypted channel (same frame14//      Android's SyncReconnectListener expects before it will treat the15//      connection as sync traffic rather than an unknown caller). Our16//      identity is bound to the static key authenticated by KK.17//   5. Hand the caller a `SyncSession` for the SyncMessage protocol.18//19// Subnet probe falls back to scanning the /24 returned by `getLocalSubnet`20// when the cached last-known IP fails. We use a short connection timeout21// (~500ms per IP) and parallel batches of 16 to keep the worst-case scan22// under ~10 seconds.2324import { noiseKKInitiator, type NoiseNKSession } from "./noise-js";25import { SYNC_LIMITS } from "./sync-limits";26import { hexToPubkey } from "./sync-identity";27import { buildRendezvousUrl } from "./rendezvous";28import type { DeviceType } from "./sync-types";2930const DEFAULT_PROBE_TIMEOUT_MS = 500;31const DEFAULT_PROBE_BATCH = 16;32const DEFAULT_CONNECT_TIMEOUT_MS = 4_000;33const DEFAULT_HANDSHAKE_TIMEOUT_MS = 10_000;34/** v2 brief §5.2 — subnet probe is rate-limited to once per 60 seconds. */35export const SUBNET_PROBE_COOLDOWN_MS = 60_000;3637export interface SyncSession {38  send(plaintext: Uint8Array): void;39  receive(): Promise<Uint8Array>;40  close(): Promise<void>;41  remoteStaticPublicKey: Uint8Array;42}4344export interface SyncEndpoint {45  ip: string;46  port: number;47}4849export interface ProbedSyncEndpoint extends SyncEndpoint {50  /** Exact clock-skew window capability that completed KK at this endpoint. */51  rendezvousId: string;52}5354/** Our identity, repeated in the post-handshake PAIR_IDENTITY frame and bound55 *  to the local static key already authenticated by KK. */56export interface SyncPeerIdentity {57  deviceId: string;58  deviceName: string;59  deviceType: DeviceType;60  longTermPublicKeyHex: string;61  /** Local-only key material for the KK reconnect handshake; never encoded. */62  longTermPrivateKey: Uint8Array;63  signingPublicKeyHex?: string;64}6566export class SyncFrameTooLargeError extends Error {67  constructor(message = "sync frame too large (WebSocket close 1009)") {68    super(message);69    this.name = "SyncFrameTooLargeError";70  }71}7273export interface SyncSessionOptions {74  connectTimeoutMs?: number;75  handshakeTimeoutMs?: number;76}7778function encodePairIdentity(id: SyncPeerIdentity): Uint8Array {79  return new TextEncoder().encode(JSON.stringify({80    type: "PAIR_IDENTITY",81    device_id: id.deviceId,82    device_name: id.deviceName,83    device_type: id.deviceType,84    long_term_public_key_hex: id.longTermPublicKeyHex,85    signing_public_key_hex: id.signingPublicKeyHex,86  }));87}8889/**90 * Open a WebSocket to (ip, port), run Noise KK against both paired static91 * keys, exchange bound PAIR_IDENTITY frames, and return a ready-to-use92 * session.93 *94 * On a successful return, `setLastKnownIp` in the caller can persist95 * (ip, port) for the next session's fast path.96 */97export async function openSyncSession(98  endpoint: SyncEndpoint,99  ourIdentity: SyncPeerIdentity,100  expectedRemotePubkeyHex: string,101  rendezvousId: string,102  options: SyncSessionOptions = {},103): Promise<SyncSession> {104  return openSyncSessionAtUrl(105    `ws://${endpoint.ip}:${endpoint.port}/`,106    "LAN peer",107    ourIdentity,108    expectedRemotePubkeyHex,109    rendezvousId,110    options,111  );112}113114/**115 * Relay reconnect preserves the exact LAN wire protocol; only the WebSocket116 * target changes. The URL builder enforces credential-free `wss://`.117 */118export async function openRelaySyncSession(119  relayUrl: string,120  ourIdentity: SyncPeerIdentity,121  expectedRemotePubkeyHex: string,122  rendezvousId: string,123  options: SyncSessionOptions = {},124): Promise<SyncSession> {125  return openSyncSessionAtUrl(126    buildRendezvousUrl(relayUrl, rendezvousId),127    "relay",128    ourIdentity,129    expectedRemotePubkeyHex,130    rendezvousId,131    options,132  );133}134135async function openSyncSessionAtUrl(136  url: string,137  targetLabel: string,138  ourIdentity: SyncPeerIdentity,139  expectedRemotePubkeyHex: string,140  rendezvousId: string,141  options: SyncSessionOptions,142): Promise<SyncSession> {143  if (!/^[A-Za-z0-9_-]{22,128}$/.test(rendezvousId)) {144    throw new Error("reconnect requires a valid rotating rendezvous id");145  }146  const ws = await connectWebSocketUrl(147    url,148    targetLabel,149    options.connectTimeoutMs ?? DEFAULT_CONNECT_TIMEOUT_MS,150  );151  const inbox: Uint8Array[] = [];152  const readers: Array<(b: Uint8Array) => void> = [];153  let closed = false;154  let closeError: Error = new Error("WebSocket closed");155156  ws.binaryType = "arraybuffer";157  ws.onmessage = (ev: MessageEvent) => {158    const data = ev.data;159    if (data instanceof ArrayBuffer) {160      // MV-3: refuse oversized frames before they are buffered or decrypted.161      // A peer that sends one is misbehaving — tear the session down rather162      // than letting the inbox grow toward OOM.163      if (data.byteLength > SYNC_LIMITS.MAX_FRAME_BYTES) {164        // MV-3: oversized frame — tear the session down. close() routes through165        // onclose, which rejects any parked receive(), so no error array is166        // needed (PCH-01-026 #9).167        console.warn(168          `[Peach] sync frame exceeds max size (${SYNC_LIMITS.MAX_FRAME_BYTES} bytes); closing`,169        );170        closeError = new SyncFrameTooLargeError(171          `sync frame exceeds ${SYNC_LIMITS.MAX_FRAME_BYTES} bytes`,172        );173        try { ws.close(1009, "sync frame too large"); } catch { /* ignore */ }174        return;175      }176      const bytes = new Uint8Array(data);177      const r = readers.shift();178      if (r) r(bytes);179      else inbox.push(bytes);180    }181  };182  ws.onerror = () => {183    // PCH-01-026 #9: surface the error and unblock any parked receive() by184    // closing. Browsers usually follow error with close, but not always; an185    // explicit close guarantees the handshake fails fast instead of hanging.186    console.warn("[Peach] sync WebSocket error; closing transport");187    try { ws.close(); } catch { /* ignore */ }188  };189  ws.onclose = (event: CloseEvent) => {190    closed = true;191    if (event.code === 1009) {192      closeError = new SyncFrameTooLargeError();193    }194    while (readers.length > 0) {195      const r = readers.shift()!;196      r(new Uint8Array(0));197    }198  };199200  function send(bytes: Uint8Array): Promise<void> {201    return new Promise((resolve, reject) => {202      if (bytes.byteLength > SYNC_LIMITS.MAX_FRAME_BYTES) {203        reject(204          new SyncFrameTooLargeError(205            `sync frame exceeds ${SYNC_LIMITS.MAX_FRAME_BYTES} bytes`,206          ),207        );208        return;209      }210      try {211        ws.send(bytes as Uint8Array<ArrayBuffer>);212        resolve();213      } catch (err) {214        reject(err as Error);215      }216    });217  }218  function receive(): Promise<Uint8Array> {219    if (closed) return Promise.reject(closeError);220    const v = inbox.shift();221    if (v) return Promise.resolve(v);222    return new Promise<Uint8Array>((resolve, reject) => {223      readers.push((b) => {224        if (b.length === 0 && closed) reject(closeError);225        else resolve(b);226      });227    });228  }229230  const handshakeTimer = setTimeout(() => {231    closeError = new Error("sync_handshake_timeout");232    try { ws.close(1000, "sync handshake timeout"); } catch { /* ignore */ }233  }, options.handshakeTimeoutMs ?? DEFAULT_HANDSHAKE_TIMEOUT_MS);234  let session: NoiseNKSession;235  try {236    const ourPublicKey = hexToPubkey(ourIdentity.longTermPublicKeyHex);237    if (ourIdentity.longTermPrivateKey.length !== 32) {238      throw new Error("local reconnect private key must be exactly 32 bytes");239    }240    // This opaque rotating capability lets Android select paired local state241    // without exposing either public key to a relay. It grants no identity242    // trust; KK still proves both private keys immediately afterwards.243    await send(new TextEncoder().encode(JSON.stringify({244      type: "PEACH_RECONNECT",245      rendezvous_id: rendezvousId,246    })));247    session = await noiseKKInitiator(248      ourIdentity.longTermPrivateKey,249      ourPublicKey,250      hexToPubkey(expectedRemotePubkeyHex),251      send,252      receive,253    );254  } catch (err) {255    clearTimeout(handshakeTimer);256    try { ws.close(); } catch { /* ignore */ }257    throw err;258  }259260  try {261    await send(session.encrypt(encodePairIdentity(ourIdentity)));262    const theirFrame = JSON.parse(new TextDecoder().decode(session.decrypt(await receive())));263    if (theirFrame?.type !== "PAIR_IDENTITY") {264      throw new Error(`expected PAIR_IDENTITY, got ${theirFrame?.type as string}`);265    }266    if (theirFrame.long_term_public_key_hex !== expectedRemotePubkeyHex) {267      throw new Error("peer identity does not match authenticated KK key");268    }269    clearTimeout(handshakeTimer);270  } catch (err) {271    clearTimeout(handshakeTimer);272    try { ws.close(); } catch { /* ignore */ }273    throw err;274  }275276  return {277    send(plaintext: Uint8Array) {278      const ciphertext = session.encrypt(plaintext);279      if (ciphertext.byteLength > SYNC_LIMITS.MAX_FRAME_BYTES) {280        throw new SyncFrameTooLargeError(281          `sync frame exceeds ${SYNC_LIMITS.MAX_FRAME_BYTES} bytes`,282        );283      }284      ws.send(ciphertext as Uint8Array<ArrayBuffer>);285    },286    async receive() {287      const ct = await receive();288      return session.decrypt(ct);289    },290    async close() {291      try { ws.close(); } catch { /* ignore */ }292    },293    remoteStaticPublicKey: hexToPubkey(expectedRemotePubkeyHex),294  };295}296297function connectWebSocketUrl(298  url: string,299  targetLabel: string,300  timeoutMs: number,301): Promise<WebSocket> {302  return new Promise((resolve, reject) => {303    let ws: WebSocket;304    try {305      ws = new WebSocket(url);306    } catch (err) {307      reject(err as Error);308      return;309    }310    const t = setTimeout(() => {311      try { ws.close(); } catch { /* ignore */ }312      reject(new Error(`${targetLabel} WebSocket connect timed out`));313    }, timeoutMs);314    ws.onopen = () => {315      clearTimeout(t);316      resolve(ws);317    };318    ws.onerror = () => {319      clearTimeout(t);320      reject(new Error(`${targetLabel} WebSocket connect errored`));321    };322  });323}324325/**326 * Per-pubkey cooldown tracker for the subnet probe. v2 brief §5.2 caps327 * full /24 scans at once per minute so an extension whose cached IP has328 * gone stale doesn't hammer the LAN with handshake attempts every time329 * the popup opens.330 *331 * Keyed by the expected remote pubkey because each paired peer is probed332 * independently; the rate limit applies to "scan looking for THIS peer",333 * not "scan the subnet at all".334 */335const lastProbeAt = new Map<string, number>();336337export function subnetProbeOnCooldown(338  expectedRemotePubkeyHex: string,339  now: number = Date.now(),340  cooldownMs: number = SUBNET_PROBE_COOLDOWN_MS,341): boolean {342  const last = lastProbeAt.get(expectedRemotePubkeyHex);343  return last !== undefined && now - last < cooldownMs;344}345346/** Test-only — clear the per-pubkey probe cooldown ledger. */347export function _resetSubnetProbeCooldown(): void {348  lastProbeAt.clear();349}350351/**352 * Subnet probe — scan a /24 looking for a host that completes Noise KK353 * with the expected long-term static key. Returns the first match, or354 * null if the cooldown is still active or no host responded.355 *356 * Rate-limited to once per `SUBNET_PROBE_COOLDOWN_MS` per peer pubkey357 * (v2 brief §5.2). Callers should still attempt the cached358 * `last_known_ip`/`last_known_port` directly first — the subnet probe is359 * fallback only.360 *361 * For each IP we attempt the full Noise KK handshake, but with a tight362 * timeout: dead hosts fail at the WebSocket connect step in <1ms (browser363 * surfaces ECONNREFUSED immediately) or by the connectTimeout otherwise.364 */365export async function probeSubnetForPeer(366  subnet: string, // e.g. "192.168.1"367  port: number,368  ourIdentity: SyncPeerIdentity,369  expectedRemotePubkeyHex: string,370  rendezvousIds: readonly string[],371  options: { batch?: number; timeoutMs?: number; cooldownMs?: number; now?: number } = {},372): Promise<ProbedSyncEndpoint | null> {373  if (rendezvousIds.length === 0) {374    throw new Error("subnet probe requires at least one rendezvous id");375  }376  const now = options.now ?? Date.now();377  const cooldownMs = options.cooldownMs ?? SUBNET_PROBE_COOLDOWN_MS;378  if (subnetProbeOnCooldown(expectedRemotePubkeyHex, now, cooldownMs)) {379    return null;380  }381  lastProbeAt.set(expectedRemotePubkeyHex, now);382  const batchSize = options.batch ?? DEFAULT_PROBE_BATCH;383  const timeoutMs = options.timeoutMs ?? DEFAULT_PROBE_TIMEOUT_MS;384  for (let start = 1; start <= 254; start += batchSize) {385    const end = Math.min(start + batchSize - 1, 254);386    const ips: string[] = [];387    for (let i = start; i <= end; i++) ips.push(`${subnet}.${i}`);388389    const results = await Promise.all(ips.map((ip) => probeOne(390      { ip, port },391      ourIdentity,392      expectedRemotePubkeyHex,393      rendezvousIds,394      timeoutMs,395    )));396    const hit = results.find((r): r is ProbedSyncEndpoint => r !== null);397    if (hit) return hit;398  }399  return null;400}401402async function probeOne(403  endpoint: SyncEndpoint,404  ourIdentity: SyncPeerIdentity,405  expectedRemotePubkeyHex: string,406  rendezvousIds: readonly string[],407  timeoutMs: number,408): Promise<ProbedSyncEndpoint | null> {409  for (const rendezvousId of rendezvousIds) {410    try {411      const session = await openSyncSession(412        endpoint,413        ourIdentity,414        expectedRemotePubkeyHex,415        rendezvousId,416        { connectTimeoutMs: timeoutMs },417      );418      await session.close();419      return { ...endpoint, rendezvousId };420    } catch {421      // Try the next locally-derived clock-skew window at the same host.422    }423  }424  return null;425}