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 {27  padSyncFrame,28  unpadSyncFrame,29  MAX_NOISE_PLAINTEXT_BYTES,30  FRAME_CONTINUATION_MARKER,31  chunkPaddedSyncFrame,32} from "./sync-padding";33import { hexToPubkey } from "./sync-identity";34import { buildRendezvousUrl } from "./rendezvous";35import type { DeviceType } from "./sync-types";3637// E6 (session rekeying) is deliberately NOT wired, and this is a reality check38// against the spec rather than an omission. E6 targets "long-lived sync39// sessions"; the browser has none. Sessions opened here are one-shot — roughly40// ten frames each way, a 30s coordinator timeout (SESSION_TIMEOUT_MS), then41// sync-runner closes them — and every reconnect runs a fresh Noise KK handshake42// with fresh ephemerals. (Android's listener does park a connection open after43// a round so it can act as a relay hub for another native peer, but the browser44// drives the lifetime of every session it opens, and closes it.) That already45// gives stronger post-compromise security than intra-session Rekey() would,46// because no two sessions share key material at all (pinned by the E6 test in47// noise-kk.test.ts).48//49// Wiring Rekey() anyway would mean both ends rekeying at the identical frame50// index with no wire signal to resynchronise on. One frame of drift between51// browser and Android breaks all sync — the exact failure class this protocol52// has already been burned by twice. The primitives are exposed on NoiseSession53// (rekeyOutgoing/rekeyIncoming) if a genuinely long-lived session ever appears;54// until then the cost is real and the benefit is zero.55const DEFAULT_PROBE_TIMEOUT_MS = 500;56const DEFAULT_PROBE_BATCH = 16;57const DEFAULT_CONNECT_TIMEOUT_MS = 4_000;58const DEFAULT_HANDSHAKE_TIMEOUT_MS = 10_000;59/** v2 brief §5.2 — subnet probe is rate-limited to once per 60 seconds. */60export const SUBNET_PROBE_COOLDOWN_MS = 60_000;6162export interface SyncSession {63  send(plaintext: Uint8Array): void;64  receive(): Promise<Uint8Array>;65  close(): Promise<void>;66  remoteStaticPublicKey: Uint8Array;67}6869export interface SyncEndpoint {70  ip: string;71  port: number;72}7374export interface ProbedSyncEndpoint extends SyncEndpoint {75  /** Exact clock-skew window capability that completed KK at this endpoint. */76  rendezvousId: string;77}7879/** Our identity, repeated in the post-handshake PAIR_IDENTITY frame and bound80 *  to the local static key already authenticated by KK. */81export interface SyncPeerIdentity {82  deviceId: string;83  deviceName: string;84  deviceType: DeviceType;85  longTermPublicKeyHex: string;86  /** Local-only key material for the KK reconnect handshake; never encoded. */87  longTermPrivateKey: Uint8Array;88  signingPublicKeyHex?: string;89}9091export class SyncFrameTooLargeError extends Error {92  constructor(message = "sync frame too large (WebSocket close 1009)") {93    super(message);94    this.name = "SyncFrameTooLargeError";95  }96}9798export interface SyncSessionOptions {99  connectTimeoutMs?: number;100  handshakeTimeoutMs?: number;101}102103function encodePairIdentity(id: SyncPeerIdentity): Uint8Array {104  return new TextEncoder().encode(JSON.stringify({105    type: "PAIR_IDENTITY",106    device_id: id.deviceId,107    device_name: id.deviceName,108    device_type: id.deviceType,109    long_term_public_key_hex: id.longTermPublicKeyHex,110    signing_public_key_hex: id.signingPublicKeyHex,111  }));112}113114/**115 * Open a WebSocket to (ip, port), run Noise KK against both paired static116 * keys, exchange bound PAIR_IDENTITY frames, and return a ready-to-use117 * session.118 *119 * On a successful return, `setLastKnownIp` in the caller can persist120 * (ip, port) for the next session's fast path.121 */122export async function openSyncSession(123  endpoint: SyncEndpoint,124  ourIdentity: SyncPeerIdentity,125  expectedRemotePubkeyHex: string,126  rendezvousId: string,127  options: SyncSessionOptions = {},128): Promise<SyncSession> {129  return openSyncSessionAtUrl(130    `ws://${endpoint.ip}:${endpoint.port}/`,131    "LAN peer",132    ourIdentity,133    expectedRemotePubkeyHex,134    rendezvousId,135    options,136  );137}138139/**140 * Relay reconnect preserves the exact LAN wire protocol; only the WebSocket141 * target changes. The URL builder enforces credential-free `wss://`.142 */143export async function openRelaySyncSession(144  relayUrl: string,145  ourIdentity: SyncPeerIdentity,146  expectedRemotePubkeyHex: string,147  rendezvousId: string,148  options: SyncSessionOptions = {},149): Promise<SyncSession> {150  return openSyncSessionAtUrl(151    buildRendezvousUrl(relayUrl, rendezvousId),152    "relay",153    ourIdentity,154    expectedRemotePubkeyHex,155    rendezvousId,156    options,157  );158}159160async function openSyncSessionAtUrl(161  url: string,162  targetLabel: string,163  ourIdentity: SyncPeerIdentity,164  expectedRemotePubkeyHex: string,165  rendezvousId: string,166  options: SyncSessionOptions,167): Promise<SyncSession> {168  if (!/^[A-Za-z0-9_-]{22,128}$/.test(rendezvousId)) {169    throw new Error("reconnect requires a valid rotating rendezvous id");170  }171  const ws = await connectWebSocketUrl(172    url,173    targetLabel,174    options.connectTimeoutMs ?? DEFAULT_CONNECT_TIMEOUT_MS,175  );176  const inbox: Uint8Array[] = [];177  const readers: Array<(b: Uint8Array) => void> = [];178  let closed = false;179  let closeError: Error = new Error("WebSocket closed");180181  ws.binaryType = "arraybuffer";182  ws.onmessage = (ev: MessageEvent) => {183    const data = ev.data;184    if (data instanceof ArrayBuffer) {185      // MV-3: refuse oversized frames before they are buffered or decrypted.186      // A peer that sends one is misbehaving — tear the session down rather187      // than letting the inbox grow toward OOM.188      if (data.byteLength > SYNC_LIMITS.MAX_FRAME_BYTES) {189        // MV-3: oversized frame — tear the session down. close() routes through190        // onclose, which rejects any parked receive(), so no error array is191        // needed (PCH-01-026 #9).192        console.warn(193          `[Peach] sync frame exceeds max size (${SYNC_LIMITS.MAX_FRAME_BYTES} bytes); closing`,194        );195        closeError = new SyncFrameTooLargeError(196          `sync frame exceeds ${SYNC_LIMITS.MAX_FRAME_BYTES} bytes`,197        );198        try { ws.close(1009, "sync frame too large"); } catch { /* ignore */ }199        return;200      }201      const bytes = new Uint8Array(data);202      const r = readers.shift();203      if (r) r(bytes);204      else inbox.push(bytes);205    }206  };207  ws.onerror = () => {208    // PCH-01-026 #9: surface the error and unblock any parked receive() by209    // closing. Browsers usually follow error with close, but not always; an210    // explicit close guarantees the handshake fails fast instead of hanging.211    console.warn("[Peach] sync WebSocket error; closing transport");212    try { ws.close(); } catch { /* ignore */ }213  };214  ws.onclose = (event: CloseEvent) => {215    closed = true;216    if (event.code === 1009) {217      closeError = new SyncFrameTooLargeError();218    }219    while (readers.length > 0) {220      const r = readers.shift()!;221      r(new Uint8Array(0));222    }223  };224225  function send(bytes: Uint8Array): Promise<void> {226    return new Promise((resolve, reject) => {227      if (bytes.byteLength > SYNC_LIMITS.MAX_FRAME_BYTES) {228        reject(229          new SyncFrameTooLargeError(230            `sync frame exceeds ${SYNC_LIMITS.MAX_FRAME_BYTES} bytes`,231          ),232        );233        return;234      }235      try {236        ws.send(bytes as Uint8Array<ArrayBuffer>);237        resolve();238      } catch (err) {239        reject(err as Error);240      }241    });242  }243  function receive(): Promise<Uint8Array> {244    if (closed) return Promise.reject(closeError);245    const v = inbox.shift();246    if (v) return Promise.resolve(v);247    return new Promise<Uint8Array>((resolve, reject) => {248      readers.push((b) => {249        if (b.length === 0 && closed) reject(closeError);250        else resolve(b);251      });252    });253  }254255  const handshakeTimer = setTimeout(() => {256    closeError = new Error("sync_handshake_timeout");257    try { ws.close(1000, "sync handshake timeout"); } catch { /* ignore */ }258  }, options.handshakeTimeoutMs ?? DEFAULT_HANDSHAKE_TIMEOUT_MS);259  let session: NoiseNKSession;260  try {261    const ourPublicKey = hexToPubkey(ourIdentity.longTermPublicKeyHex);262    if (ourIdentity.longTermPrivateKey.length !== 32) {263      throw new Error("local reconnect private key must be exactly 32 bytes");264    }265    // This opaque rotating capability lets Android select paired local state266    // without exposing either public key to a relay. It grants no identity267    // trust; KK still proves both private keys immediately afterwards.268    await send(new TextEncoder().encode(JSON.stringify({269      type: "PEACH_RECONNECT",270      rendezvous_id: rendezvousId,271    })));272    session = await noiseKKInitiator(273      ourIdentity.longTermPrivateKey,274      ourPublicKey,275      hexToPubkey(expectedRemotePubkeyHex),276      send,277      receive,278    );279  } catch (err) {280    clearTimeout(handshakeTimer);281    try { ws.close(); } catch { /* ignore */ }282    throw err;283  }284285  try {286    await send(session.encrypt(padSyncFrame(encodePairIdentity(ourIdentity))));287    const theirFrame = JSON.parse(288      new TextDecoder().decode(unpadSyncFrame(session.decrypt(await receive()))),289    );290    if (theirFrame?.type !== "PAIR_IDENTITY") {291      throw new Error(`expected PAIR_IDENTITY, got ${theirFrame?.type as string}`);292    }293    if (theirFrame.long_term_public_key_hex !== expectedRemotePubkeyHex) {294      throw new Error("peer identity does not match authenticated KK key");295    }296    clearTimeout(handshakeTimer);297  } catch (err) {298    clearTimeout(handshakeTimer);299    session.close();300    try { ws.close(); } catch { /* ignore */ }301    throw err;302  }303304  return {305    async send(plaintext: Uint8Array) {306      // E7: bucket the size before encrypting, so the relay sees frame shapes307      // that no longer distinguish one operation (or vault size) from another.308      // E7b: a padded frame can legitimately be up to 24 MiB, but Noise caps a309      // single transport message at 65535 bytes (see crypto-core/src/lib.rs —310      // 65519 plaintext, 16-byte tag). We split a too-large padded frame into311      // numbered Noise messages and tag every chunk after the first with a312      // 1-byte continuation header. The receiver reassembles by reading the313      // padded frame's 4-byte length header, computing the same bucket from314      // the same ladder, and keeping reading until the bucket is full.315      const padded = padSyncFrame(plaintext);316      if (padded.byteLength > SYNC_LIMITS.MAX_FRAME_BYTES) {317        throw new SyncFrameTooLargeError(318          `sync frame exceeds ${SYNC_LIMITS.MAX_FRAME_BYTES} bytes`,319        );320      }321      // A single-chunk frame goes out as one Noise message, exactly as before322      // — no continuation byte, no per-chunk overhead, so the common small323      // case is unchanged on the wire.324      if (padded.byteLength <= MAX_NOISE_PLAINTEXT_BYTES) {325        const ct = session.encrypt(padded);326        ws.send(ct as Uint8Array<ArrayBuffer>);327        return;328      }329      for (const plaintext of chunkPaddedSyncFrame(padded)) {330        if (plaintext.byteLength > MAX_NOISE_PLAINTEXT_BYTES) {331          throw new Error(332            `chunk exceeds Noise limit after continuation byte: ${plaintext.byteLength}`,333          );334        }335        const ct = session.encrypt(plaintext);336        if (ct.byteLength > SYNC_LIMITS.MAX_FRAME_BYTES) {337          throw new SyncFrameTooLargeError(338            `sync frame exceeds ${SYNC_LIMITS.MAX_FRAME_BYTES} bytes`,339          );340        }341        ws.send(ct as Uint8Array<ArrayBuffer>);342      }343    },344    async receive() {345      // The first Noise message of a frame is the first one we read. If it346      // begins with the continuation marker, the message is one of N chunks347      // — the padded-frame length header comes inside the first chunk's348      // payload, so we read it there to learn the total.349      let buffered: Uint8Array | null = null;350      while (true) {351        const ct = await receive();352        const pt = session.decrypt(ct);353        if (buffered === null && pt.byteLength > 0 && pt[0] === FRAME_CONTINUATION_MARKER) {354          // Strip the continuation byte and keep the payload.355          buffered = pt.subarray(1);356          continue;357        }358        if (buffered !== null) {359          if (pt.byteLength > 0 && pt[0] === FRAME_CONTINUATION_MARKER) {360            const merged: Uint8Array = new Uint8Array(buffered.length + pt.byteLength - 1);361            merged.set(buffered, 0);362            merged.set(pt.subarray(1), buffered.length);363            buffered = merged;364            continue;365          }366          // Final chunk. Concatenate without a continuation byte, then fall367          // through to the unpadder.368          const reassembled = new Uint8Array(buffered.length + pt.byteLength);369          reassembled.set(buffered, 0);370          reassembled.set(pt, buffered.length);371          return unpadSyncFrame(reassembled);372        }373        // Single-chunk frame, exactly as before.374        return unpadSyncFrame(pt);375      }376    },377    async close() {378      try { ws.close(); } catch { /* ignore */ }379      session.close();380    },381    remoteStaticPublicKey: hexToPubkey(expectedRemotePubkeyHex),382  };383}384385function connectWebSocketUrl(386  url: string,387  targetLabel: string,388  timeoutMs: number,389): Promise<WebSocket> {390  return new Promise((resolve, reject) => {391    let ws: WebSocket;392    try {393      ws = new WebSocket(url);394    } catch (err) {395      reject(err as Error);396      return;397    }398    const t = setTimeout(() => {399      try { ws.close(); } catch { /* ignore */ }400      reject(new Error(`${targetLabel} WebSocket connect timed out`));401    }, timeoutMs);402    ws.onopen = () => {403      clearTimeout(t);404      resolve(ws);405    };406    ws.onerror = () => {407      clearTimeout(t);408      reject(new Error(`${targetLabel} WebSocket connect errored`));409    };410  });411}412413/**414 * Per-pubkey cooldown tracker for the subnet probe. v2 brief §5.2 caps415 * full /24 scans at once per minute so an extension whose cached IP has416 * gone stale doesn't hammer the LAN with handshake attempts every time417 * the popup opens.418 *419 * Keyed by the expected remote pubkey because each paired peer is probed420 * independently; the rate limit applies to "scan looking for THIS peer",421 * not "scan the subnet at all".422 */423const lastProbeAt = new Map<string, number>();424425export function subnetProbeOnCooldown(426  expectedRemotePubkeyHex: string,427  now: number = Date.now(),428  cooldownMs: number = SUBNET_PROBE_COOLDOWN_MS,429): boolean {430  const last = lastProbeAt.get(expectedRemotePubkeyHex);431  return last !== undefined && now - last < cooldownMs;432}433434/** Test-only — clear the per-pubkey probe cooldown ledger. */435export function _resetSubnetProbeCooldown(): void {436  lastProbeAt.clear();437}438439/**440 * Subnet probe — scan a /24 looking for a host that completes Noise KK441 * with the expected long-term static key. Returns the first match, or442 * null if the cooldown is still active or no host responded.443 *444 * Rate-limited to once per `SUBNET_PROBE_COOLDOWN_MS` per peer pubkey445 * (v2 brief §5.2). Callers should still attempt the cached446 * `last_known_ip`/`last_known_port` directly first — the subnet probe is447 * fallback only.448 *449 * For each IP we attempt the full Noise KK handshake, but with a tight450 * timeout: dead hosts fail at the WebSocket connect step in <1ms (browser451 * surfaces ECONNREFUSED immediately) or by the connectTimeout otherwise.452 */453export async function probeSubnetForPeer(454  subnet: string, // e.g. "192.168.1"455  port: number,456  ourIdentity: SyncPeerIdentity,457  expectedRemotePubkeyHex: string,458  rendezvousIds: readonly string[],459  options: { batch?: number; timeoutMs?: number; cooldownMs?: number; now?: number } = {},460): Promise<ProbedSyncEndpoint | null> {461  if (rendezvousIds.length === 0) {462    throw new Error("subnet probe requires at least one rendezvous id");463  }464  const now = options.now ?? Date.now();465  const cooldownMs = options.cooldownMs ?? SUBNET_PROBE_COOLDOWN_MS;466  if (subnetProbeOnCooldown(expectedRemotePubkeyHex, now, cooldownMs)) {467    return null;468  }469  lastProbeAt.set(expectedRemotePubkeyHex, now);470  const batchSize = options.batch ?? DEFAULT_PROBE_BATCH;471  const timeoutMs = options.timeoutMs ?? DEFAULT_PROBE_TIMEOUT_MS;472  for (let start = 1; start <= 254; start += batchSize) {473    const end = Math.min(start + batchSize - 1, 254);474    const ips: string[] = [];475    for (let i = start; i <= end; i++) ips.push(`${subnet}.${i}`);476477    const results = await Promise.all(ips.map((ip) => probeOne(478      { ip, port },479      ourIdentity,480      expectedRemotePubkeyHex,481      rendezvousIds,482      timeoutMs,483    )));484    const hit = results.find((r): r is ProbedSyncEndpoint => r !== null);485    if (hit) return hit;486  }487  return null;488}489490async function probeOne(491  endpoint: SyncEndpoint,492  ourIdentity: SyncPeerIdentity,493  expectedRemotePubkeyHex: string,494  rendezvousIds: readonly string[],495  timeoutMs: number,496): Promise<ProbedSyncEndpoint | null> {497  for (const rendezvousId of rendezvousIds) {498    try {499      const session = await openSyncSession(500        endpoint,501        ourIdentity,502        expectedRemotePubkeyHex,503        rendezvousId,504        { connectTimeoutMs: timeoutMs },505      );506      await session.close();507      return { ...endpoint, rendezvousId };508    } catch {509      // Try the next locally-derived clock-skew window at the same host.510    }511  }512  return null;513}