sync-transport.ts

1// WebSocket sync transport — standardized on Noise NK (see2// HANDSHAKE_STANDARDIZE_ON_NK.md). Android's ongoing-sync listener3// (SyncReconnectListener.kt) is a Noise NK responder on the same fixed port4// (49142) used for pairing; there is no separate Noise XX responder.5//6// The extension is always the WebSocket client; Android (or future iOS /7// desktop sync hosts) is always the server. Each binary WebSocket message8// carries exactly one Noise frame — no length prefix, no fragmentation.9//10// On connect:11//   1. Open ws://<ip>:<port>/12//   2. Run Noise NK as the initiator against the paired device's known13//      long-term static public key. Unlike XX, NK authenticates only the14//      responder during the handshake — there is no initiator static15//      keypair in the crypto itself.16//   3. Exchange PAIR_IDENTITY frames over the encrypted channel (same frame17//      Android's SyncReconnectListener expects before it will treat the18//      connection as sync traffic rather than an unknown caller). Our19//      identity is asserted, not proven, at this layer — this mirrors20//      Android's own implementation, not a gap introduced here.21//   4. Hand the caller a `SyncSession` for the SyncMessage protocol.22//23// Subnet probe falls back to scanning the /24 returned by `getLocalSubnet`24// when the cached last-known IP fails. We use a short connection timeout25// (~500ms per IP) and parallel batches of 16 to keep the worst-case scan26// under ~10 seconds.2728import { noiseNKInitiator, type NoiseNKSession } from "./noise-js";29import { SYNC_LIMITS } from "./sync-limits";30import { hexToPubkey } from "./sync-identity";31import type { DeviceType } from "./sync-types";3233const DEFAULT_PROBE_TIMEOUT_MS = 500;34const DEFAULT_PROBE_BATCH = 16;35const DEFAULT_CONNECT_TIMEOUT_MS = 4_000;36/** v2 brief §5.2 — subnet probe is rate-limited to once per 60 seconds. */37export const SUBNET_PROBE_COOLDOWN_MS = 60_000;3839export interface SyncSession {40  send(plaintext: Uint8Array): void;41  receive(): Promise<Uint8Array>;42  close(): Promise<void>;43  remoteStaticPublicKey: Uint8Array;44}4546export interface SyncEndpoint {47  ip: string;48  port: number;49}5051/** Our identity, asserted in the post-handshake PAIR_IDENTITY frame — the52 *  same shape `code-pairing.ts` sends during pairing (`device_id` etc.). */53export interface SyncPeerIdentity {54  deviceId: string;55  deviceName: string;56  deviceType: DeviceType;57  longTermPublicKeyHex: string;58  signingPublicKeyHex?: string;59}6061export class SyncFrameTooLargeError extends Error {62  constructor(message = "sync frame too large (WebSocket close 1009)") {63    super(message);64    this.name = "SyncFrameTooLargeError";65  }66}6768function encodePairIdentity(id: SyncPeerIdentity): Uint8Array {69  return new TextEncoder().encode(JSON.stringify({70    type: "PAIR_IDENTITY",71    device_id: id.deviceId,72    device_name: id.deviceName,73    device_type: id.deviceType,74    long_term_public_key_hex: id.longTermPublicKeyHex,75    signing_public_key_hex: id.signingPublicKeyHex,76  }));77}7879/**80 * Open a WebSocket to (ip, port), run Noise NK against the paired device's81 * known static key, exchange PAIR_IDENTITY, and return a ready-to-use82 * session.83 *84 * On a successful return, `setLastKnownIp` in the caller can persist85 * (ip, port) for the next session's fast path.86 */87export async function openSyncSession(88  endpoint: SyncEndpoint,89  ourIdentity: SyncPeerIdentity,90  expectedRemotePubkeyHex: string,91  options: { connectTimeoutMs?: number } = {},92): Promise<SyncSession> {93  const ws = await connectWebSocket(endpoint, options.connectTimeoutMs ?? DEFAULT_CONNECT_TIMEOUT_MS);94  const inbox: Uint8Array[] = [];95  const readers: Array<(b: Uint8Array) => void> = [];96  let closed = false;97  let closeError: Error = new Error("WebSocket closed");9899  ws.binaryType = "arraybuffer";100  ws.onmessage = (ev: MessageEvent) => {101    const data = ev.data;102    if (data instanceof ArrayBuffer) {103      // MV-3: refuse oversized frames before they are buffered or decrypted.104      // A peer that sends one is misbehaving — tear the session down rather105      // than letting the inbox grow toward OOM.106      if (data.byteLength > SYNC_LIMITS.MAX_FRAME_BYTES) {107        // MV-3: oversized frame — tear the session down. close() routes through108        // onclose, which rejects any parked receive(), so no error array is109        // needed (PCH-01-026 #9).110        console.warn(111          `[Peach] sync frame exceeds max size (${SYNC_LIMITS.MAX_FRAME_BYTES} bytes); closing`,112        );113        closeError = new SyncFrameTooLargeError(114          `sync frame exceeds ${SYNC_LIMITS.MAX_FRAME_BYTES} bytes`,115        );116        try { ws.close(1009, "sync frame too large"); } catch { /* ignore */ }117        return;118      }119      const bytes = new Uint8Array(data);120      const r = readers.shift();121      if (r) r(bytes);122      else inbox.push(bytes);123    }124  };125  ws.onerror = () => {126    // PCH-01-026 #9: surface the error and unblock any parked receive() by127    // closing. Browsers usually follow error with close, but not always; an128    // explicit close guarantees the handshake fails fast instead of hanging.129    console.warn("[Peach] sync WebSocket error; closing transport");130    try { ws.close(); } catch { /* ignore */ }131  };132  ws.onclose = (event: CloseEvent) => {133    closed = true;134    if (event.code === 1009) {135      closeError = new SyncFrameTooLargeError();136    }137    while (readers.length > 0) {138      const r = readers.shift()!;139      r(new Uint8Array(0));140    }141  };142143  function send(bytes: Uint8Array): Promise<void> {144    return new Promise((resolve, reject) => {145      if (bytes.byteLength > SYNC_LIMITS.MAX_FRAME_BYTES) {146        reject(147          new SyncFrameTooLargeError(148            `sync frame exceeds ${SYNC_LIMITS.MAX_FRAME_BYTES} bytes`,149          ),150        );151        return;152      }153      try {154        ws.send(bytes as Uint8Array<ArrayBuffer>);155        resolve();156      } catch (err) {157        reject(err as Error);158      }159    });160  }161  function receive(): Promise<Uint8Array> {162    if (closed) return Promise.reject(closeError);163    const v = inbox.shift();164    if (v) return Promise.resolve(v);165    return new Promise<Uint8Array>((resolve, reject) => {166      readers.push((b) => {167        if (b.length === 0 && closed) reject(closeError);168        else resolve(b);169      });170    });171  }172173  let session: NoiseNKSession;174  try {175    // NK's responder static key is a handshake INPUT, not something learned176    // mid-handshake — a wrong key simply fails msg2 decryption below, so177    // there is no separate mismatch check to run first (unlike XX).178    session = await noiseNKInitiator(hexToPubkey(expectedRemotePubkeyHex), send, receive);179  } catch (err) {180    try { ws.close(); } catch { /* ignore */ }181    throw err;182  }183184  try {185    await send(session.encrypt(encodePairIdentity(ourIdentity)));186    const theirFrame = JSON.parse(new TextDecoder().decode(session.decrypt(await receive())));187    if (theirFrame?.type !== "PAIR_IDENTITY") {188      throw new Error(`expected PAIR_IDENTITY, got ${theirFrame?.type as string}`);189    }190  } catch (err) {191    try { ws.close(); } catch { /* ignore */ }192    throw err;193  }194195  return {196    send(plaintext: Uint8Array) {197      const ciphertext = session.encrypt(plaintext);198      if (ciphertext.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(ciphertext as Uint8Array<ArrayBuffer>);204    },205    async receive() {206      const ct = await receive();207      return session.decrypt(ct);208    },209    async close() {210      try { ws.close(); } catch { /* ignore */ }211    },212    remoteStaticPublicKey: hexToPubkey(expectedRemotePubkeyHex),213  };214}215216function connectWebSocket(endpoint: SyncEndpoint, timeoutMs: number): Promise<WebSocket> {217  return new Promise((resolve, reject) => {218    let ws: WebSocket;219    try {220      ws = new WebSocket(`ws://${endpoint.ip}:${endpoint.port}/`);221    } catch (err) {222      reject(err as Error);223      return;224    }225    const t = setTimeout(() => {226      try { ws.close(); } catch { /* ignore */ }227      reject(new Error(`WebSocket connect to ${endpoint.ip}:${endpoint.port} timed out`));228    }, timeoutMs);229    ws.onopen = () => {230      clearTimeout(t);231      resolve(ws);232    };233    ws.onerror = () => {234      clearTimeout(t);235      reject(new Error(`WebSocket connect to ${endpoint.ip}:${endpoint.port} errored`));236    };237  });238}239240/**241 * Per-pubkey cooldown tracker for the subnet probe. v2 brief §5.2 caps242 * full /24 scans at once per minute so an extension whose cached IP has243 * gone stale doesn't hammer the LAN with handshake attempts every time244 * the popup opens.245 *246 * Keyed by the expected remote pubkey because each paired peer is probed247 * independently; the rate limit applies to "scan looking for THIS peer",248 * not "scan the subnet at all".249 */250const lastProbeAt = new Map<string, number>();251252export function subnetProbeOnCooldown(253  expectedRemotePubkeyHex: string,254  now: number = Date.now(),255  cooldownMs: number = SUBNET_PROBE_COOLDOWN_MS,256): boolean {257  const last = lastProbeAt.get(expectedRemotePubkeyHex);258  return last !== undefined && now - last < cooldownMs;259}260261/** Test-only — clear the per-pubkey probe cooldown ledger. */262export function _resetSubnetProbeCooldown(): void {263  lastProbeAt.clear();264}265266/**267 * Subnet probe — scan a /24 looking for a host that completes Noise NK268 * with the expected long-term static key. Returns the first match, or269 * null if the cooldown is still active or no host responded.270 *271 * Rate-limited to once per `SUBNET_PROBE_COOLDOWN_MS` per peer pubkey272 * (v2 brief §5.2). Callers should still attempt the cached273 * `last_known_ip`/`last_known_port` directly first — the subnet probe is274 * fallback only.275 *276 * For each IP we attempt the full Noise NK handshake, but with a tight277 * timeout: dead hosts fail at the WebSocket connect step in <1ms (browser278 * surfaces ECONNREFUSED immediately) or by the connectTimeout otherwise.279 */280export async function probeSubnetForPeer(281  subnet: string, // e.g. "192.168.1"282  port: number,283  ourIdentity: SyncPeerIdentity,284  expectedRemotePubkeyHex: string,285  options: { batch?: number; timeoutMs?: number; cooldownMs?: number; now?: number } = {},286): Promise<SyncEndpoint | null> {287  const now = options.now ?? Date.now();288  const cooldownMs = options.cooldownMs ?? SUBNET_PROBE_COOLDOWN_MS;289  if (subnetProbeOnCooldown(expectedRemotePubkeyHex, now, cooldownMs)) {290    return null;291  }292  lastProbeAt.set(expectedRemotePubkeyHex, now);293  const batchSize = options.batch ?? DEFAULT_PROBE_BATCH;294  const timeoutMs = options.timeoutMs ?? DEFAULT_PROBE_TIMEOUT_MS;295  for (let start = 1; start <= 254; start += batchSize) {296    const end = Math.min(start + batchSize - 1, 254);297    const ips: string[] = [];298    for (let i = start; i <= end; i++) ips.push(`${subnet}.${i}`);299300    const results = await Promise.all(ips.map((ip) => probeOne(301      { ip, port },302      ourIdentity,303      expectedRemotePubkeyHex,304      timeoutMs,305    )));306    const hit = results.find((r): r is SyncEndpoint => r !== null);307    if (hit) return hit;308  }309  return null;310}311312async function probeOne(313  endpoint: SyncEndpoint,314  ourIdentity: SyncPeerIdentity,315  expectedRemotePubkeyHex: string,316  timeoutMs: number,317): Promise<SyncEndpoint | null> {318  try {319    const session = await openSyncSession(320      endpoint,321      ourIdentity,322      expectedRemotePubkeyHex,323      { connectTimeoutMs: timeoutMs },324    );325    await session.close();326    return endpoint;327  } catch {328    return null;329  }330}