LumaWeb/experiments/0002-xmpp-library/parser-validation/test/integration.test.mjs

import assert from 'node:assert/strict';
import {test} from 'node:test';
import clientModule from 'stanza/Client.js';
import {installBoundedTransport, validateMessage, LIMITS} from '../bounded-ingress.mjs';
import {opening, closing, message, valid, deep, elements, attributes, boundary, invalid} from './fixtures.mjs';
const Client = clientModule.default;
const tick = () => new Promise(resolve => setImmediate(resolve));

async function harness() {
  const sockets = [];
  class FixtureSocket {
    readyState = 1; protocol = 'xmpp'; sent = []; closes = 0;
    constructor(url, protocol) { this.url = url; this.requestedProtocol = protocol; sockets.push(this); }
    send(data) { this.sent.push(data); }
    close() { this.closes++; this.readyState = 3; }
  }
  const client = new Client({autoReconnect: false, server: 'example.invalid'});
  installBoundedTransport(client, FixtureSocket);
  client.updateConfig({transports: {'bounded-websocket': 'wss://example.invalid/xmpp'},
    transportPreferenceOrder: ['bounded-websocket']});
  const received = [], errors = [], raw = [], stanzas = [], starts = [];
  client.on('stream:data', (stanza, kind) => received.push({stanza, kind}));
  client.on('stanza', stanza => stanzas.push(stanza));
  client.on('stream:start', stanza => starts.push(stanza));
  client.on('ingress:rejected', code => errors.push(code));
  client.on('raw', (...args) => raw.push(args));
  // Tracing calls the real registry.import; proves rejection before import.
  let imports = 0;
  const original = client.stanzas.import.bind(client.stanzas);
  client.stanzas.import = (...args) => { imports++; return original(...args); };
  await client.connect();
  const socket = sockets.at(-1);
  socket.onopen();
  return {client, sockets, socket, received, errors, raw, stanzas, starts,
    imports: () => imports, finish: async () => {client.transport?.disconnect(false); await tick();}};
}

for (const [name, xml, code] of invalid) {
  test(`terminal rejection: ${name}`, async () => {
    const h = await harness();
    const late = h.socket.onmessage;
    late({data: xml});
    assert.deepEqual(h.errors, [code]);
    assert.equal(h.socket.closes, 1);
    assert.equal(h.socket.onmessage, null);
    assert.equal(h.received.length, 0);
    assert.equal(h.starts.length, 0);
    if (code !== 'unknown-root') assert.equal(h.imports(), 0);
    // Old captured handlers and subsequent messages cannot revive an attempt.
    late({data: closing}); late({data: valid});
    await tick();
    assert.equal(h.stanzas.length, 0);
    assert.equal(h.received.length, 0);
    assert.deepEqual(h.errors, [code]);
    assert.equal(h.client.transport, undefined);
    assert.equal(h.raw.length, 0);
    await h.finish();
  });
}

for (const [name, xml, body] of [
  ['all aggregate limits equal', boundary().xml, boundary().body],
  ['depth equal', deep(LIMITS.depth)],
  ['element count equal', elements(LIMITS.elements)],
  ['attribute count equal', attributes(LIMITS.attributes)],
  ['attribute value equal', `<message xmlns="jabber:client" a="${'v'.repeat(256)}"/>`],
  ['name equal', `<message xmlns="jabber:client"><${'x'.repeat(128)}/></message>`],
  ['UTF-8 byte equality', message('🦉'.repeat(1000) + 'a'.repeat(4096 - message('').length - 4000)), '🦉'.repeat(1000) + 'a'.repeat(4096 - message('').length - 4000)],
  ['escaped text and CDATA', message('&lt;&amp;<![CDATA[<fixture>]]>'), '<&<fixture>'],
  ['prefixed namespace', '<c:message xmlns:c="jabber:client"><c:body>namespace</c:body></c:message>', 'namespace']
]) {
  test(`actual parser/import accepts ${name}`, async () => {
    const h = await harness();
    h.socket.onmessage({data: xml});
    await tick();
    assert.deepEqual(h.errors, []);
    assert.equal(h.received.length, 1);
    assert.equal(h.received[0].kind, 'message');
    if (body !== undefined) assert.equal(h.received[0].stanza.body, body);
    assert.equal(h.stanzas.length, 1);
    assert.ok(h.imports() > 0);
    await h.finish();
  });
}

