stack/packages/bus/tests/peercred.test.mjs

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;
});