123456789101112131415161718192021import { noiseXXInitiator, type NoiseXXSession } from "./noise-xx";22import { hexToPubkey, pubkeyToHex } from "./sync-identity";2324const DEFAULT_PROBE_TIMEOUT_MS = 500;25const DEFAULT_PROBE_BATCH = 16;26const DEFAULT_CONNECT_TIMEOUT_MS = 4_000;2728export const SUBNET_PROBE_COOLDOWN_MS = 60_000;2930export interface ChirpSyncSession {31 send(plaintext: Uint8Array): void;32 receive(): Promise<Uint8Array>;33 close(): Promise<void>;34 remoteStaticPublicKey: Uint8Array;35}3637export interface SyncEndpoint {38 ip: string;39 port: number;40}41424344454647484950export async function openChirpSyncSession(51 endpoint: SyncEndpoint,52 ourLongTermPrivateKey: Uint8Array,53 ourLongTermPublicKey: Uint8Array,54 expectedRemotePubkeyHex: string,55 options: { connectTimeoutMs?: number } = {},56): Promise<ChirpSyncSession> {57 const ws = await connectWebSocket(endpoint, options.connectTimeoutMs ?? DEFAULT_CONNECT_TIMEOUT_MS);58 const inbox: Uint8Array[] = [];59 const readers: Array<(b: Uint8Array) => void> = [];60 const errors: Error[] = [];61 let closed = false;6263 ws.binaryType = "arraybuffer";64 ws.onmessage = (ev: MessageEvent) => {65 const data = ev.data;66 if (data instanceof ArrayBuffer) {67 const bytes = new Uint8Array(data);68 const r = readers.shift();69 if (r) r(bytes);70 else inbox.push(bytes);71 }72 };73 ws.onerror = () => {74 errors.push(new Error("WebSocket error"));75 };76 ws.onclose = () => {77 closed = true;78 while (readers.length > 0) {79 const r = readers.shift()!;80 r(new Uint8Array(0));81 }82 };8384 function send(bytes: Uint8Array): Promise<void> {85 return new Promise((resolve, reject) => {86 try {87 ws.send(bytes as Uint8Array<ArrayBuffer>);88 resolve();89 } catch (err) {90 reject(err as Error);91 }92 });93 }94 function receive(): Promise<Uint8Array> {95 if (closed) return Promise.reject(new Error("WebSocket closed before frame"));96 const v = inbox.shift();97 if (v) return Promise.resolve(v);98 return new Promise<Uint8Array>((resolve, reject) => {99 readers.push((b) => {100 if (b.length === 0 && closed) reject(new Error("WebSocket closed"));101 else resolve(b);102 });103 });104 }105106 let session: NoiseXXSession;107 try {108 session = await noiseXXInitiator(109 ourLongTermPrivateKey,110 ourLongTermPublicKey,111 send,112 receive,113 );114 } catch (err) {115 try { ws.close(); } catch { }116 throw err;117 }118119 const remoteHex = pubkeyToHex(session.remoteStaticPublicKey);120 if (remoteHex !== expectedRemotePubkeyHex) {121 try { ws.close(); } catch { }122 throw new Error(123 `paired-device static key mismatch (got ${remoteHex.slice(0, 16)}…, ` +124 `expected ${expectedRemotePubkeyHex.slice(0, 16)}…)`,125 );126 }127 128 void hexToPubkey(expectedRemotePubkeyHex);129130 return {131 send(plaintext: Uint8Array) {132 const ciphertext = session.encrypt(plaintext);133 ws.send(ciphertext as Uint8Array<ArrayBuffer>);134 },135 async receive() {136 const ct = await receive();137 return session.decrypt(ct);138 },139 async close() {140 try { ws.close(); } catch { }141 },142 remoteStaticPublicKey: session.remoteStaticPublicKey,143 };144}145146function connectWebSocket(endpoint: SyncEndpoint, timeoutMs: number): Promise<WebSocket> {147 return new Promise((resolve, reject) => {148 let ws: WebSocket;149 try {150 ws = new WebSocket(`ws://${endpoint.ip}:${endpoint.port}/`);151 } catch (err) {152 reject(err as Error);153 return;154 }155 const t = setTimeout(() => {156 try { ws.close(); } catch { }157 reject(new Error(`WebSocket connect to ${endpoint.ip}:${endpoint.port} timed out`));158 }, timeoutMs);159 ws.onopen = () => {160 clearTimeout(t);161 resolve(ws);162 };163 ws.onerror = () => {164 clearTimeout(t);165 reject(new Error(`WebSocket connect to ${endpoint.ip}:${endpoint.port} errored`));166 };167 });168}169170171172173174175176177178179180const lastProbeAt = new Map<string, number>();181182export function subnetProbeOnCooldown(183 expectedRemotePubkeyHex: string,184 now: number = Date.now(),185 cooldownMs: number = SUBNET_PROBE_COOLDOWN_MS,186): boolean {187 const last = lastProbeAt.get(expectedRemotePubkeyHex);188 return last !== undefined && now - last < cooldownMs;189}190191192export function _resetSubnetProbeCooldown(): void {193 lastProbeAt.clear();194}195196197198199200201202203204205206207208209210export async function probeSubnetForPeer(211 subnet: string, 212 port: number,213 ourLongTermPrivateKey: Uint8Array,214 ourLongTermPublicKey: Uint8Array,215 expectedRemotePubkeyHex: string,216 options: { batch?: number; timeoutMs?: number; cooldownMs?: number; now?: number } = {},217): Promise<SyncEndpoint | null> {218 const now = options.now ?? Date.now();219 const cooldownMs = options.cooldownMs ?? SUBNET_PROBE_COOLDOWN_MS;220 if (subnetProbeOnCooldown(expectedRemotePubkeyHex, now, cooldownMs)) {221 return null;222 }223 lastProbeAt.set(expectedRemotePubkeyHex, now);224 const batchSize = options.batch ?? DEFAULT_PROBE_BATCH;225 const timeoutMs = options.timeoutMs ?? DEFAULT_PROBE_TIMEOUT_MS;226 for (let start = 1; start <= 254; start += batchSize) {227 const end = Math.min(start + batchSize - 1, 254);228 const ips: string[] = [];229 for (let i = start; i <= end; i++) ips.push(`${subnet}.${i}`);230231 const results = await Promise.all(ips.map((ip) => probeOne(232 { ip, port },233 ourLongTermPrivateKey,234 ourLongTermPublicKey,235 expectedRemotePubkeyHex,236 timeoutMs,237 )));238 const hit = results.find((r): r is SyncEndpoint => r !== null);239 if (hit) return hit;240 }241 return null;242}243244async function probeOne(245 endpoint: SyncEndpoint,246 ourLongTermPrivateKey: Uint8Array,247 ourLongTermPublicKey: Uint8Array,248 expectedRemotePubkeyHex: string,249 timeoutMs: number,250): Promise<SyncEndpoint | null> {251 try {252 const session = await openChirpSyncSession(253 endpoint,254 ourLongTermPrivateKey,255 ourLongTermPublicKey,256 expectedRemotePubkeyHex,257 { connectTimeoutMs: timeoutMs },258 );259 await session.close();260 return endpoint;261 } catch {262 return null;263 }264}