import test from 'node:test'; import assert from 'node:assert/strict'; import { createServer } from 'node:net'; import { fork } from 'node:child_process'; import { once } from 'node:events'; import { mkdtempSync, rmSync } from 'node:fs'; import { join } from 'node:path'; import { tmpdir } from 'node:os'; import { EventEmitter } from 'node:events'; import { PassThrough } from 'node:stream'; import { createPeerReader } from '../src/peercred.mjs'; import { inspectInterpreter } from '../src/interpreter.mjs'; const interpreter = inspectInterpreter(); test('T2 eight proof callers and a racing copier bind the exact accepted socket, which survives helper exit', {timeout:10000}, async t => { const root = mkdtempSync(join(tmpdir(),'g196-p-')); const path = join(root,'s'); const reader = createPeerReader({interpreter}); const sockets = [], pending = [], after = []; let resolveReady; const ready = new Promise(r => resolveReady = r); const server = createServer({allowHalfOpen:true}, socket => { sockets.push(socket); let text = ''; socket.on('data', chunk => { text += chunk; if (!text.includes('\n')) return; if (!socket.proof) { socket.proof = JSON.parse(text.slice(0,text.indexOf('\n'))); text = text.slice(text.indexOf('\n')+1); pending.push(socket); if (pending.length === 9) resolveReady(); } if (text.includes('after-helper\n')) after.push(socket); }); socket.on('error',()=>{}); }); const children = []; t.after(async () => { reader.close(); for(const c of children)c.kill(); for(const s of sockets)s.destroy(); await new Promise(r=>server.close(r)); rmSync(root,{recursive:true,force:true}); }); await new Promise(r=>server.listen(path,r)); const pids=[]; for(let i=0;i<9;i++) { const c=fork(new URL('../fixtures/peer-client.mjs',import.meta.url),[],{stdio:['ignore','ignore','inherit','ipc']}); children.push(c); pids.push((await once(c,'message'))[0].pid); } const replies=children.map(c=>once(c,'message')); children.forEach((c,i)=>c.send({path,...(i===8?{claimed:pids[0]}:{})})); await ready; const peers = await Promise.all(pending.map(socket => reader.read(socket))); pending.forEach(s=>s.resume()); // spawn stdio pauses inherited streams. assert.equal(new Set(peers.map(p=>p.pid)).size,9); pending.forEach((socket,i)=> { const peer=peers[i]; assert.equal(peer.uid,process.getuid()); const accepted=peer.pid===socket.proof.human.pid; assert.equal(accepted, peer.pid!==pids[8]); socket.write(JSON.stringify({accepted,pid:peer.pid})+'\n'); }); const results=await Promise.all(replies); assert.equal(results.filter(([r])=>r.reply.accepted).length,8); assert.equal(results[8][0].reply.accepted,false); await Promise.all(sockets.map(s=>s.readableEnded?Promise.resolve():once(s,'end'))); assert.equal(after.length,9, 'all sockets read after the helper closed its duplicate'); }); function fakeSpawn(output, status=0) { return () => { const child=new EventEmitter(); child.stdout=new PassThrough(); child.stderr=new PassThrough(); child.kill=()=>{queueMicrotask(()=>child.emit('close',null,'SIGKILL'));return true;}; queueMicrotask(()=>{child.stdout.end(output);child.emit('close',status,null);}); return child; }; } test('helper output and failures fail closed including unsafe gid', async () => { for(const [out,code] of [['{}',0],['{"pid":2,"uid":0,"gid":9007199254740992}',0],['{"pid":2,"uid":0,"gid":0}\nextra',0],['x'.repeat(257),0],['{"pid":2,"uid":0,"gid":0}',1]]) { const reader=createPeerReader({interpreter,spawn:fakeSpawn(out,code)}); await assert.rejects(reader.read({}),/peercred-required/);reader.close(); } const reader=createPeerReader({interpreter,spawn:()=>{throw Error('spawn');}}); await assert.rejects(reader.read({}),/peercred-required/);reader.close(); }); test('helper timeout, capacity and shutdown refuse without leaving children', async () => { const reader=createPeerReader({interpreter,timeout:25,spawn:()=>{ const c=new EventEmitter();c.stdout=new PassThrough();c.stderr=new PassThrough(); c.kill=()=>{queueMicrotask(()=>c.emit('close',null,'SIGKILL'));return true;};return c; }}); const jobs=Array.from({length:16},()=>assert.rejects(reader.read({}),/peercred-required/)); await assert.rejects(reader.read({}),/peercred-required/); await Promise.all(jobs);reader.close(); await assert.rejects(reader.read({}),/peercred-required/); }); test('missing or non-regular peer helper refuses at construction before opening the bus',()=>{ assert.throws(()=>createPeerReader({interpreter,stat:()=>{throw Error('missing');}}),/peercred-required/); assert.throws(()=>createPeerReader({interpreter,stat:()=>({isFile:()=>false,isSymbolicLink:()=>true})}),/peercred-required/); }); test('shutdown waits for killed helper close before returning',async()=>{ let reaped=0; const reader=createPeerReader({interpreter,spawn:()=>{ const c=new EventEmitter();c.stdout=new PassThrough();c.stderr=new PassThrough(); c.kill=()=>{setTimeout(()=>{reaped++;c.emit('close',null,'SIGKILL');},5);return true;};return c; }}); const pending=assert.rejects(reader.read({}),/peercred-required/); await reader.close(); assert.equal(reaped,1,'shutdown must await the helper close event'); await pending; });