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 class SyncTransportResourceLimitError extends Error {92  constructor() {93    super("sync transport resource limit exceeded");94    this.name = "SyncTransportResourceLimitError";95  }96}9798export interface SyncSessionOptions {99  connectTimeoutMs?: number;100  handshakeTimeoutMs?: number;101  signal?: AbortSignal;102}103104function encodePairIdentity(id: SyncPeerIdentity): Uint8Array {105  return new TextEncoder().encode(JSON.stringify({106    type: "PAIR_IDENTITY",107    device_id: id.deviceId,108    device_name: id.deviceName,109    device_type: id.deviceType,110    long_term_public_key_hex: id.longTermPublicKeyHex,111    signing_public_key_hex: id.signingPublicKeyHex,112  }));113}114115/**116 * Relay reconnect preserves the existing Noise KK wire protocol. The URL117 * builder enforces credential-free `wss://`.118 */119export async function openRelaySyncSession(120  relayUrl: string,121  ourIdentity: SyncPeerIdentity,122  expectedRemotePubkeyHex: string,123  rendezvousId: string,124  options: SyncSessionOptions = {},125): Promise<SyncSession> {126  return openSyncSessionAtUrl(127    buildRendezvousUrl(relayUrl, rendezvousId),128    "relay",129    ourIdentity,130    expectedRemotePubkeyHex,131    rendezvousId,132    options,133  );134}135136/**137 * Hold one relay room as the deterministic responder, then run Noise KK and138 * the bound PAIR_IDENTITY exchange. The zero-length liveness frame keeps a139 * Chrome 116+ MV3 service worker alive while the room is empty; the Peach140 * relay consumes that frame locally and never forwards it to the peer.141 */142export async function openRelaySyncResponderSession(143  relayUrl: string,144  ourIdentity: SyncPeerIdentity,145  expectedRemotePubkeyHex: string,146  rendezvousId: string,147  options: SyncSessionOptions & { keepaliveMs?: number } = {},148): Promise<SyncSession> {149  if (!/^[A-Za-z0-9_-]{22,128}$/.test(rendezvousId)) {150    throw new Error("reconnect requires a valid rotating rendezvous id");151  }152  if (ourIdentity.longTermPrivateKey.length !== 32) {153    throw new Error("local reconnect private key must be exactly 32 bytes");154  }155  const ws = await connectWebSocketUrl(156    buildRendezvousUrl(relayUrl, rendezvousId),157    "relay",158    options.connectTimeoutMs ?? DEFAULT_CONNECT_TIMEOUT_MS,159  );160  const inbox: Uint8Array[] = [];161  let inboxBytes = 0;162  const readers: Array<(b: Uint8Array) => void> = [];163  let closed = false;164  let closeError: Error = new Error("WebSocket closed");165  const keepaliveMs = options.keepaliveMs ?? RELAY_WAITER_KEEPALIVE_MS;166  const abortWaiter = () => {167    try { ws.close(1000, "relay waiter replaced"); } catch { /* ignore */ }168  };169  const keepalive = setInterval(() => {170    if (ws.readyState !== WebSocket.OPEN) return;171    try { ws.send(new Uint8Array(0)); } catch { /* onclose owns cleanup */ }172  }, keepaliveMs);173174  ws.binaryType = "arraybuffer";175  ws.onmessage = (event: MessageEvent) => {176    if (closed) return;177    if (!(event.data instanceof ArrayBuffer)) return;178    if (event.data.byteLength === 0) return;179    if (event.data.byteLength > SYNC_LIMITS.MAX_FRAME_BYTES) {180      closeError = new SyncFrameTooLargeError(181        `sync frame exceeds ${SYNC_LIMITS.MAX_FRAME_BYTES} bytes`,182      );183      try { ws.close(1009, "sync frame too large"); } catch { /* ignore */ }184      return;185    }186    const bytes = new Uint8Array(event.data);187    const reader = readers.shift();188    if (reader) reader(bytes);189    else if (190      inbox.length >= SYNC_LIMITS.MAX_INBOX_MESSAGES ||191      inboxBytes + bytes.byteLength > SYNC_LIMITS.MAX_INBOX_BYTES192    ) {193      closeError = new SyncTransportResourceLimitError();194      closed = true;195      inbox.length = 0;196      inboxBytes = 0;197      try { ws.close(1009, "sync inbox limit"); } catch { /* ignore */ }198    } else {199      inbox.push(bytes);200      inboxBytes += bytes.byteLength;201    }202  };203  ws.onerror = () => {204    try { ws.close(); } catch { /* ignore */ }205  };206  ws.onclose = (event: CloseEvent) => {207    clearInterval(keepalive);208    options.signal?.removeEventListener("abort", abortWaiter);209    closed = true;210    inbox.length = 0;211    inboxBytes = 0;212    if (event.code === 1009 && !(closeError instanceof SyncTransportResourceLimitError)) {213      closeError = new SyncFrameTooLargeError();214    }215    while (readers.length > 0) readers.shift()!(new Uint8Array(0));216  };217  // Install close handling before honoring an already-aborted signal so a218  // synchronous/mock close cannot strand the first receive() forever.219  options.signal?.addEventListener("abort", abortWaiter, { once: true });220  if (options.signal?.aborted) abortWaiter();221222  const send = async (bytes: Uint8Array): Promise<void> => {223    if (bytes.byteLength > SYNC_LIMITS.MAX_FRAME_BYTES) {224      throw new SyncFrameTooLargeError(225        `sync frame exceeds ${SYNC_LIMITS.MAX_FRAME_BYTES} bytes`,226      );227    }228    ws.send(bytes as Uint8Array<ArrayBuffer>);229  };230  const receive = (): Promise<Uint8Array> => {231    if (closed) return Promise.reject(closeError);232    const buffered = inbox.shift();233    if (buffered) {234      inboxBytes -= buffered.byteLength;235      return Promise.resolve(buffered);236    }237    return new Promise<Uint8Array>((resolve, reject) => {238      readers.push((bytes) => {239        if (bytes.length === 0 && closed) reject(closeError);240        else resolve(bytes);241      });242    });243  };244245  let noiseSession: NoiseNKSession | undefined;246  try {247    const selectorBytes = await receive();248    const selector = JSON.parse(new TextDecoder().decode(selectorBytes));249    if (250      selector?.type !== "PEACH_RECONNECT" ||251      typeof selector.rendezvous_id !== "string" ||252      !equalAscii(selector.rendezvous_id, rendezvousId)253    ) {254      throw new Error("reconnect capability does not match relay room");255    }256    const handshakeTimer = setTimeout(() => {257      closeError = new Error("sync_handshake_timeout");258      try { ws.close(1000, "sync handshake timeout"); } catch { /* ignore */ }259    }, options.handshakeTimeoutMs ?? DEFAULT_HANDSHAKE_TIMEOUT_MS);260    try {261      const ourPublicKey = hexToPubkey(ourIdentity.longTermPublicKeyHex);262      noiseSession = await noiseKKResponder(263        ourIdentity.longTermPrivateKey,264        ourPublicKey,265        hexToPubkey(expectedRemotePubkeyHex),266        send,267        receive,268      );269      const theirFrame = JSON.parse(270        new TextDecoder().decode(271          unpadSyncFrame(noiseSession.decrypt(await receive())),272        ),273      );274      if (theirFrame?.type !== "PAIR_IDENTITY") {275        throw new Error(`expected PAIR_IDENTITY, got ${theirFrame?.type as string}`);276      }277      if (theirFrame.long_term_public_key_hex !== expectedRemotePubkeyHex) {278        throw new Error("peer identity does not match authenticated KK key");279      }280      await send(noiseSession.encrypt(padSyncFrame(encodePairIdentity(ourIdentity))));281    } finally {282      clearTimeout(handshakeTimer);283    }284  } catch (error) {285    clearInterval(keepalive);286    options.signal?.removeEventListener("abort", abortWaiter);287    noiseSession?.close();288    try { ws.close(); } catch { /* ignore */ }289    throw error;290  }291292  return buildEncryptedSyncSession(293    ws,294    noiseSession,295    expectedRemotePubkeyHex,296    receive,297    () => {298      clearInterval(keepalive);299      options.signal?.removeEventListener("abort", abortWaiter);300    },301  );302}303304export async function openSyncSessionAtUrl(305  url: string,306  targetLabel: string,307  ourIdentity: SyncPeerIdentity,308  expectedRemotePubkeyHex: string,309  rendezvousId: string,310  options: SyncSessionOptions = {},311): Promise<SyncSession> {312  if (!/^[A-Za-z0-9_-]{22,128}$/.test(rendezvousId)) {313    throw new Error("reconnect requires a valid rotating rendezvous id");314  }315  const ws = await connectWebSocketUrl(316    url,317    targetLabel,318    options.connectTimeoutMs ?? DEFAULT_CONNECT_TIMEOUT_MS,319  );320  const inbox: Uint8Array[] = [];321  let inboxBytes = 0;322  const readers: Array<(b: Uint8Array) => void> = [];323  let closed = false;324  let closeError: Error = new Error("WebSocket closed");325326  ws.binaryType = "arraybuffer";327  ws.onmessage = (ev: MessageEvent) => {328    if (closed) return;329    const data = ev.data;330    if (data instanceof ArrayBuffer) {331      // MV-3: refuse oversized frames before they are buffered or decrypted.332      // A peer that sends one is misbehaving — tear the session down rather333      // than letting the inbox grow toward OOM.334      if (data.byteLength > SYNC_LIMITS.MAX_FRAME_BYTES) {335        // MV-3: oversized frame — tear the session down. close() routes through336        // onclose, which rejects any parked receive(), so no error array is337        // needed (PCH-01-026 #9).338        console.warn(339          `[Peach] sync frame exceeds max size (${SYNC_LIMITS.MAX_FRAME_BYTES} bytes); closing`,340        );341        closeError = new SyncFrameTooLargeError(342          `sync frame exceeds ${SYNC_LIMITS.MAX_FRAME_BYTES} bytes`,343        );344        try { ws.close(1009, "sync frame too large"); } catch { /* ignore */ }345        return;346      }347      const bytes = new Uint8Array(data);348      const r = readers.shift();349      if (r) r(bytes);350      else if (351        inbox.length >= SYNC_LIMITS.MAX_INBOX_MESSAGES ||352        inboxBytes + bytes.byteLength > SYNC_LIMITS.MAX_INBOX_BYTES353      ) {354        closeError = new SyncTransportResourceLimitError();355        closed = true;356        inbox.length = 0;357        inboxBytes = 0;358        try { ws.close(1009, "sync inbox limit"); } catch { /* ignore */ }359      } else {360        inbox.push(bytes);361        inboxBytes += bytes.byteLength;362      }363    }364  };365  ws.onerror = () => {366    // PCH-01-026 #9: surface the error and unblock any parked receive() by367    // closing. Browsers usually follow error with close, but not always; an368    // explicit close guarantees the handshake fails fast instead of hanging.369    console.warn("[Peach] sync WebSocket error; closing transport");370    try { ws.close(); } catch { /* ignore */ }371  };372  ws.onclose = (event: CloseEvent) => {373    closed = true;374    inbox.length = 0;375    inboxBytes = 0;376    if (event.code === 1009 && !(closeError instanceof SyncTransportResourceLimitError)) {377      closeError = new SyncFrameTooLargeError();378    }379    while (readers.length > 0) {380      const r = readers.shift()!;381      r(new Uint8Array(0));382    }383  };384385  function send(bytes: Uint8Array): Promise<void> {386    return new Promise((resolve, reject) => {387      if (bytes.byteLength > SYNC_LIMITS.MAX_FRAME_BYTES) {388        reject(389          new SyncFrameTooLargeError(390            `sync frame exceeds ${SYNC_LIMITS.MAX_FRAME_BYTES} bytes`,391          ),392        );393        return;394      }395      try {396        ws.send(bytes as Uint8Array<ArrayBuffer>);397        resolve();398      } catch (err) {399        reject(err as Error);400      }401    });402  }403  function receive(): Promise<Uint8Array> {404    if (closed) return Promise.reject(closeError);405    const v = inbox.shift();406    if (v) {407      inboxBytes -= v.byteLength;408      return Promise.resolve(v);409    }410    return new Promise<Uint8Array>((resolve, reject) => {411      readers.push((b) => {412        if (b.length === 0 && closed) reject(closeError);413        else resolve(b);414      });415    });416  }417418  const handshakeTimer = setTimeout(() => {419    closeError = new Error("sync_handshake_timeout");420    try { ws.close(1000, "sync handshake timeout"); } catch { /* ignore */ }421  }, options.handshakeTimeoutMs ?? DEFAULT_HANDSHAKE_TIMEOUT_MS);422  let session: NoiseNKSession;423  try {424    const ourPublicKey = hexToPubkey(ourIdentity.longTermPublicKeyHex);425    if (ourIdentity.longTermPrivateKey.length !== 32) {426      throw new Error("local reconnect private key must be exactly 32 bytes");427    }428    // This opaque rotating capability lets Android select paired local state429    // without exposing either public key to a relay. It grants no identity430    // trust; KK still proves both private keys immediately afterwards.431    await send(new TextEncoder().encode(JSON.stringify({432      type: "PEACH_RECONNECT",433      rendezvous_id: rendezvousId,434    })));435    session = await noiseKKInitiator(436      ourIdentity.longTermPrivateKey,437      ourPublicKey,438      hexToPubkey(expectedRemotePubkeyHex),439      send,440      receive,441    );442  } catch (err) {443    clearTimeout(handshakeTimer);444    try { ws.close(); } catch { /* ignore */ }445    throw err;446  }447448  try {449    await send(session.encrypt(padSyncFrame(encodePairIdentity(ourIdentity))));450    const theirFrame = JSON.parse(451      new TextDecoder().decode(unpadSyncFrame(session.decrypt(await receive()))),452    );453    if (theirFrame?.type !== "PAIR_IDENTITY") {454      throw new Error(`expected PAIR_IDENTITY, got ${theirFrame?.type as string}`);455    }456    if (theirFrame.long_term_public_key_hex !== expectedRemotePubkeyHex) {457      throw new Error("peer identity does not match authenticated KK key");458    }459    clearTimeout(handshakeTimer);460  } catch (err) {461    clearTimeout(handshakeTimer);462    session.close();463    try { ws.close(); } catch { /* ignore */ }464    throw err;465  }466467  return buildEncryptedSyncSession(468    ws,469    session,470    expectedRemotePubkeyHex,471    receive,472  );473}474475// Exported for the PB-26-002 reassembly-bound regression test.476export function buildEncryptedSyncSession(477  ws: WebSocket,478  session: NoiseNKSession,479  expectedRemotePubkeyHex: string,480  receiveCiphertext: () => Promise<Uint8Array>,481  beforeClose: () => void = () => undefined,482): SyncSession {483  return {484    async send(plaintext: Uint8Array) {485      // E7: bucket the size before encrypting, so the relay sees frame shapes486      // that no longer distinguish one operation (or vault size) from another.487      // E7b: a padded frame can legitimately be up to 24 MiB, but Noise caps a488      // single transport message at 65535 bytes (see crypto-core/src/lib.rs —489      // 65519 plaintext, 16-byte tag). We split a too-large padded frame into490      // numbered Noise messages and tag every chunk after the first with a491      // 1-byte continuation header. The receiver reassembles by reading the492      // padded frame's 4-byte length header, computing the same bucket from493      // the same ladder, and keeping reading until the bucket is full.494      const padded = padSyncFrame(plaintext);495      if (padded.byteLength > SYNC_LIMITS.MAX_FRAME_BYTES) {496        throw new SyncFrameTooLargeError(497          `sync frame exceeds ${SYNC_LIMITS.MAX_FRAME_BYTES} bytes`,498        );499      }500      // A single-chunk frame goes out as one Noise message, exactly as before501      // — no continuation byte, no per-chunk overhead, so the common small502      // case is unchanged on the wire.503      if (padded.byteLength <= MAX_NOISE_PLAINTEXT_BYTES) {504        const ct = session.encrypt(padded);505        ws.send(ct as Uint8Array<ArrayBuffer>);506        return;507      }508      for (const plaintext of chunkPaddedSyncFrame(padded)) {509        if (plaintext.byteLength > MAX_NOISE_PLAINTEXT_BYTES) {510          throw new Error(511            `chunk exceeds Noise limit after continuation byte: ${plaintext.byteLength}`,512          );513        }514        const ct = session.encrypt(plaintext);515        if (ct.byteLength > SYNC_LIMITS.MAX_FRAME_BYTES) {516          throw new SyncFrameTooLargeError(517            `sync frame exceeds ${SYNC_LIMITS.MAX_FRAME_BYTES} bytes`,518          );519        }520        ws.send(ct as Uint8Array<ArrayBuffer>);521      }522    },523    async receive() {524      // The first Noise message of a frame is the first one we read. If it525      // begins with the continuation marker, the message is one of N chunks526      // — the padded-frame length header comes inside the first chunk's527      // payload, so we read it there to learn the total.528      let buffered: Uint8Array | null = null;529      while (true) {530        const ct = await receiveCiphertext();531        const pt = session.decrypt(ct);532        if (buffered === null && pt.byteLength > 0 && pt[0] === FRAME_CONTINUATION_MARKER) {533          // Strip the continuation byte and keep the payload.534          buffered = pt.subarray(1);535          continue;536        }537        if (buffered !== null) {538          if (pt.byteLength > 0 && pt[0] === FRAME_CONTINUATION_MARKER) {539            const merged: Uint8Array = new Uint8Array(buffered.length + pt.byteLength - 1);540            merged.set(buffered, 0);541            merged.set(pt.subarray(1), buffered.length);542            buffered = merged;543            // PB-26-002: the frame bound must hold DURING reassembly. The544            // post-receive check in the coordinator runs only after the full545            // frame is in memory, so a peer that completes the Noise KK546            // handshake could otherwise stream continuation chunks without547            // bound inside the session window.548            if (buffered.length > SYNC_LIMITS.MAX_FRAME_BYTES) {549              throw new SyncFrameTooLargeError(550                `sync frame exceeds ${SYNC_LIMITS.MAX_FRAME_BYTES} bytes`,551              );552            }553            continue;554          }555          // Final chunk. PB-26-602: the continuation loop only bounds what is556          // buffered, so the last chunk could push the frame up to 64 KiB past557          // the cap. Bound the concatenation, not just the buffer.558          if (buffered.length + pt.byteLength > SYNC_LIMITS.MAX_FRAME_BYTES) {559            throw new SyncFrameTooLargeError(560              `sync frame exceeds ${SYNC_LIMITS.MAX_FRAME_BYTES} bytes`,561            );562          }563          const reassembled = new Uint8Array(buffered.length + pt.byteLength);564          reassembled.set(buffered, 0);565          reassembled.set(pt, buffered.length);566          return unpadSyncFrame(reassembled);567        }568        // Single-chunk frame, exactly as before.569        return unpadSyncFrame(pt);570      }571    },572    async close() {573      beforeClose();574      try { ws.close(); } catch { /* ignore */ }575      session.close();576    },577    remoteStaticPublicKey: hexToPubkey(expectedRemotePubkeyHex),578  };579}580581function equalAscii(left: string, right: string): boolean {582  const a = new TextEncoder().encode(left);583  const b = new TextEncoder().encode(right);584  if (a.length !== b.length) return false;585  let diff = 0;586  for (let i = 0; i < a.length; i += 1) diff |= a[i] ^ b[i];587  return diff === 0;588}589590function connectWebSocketUrl(591  url: string,592  targetLabel: string,593  timeoutMs: number,594): Promise<WebSocket> {595  return new Promise((resolve, reject) => {596    let ws: WebSocket;597    try {598      ws = new WebSocket(url);599    } catch (err) {600      reject(err as Error);601      return;602    }603    const t = setTimeout(() => {604      try { ws.close(); } catch { /* ignore */ }605      reject(new Error(`${targetLabel} WebSocket connect timed out`));606    }, timeoutMs);607    ws.onopen = () => {608      clearTimeout(t);609      resolve(ws);610    };611    ws.onerror = () => {612      clearTimeout(t);613      reject(new Error(`${targetLabel} WebSocket connect errored`));614    };615  });616}