From d61c5441fc503ad0ea28c43b4c4bfc8aaf6503d8 Mon Sep 17 00:00:00 2001 From: ms-lead-reviewer Date: Fri, 17 Jul 2026 20:44:37 -0500 Subject: [PATCH] test(#828): define lease broker security contract Add red-first acceptance coverage for kernel peer credentials, broker-minted sessions, ancestry and generation fencing, crypto tokens, protected socket/state posture, and fail-closed persistence boundaries. --- .../lease-broker.acceptance.spec.ts | 579 ++++++++++++++++++ .../src/lease-broker/state_store_unittest.py | 109 ++++ 2 files changed, 688 insertions(+) create mode 100644 packages/mosaic/src/lease-broker/lease-broker.acceptance.spec.ts create mode 100644 packages/mosaic/src/lease-broker/state_store_unittest.py diff --git a/packages/mosaic/src/lease-broker/lease-broker.acceptance.spec.ts b/packages/mosaic/src/lease-broker/lease-broker.acceptance.spec.ts new file mode 100644 index 0000000..db9b6ec --- /dev/null +++ b/packages/mosaic/src/lease-broker/lease-broker.acceptance.spec.ts @@ -0,0 +1,579 @@ +import { randomUUID } from 'node:crypto'; +import { chmod, mkdtemp, readFile, stat, symlink, writeFile } from 'node:fs/promises'; +import { createConnection, type Socket } from 'node:net'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { spawn, spawnSync, type ChildProcess } from 'node:child_process'; + +import { afterEach, describe, expect, test, vi } from 'vitest'; + +interface BrokerReply { + ok: boolean; + code?: string; + session_id?: string; + peer?: { pid: number; uid: number; gid: number; starttime: string }; + token?: string; +} + +const daemonPath = new URL('../../framework/tools/lease-broker/daemon.py', import.meta.url) + .pathname; +const children: ChildProcess[] = []; + +async function withTimeout( + promise: Promise, + label: string, + milliseconds = 3_000, +): Promise { + let timer: NodeJS.Timeout | undefined; + try { + return await Promise.race([ + promise, + new Promise((_, reject) => { + timer = setTimeout(() => reject(new Error(`${label} timed out`)), milliseconds); + }), + ]); + } finally { + if (timer !== undefined) clearTimeout(timer); + } +} + +async function rawRequest( + socketPath: string, + write: (socket: Socket) => void, +): Promise { + return await withTimeout( + new Promise((resolve, reject) => { + const socket = createConnection(socketPath); + let response = ''; + socket.setEncoding('utf8'); + socket.once('error', reject); + socket.on('data', (chunk: string) => { + response += chunk; + }); + socket.once('end', () => resolve(JSON.parse(response) as BrokerReply)); + socket.once('connect', () => write(socket)); + }), + 'raw broker request', + ); +} + +async function request(socketPath: string, requestValue: object): Promise { + return await new Promise((resolve, reject) => { + const socket = createConnection(socketPath); + let response = ''; + socket.setEncoding('utf8'); + socket.once('error', reject); + socket.on('data', (chunk: string) => { + response += chunk; + }); + socket.once('end', () => resolve(JSON.parse(response) as BrokerReply)); + socket.once('connect', () => socket.end(`${JSON.stringify(requestValue)}\n`)); + }); +} + +async function startBroker( + parentMode = 0o700, +): Promise<{ root: string; socket: string; state: string }> { + const root = await mkdtemp(join(tmpdir(), 'mosaic-lease-broker-')); + await chmod(root, parentMode); + const socket = join(root, 'broker.sock'); + const state = join(root, 'state.json'); + const child = spawn('python3', [daemonPath, '--socket', socket, '--state', state], { + stdio: ['ignore', 'pipe', 'pipe'], + }); + children.push(child); + await new Promise((resolve, reject) => { + let stderr = ''; + child.stderr?.setEncoding('utf8'); + child.stderr?.on('data', (chunk: string) => (stderr += chunk)); + child.once('error', reject); + child.once('exit', (code: number | null) => + reject(new Error(`broker exited ${code}: ${stderr}`)), + ); + child.stdout?.once('data', () => resolve()); + }); + return { root, socket, state }; +} + +async function startBrokerWithState(stateValue: string): Promise { + const root = await mkdtemp(join(tmpdir(), 'mosaic-lease-broker-')); + await chmod(root, 0o700); + const state = join(root, 'state.json'); + await writeFile(state, stateValue, { mode: 0o600 }); + const child = spawn('python3', [ + daemonPath, + '--socket', + join(root, 'broker.sock'), + '--state', + state, + ]); + children.push(child); + return await new Promise((resolve) => { + let raw = ''; + child.stderr?.on('data', (chunk: Buffer) => (raw += chunk.toString())); + child.once('exit', () => resolve(raw)); + }); +} + +afterEach(() => { + for (const child of children.splice(0)) child.kill('SIGTERM'); + vi.restoreAllMocks(); +}); + +describe('authenticated external lease broker', () => { + test('peercred returns true kernel (pid,starttime)', async () => { + const getuid = process.getuid; + const getgid = process.getgid; + if (getuid === undefined || getgid === undefined) { + throw new Error('Linux peer credentials require process.getuid() and process.getgid()'); + } + + const { socket } = await startBroker(); + const reply = await request(socket, { action: 'register_anchor', runtime_generation: 1 }); + const statText = await readFile(`/proc/${process.pid}/stat`, 'utf8'); + const fields = statText.slice(statText.lastIndexOf(')') + 2).split(' '); + expect(reply).toMatchObject({ + ok: true, + peer: { + pid: process.pid, + uid: getuid(), + gid: getgid(), + starttime: fields[19], + }, + }); + }); + + test.each([null, '', 'chosen'])('caller-asserted session_id refused (%j)', async (session_id) => { + const { socket } = await startBroker(); + const reply = await request(socket, { + action: 'register_anchor', + runtime_generation: 1, + session_id, + }); + expect(reply).toMatchObject({ ok: false, code: 'CALLER_SESSION_ID_REFUSED' }); + }); + + test('sibling-substitution rejected', async () => { + const { socket } = await startBroker(); + const launcher = spawn( + process.execPath, + [ + '-e', + `const n=require('net');const s=n.connect(${JSON.stringify(socket)},()=>s.end(JSON.stringify({action:'register_anchor',runtime_generation:1})+'\\n'));s.on('data',d=>{process.send(JSON.parse(d));setInterval(()=>{},1000)})`, + ], + { stdio: ['ignore', 'ignore', 'ignore', 'ipc'] }, + ); + children.push(launcher); + const registration = await new Promise((resolve) => + launcher.once('message', (message) => resolve(message as BrokerReply)), + ); + const attacker = spawn( + process.execPath, + [ + '-e', + `const n=require('net');const s=n.connect(${JSON.stringify(socket)},()=>s.end(JSON.stringify({action:'authenticate',session_id:${JSON.stringify(registration.session_id)},runtime_generation:1})+'\\n'));s.pipe(process.stdout)`, + ], + { stdio: ['ignore', 'pipe', 'ignore'] }, + ); + children.push(attacker); + let raw = ''; + attacker.stdout?.setEncoding('utf8'); + attacker.stdout?.on('data', (chunk: string) => (raw += chunk)); + await new Promise((resolve) => attacker.once('exit', () => resolve())); + expect(JSON.parse(raw)).toMatchObject({ ok: false, code: 'ANCESTRY_MISMATCH' }); + }); + + test('generation bump revokes prior incarnation', async () => { + const { socket } = await startBroker(); + const registered = await request(socket, { action: 'register_anchor', runtime_generation: 1 }); + expect( + await request(socket, { + action: 'authenticate', + session_id: registered.session_id, + runtime_generation: 2, + }), + ).toMatchObject({ ok: true }); + expect( + await request(socket, { + action: 'authenticate', + session_id: registered.session_id, + runtime_generation: 1, + }), + ).toMatchObject({ ok: false, code: 'STALE_GENERATION' }); + }); + + test('same anchor re-registration reuses its session and revokes the prior incarnation', async () => { + const { socket } = await startBroker(); + const first = await request(socket, { action: 'register_anchor', runtime_generation: 1 }); + const binding = { + compaction_epoch: 2, + request_epoch: 3, + h_source: 'a'.repeat(64), + h_payload: 'b'.repeat(64), + schema_version: 1, + }; + const minted = await request(socket, { + action: 'mint_token', + session_id: first.session_id, + runtime_generation: 1, + binding, + }); + + const bumped = await request(socket, { action: 'register_anchor', runtime_generation: 2 }); + const repeated = await request(socket, { action: 'register_anchor', runtime_generation: 2 }); + const lower = await request(socket, { action: 'register_anchor', runtime_generation: 1 }); + + expect(bumped).toMatchObject({ ok: true, session_id: first.session_id }); + expect(repeated).toMatchObject({ ok: true, session_id: first.session_id }); + expect(lower).toMatchObject({ ok: false, code: 'STALE_GENERATION' }); + expect( + await request(socket, { + action: 'authenticate', + session_id: first.session_id, + runtime_generation: 1, + }), + ).toMatchObject({ ok: false, code: 'STALE_GENERATION' }); + expect( + await request(socket, { + action: 'consume_token', + session_id: first.session_id, + runtime_generation: 2, + token: minted.token, + }), + ).toMatchObject({ ok: false, code: 'TOKEN_REPLAY' }); + }); + + test('crypto token path works when Math.random is poisoned', async () => { + const { socket } = await startBroker(); + vi.spyOn(Math, 'random').mockImplementation(() => { + throw new Error('Math.random forbidden'); + }); + const registered = await request(socket, { action: 'register_anchor', runtime_generation: 1 }); + const binding = { + compaction_epoch: 2, + request_epoch: 3, + h_source: 'a'.repeat(64), + h_payload: 'b'.repeat(64), + schema_version: 1, + }; + const first = await request(socket, { + action: 'mint_token', + session_id: registered.session_id, + runtime_generation: 1, + binding, + }); + const second = await request(socket, { + action: 'mint_token', + session_id: registered.session_id, + runtime_generation: 1, + binding, + }); + expect(first.token).toMatch(/^[a-f0-9]{64}$/); + expect(second.token).not.toBe(first.token); + expect( + await request(socket, { + action: 'consume_token', + session_id: registered.session_id, + runtime_generation: 1, + token: first.token, + }), + ).toMatchObject({ ok: true }); + expect( + await request(socket, { + action: 'consume_token', + session_id: registered.session_id, + runtime_generation: 1, + token: first.token, + }), + ).toMatchObject({ ok: false, code: 'TOKEN_REPLAY' }); + }); + + test('socket parent 0700 and socket 0600 enforced', async () => { + const { root, socket, state } = await startBroker(); + await request(socket, { action: 'register_anchor', runtime_generation: 1 }); + expect((await stat(root)).mode & 0o777).toBe(0o700); + expect((await stat(socket)).mode & 0o777).toBe(0o600); + expect((await stat(state)).mode & 0o777).toBe(0o600); + }); + + test('insecure existing posture refused', async () => { + await expect(startBroker(0o755)).rejects.toThrow(/INSECURE_PARENT_MODE/); + }); + + test('malformed and oversized frames fail closed without killing broker', async () => { + const { socket } = await startBroker(); + const malformed = await new Promise((resolve, reject) => { + const connection = createConnection(socket, () => connection.end('{nope}\n')); + let raw = ''; + connection.on('data', (chunk: Buffer) => (raw += chunk.toString())); + connection.once('end', () => resolve(raw)); + connection.once('error', reject); + }); + expect(JSON.parse(malformed)).toMatchObject({ ok: false, code: 'MALFORMED_REQUEST' }); + const registered = await request(socket, { + action: 'register_anchor', + runtime_generation: 1, + nonce: randomUUID(), + }); + expect(registered.ok).toBe(true); + }); + + test('silent connection deadline cannot prevent the next valid registration', async () => { + const { socket } = await startBroker(); + const silent = createConnection(socket); + await new Promise((resolve, reject) => { + silent.once('connect', resolve); + silent.once('error', reject); + }); + const registered = await withTimeout( + request(socket, { action: 'register_anchor', runtime_generation: 1 }), + 'registration behind silent connection', + ); + expect(registered.ok).toBe(true); + silent.destroy(); + }); + + test('newline-only client without half-close gets no success and cannot block next request', async () => { + const { socket } = await startBroker(); + const incomplete = createConnection(socket); + let raw = ''; + incomplete.setEncoding('utf8'); + incomplete.on('data', (chunk: string) => (raw += chunk)); + await new Promise((resolve, reject) => { + incomplete.once('connect', () => { + incomplete.write( + `${JSON.stringify({ action: 'register_anchor', runtime_generation: 1 })}\n`, + ); + resolve(); + }); + incomplete.once('error', reject); + }); + await new Promise((resolve) => setTimeout(resolve, 1_100)); + expect(raw).not.toContain('"ok":true'); + const registered = await withTimeout( + request(socket, { action: 'register_anchor', runtime_generation: 1 }), + 'registration after non-half-closed client', + ); + expect(registered.ok).toBe(true); + incomplete.destroy(); + }); + + test('client disconnect cannot prevent the next valid authentication', async () => { + const { socket } = await startBroker(); + const registered = await request(socket, { action: 'register_anchor', runtime_generation: 1 }); + const reset = createConnection(socket); + await new Promise((resolve, reject) => { + reset.once('connect', () => { + reset.write(`${JSON.stringify({ action: 'register_anchor', runtime_generation: 1 })}\n`); + reset.destroy(); + resolve(); + }); + reset.once('error', reject); + }); + const authenticated = await withTimeout( + request(socket, { + action: 'authenticate', + session_id: registered.session_id, + runtime_generation: 1, + }), + 'authentication after client disconnect', + ); + expect(authenticated.ok).toBe(true); + }); + + test('delayed second frame is rejected and the next request succeeds', async () => { + const { socket } = await startBroker(); + const reply = await rawRequest(socket, (connection) => { + connection.write(`${JSON.stringify({ action: 'register_anchor', runtime_generation: 1 })}\n`); + setTimeout(() => connection.end('{}\n'), 50); + }); + expect(reply).toMatchObject({ ok: false, code: 'MALFORMED_REQUEST' }); + expect( + await request(socket, { action: 'register_anchor', runtime_generation: 1 }), + ).toMatchObject({ + ok: true, + }); + }); + + test('unterminated frame is rejected and the next request succeeds', async () => { + const { socket } = await startBroker(); + const reply = await rawRequest(socket, (connection) => connection.end('{}')); + expect(reply).toMatchObject({ ok: false, code: 'MALFORMED_REQUEST' }); + expect( + await request(socket, { action: 'register_anchor', runtime_generation: 1 }), + ).toMatchObject({ + ok: true, + }); + }); + + test('genuinely oversized frame is rejected and the next request succeeds', async () => { + const { socket } = await startBroker(); + const reply = await rawRequest(socket, (connection) => + connection.end(`${JSON.stringify({ padding: 'x'.repeat(64 * 1024) })}\n`), + ); + expect(reply).toMatchObject({ ok: false, code: 'MALFORMED_REQUEST' }); + expect( + await request(socket, { action: 'register_anchor', runtime_generation: 1 }), + ).toMatchObject({ + ok: true, + }); + }); + + test('boolean runtime generations fail closed', async () => { + const { socket } = await startBroker(); + expect( + await request(socket, { action: 'register_anchor', runtime_generation: true }), + ).toMatchObject({ ok: false, code: 'INVALID_GENERATION' }); + const registered = await request(socket, { action: 'register_anchor', runtime_generation: 1 }); + expect( + await request(socket, { + action: 'authenticate', + session_id: registered.session_id, + runtime_generation: false, + }), + ).toMatchObject({ ok: false, code: 'INVALID_IDENTITY' }); + }); + + test.each([ + { compaction_epoch: true, request_epoch: 0, schema_version: 1 }, + { compaction_epoch: 0, request_epoch: -1, schema_version: 1 }, + { compaction_epoch: 0, request_epoch: 0, schema_version: false }, + { compaction_epoch: 0, request_epoch: 0, schema_version: -1 }, + { compaction_epoch: 0, request_epoch: 0, schema_version: 1, h_source: 'A'.repeat(64) }, + { compaction_epoch: 0, request_epoch: 0, schema_version: 1, h_payload: 'a'.repeat(63) }, + ])('invalid cycle binding fails closed without persisting a token (%j)', async (override) => { + const { socket, state } = await startBroker(); + const registered = await request(socket, { action: 'register_anchor', runtime_generation: 1 }); + const binding = Object.assign( + { + compaction_epoch: 0, + request_epoch: 0, + h_source: 'a'.repeat(64), + h_payload: 'b'.repeat(64), + schema_version: 1, + }, + override, + ); + expect( + await request(socket, { + action: 'mint_token', + session_id: registered.session_id, + runtime_generation: 1, + binding, + }), + ).toMatchObject({ ok: false, code: 'INVALID_BINDING' }); + const persisted = JSON.parse(await readFile(state, 'utf8')) as { tokens: object }; + expect(persisted.tokens).toEqual({}); + }); + + test('StateStore write-all unit path handles partial writes and cleans failed temp files', () => { + const result = spawnSync('python3', [join(import.meta.dirname, 'state_store_unittest.py')], { + encoding: 'utf8', + }); + expect(result.status, result.stderr).toBe(0); + }); + + test('persistence integrity failure refuses startup', async () => { + expect(await startBrokerWithState('{corrupt')).toContain('STATE_INTEGRITY'); + }); + + test.each([ + { version: 1, sessions: {}, tokens: {}, unexpected: true }, + { version: 1, sessions: { bad: {} }, tokens: {} }, + { + version: 1, + sessions: { + ['a'.repeat(64)]: { anchor_pid: true, anchor_starttime: '1', runtime_generation: 0 }, + }, + tokens: {}, + }, + { + version: 1, + sessions: { + ['a'.repeat(64)]: { anchor_pid: 1, anchor_starttime: '01', runtime_generation: 0 }, + }, + tokens: {}, + }, + { + version: 1, + sessions: { + ['a'.repeat(64)]: { anchor_pid: 1, anchor_starttime: '1', runtime_generation: 0 }, + ['b'.repeat(64)]: { anchor_pid: 1, anchor_starttime: '1', runtime_generation: 1 }, + }, + tokens: {}, + }, + { + version: 1, + sessions: { + ['a'.repeat(64)]: { anchor_pid: 1, anchor_starttime: '1', runtime_generation: 0 }, + }, + tokens: { + ['b'.repeat(64)]: { + session_id: 'c'.repeat(64), + runtime_generation: 0, + binding: { + compaction_epoch: 0, + request_epoch: 0, + h_source: 'd'.repeat(64), + h_payload: 'e'.repeat(64), + schema_version: 1, + }, + consumed: false, + }, + }, + }, + { + version: 1, + sessions: { + ['a'.repeat(64)]: { anchor_pid: 1, anchor_starttime: '1', runtime_generation: 1 }, + }, + tokens: { + ['b'.repeat(64)]: { + session_id: 'a'.repeat(64), + runtime_generation: 2, + binding: { + compaction_epoch: 0, + request_epoch: 0, + h_source: 'd'.repeat(64), + h_payload: 'e'.repeat(64), + schema_version: 1, + }, + consumed: false, + }, + }, + }, + ])('nested corrupt state refuses startup (%#)', async (stateValue) => { + expect(await startBrokerWithState(JSON.stringify(stateValue))).toContain('STATE_INTEGRITY'); + }); + + test('symlink state refuses startup', async () => { + const root = await mkdtemp(join(tmpdir(), 'mosaic-lease-broker-')); + await chmod(root, 0o700); + const target = join(root, 'target.json'); + const state = join(root, 'state.json'); + await writeFile(target, JSON.stringify({ version: 1, sessions: {}, tokens: {} }), { + mode: 0o600, + }); + await symlink(target, state); + const child = spawn('python3', [ + daemonPath, + '--socket', + join(root, 'broker.sock'), + '--state', + state, + ]); + children.push(child); + const stderr = await new Promise((resolve) => { + let raw = ''; + child.stderr?.on('data', (chunk: Buffer) => (raw += chunk.toString())); + child.once('exit', () => resolve(raw)); + }); + expect(stderr).toContain('STATE_INTEGRITY'); + }); + + test('oversized state refuses startup', async () => { + expect(await startBrokerWithState(' '.repeat(4 * 1024 * 1024 + 1))).toContain( + 'STATE_INTEGRITY', + ); + }); +}); diff --git a/packages/mosaic/src/lease-broker/state_store_unittest.py b/packages/mosaic/src/lease-broker/state_store_unittest.py new file mode 100644 index 0000000..bceb696 --- /dev/null +++ b/packages/mosaic/src/lease-broker/state_store_unittest.py @@ -0,0 +1,109 @@ +#!/usr/bin/env python3 +"""Standard-library edge tests for lease-broker atomic state persistence.""" + +from __future__ import annotations + +import importlib.util +import json +import os +import tempfile +import unittest +from pathlib import Path +from unittest.mock import patch + + +DAEMON_PATH = Path(__file__).parents[2] / "framework/tools/lease-broker/daemon.py" +SPEC = importlib.util.spec_from_file_location("lease_broker_daemon", DAEMON_PATH) +if SPEC is None or SPEC.loader is None: + raise RuntimeError("unable to load lease broker daemon") +DAEMON = importlib.util.module_from_spec(SPEC) +SPEC.loader.exec_module(DAEMON) + + +class StateStoreCommitTest(unittest.TestCase): + def make_store(self, root: Path): + os.chmod(root, 0o700) + store = DAEMON.StateStore(root / "state.json") + store.value["marker"] = "partial-write-proof" + return store + + def test_partial_writes_persist_the_complete_payload(self) -> None: + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + store = self.make_store(root) + real_write = os.write + + def partial_write(descriptor: int, payload: bytes) -> int: + return real_write(descriptor, payload[: max(1, len(payload) // 3)]) + + with patch.object(DAEMON.os, "write", side_effect=partial_write): + store.commit() + + self.assertEqual(json.loads(store.path.read_text()), store.value) + + def test_zero_progress_removes_owned_temporary_file(self) -> None: + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + store = self.make_store(root) + with patch.object(DAEMON.os, "write", return_value=0): + with self.assertRaises(OSError): + store.commit() + + self.assertFalse(store.path.exists()) + self.assertEqual(list(root.glob(".*.tmp")), []) + + +class BrokerBehaviorTest(unittest.TestCase): + def make_broker(self, root: Path): + os.chmod(root, 0o700) + return DAEMON.Broker(DAEMON.StateStore(root / "state.json")) + + def test_anchor_generation_bump_reuses_session_and_revokes_token(self) -> None: + binding = { + "compaction_epoch": 0, + "request_epoch": 0, + "h_source": "a" * 64, + "h_payload": "b" * 64, + "schema_version": 1, + } + with tempfile.TemporaryDirectory() as directory: + broker = self.make_broker(Path(directory)) + with ( + patch.object( + DAEMON, + "proc_node", + return_value={"pid": 123, "ppid": 1, "starttime": "456"}, + ), + patch.object(DAEMON, "verified_ancestry", return_value=True), + ): + first = broker.handle((123, 1000, 1000), { + "action": "register_anchor", + "runtime_generation": 1, + }) + minted = broker.handle((123, 1000, 1000), { + "action": "mint_token", + "session_id": first["session_id"], + "runtime_generation": 1, + "binding": binding, + }) + bumped = broker.handle((123, 1000, 1000), { + "action": "register_anchor", + "runtime_generation": 2, + }) + repeated = broker.handle((123, 1000, 1000), { + "action": "register_anchor", + "runtime_generation": 2, + }) + + self.assertEqual(bumped["session_id"], first["session_id"]) + self.assertEqual(repeated["session_id"], first["session_id"]) + self.assertTrue(broker.store.tokens()[minted["token"]]["consumed"]) + with self.assertRaisesRegex(DAEMON.BrokerFailure, "STALE_GENERATION"): + with patch.object(DAEMON, "proc_node", return_value={"starttime": "456"}): + broker.handle((123, 1000, 1000), { + "action": "register_anchor", + "runtime_generation": 1, + }) + +if __name__ == "__main__": + unittest.main()