safetynet/tests/broker-framed-channel.test.mjs

import test from 'node:test';
import assert from 'node:assert/strict';
import {EventEmitter} from 'node:events';
import {createRequire} from 'node:module';
import {PassThrough} from 'node:stream';
import {randomUUID} from 'node:crypto';
const {createFramedChannel}=createRequire(import.meta.url)('../apps/network-broker/framed-channel.cjs');
function frame(value){const bytes=Buffer.from(typeof value==='string'?value:JSON.stringify(value)),result=Buffer.alloc(bytes.length+4);result.writeUInt32LE(bytes.length);bytes.copy(result,4);return result;}
function fixture(options={}){
 const read=new EventEmitter(),write=new EventEmitter(),writes=[],callbacks=[];
 read.destroy=()=>{};write.destroy=()=>{};write.write=(bytes,callback)=>{writes.push(Buffer.from(bytes));callbacks.push(callback);};
 const channel=createFramedChannel({readable:read,writable:write,...options}),messages=[];let disconnected=0;
 channel.on('message',value=>messages.push(value));channel.on('disconnect',()=>disconnected++);
 return {read,write,channel,writes,callbacks,messages,disconnected:()=>disconnected};
}
test('fragmented headers and UTF8 plus coalesced frames preserve exact bounded objects',()=>{
 const f=fixture(),bytes=Buffer.concat([frame({text:'ä漢字'}),frame({type:'stop'})]);
 for(const byte of bytes)f.read.emit('data',Buffer.from([byte]));
 assert.deepEqual(f.messages,[{text:'ä漢字'},{type:'stop'}]);assert.equal(f.channel.connected,true);
 f.read.emit('data',Buffer.concat([frame({n:1}),frame({n:2})]));assert.equal(f.messages.length,4);
 f.channel.disconnect();assert.equal(f.disconnected(),1);
});
test('first bootstrap alone may exceed64KiB; later oversize rejected before allocation',()=>{
 const f=fixture({inboundFirstLimit:2097152});f.read.emit('data',frame({payload:'x'.repeat(70000)}));assert.equal(f.messages.length,1);
 const header=Buffer.alloc(4);header.writeUInt32LE(65537);f.read.emit('data',header);assert.equal(f.channel.connected,false);assert.equal(f.messages.length,1);
 for(const length of [0,2097153,0xffffffff]){const g=fixture({inboundFirstLimit:2097152}),h=Buffer.alloc(4);h.writeUInt32LE(length);g.read.emit('data',h);assert.equal(g.disconnected(),1);}
});
test('invalid UTF8, primitive JSON and malformed JSON disconnect without message or raw error',()=>{
 const invalid=Buffer.from([2,0,0,0,0xc0,0xaf]);
 for(const bytes of [invalid,frame('null'),frame('[]'),frame('"secret"'),frame('{broken')]){const f=fixture();f.read.emit('data',bytes);assert.equal(f.messages.length,0);assert.equal(f.disconnected(),1);}
 const f=fixture();f.read.emit('data',frame({valid:true}).subarray(0,6));f.read.emit('end');assert.equal(f.disconnected(),1);assert.equal(f.messages.length,0);
});
test('writes are framed with first-only bootstrap limit and at most8 unsettled writes',()=>{
 const f=fixture({outboundFirstLimit:2097152}),results=[];
 f.channel.send({payload:'x'.repeat(70000)},error=>results.push(error));assert.equal(f.writes[0].readUInt32LE(0),70014);f.callbacks.shift()();assert.equal(results[0],undefined);
 f.channel.send({payload:'x'.repeat(70000)},error=>results.push(error));assert.equal(f.disconnected(),1);assert.equal(results[1].message,'broker-channel-unavailable');assert.equal(f.writes.length,1);
 const g=fixture();for(let i=0;i<8;i++)g.channel.send({n:i});assert.equal(g.writes.length,8);let failure;
 g.channel.send({n:8},error=>{failure=error;});assert.equal(failure.message,'broker-channel-unavailable');assert.equal(g.disconnected(),1);assert.equal(g.writes.length,8);
});
test('stream failure yields fixed callback error and exactly one disconnect',()=>{
 const f=fixture();let failure;f.channel.send({type:'ready'},error=>{failure=error;});f.callbacks[0](Error('PRIVATE_FAILURE'));assert.equal(failure.message,'broker-channel-unavailable');
 f.read.emit('error',Error('PRIVATE_FAILURE'));f.write.emit('error',Error('PRIVATE_FAILURE'));assert.equal(f.disconnected(),1);
});
test('disconnect fails pending callbacks exactly once; late write success grants no success',()=>{
 const f=fixture(),results=[];f.channel.send({type:'audit'},error=>results.push(error?.message||'success'));
 f.channel.disconnect();assert.deepEqual(results,['broker-channel-unavailable']);
 f.callbacks[0]();f.callbacks[0](Error('late'));assert.deepEqual(results,['broker-channel-unavailable']);assert.equal(f.disconnected(),1);
});
test('throwing write callbacks cannot prevent fail-closed disconnection',()=>{
 const f=fixture();f.channel.send({type:'audit'},()=>{throw Error('private callback');});
 assert.doesNotThrow(()=>f.callbacks[0]());assert.equal(f.channel.connected,false);
 assert.doesNotThrow(()=>f.channel.send({type:'audit'},()=>{throw Error('private callback');}));assert.equal(f.disconnected(),1);
});
test('actual duplex framing carries production protocol and exact durable receipt',async t=>{
 const {runProductionProcess}=createRequire(import.meta.url)('../apps/network-broker/production-entry.cjs');
 const toChild=new PassThrough(),toParent=new PassThrough();
 const child=createFramedChannel({readable:toChild,writable:toParent,inboundFirstLimit:2097152});
 const parent=createFramedChannel({readable:toParent,writable:toChild,outboundFirstLimit:2097152});
 t.after(()=>parent.disconnect());
 let callbacks,closed=0;
 const processControl=runProductionProcess(async(_config,value)=>{callbacks=value;return {authMode:'basic',credentials:{username:'safetynet-'+'a'.repeat(32),password:'b'.repeat(43)},listen:async()=>43100,close:async()=>closed++};},child);
 const received=[];parent.on('message',message=>received.push(message));
 const context={version:1,generation:randomUUID(),policyVersion:1};
 parent.send({type:'start',...context,config:{brokerPort:43100}});
 await new Promise(resolve=>setImmediate(resolve));assert.equal(received[0].type,'ready');
 const event={eventId:randomUUID(),kind:'upload.authorized',ruleId:'broker-approved',policyVersion:1,request:{bytes:3,detectedType:'text'}};
 let committed=false;const pending=callbacks.persistAudit(event).then(receipt=>{committed=true;return receipt;});
 assert.equal(committed,false);assert.equal(received[1].type,'audit');
 parent.send({type:'audit-receipt',...context,requestId:received[1].requestId,sequence:received[1].sequence,receipt:{persisted:true,eventId:event.eventId}});
 assert.deepEqual(await pending,{persisted:true,eventId:event.eventId});
 // Receipt replay is invalid even though its frame, generation and UUID match.
 parent.send({type:'audit-receipt',...context,requestId:received[1].requestId,sequence:received[1].sequence,receipt:{persisted:true,eventId:event.eventId}});
 await processControl.close();assert.equal(child.connected,false);assert.equal(closed,1);
});