Something went wrong. Try again.
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252import { el, parseXml, serialize, type XmlElement } from "./xml";
/** * An XMPP client over WebSocket (RFC 7395), for the extension and for the central system. SASL PLAIN (the credential is a bearer token over a loopback or tunnelled * connection, docs/design/distributed-network-and-identity.md), resource bind, stream management with resume (XEP-0198), pings (XEP-0199) that close the connection when * unanswered, and strict parsing of everything the server sends (shared/xml.ts). It reconnects nothing by itself: the owner decides, from `onClose` and from * `XmppAuthError`. Platform details (the WebSocket, timers, the clock) are passed in, so it runs in a service worker, an extension page and Node. */
export interface WebSocketLike { send(data: string): void; close(): void; onopen: (() => void) | null; onmessage: ((e: { data: unknown }) => void) | null; onclose: ((e: { code: number; reason: string }) => void) | null; onerror: ((e: unknown) => void) | null;}
export class XmppAuthError extends Error { override name = "XmppAuthError";}
export interface XmppClientOptions { url: string; domain: string; user: string; pass: string; resource: string; webSocket(url: string, protocol: string): WebSocketLike; /** Ping the server this often; with none, no pings. */ pingMs?: number; /** Close the connection if a ping is not answered within this (default 10 s). */ pingTimeoutMs?: number; connectTimeoutMs?: number; /** Send initial presence once bound (default true). A session that has not is not "available", and the server does not deliver messages to its bare address. */ announce?: boolean; /** A stream to resume, as `sm()` reported it on an earlier connection. */ resume?: { id: string; handled: number }; onStanza?(stanza: XmlElement): void; onClose?(info: { reason: string }): void; setTimer(fn: () => void, ms: number): unknown; clearTimer(handle: unknown): void; now(): number;}
const FRAMING = "urn:ietf:params:xml:ns:xmpp-framing";const SASL = "urn:ietf:params:xml:ns:xmpp-sasl";const BIND = "urn:ietf:params:xml:ns:xmpp-bind";const SM = "urn:xmpp:sm:3";const PING = "urn:xmpp:ping";
const b64 = (s: string): string => { let bin = ""; for (const b of new TextEncoder().encode(s)) bin += String.fromCharCode(b); return btoa(bin);};
export class XmppClient { jid = ""; pings = 0; pongs = 0; closed = false; private ws!: WebSocketLike; private backlog: XmlElement[] = []; private waiters: Array<{ test(e: XmlElement): boolean; resolve(e: XmlElement): void; reject(err: Error): void }> = []; private smId: string | undefined; private handled = 0; private smOn = false; private pingTimer: unknown; private deadline: unknown; private streamError: string | undefined; private finished = false;
constructor(private readonly o: XmppClientOptions) {}
/** The stream management state to save, so a later connection can resume. */ sm(): { id: string; handled: number } | undefined { return this.smOn && this.smId !== undefined ? { id: this.smId, handled: this.handled } : undefined; }
private write(e: XmlElement): void { if (this.closed) throw new Error("the connection is closed"); this.ws.send(serialize(e)); }
send(stanza: XmlElement): void { this.write({ ...stanza, attrs: { xmlns: "jabber:client", ...stanza.attrs } }); }
private open(): void { this.write(el("open", { xmlns: FRAMING, to: this.o.domain, version: "1.0" })); }
private route(e: XmlElement): void { const i = this.waiters.findIndex((w) => w.test(e)); if (i >= 0) this.waiters.splice(i, 1)[0]!.resolve(e); // only login steps wait for frames; once bound, a frame nobody waits for is not kept (a peer could otherwise grow this without limit) else if (!this.bound && this.backlog.length < 50) this.backlog.push(e); }
private waitFor(test: (e: XmlElement) => boolean, what: string): Promise<XmlElement> { const hit = this.backlog.findIndex(test); if (hit >= 0) return Promise.resolve(this.backlog.splice(hit, 1)[0]!); return new Promise((resolve, reject) => { const timer = this.o.setTimer(() => { this.waiters = this.waiters.filter((w) => w !== entry); reject(new Error(`timed out waiting for ${what}`)); }, this.o.connectTimeoutMs ?? 10_000); const entry = { test, resolve: (e: XmlElement) => (this.o.clearTimer(timer), resolve(e)), reject: (err: Error) => (this.o.clearTimer(timer), reject(err)) }; this.waiters.push(entry); }); }
private finish(reason: string): void { if (this.finished) return; this.finished = true; this.closed = true; if (this.pingTimer !== undefined) this.o.clearTimer(this.pingTimer); if (this.deadline !== undefined) this.o.clearTimer(this.deadline); for (const w of this.waiters.splice(0)) w.reject(new Error(reason)); try { this.ws.close(); } catch { /* already closed */ } this.o.onClose?.({ reason }); }
private onFrame(raw: unknown): void { let e: XmlElement; try { e = parseXml(String(raw)); } catch { return; // not a frame this client understands: dropped, the connection goes on } if (this.smOn && (e.name === "message" || e.name === "presence" || (e.name === "iq" && !this.isPingFromServer(e)))) this.handled++; if (e.name === "r" && e.ns === SM) return this.write(el("a", { xmlns: SM, h: String(this.handled) })); if (e.name === "a" && e.ns === SM) return; if (e.name === "iq" && this.isPingFromServer(e)) return this.write(el("iq", { xmlns: "jabber:client", type: "result", id: e.attrs.id ?? "", ...(e.attrs.from !== undefined ? { to: e.attrs.from } : {}) })); if (e.name === "iq" && e.attrs.type === "result" && (e.attrs.id ?? "").startsWith("aite-ping-")) { this.pongs++; if (this.deadline !== undefined) this.o.clearTimer(this.deadline); this.deadline = undefined; return; } if (e.name === "error" && e.prefix === "stream") { this.streamError = `stream error: ${e.children[0]?.name ?? "unknown"}`; return; } if (e.name === "close" && e.ns === FRAMING) return this.finish(this.streamError ?? "the server closed the stream"); if (["message", "presence"].includes(e.name) || (e.name === "iq" && this.bound)) { this.route(e); this.o.onStanza?.(e); return; } this.route(e); }
private bound = false; private isPingFromServer(e: XmlElement): boolean { return e.attrs.type === "get" && e.children[0]?.name === "ping" && e.children[0].ns === PING; }
async connect(): Promise<{ jid: string; resumed: boolean }> { this.ws = this.o.webSocket(this.o.url, "xmpp"); this.ws.onmessage = (ev) => this.onFrame(ev.data); this.ws.onclose = (ev) => this.finish(this.streamError ?? `the connection closed (code ${ev.code})`); this.ws.onerror = () => undefined; // the close that follows says why await new Promise<void>((resolve, reject) => { this.ws.onopen = () => resolve(); const prior = this.ws.onclose; this.ws.onclose = (ev) => { reject(new Error("could not open the connection")); prior?.(ev); }; }); this.ws.onclose = (ev) => this.finish(this.streamError ?? `the connection closed (code ${ev.code})`); try { this.open(); const features = await this.waitFor((e) => e.name === "features", "the server's features"); const mechanisms = features.children.find((c) => c.name === "mechanisms")?.children.map((m) => m.text) ?? []; if (!mechanisms.includes("PLAIN")) throw new Error("the server does not offer SASL PLAIN"); this.write(el("auth", { xmlns: SASL, mechanism: "PLAIN" }, b64(`\0${this.o.user}\0${this.o.pass}`))); const result = await this.waitFor((e) => e.name === "success" || e.name === "failure", "the login result"); if (result.name === "failure") throw new XmppAuthError(`login refused${result.children[0] ? ` (${result.children[0].name})` : ""}`); this.open(); await this.waitFor((e) => e.name === "features" && e.children.some((c) => c.name === "bind"), "the features after login");
let resumed = false; if (this.o.resume !== undefined) { this.write(el("resume", { xmlns: SM, previd: this.o.resume.id, h: String(this.o.resume.handled) })); const r = await this.waitFor((e) => e.name === "resumed" || e.name === "failed", "the resume answer"); if (r.name === "resumed") { resumed = true; this.smId = this.o.resume.id; this.handled = this.o.resume.handled; this.smOn = true; this.jid = `${this.o.user}@${this.o.domain}/${this.o.resource}`; } } if (!resumed) { this.write(el("iq", { xmlns: "jabber:client", type: "set", id: "aite-bind" }, el("bind", { xmlns: BIND }, el("resource", {}, this.o.resource)))); const bound = await this.waitFor((e) => e.name === "iq" && e.attrs.id === "aite-bind", "the bind answer"); if (bound.attrs.type !== "result") throw new Error("the server refused to bind a resource"); this.jid = bound.children[0]?.children.find((c) => c.name === "jid")?.text ?? `${this.o.user}@${this.o.domain}/${this.o.resource}`; this.write(el("enable", { xmlns: SM, resume: "true" })); const en = await this.waitFor((e) => e.name === "enabled" || e.name === "failed", "stream management").catch(() => undefined); if (en?.name === "enabled") { this.smOn = true; this.smId = en.attrs.id; this.handled = 0; } } this.bound = true; if (!resumed && this.o.announce !== false) this.write(el("presence", { xmlns: "jabber:client" })); this.startPing(); return { jid: this.jid, resumed }; } catch (err) { this.finish(err instanceof Error ? err.message : String(err)); throw err; } }
private startPing(): void { if (this.o.pingMs === undefined) return; const every = this.o.pingMs; const timeout = this.o.pingTimeoutMs ?? 10_000; const beat = (): void => { if (this.closed) return; this.pings++; try { this.write(el("iq", { xmlns: "jabber:client", type: "get", id: `aite-ping-${this.pings}`, to: this.o.domain }, el("ping", { xmlns: PING }))); } catch { return; } this.deadline = this.o.setTimer(() => this.finish("a ping was not answered"), timeout); this.pingTimer = this.o.setTimer(beat, every); }; this.pingTimer = this.o.setTimer(beat, every); }
close(): void { if (this.finished) return; try { this.write(el("close", { xmlns: FRAMING })); } catch { /* already closed */ } this.finish("closed by this side"); }}