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} from "./sync-padding";32import { hexToPubkey } from "./sync-identity";33import { buildRendezvousUrl } from "./rendezvous";34import type { DeviceType } from "./sync-types";3536// E6 (session rekeying) is deliberately NOT wired, and this is a reality check37// against the spec rather than an omission. E6 targets "long-lived sync38// sessions"; the browser has none. Sessions opened here are one-shot — roughly39// ten frames each way, a 30s coordinator timeout (SESSION_TIMEOUT_MS), then40// sync-runner closes them — and every reconnect runs a fresh Noise KK handshake41// with fresh ephemerals. (Android's listener does park a connection open after42// a round so it can act as a relay hub for another native peer, but the browser43// drives the lifetime of every session it opens, and closes it.) That already44// gives stronger post-compromise security than intra-session Rekey() would,45// because no two sessions share key material at all (pinned by the E6 test in46// noise-kk.test.ts).47//48// Wiring Rekey() anyway would mean both ends rekeying at the identical frame49// index with no wire signal to resynchronise on. One frame of drift between50// browser and Android breaks all sync — the exact failure class this protocol51// has already been burned by twice. The primitives are exposed on NoiseSession52// (rekeyOutgoing/rekeyIncoming) if a genuinely long-lived session ever appears;53// until then the cost is real and the benefit is zero.54const DEFAULT_PROBE_TIMEOUT_MS = 500;55const DEFAULT_PROBE_BATCH = 16;56const DEFAULT_CONNECT_TIMEOUT_MS = 4_000;57const DEFAULT_HANDSHAKE_TIMEOUT_MS = 10_000;58/** v2 brief §5.2 — subnet probe is rate-limited to once per 60 seconds. */59export const SUBNET_PROBE_COOLDOWN_MS = 60_000;6061export interface SyncSession {62  send(plaintext: Uint8Array): void;63  receive(): Promise<Uint8Array>;64  close(): Promise<void>;65  remoteStaticPublicKey: Uint8Array;66}6768export interface SyncEndpoint {69  ip: string;70  port: number;71}7273export interface ProbedSyncEndpoint extends SyncEndpoint {74  /** Exact clock-skew window capability that completed KK at this endpoint. */75  rendezvousId: string;76}7778/** Our identity, repeated in the post-handshake PAIR_IDENTITY frame and bound79 *  to the local static key already authenticated by KK. */80export interface SyncPeerIdentity {81  deviceId: string;82  deviceName: string;83  deviceType: DeviceType;84  longTermPublicKeyHex: string;85  /** Local-only key material for the KK reconnect handshake; never encoded. */86  longTermPrivateKey: Uint8Array;87  signingPublicKeyHex?: string;88}8990export class SyncFrameTooLargeError extends Error {91  constructor(message = "sync frame too large (WebSocket close 1009)") {92    super(message);93    this.name = "SyncFrameTooLargeError";94  }95}9697export interface SyncSessionOptions {98  connectTimeoutMs?: number;99  handshakeTimeoutMs?: number;100}101102function encodePairIdentity(id: SyncPeerIdentity): Uint8Array {103  return new TextEncoder().encode(JSON.stringify({104    type: "PAIR_IDENTITY",105    device_id: id.deviceId,106    device_name: id.deviceName,107    device_type: id.deviceType,108    long_term_public_key_hex: id.longTermPublicKeyHex,109    signing_public_key_hex: id.signingPublicKeyHex,110  }));111}112113/**114 * Open a WebSocket to (ip, port), run Noise KK against both paired static115 * keys, exchange bound PAIR_IDENTITY frames, and return a ready-to-use116 * session.117 *118 * On a successful return, `setLastKnownIp` in the caller can persist119 * (ip, port) for the next session's fast path.120 */121export async function openSyncSession(122  endpoint: SyncEndpoint,123  ourIdentity: SyncPeerIdentity,124  expectedRemotePubkeyHex: string,125  rendezvousId: string,126  options: SyncSessionOptions = {},127): Promise<SyncSession> {128  return openSyncSessionAtUrl(129    `ws://${endpoint.ip}:${endpoint.port}/`,130    "LAN peer",131    ourIdentity,132    expectedRemotePubkeyHex,133    rendezvousId,134    options,135  );136}137138/**139 * Relay reconnect preserves the exact LAN wire protocol; only the WebSocket140 * target changes. The URL builder enforces credential-free `wss://`.141 */142export async function openRelaySyncSession(143  relayUrl: string,144  ourIdentity: SyncPeerIdentity,145  expectedRemotePubkeyHex: string,146  rendezvousId: string,147  options: SyncSessionOptions = {},148): Promise<SyncSession> {149  return openSyncSessionAtUrl(150    buildRendezvousUrl(relayUrl, rendezvousId),151    "relay",152    ourIdentity,153    expectedRemotePubkeyHex,154    rendezvousId,155    options,156  );157}158159async function openSyncSessionAtUrl(160  url: string,161  targetLabel: string,162  ourIdentity: SyncPeerIdentity,163  expectedRemotePubkeyHex: string,164  rendezvousId: string,165  options: SyncSessionOptions,166): Promise<SyncSession> {167  if (!/^[A-Za-z0-9_-]{22,128}$/.test(rendezvousId)) {168    throw new Error("reconnect requires a valid rotating rendezvous id");169  }170  const ws = await connectWebSocketUrl(171    url,172    targetLabel,173    options.connectTimeoutMs ?? DEFAULT_CONNECT_TIMEOUT_MS,174  );175  const inbox: Uint8Array[] = [];176  const readers: Array<(b: Uint8Array) => void> = [];177  let closed = false;178  let closeError: Error = new Error("WebSocket closed");179180  ws.binaryType = "arraybuffer";181  ws.onmessage = (ev: MessageEvent) => {182    const data = ev.data;183    if (data instanceof ArrayBuffer) {184      // MV-3: refuse oversized frames before they are buffered or decrypted.185      // A peer that sends one is misbehaving — tear the session down rather186      // than letting the inbox grow toward OOM.187      if (data.byteLength > SYNC_LIMITS.MAX_FRAME_BYTES) {188        // MV-3: oversized frame — tear the session down. close() routes through189        // onclose, which rejects any parked receive(), so no error array is190        // needed (PCH-01-026 #9).191        console.warn(192          `[Peach] sync frame exceeds max size (${SYNC_LIMITS.MAX_FRAME_BYTES} bytes); closing`,193        );194        closeError = new SyncFrameTooLargeError(195          `sync frame exceeds ${SYNC_LIMITS.MAX_FRAME_BYTES} bytes`,196        );197        try { ws.close(1009, "sync frame too large"); } catch { /* ignore */ }198        return;199      }200      const bytes = new Uint8Array(data);201      const r = readers.shift();202      if (r) r(bytes);203      else inbox.push(bytes);204    }205  };206  ws.onerror = () => {207    // PCH-01-026 #9: surface the error and unblock any parked receive() by208    // closing. Browsers usually follow error with close, but not always; an209    // explicit close guarantees the handshake fails fast instead of hanging.210    console.warn("[Peach] sync WebSocket error; closing transport");211    try { ws.close(); } catch { /* ignore */ }212  };213  ws.onclose = (event: CloseEvent) => {214    closed = true;215    if (event.code === 1009) {216      closeError = new SyncFrameTooLargeError();217    }218    while (readers.length > 0) {219      const r = readers.shift()!;220      r(new Uint8Array(0));221    }222  };223224  function send(bytes: Uint8Array): Promise<void> {225    return new Promise((resolve, reject) => {226      if (bytes.byteLength > SYNC_LIMITS.MAX_FRAME_BYTES) {227        reject(228          new SyncFrameTooLargeError(229            `sync frame exceeds ${SYNC_LIMITS.MAX_FRAME_BYTES} bytes`,230          ),231        );232        return;233      }234      try {235        ws.send(bytes as Uint8Array<ArrayBuffer>);236        resolve();237      } catch (err) {238        reject(err as Error);239      }240    });241  }242  function receive(): Promise<Uint8Array> {243    if (closed) return Promise.reject(closeError);244    const v = inbox.shift();245    if (v) return Promise.resolve(v);246    return new Promise<Uint8Array>((resolve, reject) => {247      readers.push((b) => {248        if (b.length === 0 && closed) reject(closeError);249        else resolve(b);250      });251    });252  }253254  const handshakeTimer = setTimeout(() => {255    closeError = new Error("sync_handshake_timeout");256    try { ws.close(1000, "sync handshake timeout"); } catch { /* ignore */ }257  }, options.handshakeTimeoutMs ?? DEFAULT_HANDSHAKE_TIMEOUT_MS);258  let session: NoiseNKSession;259  try {260    const ourPublicKey = hexToPubkey(ourIdentity.longTermPublicKeyHex);261    if (ourIdentity.longTermPrivateKey.length !== 32) {262      throw new Error("local reconnect private key must be exactly 32 bytes");263    }264    // This opaque rotating capability lets Android select paired local state265    // without exposing either public key to a relay. It grants no identity266    // trust; KK still proves both private keys immediately afterwards.267    await send(new TextEncoder().encode(JSON.stringify({268      type: "PEACH_RECONNECT",269      rendezvous_id: rendezvousId,270    })));271    session = await noiseKKInitiator(272      ourIdentity.longTermPrivateKey,273      ourPublicKey,274      hexToPubkey(expectedRemotePubkeyHex),275      send,276      receive,277    );278  } catch (err) {279    clearTimeout(handshakeTimer);280    try { ws.close(); } catch { /* ignore */ }281    throw err;282  }283284  try {285    await send(session.encrypt(padSyncFrame(encodePairIdentity(ourIdentity))));286    const theirFrame = JSON.parse(287      new TextDecoder().decode(unpadSyncFrame(session.decrypt(await receive()))),288    );289    if (theirFrame?.type !== "PAIR_IDENTITY") {290      throw new Error(`expected PAIR_IDENTITY, got ${theirFrame?.type as string}`);291    }292    if (theirFrame.long_term_public_key_hex !== expectedRemotePubkeyHex) {293      throw new Error("peer identity does not match authenticated KK key");294    }295    clearTimeout(handshakeTimer);296  } catch (err) {297    clearTimeout(handshakeTimer);298    session.close();299    try { ws.close(); } catch { /* ignore */ }300    throw err;301  }302303  return {304    async send(plaintext: Uint8Array) {305      // E7: bucket the size before encrypting, so the relay sees frame shapes306      // that no longer distinguish one operation (or vault size) from another.307      // E7b: a padded frame can legitimately be up to 24 MiB, but Noise caps a308      // single transport message at 65535 bytes (see crypto-core/src/lib.rs —309      // 65519 plaintext, 16-byte tag). We split a too-large padded frame into310      // numbered Noise messages and tag every chunk after the first with a311      // 1-byte continuation header. The receiver reassembles by reading the312      // padded frame's 4-byte length header, computing the same bucket from313      // the same ladder, and keeping reading until the bucket is full.314      const padded = padSyncFrame(plaintext);315      if (padded.byteLength > SYNC_LIMITS.MAX_FRAME_BYTES) {316        throw new SyncFrameTooLargeError(317          `sync frame exceeds ${SYNC_LIMITS.MAX_FRAME_BYTES} bytes`,318        );319      }320      // A single-chunk frame goes out as one Noise message, exactly as before321      // — no continuation byte, no per-chunk overhead, so the common small322      // case is unchanged on the wire.323      if (padded.byteLength <= MAX_NOISE_PLAINTEXT_BYTES) {324        const ct = session.encrypt(padded);325        ws.send(ct as Uint8Array<ArrayBuffer>);326        return;327      }328      const total = padded.byteLength;329      const payload = padded.subarray(0);330      for (let offset = 0; offset < total; ) {331        const remaining = total - offset;332        const chunkPlaintext =333          remaining <= MAX_NOISE_PLAINTEXT_BYTES334            ? payload.subarray(offset, total)335            : payload.subarray(offset, offset + MAX_NOISE_PLAINTEXT_BYTES);336        const chunkBytes = chunkPlaintext.byteLength;337        const isLast =338          offset + chunkBytes >= total && chunkBytes <= MAX_NOISE_PLAINTEXT_BYTES;339        // First chunk of a multi-chunk frame: the receiver already has the340        // length header in the payload, so a continuation byte is enough to341        // signal "more coming, don't try to use this yet". Last (and342        // only) chunks do not carry the byte — the length header tells the343        // receiver the frame is complete.344        const plaintext = isLast345          ? chunkPlaintext346          : concatBytes(FRAME_CONTINUATION_MARKER, chunkPlaintext);347        if (plaintext.byteLength > MAX_NOISE_PLAINTEXT_BYTES) {348          // Cannot happen: a single chunk is at most the limit, and one349          // continuation byte over a sublimit chunk is still under the350          // limit by 1. Defended against in case a future helper changes351          // either bound.352          throw new Error(353            `chunk exceeds Noise limit after continuation byte: ${plaintext.byteLength}`,354          );355        }356        const ct = session.encrypt(plaintext);357        if (ct.byteLength > SYNC_LIMITS.MAX_FRAME_BYTES) {358          throw new SyncFrameTooLargeError(359            `sync frame exceeds ${SYNC_LIMITS.MAX_FRAME_BYTES} bytes`,360          );361        }362        ws.send(ct as Uint8Array<ArrayBuffer>);363        offset += chunkBytes;364      }365    },366    async receive() {367      // The first Noise message of a frame is the first one we read. If it368      // begins with the continuation marker, the message is one of N chunks369      // — the padded-frame length header comes inside the first chunk's370      // payload, so we read it there to learn the total.371      let buffered: Uint8Array | null = null;372      while (true) {373        const ct = await receive();374        const pt = session.decrypt(ct);375        if (buffered === null && pt.byteLength > 0 && pt[0] === FRAME_CONTINUATION_MARKER) {376          // Strip the continuation byte and keep the payload.377          buffered = pt.subarray(1);378          continue;379        }380        if (buffered !== null) {381          if (pt.byteLength > 0 && pt[0] === FRAME_CONTINUATION_MARKER) {382            const merged: Uint8Array = new Uint8Array(buffered.length + pt.byteLength - 1);383            merged.set(buffered, 0);384            merged.set(pt.subarray(1), buffered.length);385            buffered = merged;386            continue;387          }388          // Final chunk. Concatenate without a continuation byte, then fall389          // through to the unpadder.390          const reassembled = new Uint8Array(buffered.length + pt.byteLength);391          reassembled.set(buffered, 0);392          reassembled.set(pt, buffered.length);393          return unpadSyncFrame(reassembled);394        }395        // Single-chunk frame, exactly as before.396        return unpadSyncFrame(pt);397      }398    },399    async close() {400      try { ws.close(); } catch { /* ignore */ }401      session.close();402    },403    remoteStaticPublicKey: hexToPubkey(expectedRemotePubkeyHex),404  };405}406407function concatBytes(prefix: number, payload: Uint8Array): Uint8Array {408  const out = new Uint8Array(1 + payload.byteLength);409  out[0] = prefix;410  out.set(payload, 1);411  return out;412}413414function connectWebSocketUrl(415  url: string,416  targetLabel: string,417  timeoutMs: number,418): Promise<WebSocket> {419  return new Promise((resolve, reject) => {420    let ws: WebSocket;421    try {422      ws = new WebSocket(url);423    } catch (err) {424      reject(err as Error);425      return;426    }427    const t = setTimeout(() => {428      try { ws.close(); } catch { /* ignore */ }429      reject(new Error(`${targetLabel} WebSocket connect timed out`));430    }, timeoutMs);431    ws.onopen = () => {432      clearTimeout(t);433      resolve(ws);434    };435    ws.onerror = () => {436      clearTimeout(t);437      reject(new Error(`${targetLabel} WebSocket connect errored`));438    };439  });440}441442/**443 * Per-pubkey cooldown tracker for the subnet probe. v2 brief §5.2 caps444 * full /24 scans at once per minute so an extension whose cached IP has445 * gone stale doesn't hammer the LAN with handshake attempts every time446 * the popup opens.447 *448 * Keyed by the expected remote pubkey because each paired peer is probed449 * independently; the rate limit applies to "scan looking for THIS peer",450 * not "scan the subnet at all".451 */452const lastProbeAt = new Map<string, number>();453454export function subnetProbeOnCooldown(455  expectedRemotePubkeyHex: string,456  now: number = Date.now(),457  cooldownMs: number = SUBNET_PROBE_COOLDOWN_MS,458): boolean {459  const last = lastProbeAt.get(expectedRemotePubkeyHex);460  return last !== undefined && now - last < cooldownMs;461}462463/** Test-only — clear the per-pubkey probe cooldown ledger. */464export function _resetSubnetProbeCooldown(): void {465  lastProbeAt.clear();466}467468/**469 * Subnet probe — scan a /24 looking for a host that completes Noise KK470 * with the expected long-term static key. Returns the first match, or471 * null if the cooldown is still active or no host responded.472 *473 * Rate-limited to once per `SUBNET_PROBE_COOLDOWN_MS` per peer pubkey474 * (v2 brief §5.2). Callers should still attempt the cached475 * `last_known_ip`/`last_known_port` directly first — the subnet probe is476 * fallback only.477 *478 * For each IP we attempt the full Noise KK handshake, but with a tight479 * timeout: dead hosts fail at the WebSocket connect step in <1ms (browser480 * surfaces ECONNREFUSED immediately) or by the connectTimeout otherwise.481 */482export async function probeSubnetForPeer(483  subnet: string, // e.g. "192.168.1"484  port: number,485  ourIdentity: SyncPeerIdentity,486  expectedRemotePubkeyHex: string,487  rendezvousIds: readonly string[],488  options: { batch?: number; timeoutMs?: number; cooldownMs?: number; now?: number } = {},489): Promise<ProbedSyncEndpoint | null> {490  if (rendezvousIds.length === 0) {491    throw new Error("subnet probe requires at least one rendezvous id");492  }493  const now = options.now ?? Date.now();494  const cooldownMs = options.cooldownMs ?? SUBNET_PROBE_COOLDOWN_MS;495  if (subnetProbeOnCooldown(expectedRemotePubkeyHex, now, cooldownMs)) {496    return null;497  }498  lastProbeAt.set(expectedRemotePubkeyHex, now);499  const batchSize = options.batch ?? DEFAULT_PROBE_BATCH;500  const timeoutMs = options.timeoutMs ?? DEFAULT_PROBE_TIMEOUT_MS;501  for (let start = 1; start <= 254; start += batchSize) {502    const end = Math.min(start + batchSize - 1, 254);503    const ips: string[] = [];504    for (let i = start; i <= end; i++) ips.push(`${subnet}.${i}`);505506    const results = await Promise.all(ips.map((ip) => probeOne(507      { ip, port },508      ourIdentity,509      expectedRemotePubkeyHex,510      rendezvousIds,511      timeoutMs,512    )));513    const hit = results.find((r): r is ProbedSyncEndpoint => r !== null);514    if (hit) return hit;515  }516  return null;517}518519async function probeOne(520  endpoint: SyncEndpoint,521  ourIdentity: SyncPeerIdentity,522  expectedRemotePubkeyHex: string,523  rendezvousIds: readonly string[],524  timeoutMs: number,525): Promise<ProbedSyncEndpoint | null> {526  for (const rendezvousId of rendezvousIds) {527    try {528      const session = await openSyncSession(529        endpoint,530        ourIdentity,531        expectedRemotePubkeyHex,532        rendezvousId,533        { connectTimeoutMs: timeoutMs },534      );535      await session.close();536      return { ...endpoint, rendezvousId };537    } catch {538      // Try the next locally-derived clock-skew window at the same host.539    }540  }541  return null;542}