test('bytes one over and UTF-8 one over reject before import', async () => {
  for (const xml of [boundary().xml + ' ', message('🦉'.repeat(1000) + 'a'.repeat(4097 - message('').length - 4000))]) {
    const h = await harness(); h.socket.onmessage({data: xml});
    assert.deepEqual(h.errors, ['bytes']); assert.equal(h.imports(), 0);
    await h.finish();
  }
});

test('stock 18-message partial document never accumulates or revives; fresh connect works', async () => {
  const h = await harness(); const old = h.socket.onmessage;
  old({data: opening});
  for (let i = 0; i < 16; i++) old({data: 'a'.repeat(4096)});
  old({data: closing});
  await tick();
  assert.equal(h.received.length, 0); assert.equal(h.imports(), 0);
  assert.deepEqual(h.errors, ['invalid-xml']);
  await h.client.connect(); const fresh = h.sockets.at(-1); fresh.onopen();
  old({data: valid}); assert.equal(h.received.length, 0);
  fresh.onmessage({data: valid}); await tick();
  assert.equal(h.received.length, 1); assert.equal(h.stanzas.length, 1);
  assert.equal(h.received[0].stanza.body, 'finite fixture');
  await h.finish();
});

test('successive complete documents have no shared namespace/parser state', async () => {
  const h = await harness();
  h.socket.onmessage({data: valid}); h.socket.onmessage({data: valid}); await tick();
  assert.equal(h.stanzas.length, 2);
  h.socket.onmessage({data: '<c:message/>'}); await tick();
  assert.deepEqual(h.errors, ['invalid-xml']); assert.equal(h.stanzas.length, 2);
  await h.finish();
});

test('open, restart, message, outgoing export and close use actual registry', async () => {
  const h = await harness();
  h.socket.onmessage({data: '<open xmlns="urn:ietf:params:xml:ns:xmpp-framing" from="example.invalid" version="1.0"/>'});
  assert.equal(h.client.transport.hasStream, true); assert.equal(h.starts.length, 1);
  h.client.transport.restart(); assert.equal(h.client.transport.hasStream, false);
  await h.client.transport.send('message', {body: 'outgoing fixture'});
  assert.ok(h.socket.sent.at(-1).includes('outgoing fixture'));
  h.socket.onmessage({data: '<close xmlns="urn:ietf:params:xml:ns:xmpp-framing"/>'});
  await tick(); assert.equal(h.socket.closes, 1); assert.equal(h.client.transport, undefined);
  assert.equal(h.errors.length, 0);
  await h.finish();
});

test('failed subprotocol, socket error, clean disconnect and stale callbacks are terminal', async () => {
  for (const mode of ['subprotocol', 'error', 'clean']) {
    const h = await harness(); const old = h.socket.onmessage;
    if (mode === 'subprotocol') {h.socket.protocol = 'wrong'; h.socket.onopen();}
    if (mode === 'error') h.socket.onerror(new Error('fixture'));
    if (mode === 'clean') {
      h.socket.onmessage({data: '<open xmlns="urn:ietf:params:xml:ns:xmpp-framing"/>'});
      h.client.transport.disconnect(); assert.ok(h.socket.sent.at(-1).includes('close'));
    }
    old({data: valid}); await tick();
    assert.equal(h.received.length, 0); assert.equal(h.socket.closes, 1);
    await h.finish();
  }
});

test('SAX framing validation retains no attempt object', () => {
  assert.throws(() => validateMessage(opening));
  assert.throws(() => validateMessage(closing));
  assert.doesNotThrow(() => validateMessage(valid));
});