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 type { DeviceType } from "./sync-types";2829const DEFAULT_PROBE_TIMEOUT_MS = 500;30const DEFAULT_PROBE_BATCH = 16;31const DEFAULT_CONNECT_TIMEOUT_MS = 4_000;32/** v2 brief §5.2 — subnet probe is rate-limited to once per 60 seconds. */33export const SUBNET_PROBE_COOLDOWN_MS = 60_000;3435export interface SyncSession {36  send(plaintext: Uint8Array): void;37  receive(): Promise<Uint8Array>;38  close(): Promise<void>;39  remoteStaticPublicKey: Uint8Array;40}4142export interface SyncEndpoint {43  ip: string;44  port: number;45}4647export interface ProbedSyncEndpoint extends SyncEndpoint {48  /** Exact clock-skew window capability that completed KK at this endpoint. */49  rendezvousId: string;50}5152/** Our identity, repeated in the post-handshake PAIR_IDENTITY frame and bound53 *  to the local static key already authenticated by KK. */54export interface SyncPeerIdentity {55  deviceId: string;56  deviceName: string;57  deviceType: DeviceType;58  longTermPublicKeyHex: string;59  /** Local-only key material for the KK reconnect handshake; never encoded. */60  longTermPrivateKey: Uint8Array;61  signingPublicKeyHex?: string;62}6364export class SyncFrameTooLargeError extends Error {65  constructor(message = "sync frame too large (WebSocket close 1009)") {66    super(message);67    this.name = "SyncFrameTooLargeError";68  }69}7071function encodePairIdentity(id: SyncPeerIdentity): Uint8Array {72  return new TextEncoder().encode(JSON.stringify({73    type: "PAIR_IDENTITY",74    device_id: id.deviceId,75    device_name: id.deviceName,76    device_type: id.deviceType,77    long_term_public_key_hex: id.longTermPublicKeyHex,78    signing_public_key_hex: id.signingPublicKeyHex,79  }));80}8182/**83 * Open a WebSocket to (ip, port), run Noise KK against both paired static84 * keys, exchange bound PAIR_IDENTITY frames, and return a ready-to-use85 * session.86 *87 * On a successful return, `setLastKnownIp` in the caller can persist88 * (ip, port) for the next session's fast path.89 */90export async function openSyncSession(91  endpoint: SyncEndpoint,92  ourIdentity: SyncPeerIdentity,93  expectedRemotePubkeyHex: string,94  rendezvousId: string,95  options: { connectTimeoutMs?: number } = {},96): Promise<SyncSession> {97  if (!/^[A-Za-z0-9_-]{22,128}$/.test(rendezvousId)) {98    throw new Error("reconnect requires a valid rotating rendezvous id");99  }100  const ws = await connectWebSocket(endpoint, options.connectTimeoutMs ?? DEFAULT_CONNECT_TIMEOUT_MS);101  const inbox: Uint8Array[] = [];102  const readers: Array<(b: Uint8Array) => void> = [];103  let closed = false;104  let closeError: Error = new Error("WebSocket closed");105106  ws.binaryType = "arraybuffer";107  ws.onmessage = (ev: MessageEvent) => {108    const data = ev.data;109    if (data instanceof ArrayBuffer) {110      // MV-3: refuse oversized frames before they are buffered or decrypted.111      // A peer that sends one is misbehaving — tear the session down rather112      // than letting the inbox grow toward OOM.113      if (data.byteLength > SYNC_LIMITS.MAX_FRAME_BYTES) {114        // MV-3: oversized frame — tear the session down. close() routes through115        // onclose, which rejects any parked receive(), so no error array is116        // needed (PCH-01-026 #9).117        console.warn(118          `[Peach] sync frame exceeds max size (${SYNC_LIMITS.MAX_FRAME_BYTES} bytes); closing`,119        );120        closeError = new SyncFrameTooLargeError(121          `sync frame exceeds ${SYNC_LIMITS.MAX_FRAME_BYTES} bytes`,122        );123        try { ws.close(1009, "sync frame too large"); } catch { /* ignore */ }124        return;125      }126      const bytes = new Uint8Array(data);127      const r = readers.shift();128      if (r) r(bytes);129      else inbox.push(bytes);130    }131  };132  ws.onerror = () => {133    // PCH-01-026 #9: surface the error and unblock any parked receive() by134    // closing. Browsers usually follow error with close, but not always; an135    // explicit close guarantees the handshake fails fast instead of hanging.136    console.warn("[Peach] sync WebSocket error; closing transport");137    try { ws.close(); } catch { /* ignore */ }138  };139  ws.onclose = (event: CloseEvent) => {140    closed = true;141    if (event.code === 1009) {142      closeError = new SyncFrameTooLargeError();143    }144    while (readers.length > 0) {145      const r = readers.shift()!;146      r(new Uint8Array(0));147    }148  };149150  function send(bytes: Uint8Array): Promise<void> {151    return new Promise((resolve, reject) => {152      if (bytes.byteLength > SYNC_LIMITS.MAX_FRAME_BYTES) {153        reject(154          new SyncFrameTooLargeError(155            `sync frame exceeds ${SYNC_LIMITS.MAX_FRAME_BYTES} bytes`,156          ),157        );158        return;159      }160      try {161        ws.send(bytes as Uint8Array<ArrayBuffer>);162        resolve();163      } catch (err) {164        reject(err as Error);165      }166    });167  }168  function receive(): Promise<Uint8Array> {169    if (closed) return Promise.reject(closeError);170    const v = inbox.shift();171    if (v) return Promise.resolve(v);172    return new Promise<Uint8Array>((resolve, reject) => {173      readers.push((b) => {174        if (b.length === 0 && closed) reject(closeError);175        else resolve(b);176      });177    });178  }179180  let session: NoiseNKSession;181  try {182    const ourPublicKey = hexToPubkey(ourIdentity.longTermPublicKeyHex);183    if (ourIdentity.longTermPrivateKey.length !== 32) {184      throw new Error("local reconnect private key must be exactly 32 bytes");185    }186    // This opaque rotating capability lets Android select paired local state187    // without exposing either public key to a relay. It grants no identity188    // trust; KK still proves both private keys immediately afterwards.189    await send(new TextEncoder().encode(JSON.stringify({190      type: "PEACH_RECONNECT",191      rendezvous_id: rendezvousId,192    })));193    session = await noiseKKInitiator(194      ourIdentity.longTermPrivateKey,195      ourPublicKey,196      hexToPubkey(expectedRemotePubkeyHex),197      send,198      receive,199    );200  } catch (err) {201    try { ws.close(); } catch { /* ignore */ }202    throw err;203  }204205  try {206    await send(session.encrypt(encodePairIdentity(ourIdentity)));207    const theirFrame = JSON.parse(new TextDecoder().decode(session.decrypt(await receive())));208    if (theirFrame?.type !== "PAIR_IDENTITY") {209      throw new Error(`expected PAIR_IDENTITY, got ${theirFrame?.type as string}`);210    }211    if (theirFrame.long_term_public_key_hex !== expectedRemotePubkeyHex) {212      throw new Error("peer identity does not match authenticated KK key");213    }214  } catch (err) {215    try { ws.close(); } catch { /* ignore */ }216    throw err;217  }218219  return {220    send(plaintext: Uint8Array) {221      const ciphertext = session.encrypt(plaintext);222      if (ciphertext.byteLength > SYNC_LIMITS.MAX_FRAME_BYTES) {223        throw new SyncFrameTooLargeError(224          `sync frame exceeds ${SYNC_LIMITS.MAX_FRAME_BYTES} bytes`,225        );226      }227      ws.send(ciphertext as Uint8Array<ArrayBuffer>);228    },229    async receive() {230      const ct = await receive();231      return session.decrypt(ct);232    },233    async close() {234      try { ws.close(); } catch { /* ignore */ }235    },236    remoteStaticPublicKey: hexToPubkey(expectedRemotePubkeyHex),237  };238}239240function connectWebSocket(endpoint: SyncEndpoint, timeoutMs: number): Promise<WebSocket> {241  return new Promise((resolve, reject) => {242    let ws: WebSocket;243    try {244      ws = new WebSocket(`ws://${endpoint.ip}:${endpoint.port}/`);245    } catch (err) {246      reject(err as Error);247      return;248    }249    const t = setTimeout(() => {250      try { ws.close(); } catch { /* ignore */ }251      reject(new Error(`WebSocket connect to ${endpoint.ip}:${endpoint.port} timed out`));252    }, timeoutMs);253    ws.onopen = () => {254      clearTimeout(t);255      resolve(ws);256    };257    ws.onerror = () => {258      clearTimeout(t);259      reject(new Error(`WebSocket connect to ${endpoint.ip}:${endpoint.port} errored`));260    };261  });262}263264/**265 * Per-pubkey cooldown tracker for the subnet probe. v2 brief §5.2 caps266 * full /24 scans at once per minute so an extension whose cached IP has267 * gone stale doesn't hammer the LAN with handshake attempts every time268 * the popup opens.269 *270 * Keyed by the expected remote pubkey because each paired peer is probed271 * independently; the rate limit applies to "scan looking for THIS peer",272 * not "scan the subnet at all".273 */274const lastProbeAt = new Map<string, number>();275276export function subnetProbeOnCooldown(277  expectedRemotePubkeyHex: string,278  now: number = Date.now(),279  cooldownMs: number = SUBNET_PROBE_COOLDOWN_MS,280): boolean {281  const last = lastProbeAt.get(expectedRemotePubkeyHex);282  return last !== undefined && now - last < cooldownMs;283}284285/** Test-only — clear the per-pubkey probe cooldown ledger. */286export function _resetSubnetProbeCooldown(): void {287  lastProbeAt.clear();288}289290/**291 * Subnet probe — scan a /24 looking for a host that completes Noise KK292 * with the expected long-term static key. Returns the first match, or293 * null if the cooldown is still active or no host responded.294 *295 * Rate-limited to once per `SUBNET_PROBE_COOLDOWN_MS` per peer pubkey296 * (v2 brief §5.2). Callers should still attempt the cached297 * `last_known_ip`/`last_known_port` directly first — the subnet probe is298 * fallback only.299 *300 * For each IP we attempt the full Noise KK handshake, but with a tight301 * timeout: dead hosts fail at the WebSocket connect step in <1ms (browser302 * surfaces ECONNREFUSED immediately) or by the connectTimeout otherwise.303 */304export async function probeSubnetForPeer(305  subnet: string, // e.g. "192.168.1"306  port: number,307  ourIdentity: SyncPeerIdentity,308  expectedRemotePubkeyHex: string,309  rendezvousIds: readonly string[],310  options: { batch?: number; timeoutMs?: number; cooldownMs?: number; now?: number } = {},311): Promise<ProbedSyncEndpoint | null> {312  if (rendezvousIds.length === 0) {313    throw new Error("subnet probe requires at least one rendezvous id");314  }315  const now = options.now ?? Date.now();316  const cooldownMs = options.cooldownMs ?? SUBNET_PROBE_COOLDOWN_MS;317  if (subnetProbeOnCooldown(expectedRemotePubkeyHex, now, cooldownMs)) {318    return null;319  }320  lastProbeAt.set(expectedRemotePubkeyHex, now);321  const batchSize = options.batch ?? DEFAULT_PROBE_BATCH;322  const timeoutMs = options.timeoutMs ?? DEFAULT_PROBE_TIMEOUT_MS;323  for (let start = 1; start <= 254; start += batchSize) {324    const end = Math.min(start + batchSize - 1, 254);325    const ips: string[] = [];326    for (let i = start; i <= end; i++) ips.push(`${subnet}.${i}`);327328    const results = await Promise.all(ips.map((ip) => probeOne(329      { ip, port },330      ourIdentity,331      expectedRemotePubkeyHex,332      rendezvousIds,333      timeoutMs,334    )));335    const hit = results.find((r): r is ProbedSyncEndpoint => r !== null);336    if (hit) return hit;337  }338  return null;339}340341async function probeOne(342  endpoint: SyncEndpoint,343  ourIdentity: SyncPeerIdentity,344  expectedRemotePubkeyHex: string,345  rendezvousIds: readonly string[],346  timeoutMs: number,347): Promise<ProbedSyncEndpoint | null> {348  for (const rendezvousId of rendezvousIds) {349    try {350      const session = await openSyncSession(351        endpoint,352        ourIdentity,353        expectedRemotePubkeyHex,354        rendezvousId,355        { connectTimeoutMs: timeoutMs },356      );357      await session.close();358      return { ...endpoint, rendezvousId };359    } catch {360      // Try the next locally-derived clock-skew window at the same host.361    }362  }363  return null;364}