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 lower long-term public key is the WebSocket/Noise initiator; the higher5// key holds the relay room as responder. Device type does not affect the role,6// so two extensions can sync directly without a native hub. Each binary7// WebSocket message carries one Noise frame (large padded application frames8// are split across multiple Noise messages below).9//10// On connect:11//   1. Open wss://<relay>/<rendezvous-id>/12//   2. Send PEACH_RECONNECT with an opaque rotating rendezvous capability so13//      the listener can select paired local state without exposing identities.14//   3. Run standard Noise KK using both locally-known static keys.15//   4. Exchange PAIR_IDENTITY frames over the encrypted channel (same frame16//      Android's SyncReconnectListener expects before it will treat the17//      connection as sync traffic rather than an unknown caller). Our18//      identity is bound to the static key authenticated by KK.19//   5. Hand the caller a `SyncSession` for the SyncMessage protocol.20//21// Subnet probe falls back to scanning the /24 returned by `getLocalSubnet`22// when the cached last-known IP fails. We use a short connection timeout23// (~500ms per IP) and parallel batches of 16 to keep the worst-case scan24// under ~10 seconds.2526import {27  noiseKKInitiator,28  noiseKKResponder,29  type NoiseNKSession,30} from "./noise-js";31import { SYNC_LIMITS } from "./sync-limits";32import {33  padSyncFrame,34  unpadSyncFrame,35  MAX_NOISE_PLAINTEXT_BYTES,36  FRAME_CONTINUATION_MARKER,37  chunkPaddedSyncFrame,38} from "./sync-padding";39import { hexToPubkey } from "./sync-identity";40import { buildRendezvousUrl } from "./rendezvous";41import type { DeviceType } from "./sync-types";4243// E6 (session rekeying) is deliberately NOT wired, and this is a reality check44// against the spec rather than an omission. E6 targets "long-lived sync45// sessions"; the browser has none. Sessions opened here are one-shot — roughly46// ten frames each way, a 30s coordinator timeout (SESSION_TIMEOUT_MS), then47// sync-runner closes them — and every reconnect runs a fresh Noise KK handshake48// with fresh ephemerals. (Android's listener does park a connection open after49// a round so it can act as a relay hub for another native peer, but the browser50// drives the lifetime of every session it opens, and closes it.) That already51// gives stronger post-compromise security than intra-session Rekey() would,52// because no two sessions share key material at all (pinned by the E6 test in53// noise-kk.test.ts).54//55// Wiring Rekey() anyway would mean both ends rekeying at the identical frame56// index with no wire signal to resynchronise on. One frame of drift between57// browser and Android breaks all sync — the exact failure class this protocol58// has already been burned by twice. The primitives are exposed on NoiseSession59// (rekeyOutgoing/rekeyIncoming) if a genuinely long-lived session ever appears;60// until then the cost is real and the benefit is zero.61const DEFAULT_CONNECT_TIMEOUT_MS = 4_000;62const DEFAULT_HANDSHAKE_TIMEOUT_MS = 10_000;63export const RELAY_WAITER_KEEPALIVE_MS = 20_000;6465export interface SyncSession {66  send(plaintext: Uint8Array): void;67  receive(): Promise<Uint8Array>;68  close(): Promise<void>;69  remoteStaticPublicKey: Uint8Array;70}7172/** Our identity, repeated in the post-handshake PAIR_IDENTITY frame and bound73 *  to the local static key already authenticated by KK. */74export interface SyncPeerIdentity {75  deviceId: string;76  deviceName: string;77  deviceType: DeviceType;78  longTermPublicKeyHex: string;79  /** Local-only key material for the KK reconnect handshake; never encoded. */80  longTermPrivateKey: Uint8Array;81  signingPublicKeyHex?: string;82}8384export class SyncFrameTooLargeError extends Error {85  constructor(message = "sync frame too large (WebSocket close 1009)") {86    super(message);87    this.name = "SyncFrameTooLargeError";88  }89}9091export interface SyncSessionOptions {92  connectTimeoutMs?: number;93  handshakeTimeoutMs?: number;94  signal?: AbortSignal;95}9697function encodePairIdentity(id: SyncPeerIdentity): Uint8Array {98  return new TextEncoder().encode(JSON.stringify({99    type: "PAIR_IDENTITY",100    device_id: id.deviceId,101    device_name: id.deviceName,102    device_type: id.deviceType,103    long_term_public_key_hex: id.longTermPublicKeyHex,104    signing_public_key_hex: id.signingPublicKeyHex,105  }));106}107108/**109 * Relay reconnect preserves the existing Noise KK wire protocol. The URL110 * builder enforces credential-free `wss://`.111 */112export async function openRelaySyncSession(113  relayUrl: string,114  ourIdentity: SyncPeerIdentity,115  expectedRemotePubkeyHex: string,116  rendezvousId: string,117  options: SyncSessionOptions = {},118): Promise<SyncSession> {119  return openSyncSessionAtUrl(120    buildRendezvousUrl(relayUrl, rendezvousId),121    "relay",122    ourIdentity,123    expectedRemotePubkeyHex,124    rendezvousId,125    options,126  );127}128129/**130 * Hold one relay room as the deterministic responder, then run Noise KK and131 * the bound PAIR_IDENTITY exchange. The zero-length liveness frame keeps a132 * Chrome 116+ MV3 service worker alive while the room is empty; the Peach133 * relay consumes that frame locally and never forwards it to the peer.134 */135export async function openRelaySyncResponderSession(136  relayUrl: string,137  ourIdentity: SyncPeerIdentity,138  expectedRemotePubkeyHex: string,139  rendezvousId: string,140  options: SyncSessionOptions & { keepaliveMs?: number } = {},141): Promise<SyncSession> {142  if (!/^[A-Za-z0-9_-]{22,128}$/.test(rendezvousId)) {143    throw new Error("reconnect requires a valid rotating rendezvous id");144  }145  if (ourIdentity.longTermPrivateKey.length !== 32) {146    throw new Error("local reconnect private key must be exactly 32 bytes");147  }148  const ws = await connectWebSocketUrl(149    buildRendezvousUrl(relayUrl, rendezvousId),150    "relay",151    options.connectTimeoutMs ?? DEFAULT_CONNECT_TIMEOUT_MS,152  );153  const inbox: Uint8Array[] = [];154  const readers: Array<(b: Uint8Array) => void> = [];155  let closed = false;156  let closeError: Error = new Error("WebSocket closed");157  const keepaliveMs = options.keepaliveMs ?? RELAY_WAITER_KEEPALIVE_MS;158  const abortWaiter = () => {159    try { ws.close(1000, "relay waiter replaced"); } catch { /* ignore */ }160  };161  const keepalive = setInterval(() => {162    if (ws.readyState !== WebSocket.OPEN) return;163    try { ws.send(new Uint8Array(0)); } catch { /* onclose owns cleanup */ }164  }, keepaliveMs);165166  ws.binaryType = "arraybuffer";167  ws.onmessage = (event: MessageEvent) => {168    if (!(event.data instanceof ArrayBuffer)) return;169    if (event.data.byteLength === 0) return;170    if (event.data.byteLength > SYNC_LIMITS.MAX_FRAME_BYTES) {171      closeError = new SyncFrameTooLargeError(172        `sync frame exceeds ${SYNC_LIMITS.MAX_FRAME_BYTES} bytes`,173      );174      try { ws.close(1009, "sync frame too large"); } catch { /* ignore */ }175      return;176    }177    const bytes = new Uint8Array(event.data);178    const reader = readers.shift();179    if (reader) reader(bytes);180    else inbox.push(bytes);181  };182  ws.onerror = () => {183    try { ws.close(); } catch { /* ignore */ }184  };185  ws.onclose = (event: CloseEvent) => {186    clearInterval(keepalive);187    options.signal?.removeEventListener("abort", abortWaiter);188    closed = true;189    if (event.code === 1009) closeError = new SyncFrameTooLargeError();190    while (readers.length > 0) readers.shift()!(new Uint8Array(0));191  };192  // Install close handling before honoring an already-aborted signal so a193  // synchronous/mock close cannot strand the first receive() forever.194  options.signal?.addEventListener("abort", abortWaiter, { once: true });195  if (options.signal?.aborted) abortWaiter();196197  const send = async (bytes: Uint8Array): Promise<void> => {198    if (bytes.byteLength > SYNC_LIMITS.MAX_FRAME_BYTES) {199      throw new SyncFrameTooLargeError(200        `sync frame exceeds ${SYNC_LIMITS.MAX_FRAME_BYTES} bytes`,201      );202    }203    ws.send(bytes as Uint8Array<ArrayBuffer>);204  };205  const receive = (): Promise<Uint8Array> => {206    if (closed) return Promise.reject(closeError);207    const buffered = inbox.shift();208    if (buffered) return Promise.resolve(buffered);209    return new Promise<Uint8Array>((resolve, reject) => {210      readers.push((bytes) => {211        if (bytes.length === 0 && closed) reject(closeError);212        else resolve(bytes);213      });214    });215  };216217  let noiseSession: NoiseNKSession | undefined;218  try {219    const selectorBytes = await receive();220    const selector = JSON.parse(new TextDecoder().decode(selectorBytes));221    if (222      selector?.type !== "PEACH_RECONNECT" ||223      typeof selector.rendezvous_id !== "string" ||224      !equalAscii(selector.rendezvous_id, rendezvousId)225    ) {226      throw new Error("reconnect capability does not match relay room");227    }228    const handshakeTimer = setTimeout(() => {229      closeError = new Error("sync_handshake_timeout");230      try { ws.close(1000, "sync handshake timeout"); } catch { /* ignore */ }231    }, options.handshakeTimeoutMs ?? DEFAULT_HANDSHAKE_TIMEOUT_MS);232    try {233      const ourPublicKey = hexToPubkey(ourIdentity.longTermPublicKeyHex);234      noiseSession = await noiseKKResponder(235        ourIdentity.longTermPrivateKey,236        ourPublicKey,237        hexToPubkey(expectedRemotePubkeyHex),238        send,239        receive,240      );241      const theirFrame = JSON.parse(242        new TextDecoder().decode(243          unpadSyncFrame(noiseSession.decrypt(await receive())),244        ),245      );246      if (theirFrame?.type !== "PAIR_IDENTITY") {247        throw new Error(`expected PAIR_IDENTITY, got ${theirFrame?.type as string}`);248      }249      if (theirFrame.long_term_public_key_hex !== expectedRemotePubkeyHex) {250        throw new Error("peer identity does not match authenticated KK key");251      }252      await send(noiseSession.encrypt(padSyncFrame(encodePairIdentity(ourIdentity))));253    } finally {254      clearTimeout(handshakeTimer);255    }256  } catch (error) {257    clearInterval(keepalive);258    options.signal?.removeEventListener("abort", abortWaiter);259    noiseSession?.close();260    try { ws.close(); } catch { /* ignore */ }261    throw error;262  }263264  return buildEncryptedSyncSession(265    ws,266    noiseSession,267    expectedRemotePubkeyHex,268    receive,269    () => {270      clearInterval(keepalive);271      options.signal?.removeEventListener("abort", abortWaiter);272    },273  );274}275276export async function openSyncSessionAtUrl(277  url: string,278  targetLabel: string,279  ourIdentity: SyncPeerIdentity,280  expectedRemotePubkeyHex: string,281  rendezvousId: string,282  options: SyncSessionOptions = {},283): Promise<SyncSession> {284  if (!/^[A-Za-z0-9_-]{22,128}$/.test(rendezvousId)) {285    throw new Error("reconnect requires a valid rotating rendezvous id");286  }287  const ws = await connectWebSocketUrl(288    url,289    targetLabel,290    options.connectTimeoutMs ?? DEFAULT_CONNECT_TIMEOUT_MS,291  );292  const inbox: Uint8Array[] = [];293  const readers: Array<(b: Uint8Array) => void> = [];294  let closed = false;295  let closeError: Error = new Error("WebSocket closed");296297  ws.binaryType = "arraybuffer";298  ws.onmessage = (ev: MessageEvent) => {299    const data = ev.data;300    if (data instanceof ArrayBuffer) {301      // MV-3: refuse oversized frames before they are buffered or decrypted.302      // A peer that sends one is misbehaving — tear the session down rather303      // than letting the inbox grow toward OOM.304      if (data.byteLength > SYNC_LIMITS.MAX_FRAME_BYTES) {305        // MV-3: oversized frame — tear the session down. close() routes through306        // onclose, which rejects any parked receive(), so no error array is307        // needed (PCH-01-026 #9).308        console.warn(309          `[Peach] sync frame exceeds max size (${SYNC_LIMITS.MAX_FRAME_BYTES} bytes); closing`,310        );311        closeError = new SyncFrameTooLargeError(312          `sync frame exceeds ${SYNC_LIMITS.MAX_FRAME_BYTES} bytes`,313        );314        try { ws.close(1009, "sync frame too large"); } catch { /* ignore */ }315        return;316      }317      const bytes = new Uint8Array(data);318      const r = readers.shift();319      if (r) r(bytes);320      else inbox.push(bytes);321    }322  };323  ws.onerror = () => {324    // PCH-01-026 #9: surface the error and unblock any parked receive() by325    // closing. Browsers usually follow error with close, but not always; an326    // explicit close guarantees the handshake fails fast instead of hanging.327    console.warn("[Peach] sync WebSocket error; closing transport");328    try { ws.close(); } catch { /* ignore */ }329  };330  ws.onclose = (event: CloseEvent) => {331    closed = true;332    if (event.code === 1009) {333      closeError = new SyncFrameTooLargeError();334    }335    while (readers.length > 0) {336      const r = readers.shift()!;337      r(new Uint8Array(0));338    }339  };340341  function send(bytes: Uint8Array): Promise<void> {342    return new Promise((resolve, reject) => {343      if (bytes.byteLength > SYNC_LIMITS.MAX_FRAME_BYTES) {344        reject(345          new SyncFrameTooLargeError(346            `sync frame exceeds ${SYNC_LIMITS.MAX_FRAME_BYTES} bytes`,347          ),348        );349        return;350      }351      try {352        ws.send(bytes as Uint8Array<ArrayBuffer>);353        resolve();354      } catch (err) {355        reject(err as Error);356      }357    });358  }359  function receive(): Promise<Uint8Array> {360    if (closed) return Promise.reject(closeError);361    const v = inbox.shift();362    if (v) return Promise.resolve(v);363    return new Promise<Uint8Array>((resolve, reject) => {364      readers.push((b) => {365        if (b.length === 0 && closed) reject(closeError);366        else resolve(b);367      });368    });369  }370371  const handshakeTimer = setTimeout(() => {372    closeError = new Error("sync_handshake_timeout");373    try { ws.close(1000, "sync handshake timeout"); } catch { /* ignore */ }374  }, options.handshakeTimeoutMs ?? DEFAULT_HANDSHAKE_TIMEOUT_MS);375  let session: NoiseNKSession;376  try {377    const ourPublicKey = hexToPubkey(ourIdentity.longTermPublicKeyHex);378    if (ourIdentity.longTermPrivateKey.length !== 32) {379      throw new Error("local reconnect private key must be exactly 32 bytes");380    }381    // This opaque rotating capability lets Android select paired local state382    // without exposing either public key to a relay. It grants no identity383    // trust; KK still proves both private keys immediately afterwards.384    await send(new TextEncoder().encode(JSON.stringify({385      type: "PEACH_RECONNECT",386      rendezvous_id: rendezvousId,387    })));388    session = await noiseKKInitiator(389      ourIdentity.longTermPrivateKey,390      ourPublicKey,391      hexToPubkey(expectedRemotePubkeyHex),392      send,393      receive,394    );395  } catch (err) {396    clearTimeout(handshakeTimer);397    try { ws.close(); } catch { /* ignore */ }398    throw err;399  }400401  try {402    await send(session.encrypt(padSyncFrame(encodePairIdentity(ourIdentity))));403    const theirFrame = JSON.parse(404      new TextDecoder().decode(unpadSyncFrame(session.decrypt(await receive()))),405    );406    if (theirFrame?.type !== "PAIR_IDENTITY") {407      throw new Error(`expected PAIR_IDENTITY, got ${theirFrame?.type as string}`);408    }409    if (theirFrame.long_term_public_key_hex !== expectedRemotePubkeyHex) {410      throw new Error("peer identity does not match authenticated KK key");411    }412    clearTimeout(handshakeTimer);413  } catch (err) {414    clearTimeout(handshakeTimer);415    session.close();416    try { ws.close(); } catch { /* ignore */ }417    throw err;418  }419420  return buildEncryptedSyncSession(421    ws,422    session,423    expectedRemotePubkeyHex,424    receive,425  );426}427428// Exported for the PB-26-002 reassembly-bound regression test.429export function buildEncryptedSyncSession(430  ws: WebSocket,431  session: NoiseNKSession,432  expectedRemotePubkeyHex: string,433  receiveCiphertext: () => Promise<Uint8Array>,434  beforeClose: () => void = () => undefined,435): SyncSession {436  return {437    async send(plaintext: Uint8Array) {438      // E7: bucket the size before encrypting, so the relay sees frame shapes439      // that no longer distinguish one operation (or vault size) from another.440      // E7b: a padded frame can legitimately be up to 24 MiB, but Noise caps a441      // single transport message at 65535 bytes (see crypto-core/src/lib.rs —442      // 65519 plaintext, 16-byte tag). We split a too-large padded frame into443      // numbered Noise messages and tag every chunk after the first with a444      // 1-byte continuation header. The receiver reassembles by reading the445      // padded frame's 4-byte length header, computing the same bucket from446      // the same ladder, and keeping reading until the bucket is full.447      const padded = padSyncFrame(plaintext);448      if (padded.byteLength > SYNC_LIMITS.MAX_FRAME_BYTES) {449        throw new SyncFrameTooLargeError(450          `sync frame exceeds ${SYNC_LIMITS.MAX_FRAME_BYTES} bytes`,451        );452      }453      // A single-chunk frame goes out as one Noise message, exactly as before454      // — no continuation byte, no per-chunk overhead, so the common small455      // case is unchanged on the wire.456      if (padded.byteLength <= MAX_NOISE_PLAINTEXT_BYTES) {457        const ct = session.encrypt(padded);458        ws.send(ct as Uint8Array<ArrayBuffer>);459        return;460      }461      for (const plaintext of chunkPaddedSyncFrame(padded)) {462        if (plaintext.byteLength > MAX_NOISE_PLAINTEXT_BYTES) {463          throw new Error(464            `chunk exceeds Noise limit after continuation byte: ${plaintext.byteLength}`,465          );466        }467        const ct = session.encrypt(plaintext);468        if (ct.byteLength > SYNC_LIMITS.MAX_FRAME_BYTES) {469          throw new SyncFrameTooLargeError(470            `sync frame exceeds ${SYNC_LIMITS.MAX_FRAME_BYTES} bytes`,471          );472        }473        ws.send(ct as Uint8Array<ArrayBuffer>);474      }475    },476    async receive() {477      // The first Noise message of a frame is the first one we read. If it478      // begins with the continuation marker, the message is one of N chunks479      // — the padded-frame length header comes inside the first chunk's480      // payload, so we read it there to learn the total.481      let buffered: Uint8Array | null = null;482      while (true) {483        const ct = await receiveCiphertext();484        const pt = session.decrypt(ct);485        if (buffered === null && pt.byteLength > 0 && pt[0] === FRAME_CONTINUATION_MARKER) {486          // Strip the continuation byte and keep the payload.487          buffered = pt.subarray(1);488          continue;489        }490        if (buffered !== null) {491          if (pt.byteLength > 0 && pt[0] === FRAME_CONTINUATION_MARKER) {492            const merged: Uint8Array = new Uint8Array(buffered.length + pt.byteLength - 1);493            merged.set(buffered, 0);494            merged.set(pt.subarray(1), buffered.length);495            buffered = merged;496            // PB-26-002: the frame bound must hold DURING reassembly. The497            // post-receive check in the coordinator runs only after the full498            // frame is in memory, so a peer that completes the Noise KK499            // handshake could otherwise stream continuation chunks without500            // bound inside the session window.501            if (buffered.length > SYNC_LIMITS.MAX_FRAME_BYTES) {502              throw new SyncFrameTooLargeError(503                `sync frame exceeds ${SYNC_LIMITS.MAX_FRAME_BYTES} bytes`,504              );505            }506            continue;507          }508          // Final chunk. PB-26-602: the continuation loop only bounds what is509          // buffered, so the last chunk could push the frame up to 64 KiB past510          // the cap. Bound the concatenation, not just the buffer.511          if (buffered.length + pt.byteLength > SYNC_LIMITS.MAX_FRAME_BYTES) {512            throw new SyncFrameTooLargeError(513              `sync frame exceeds ${SYNC_LIMITS.MAX_FRAME_BYTES} bytes`,514            );515          }516          const reassembled = new Uint8Array(buffered.length + pt.byteLength);517          reassembled.set(buffered, 0);518          reassembled.set(pt, buffered.length);519          return unpadSyncFrame(reassembled);520        }521        // Single-chunk frame, exactly as before.522        return unpadSyncFrame(pt);523      }524    },525    async close() {526      beforeClose();527      try { ws.close(); } catch { /* ignore */ }528      session.close();529    },530    remoteStaticPublicKey: hexToPubkey(expectedRemotePubkeyHex),531  };532}533534function equalAscii(left: string, right: string): boolean {535  const a = new TextEncoder().encode(left);536  const b = new TextEncoder().encode(right);537  if (a.length !== b.length) return false;538  let diff = 0;539  for (let i = 0; i < a.length; i += 1) diff |= a[i] ^ b[i];540  return diff === 0;541}542543function connectWebSocketUrl(544  url: string,545  targetLabel: string,546  timeoutMs: number,547): Promise<WebSocket> {548  return new Promise((resolve, reject) => {549    let ws: WebSocket;550    try {551      ws = new WebSocket(url);552    } catch (err) {553      reject(err as Error);554      return;555    }556    const t = setTimeout(() => {557      try { ws.close(); } catch { /* ignore */ }558      reject(new Error(`${targetLabel} WebSocket connect timed out`));559    }, timeoutMs);560    ws.onopen = () => {561      clearTimeout(t);562      resolve(ws);563    };564    ws.onerror = () => {565      clearTimeout(t);566      reject(new Error(`${targetLabel} WebSocket connect errored`));567    };568  });569}