LumaWeb/experiments/0002-xmpp-library/parser-validation/bounded-ingress.mjs

import {SaxesParser} from 'saxes';
import jxt from 'stanza/jxt/index.js';

// Experiment policy only. The byte ceiling bounds SAX token/namespace storage
// even before callbacks can inspect a completed name or attribute value.
export const LIMITS = Object.freeze({bytes: 4096, depth: 8, elements: 32,
  attributes: 16, attributeChars: 1024, nameChars: 128, valueChars: 256});
class IngressError extends Error {
  constructor(code) { super(code); this.code = code; }
}
const fail = code => { throw new IngressError(code); };

export function validateMessage(data) {
  if (typeof data !== 'string') fail('text-required');
  // Avoid encoding an attacker-sized string. UTF-8 bytes >= UTF-16 units for
  // valid XML; isolated surrogates are subsequently rejected by saxes.
  if (data.length > LIMITS.bytes) fail('bytes');
  if (new TextEncoder().encode(data).length > LIMITS.bytes) fail('bytes');
  if (!data.startsWith('<')) fail('framing');
  let depth = 0, elements = 0, attributes = 0, attributeChars = 0, ended = false;
  const parser = new SaxesParser({xmlns: true, fragment: false,
    defaultXMLVersion: '1.0', forceXMLVersion: true});
  parser.on('error', () => fail('invalid-xml'));
  for (const event of ['doctype', 'comment', 'processinginstruction', 'xmldecl']) {
    parser.on(event, () => fail('restricted-xml'));
  }
  parser.on('opentagstart', tag => {
    if (++depth > LIMITS.depth) fail('depth');
    if (++elements > LIMITS.elements) fail('elements');
    if (tag.name.length > LIMITS.nameChars) fail('name');
  });
  parser.on('attribute', attr => {
    if (++attributes > LIMITS.attributes) fail('attributes');
    if (attr.name.length > LIMITS.nameChars) fail('name');
    if (attr.value.length > LIMITS.valueChars) fail('attribute-value');
    attributeChars += attr.name.length + attr.value.length;
    if (attributeChars > LIMITS.attributeChars) fail('attribute-chars');
  });
  parser.on('closetag', () => { depth--; });
  parser.on('end', () => { ended = true; });
  // close() detects unfinished XML. Throwing from callbacks aborts immediately;
  // there is no recovery listener, DTD fetch, custom entity or tree construction.
  parser.write(data).close();
  if (!ended || depth !== 0 || elements === 0) fail('framing');
}

export function importMessage(data, registry, config) {
  validateMessage(data);
  // Only bounded, complete documents reach the actual candidate parser/import.
  const xml = jxt.parse(data, {allowComments: false});
  if (!xml) fail('invalid-xml');
  const kind = registry.getImportKey(xml);
  const stanza = registry.import(xml, {lang: config.lang,
    acceptLanguages: config.acceptLanguages});
  if (!kind || !stanza) fail('unknown-root');
  return {kind, stanza};
}

// Public Transport interface and Client.transports constructor registry.
// Socket injection is solely for finite fixtures; default is browser WebSocket.
export function installBoundedTransport(client, Socket = globalThis.WebSocket) {
  class BoundedTransport {
    #client; #registry; #sm; #socket; #config; #closed = true;
    hasStream = false;
    stream;
    constructor(agent, sm, registry) {
      this.#client = agent; this.#sm = sm; this.#registry = registry;
    }
    connect(config) {
      if (this.#socket || !this.#closed) throw new Error('Fresh transport required');
      if (new URL(config.url).protocol !== 'wss:') throw new Error('WSS required');
      this.#config = config;
      this.#closed = false;
      const socket = this.#socket = new Socket(config.url, 'xmpp');
      socket.onopen = () => {
        if (this.#closed) return;
        if (socket.protocol !== 'xmpp') return this.#reject('subprotocol');
        this.#sm.started = false;
        this.#client.emit('connected');
        this.restart();
      };
      socket.onmessage = event => {
        if (this.#closed) return;
        let parsed;
        try { parsed = importMessage(event.data, this.#registry, this.#config); }
        catch (error) { return this.#reject(error instanceof IngressError ? error.code : 'invalid-xml'); }
        // No raw payload event/log or async parser continuation escapes ingress.
        if (parsed.kind === 'stream') {
          if (parsed.stanza.action === 'open') {
            this.hasStream = true;
            this.stream = parsed.stanza;
            this.#client.emit('stream:start', parsed.stanza);
          } else {
            this.#client.emit('stream:end');
            this.disconnect(false);
          }
        } else {
          this.#client.emit('stream:data', parsed.stanza, parsed.kind);
        }
      };
      socket.onerror = () => this.#reject('socket-error');
      socket.onclose = () => this.disconnect(false);
    }
    #reject(code) {
      if (this.#closed) return;
      // Close/clear before notifying consumers; a reentrant callback cannot
      // revive the attempt. Never expose parser error text containing input.
      this.disconnect(false);
      this.#client.emit('ingress:rejected', code);
    }
    disconnect(clean = true) {
      if (this.#closed) return;
      this.#closed = true;
      const socket = this.#socket;
      this.#socket = undefined;
      if (socket) {
        socket.onopen = socket.onmessage = socket.onerror = socket.onclose = null;
        if (clean && this.hasStream && socket.readyState === 1) {
          socket.send(this.#registry.export('stream', {action: 'close'}).toString());
        }
        socket.close();
      }
      this.hasStream = false;
      this.stream = undefined;
      this.#config = undefined;
      if (this.#client.transport === this) this.#client.emit('--transport-disconnected');
    }
    restart() {
      if (this.#closed) return;
      this.hasStream = false;
      this.stream = undefined;
      this.#write(this.#registry.export('stream', {action: 'open',
        lang: this.#config.lang, to: this.#config.server, version: '1.0'}).toString());
    }
    #write(data) {
      if (this.#closed || this.#socket?.readyState !== 1) throw new Error('Socket closed');
      this.#socket.send(data);
    }
    async send(name, data) {
      const xml = this.#registry.export(name, data);
      if (!xml) throw new Error('Unknown outgoing stanza');
      this.#write(xml.toString());
    }
  }
  client.transports['bounded-websocket'] = BoundedTransport;
}