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    session.close();273    try { ws.close(); } catch { /* ignore */ }274    throw err;275  }276277  return {278    send(plaintext: Uint8Array) {279      const ciphertext = session.encrypt(plaintext);280      if (ciphertext.byteLength > SYNC_LIMITS.MAX_FRAME_BYTES) {281        throw new SyncFrameTooLargeError(282          `sync frame exceeds ${SYNC_LIMITS.MAX_FRAME_BYTES} bytes`,283        );284      }285      ws.send(ciphertext as Uint8Array<ArrayBuffer>);286    },287    async receive() {288      const ct = await receive();289      return session.decrypt(ct);290    },291    async close() {292      try { ws.close(); } catch { /* ignore */ }293      session.close();294    },295    remoteStaticPublicKey: hexToPubkey(expectedRemotePubkeyHex),296  };297}298299function connectWebSocketUrl(300  url: string,301  targetLabel: string,302  timeoutMs: number,303): Promise<WebSocket> {304  return new Promise((resolve, reject) => {305    let ws: WebSocket;306    try {307      ws = new WebSocket(url);308    } catch (err) {309      reject(err as Error);310      return;311    }312    const t = setTimeout(() => {313      try { ws.close(); } catch { /* ignore */ }314      reject(new Error(`${targetLabel} WebSocket connect timed out`));315    }, timeoutMs);316    ws.onopen = () => {317      clearTimeout(t);318      resolve(ws);319    };320    ws.onerror = () => {321      clearTimeout(t);322      reject(new Error(`${targetLabel} WebSocket connect errored`));323    };324  });325}326327/**328 * Per-pubkey cooldown tracker for the subnet probe. v2 brief §5.2 caps329 * full /24 scans at once per minute so an extension whose cached IP has330 * gone stale doesn't hammer the LAN with handshake attempts every time331 * the popup opens.332 *333 * Keyed by the expected remote pubkey because each paired peer is probed334 * independently; the rate limit applies to "scan looking for THIS peer",335 * not "scan the subnet at all".336 */337const lastProbeAt = new Map<string, number>();338339export function subnetProbeOnCooldown(340  expectedRemotePubkeyHex: string,341  now: number = Date.now(),342  cooldownMs: number = SUBNET_PROBE_COOLDOWN_MS,343): boolean {344  const last = lastProbeAt.get(expectedRemotePubkeyHex);345  return last !== undefined && now - last < cooldownMs;346}347348/** Test-only — clear the per-pubkey probe cooldown ledger. */349export function _resetSubnetProbeCooldown(): void {350  lastProbeAt.clear();351}352353/**354 * Subnet probe — scan a /24 looking for a host that completes Noise KK355 * with the expected long-term static key. Returns the first match, or356 * null if the cooldown is still active or no host responded.357 *358 * Rate-limited to once per `SUBNET_PROBE_COOLDOWN_MS` per peer pubkey359 * (v2 brief §5.2). Callers should still attempt the cached360 * `last_known_ip`/`last_known_port` directly first — the subnet probe is361 * fallback only.362 *363 * For each IP we attempt the full Noise KK handshake, but with a tight364 * timeout: dead hosts fail at the WebSocket connect step in <1ms (browser365 * surfaces ECONNREFUSED immediately) or by the connectTimeout otherwise.366 */367export async function probeSubnetForPeer(368  subnet: string, // e.g. "192.168.1"369  port: number,370  ourIdentity: SyncPeerIdentity,371  expectedRemotePubkeyHex: string,372  rendezvousIds: readonly string[],373  options: { batch?: number; timeoutMs?: number; cooldownMs?: number; now?: number } = {},374): Promise<ProbedSyncEndpoint | null> {375  if (rendezvousIds.length === 0) {376    throw new Error("subnet probe requires at least one rendezvous id");377  }378  const now = options.now ?? Date.now();379  const cooldownMs = options.cooldownMs ?? SUBNET_PROBE_COOLDOWN_MS;380  if (subnetProbeOnCooldown(expectedRemotePubkeyHex, now, cooldownMs)) {381    return null;382  }383  lastProbeAt.set(expectedRemotePubkeyHex, now);384  const batchSize = options.batch ?? DEFAULT_PROBE_BATCH;385  const timeoutMs = options.timeoutMs ?? DEFAULT_PROBE_TIMEOUT_MS;386  for (let start = 1; start <= 254; start += batchSize) {387    const end = Math.min(start + batchSize - 1, 254);388    const ips: string[] = [];389    for (let i = start; i <= end; i++) ips.push(`${subnet}.${i}`);390391    const results = await Promise.all(ips.map((ip) => probeOne(392      { ip, port },393      ourIdentity,394      expectedRemotePubkeyHex,395      rendezvousIds,396      timeoutMs,397    )));398    const hit = results.find((r): r is ProbedSyncEndpoint => r !== null);399    if (hit) return hit;400  }401  return null;402}403404async function probeOne(405  endpoint: SyncEndpoint,406  ourIdentity: SyncPeerIdentity,407  expectedRemotePubkeyHex: string,408  rendezvousIds: readonly string[],409  timeoutMs: number,410): Promise<ProbedSyncEndpoint | null> {411  for (const rendezvousId of rendezvousIds) {412    try {413      const session = await openSyncSession(414        endpoint,415        ourIdentity,416        expectedRemotePubkeyHex,417        rendezvousId,418        { connectTimeoutMs: timeoutMs },419      );420      await session.close();421      return { ...endpoint, rendezvousId };422    } catch {423      // Try the next locally-derived clock-skew window at the same host.424    }425  }426  return null;427}