chore: consolidate new foundation and archive v1 (#1495)

This commit is contained in:
2026-09-07 12:32:57 -05:00
3511 changed files with 727899 additions and 10 deletions
@@ -0,0 +1,237 @@
import { createServer, type Server } from 'node:net';
import { mkdir, mkdtemp, readFile, rm, stat, writeFile } from 'node:fs/promises';
import { tmpdir } from 'node:os';
import { join } from 'node:path';
import { afterEach, describe, expect, it } from 'vitest';
import {
applyBrokerSupervisor,
checkBrokerSupervisorHealth,
isBrokerSupervisorHealthy,
resolveBrokerSupervisorPaths,
resolveLeaseBrokerSocketPath,
type BrokerSupervisorPaths,
} from './broker-supervisor.js';
const REAL_FRAMEWORK_ROOT = new URL('../../framework/', import.meta.url).pathname;
const cleanupDirs: string[] = [];
const cleanupServers: Server[] = [];
async function tempDir(prefix: string): Promise<string> {
const dir = await mkdtemp(join(tmpdir(), prefix));
cleanupDirs.push(dir);
return dir;
}
afterEach(async () => {
for (const server of cleanupServers.splice(0)) {
await new Promise<void>((resolve) => server.close(() => resolve()));
}
for (const dir of cleanupDirs.splice(0)) {
await rm(dir, { recursive: true, force: true });
}
});
describe('resolveLeaseBrokerSocketPath', () => {
it('honors an explicit MOSAIC_LEASE_BROKER_SOCKET override', () => {
expect(resolveLeaseBrokerSocketPath({ MOSAIC_LEASE_BROKER_SOCKET: '/tmp/explicit.sock' })).toBe(
'/tmp/explicit.sock',
);
});
it('falls back to $XDG_RUNTIME_DIR/mosaic-lease/broker.sock', () => {
expect(resolveLeaseBrokerSocketPath({ XDG_RUNTIME_DIR: '/run/user/1000' })).toBe(
join('/run/user/1000', 'mosaic-lease', 'broker.sock'),
);
});
it('falls back to /run/user/<uid>/mosaic-lease/broker.sock as a last resort', () => {
expect(resolveLeaseBrokerSocketPath({}, 4242)).toBe(
join('/run/user', '4242', 'mosaic-lease', 'broker.sock'),
);
});
it('prefers the explicit override over XDG_RUNTIME_DIR', () => {
expect(
resolveLeaseBrokerSocketPath({
MOSAIC_LEASE_BROKER_SOCKET: '/explicit.sock',
XDG_RUNTIME_DIR: '/run/user/1000',
}),
).toBe('/explicit.sock');
});
});
describe('resolveBrokerSupervisorPaths', () => {
it('colocates the state file next to the resolved socket', () => {
const paths = resolveBrokerSupervisorPaths({
mosaicHome: '/home/x/.config/mosaic',
frameworkRoot: '/repo/framework',
env: { XDG_RUNTIME_DIR: '/run/user/1000' },
});
expect(paths.socketPath).toBe(join('/run/user/1000', 'mosaic-lease', 'broker.sock'));
expect(paths.statePath).toBe(join('/run/user/1000', 'mosaic-lease', 'state.json'));
});
it('targets the systemd --user dir under the given home, not mosaicHome', () => {
const paths = resolveBrokerSupervisorPaths({
mosaicHome: '/somewhere-else/.config/mosaic',
frameworkRoot: '/repo/framework',
homeDir: '/home/canary',
env: {},
uid: 0,
});
expect(paths.systemdUserDir).toBe(join('/home/canary', '.config', 'systemd', 'user'));
expect(paths.unitTargetPath).toBe(
join('/home/canary', '.config', 'systemd', 'user', 'mosaic-lease-broker.service'),
);
});
it('is a pure function: identical options resolve to identical paths', () => {
const options = {
mosaicHome: '/h/.config/mosaic',
frameworkRoot: '/repo/framework',
env: { XDG_RUNTIME_DIR: '/run/user/1000' },
};
expect(resolveBrokerSupervisorPaths(options)).toEqual(resolveBrokerSupervisorPaths(options));
});
});
describe('applyBrokerSupervisor', () => {
async function fakePaths(): Promise<BrokerSupervisorPaths> {
const home = await tempDir('mosaic-broker-supervisor-home-');
const mosaicHome = join(home, '.config', 'mosaic');
const runtimeDir = await tempDir('mosaic-broker-supervisor-runtime-');
return resolveBrokerSupervisorPaths({
mosaicHome,
frameworkRoot: REAL_FRAMEWORK_ROOT,
homeDir: home,
env: { XDG_RUNTIME_DIR: runtimeDir },
});
}
it('renders a unit that references the installed wrapper script and hardens the runtime dir', async () => {
const paths = await fakePaths();
const unitSource = await readFile(paths.unitSourcePath, 'utf8');
expect(unitSource).toContain('ExecStart=');
expect(unitSource).toContain('%h/.config/mosaic/tools/lease-broker/start-lease-broker.sh');
expect(unitSource).toContain('RuntimeDirectory=mosaic-lease');
expect(unitSource).toContain('RuntimeDirectoryMode=0700');
expect(unitSource).toContain('Restart=on-failure');
expect(unitSource).toContain('WantedBy=default.target');
// No ambient environment file preload, matching the other fleet units'
// strict-parsing convention.
expect(unitSource).not.toMatch(/^Environment(File)?=/m);
});
it('materializes the unit, wrapper script, and daemon sources on first apply', async () => {
const paths = await fakePaths();
const result = await applyBrokerSupervisor(paths);
expect(result.installedFiles).toContain(paths.unitTargetPath);
expect(result.installedFiles).toContain(paths.wrapperTargetPath);
for (const target of paths.daemonTargetPaths) {
expect(result.installedFiles).toContain(target);
}
const unitTargetContent = await readFile(paths.unitTargetPath, 'utf8');
const unitSourceContent = await readFile(paths.unitSourcePath, 'utf8');
expect(unitTargetContent).toBe(unitSourceContent);
const wrapperMode = (await stat(paths.wrapperTargetPath)).mode & 0o777;
expect(wrapperMode).toBe(0o755);
for (const target of paths.daemonTargetPaths) {
await expect(stat(target)).resolves.toBeDefined();
}
});
it('is idempotent: applying twice reproduces identical files with no error', async () => {
const paths = await fakePaths();
await applyBrokerSupervisor(paths);
const firstUnit = await readFile(paths.unitTargetPath, 'utf8');
const firstWrapper = await readFile(paths.wrapperTargetPath, 'utf8');
const firstWrapperMode = (await stat(paths.wrapperTargetPath)).mode & 0o777;
await expect(applyBrokerSupervisor(paths)).resolves.toBeDefined();
const secondUnit = await readFile(paths.unitTargetPath, 'utf8');
const secondWrapper = await readFile(paths.wrapperTargetPath, 'utf8');
const secondWrapperMode = (await stat(paths.wrapperTargetPath)).mode & 0o777;
expect(secondUnit).toBe(firstUnit);
expect(secondWrapper).toBe(firstWrapper);
expect(secondWrapperMode).toBe(firstWrapperMode);
});
it('never touches the real host: only writes under the supplied temp dirs', async () => {
const paths = await fakePaths();
await applyBrokerSupervisor(paths);
expect(paths.systemdUserDir.startsWith(tmpdir())).toBe(true);
expect(paths.mosaicHome.startsWith(tmpdir())).toBe(true);
});
});
describe('checkBrokerSupervisorHealth / isBrokerSupervisorHealthy', () => {
async function fakeHealthPaths(): Promise<
Pick<BrokerSupervisorPaths, 'unitTargetPath' | 'socketPath'>
> {
const runtimeDir = await tempDir('mosaic-broker-supervisor-health-');
await mkdir(join(runtimeDir, 'systemd-user'), { recursive: true });
return {
unitTargetPath: join(runtimeDir, 'systemd-user', 'mosaic-lease-broker.service'),
socketPath: join(runtimeDir, 'broker.sock'),
};
}
it('reports unhealthy when neither the unit nor the socket exist', async () => {
const paths = await fakeHealthPaths();
const health = await checkBrokerSupervisorHealth(paths);
expect(health).toEqual({ unitInstalled: false, socketPresent: false, healthy: false });
expect(await isBrokerSupervisorHealthy(paths)).toBe(false);
});
it('reports unhealthy when the unit is installed but no socket is listening', async () => {
const paths = await fakeHealthPaths();
await writeFile(paths.unitTargetPath, '[Unit]\n');
const health = await checkBrokerSupervisorHealth(paths);
expect(health.unitInstalled).toBe(true);
expect(health.socketPresent).toBe(false);
expect(health.healthy).toBe(false);
});
it('reports healthy=true once a real Unix socket exists at the resolved path, and false again once removed', async () => {
const paths = await fakeHealthPaths();
const server = createServer();
cleanupServers.push(server);
await new Promise<void>((resolve, reject) => {
server.once('error', reject);
server.listen(paths.socketPath, resolve);
});
expect(await isBrokerSupervisorHealthy(paths)).toBe(true);
const health = await checkBrokerSupervisorHealth(paths);
expect(health.socketPresent).toBe(true);
expect(health.healthy).toBe(true);
await new Promise<void>((resolve) => server.close(() => resolve()));
await rm(paths.socketPath, { force: true });
expect(await isBrokerSupervisorHealthy(paths)).toBe(false);
});
it('does not confuse a stale regular file at the socket path with a live socket', async () => {
const paths = await fakeHealthPaths();
await writeFile(paths.socketPath, 'not actually a socket');
expect(await isBrokerSupervisorHealthy(paths)).toBe(false);
});
});
@@ -0,0 +1,223 @@
/**
* Activation-side supervisor for the Mosaic lease broker (issue #869, Point-1
* C3). #828 shipped fail-closed enforcement hooks (`mutator-gate.py`,
* `receipt-observer-client.py`) with nothing that guaranteed `daemon.py` was
* running or that its socket existed before a gated runtime started. This
* module:
*
* - resolves the broker socket/state paths and the on-disk locations of the
* supervisor artifacts, deterministically and consistently with
* `defaultLeaseBrokerSocket` in `../commands/launch.ts`;
* - idempotently applies (materializes) a systemd `--user` unit plus the
* wrapper script and daemon sources it execs, mirroring the tmux fleet
* unit convention in `framework/systemd/user/`;
* - exposes a health predicate other cards (e.g. the C1 activation probe)
* can call to learn whether a broker supervisor is present and healthy.
*
* `applyBrokerSupervisor` only writes files under the paths it is given. It
* never runs `systemctl`, never starts `daemon.py`, and never touches a real
* host's `~/.config` unless the caller explicitly resolves paths there.
* Enabling/starting the unit is a separate, later, out-of-scope step.
*/
import { chmod, copyFile, mkdir, stat } from 'node:fs/promises';
import { homedir } from 'node:os';
import { dirname, join } from 'node:path';
const UNIT_NAME = 'mosaic-lease-broker.service';
const WRAPPER_SCRIPT_NAME = 'start-lease-broker.sh';
/** Co-located modules `daemon.py` imports at runtime; kept alongside it. */
const DAEMON_SOURCE_FILE_NAMES = [
'daemon.py',
'lease_generation.py',
'normative_fragments.py',
'receipt_challenge.py',
'receipt_observer.py',
] as const;
export interface ResolveBrokerSupervisorPathsOptions {
/** `~/.config/mosaic` (or an override) — where installed tool copies live. */
mosaicHome: string;
/** Root of the checked-out `framework/` directory (canonical file source). */
frameworkRoot: string;
/** Defaults to `process.env`; pass a fake for tests. */
env?: NodeJS.ProcessEnv;
/** Defaults to `os.homedir()`; pass a temp dir in tests. */
homeDir?: string;
/** Defaults to `process.getuid()` (or 0); pass a fake for tests. */
uid?: number;
}
export interface BrokerSupervisorPaths {
readonly mosaicHome: string;
readonly frameworkRoot: string;
readonly systemdUserDir: string;
readonly leaseBrokerToolsDir: string;
readonly unitSourcePath: string;
readonly unitTargetPath: string;
readonly wrapperSourcePath: string;
readonly wrapperTargetPath: string;
readonly daemonSourcePaths: readonly string[];
readonly daemonTargetPaths: readonly string[];
/**
* Resolved with the same precedence as `defaultLeaseBrokerSocket` in
* `../commands/launch.ts`: an explicit `MOSAIC_LEASE_BROKER_SOCKET`, else
* `$XDG_RUNTIME_DIR/mosaic-lease/broker.sock`, else
* `/run/user/<uid>/mosaic-lease/broker.sock`.
*/
readonly socketPath: string;
/** Colocated next to the socket, matching the broker's own generation-file convention. */
readonly statePath: string;
}
/**
* Resolve the lease broker socket path alone, with the same precedence as
* `defaultLeaseBrokerSocket` in `../commands/launch.ts`. Exported so callers
* (and tests) can assert the two stay in agreement without importing the CLI
* command module.
*/
export function resolveLeaseBrokerSocketPath(
env: NodeJS.ProcessEnv = process.env,
uid: number = typeof process.getuid === 'function' ? process.getuid() : 0,
): string {
const explicit = env['MOSAIC_LEASE_BROKER_SOCKET'];
if (explicit) return explicit;
const runtimeDir = env['XDG_RUNTIME_DIR'];
if (runtimeDir) return join(runtimeDir, 'mosaic-lease', 'broker.sock');
return join('/run/user', String(uid), 'mosaic-lease', 'broker.sock');
}
/** Resolve every path the supervisor apply/health functions need, deterministically. */
export function resolveBrokerSupervisorPaths(
options: ResolveBrokerSupervisorPathsOptions,
): BrokerSupervisorPaths {
const { mosaicHome, frameworkRoot } = options;
const env = options.env ?? process.env;
const homeDir = options.homeDir ?? homedir();
const systemdUserDir = join(homeDir, '.config', 'systemd', 'user');
const leaseBrokerToolsDir = join(mosaicHome, 'tools', 'lease-broker');
const frameworkLeaseBrokerDir = join(frameworkRoot, 'tools', 'lease-broker');
const socketPath = resolveLeaseBrokerSocketPath(env, options.uid);
const statePath = join(dirname(socketPath), 'state.json');
return {
mosaicHome,
frameworkRoot,
systemdUserDir,
leaseBrokerToolsDir,
unitSourcePath: join(frameworkRoot, 'systemd', 'user', UNIT_NAME),
unitTargetPath: join(systemdUserDir, UNIT_NAME),
wrapperSourcePath: join(frameworkLeaseBrokerDir, WRAPPER_SCRIPT_NAME),
wrapperTargetPath: join(leaseBrokerToolsDir, WRAPPER_SCRIPT_NAME),
daemonSourcePaths: DAEMON_SOURCE_FILE_NAMES.map((name) => join(frameworkLeaseBrokerDir, name)),
daemonTargetPaths: DAEMON_SOURCE_FILE_NAMES.map((name) => join(leaseBrokerToolsDir, name)),
socketPath,
statePath,
};
}
export interface ApplyBrokerSupervisorResult {
readonly installedFiles: readonly string[];
}
/**
* Idempotently materialize the supervisor unit, its wrapper script, and the
* daemon sources it execs. Safe to call on every reseed: every write is a
* deterministic overwrite of the same target path from the same source, so a
* second call reproduces identical bytes/modes and never errors.
*
* Never runs `systemctl`; the caller decides separately whether/when to
* `daemon-reload`/`enable`/`start` the installed unit.
*/
export async function applyBrokerSupervisor(
paths: BrokerSupervisorPaths,
): Promise<ApplyBrokerSupervisorResult> {
await mkdir(paths.leaseBrokerToolsDir, { recursive: true });
await mkdir(paths.systemdUserDir, { recursive: true });
const installedFiles: string[] = [];
for (let index = 0; index < paths.daemonSourcePaths.length; index += 1) {
const source = paths.daemonSourcePaths[index];
const target = paths.daemonTargetPaths[index];
if (source === undefined || target === undefined) continue;
await copyFile(source, target);
await chmod(target, 0o644);
installedFiles.push(target);
}
await copyFile(paths.wrapperSourcePath, paths.wrapperTargetPath);
await chmod(paths.wrapperTargetPath, 0o755);
installedFiles.push(paths.wrapperTargetPath);
await copyFile(paths.unitSourcePath, paths.unitTargetPath);
await chmod(paths.unitTargetPath, 0o644);
installedFiles.push(paths.unitTargetPath);
return { installedFiles };
}
export interface BrokerSupervisorHealth {
/** Whether the systemd unit file has been materialized at its target path. */
readonly unitInstalled: boolean;
/** Whether a Unix domain socket currently exists at the resolved socket path. */
readonly socketPresent: boolean;
/**
* The signal other cards (e.g. C1's activation probe) should treat as
* "a broker supervisor is present and healthy". Presence of a live socket
* is the authoritative signal: a gated runtime can only ever succeed by
* connecting to it, so this is what fail-closed callers must check.
*/
readonly healthy: boolean;
}
/**
* Report the supervisor's on-disk/health signals. Never throws for an
* absent unit or socket — both simply report `false`; unexpected filesystem
* errors (permission issues, etc.) still propagate.
*/
export async function checkBrokerSupervisorHealth(
paths: Pick<BrokerSupervisorPaths, 'unitTargetPath' | 'socketPath'>,
): Promise<BrokerSupervisorHealth> {
const [unitInstalled, socketPresent] = await Promise.all([
pathExists(paths.unitTargetPath),
isUnixSocket(paths.socketPath),
]);
return { unitInstalled, socketPresent, healthy: socketPresent };
}
/** Convenience boolean form of {@link checkBrokerSupervisorHealth} for simple call sites. */
export async function isBrokerSupervisorHealthy(
paths: Pick<BrokerSupervisorPaths, 'unitTargetPath' | 'socketPath'>,
): Promise<boolean> {
return (await checkBrokerSupervisorHealth(paths)).healthy;
}
async function pathExists(path: string): Promise<boolean> {
try {
await stat(path);
return true;
} catch (error) {
if (isEnoent(error)) return false;
throw error;
}
}
async function isUnixSocket(path: string): Promise<boolean> {
try {
const info = await stat(path);
return info.isSocket();
} catch (error) {
if (isEnoent(error)) return false;
throw error;
}
}
function isEnoent(error: unknown): boolean {
return (
typeof error === 'object' &&
error !== null &&
'code' in error &&
(error as NodeJS.ErrnoException).code === 'ENOENT'
);
}
@@ -0,0 +1,181 @@
import { mkdtemp, rm } from 'node:fs/promises';
import { createServer, type Server, type Socket } from 'node:net';
import { tmpdir } from 'node:os';
import { join } from 'node:path';
import { afterEach, describe, expect, test } from 'vitest';
import { BrokerTransportError, readBrokerReply, requestBrokerReply } from './broker-test-client.js';
const roots: string[] = [];
const servers: Server[] = [];
const sockets: Socket[] = [];
async function scriptedBroker(
replies: ReadonlyArray<ReadonlyArray<Buffer> | 'hang'>,
): Promise<{ socketPath: string; connections: () => number }> {
const root = await mkdtemp(join(tmpdir(), 'mosaic-broker-client-'));
roots.push(root);
const socketPath = join(root, 'broker.sock');
let connections = 0;
const server = createServer({ allowHalfOpen: true }, (socket) => {
sockets.push(socket);
const chunks = replies[connections] ?? replies.at(-1) ?? [];
connections += 1;
socket.once('end', () => {
if (chunks === 'hang') return;
void (async () => {
for (const chunk of chunks) {
socket.write(chunk);
await new Promise<void>((resolve) => setTimeout(resolve, 5));
}
socket.end();
})();
});
socket.resume();
});
servers.push(server);
await new Promise<void>((resolve, reject) => {
server.once('error', reject);
server.listen(socketPath, resolve);
});
return { socketPath, connections: () => connections };
}
afterEach(async () => {
for (const socket of sockets.splice(0)) socket.destroy();
await Promise.all(
servers
.splice(0)
.map(
(server) =>
new Promise<void>((resolve, reject) =>
server.close((error) => (error ? reject(error) : resolve())),
),
),
);
await Promise.all(roots.splice(0).map((root) => rm(root, { recursive: true, force: true })));
});
describe('newline-framed broker test client', () => {
test('rejects an empty early close without retrying into a false green', async () => {
const broker = await scriptedBroker([[], [Buffer.from('{"ok":true}\n')]]);
const failure = await requestBrokerReply(broker.socketPath, { action: 'probe' }).catch(
(error: unknown) => error,
);
expect(failure).toBeInstanceOf(BrokerTransportError);
expect(failure).toMatchObject({
kind: 'early-close',
attempts: 1,
responseLength: 0,
responseHex: '',
});
expect(failure).toHaveProperty(
'message',
expect.stringMatching(/closed before newline.*length=0.*hex=<empty>/i),
);
expect(broker.connections()).toBe(1);
});
test('rejects repeated truncated early closes with response bytes and length', async () => {
const truncated = Buffer.from('{"ok":');
const broker = await scriptedBroker([[truncated], [Buffer.from('{"ok":true}\n')]]);
const failure = await requestBrokerReply(broker.socketPath, { action: 'probe' }).catch(
(error: unknown) => error,
);
expect(failure).toBeInstanceOf(BrokerTransportError);
expect(failure).toMatchObject({
kind: 'early-close',
attempts: 1,
responseLength: 6,
responseHex: '7b226f6b223a',
});
expect(failure).toHaveProperty(
'message',
expect.stringMatching(/closed before newline.*length=6.*bytes=.*ok/i),
);
expect(broker.connections()).toBe(1);
});
test('rejects a newline-terminated malformed broker reply without retrying', async () => {
const broker = await scriptedBroker([[Buffer.from('{bad}\n')]]);
await expect(requestBrokerReply(broker.socketPath, { action: 'probe' })).rejects.toThrow(
/malformed broker reply.*length=6.*bytes="\{bad\}\\n"/i,
);
expect(broker.connections()).toBe(1);
});
test('rejects trailing bytes delivered after a complete frame in a later data event', async () => {
const broker = await scriptedBroker([[Buffer.from('{"ok":true}\n'), Buffer.from('extra')]]);
const failure = await requestBrokerReply(broker.socketPath, { action: 'probe' }).catch(
(error: unknown) => error,
);
expect(failure).toBeInstanceOf(BrokerTransportError);
expect(failure).toMatchObject({ kind: 'malformed-reply', responseLength: 17 });
expect(failure).toHaveProperty(
'message',
expect.stringMatching(/bytes after the newline terminator/i),
);
});
test('redacts security tokens from typed transport diagnostics', async () => {
const token = 'a'.repeat(64);
const reply = Buffer.from(`{"ok":true,"promotion_token":"${token}`);
const broker = await scriptedBroker([[reply]]);
const failure = await requestBrokerReply(broker.socketPath, { action: 'probe' }).catch(
(error: unknown) => error,
);
expect(failure).toBeInstanceOf(BrokerTransportError);
expect(failure).toMatchObject({
kind: 'early-close',
responseLength: reply.length,
responsePreview: '<redacted-sensitive-reply>',
});
expect(String((failure as Error).message)).not.toContain(token);
expect(JSON.stringify(failure)).not.toContain(token);
});
test.each([
['trailing bytes', Buffer.from('{"ok":true}\nextra'), /bytes after the newline/i],
['non-object JSON', Buffer.from('[]\n'), /JSON value is not an object/i],
['oversized frame', Buffer.alloc(64 * 1024 + 1, 0x78), /exceeds 65536 bytes/i],
])('rejects %s with deterministic framing context', async (_label, reply, message) => {
const broker = await scriptedBroker([[reply as Buffer]]);
await expect(requestBrokerReply(broker.socketPath, { action: 'probe' })).rejects.toThrow(
message as RegExp,
);
expect(broker.connections()).toBe(1);
});
test('reports timeout, connection, and writer failures as promise rejections', async () => {
const hanging = await scriptedBroker(['hang']);
await expect(
requestBrokerReply(hanging.socketPath, { action: 'probe' }, { timeoutMs: 10 }),
).rejects.toThrow(/timed out before newline.*length=0.*hex=<empty>/i);
await expect(
requestBrokerReply(join(rootForMissingSocket(), 'missing.sock'), {}),
).rejects.toThrow(/socket error before newline.*length=0/i);
const writerFailure = await scriptedBroker(['hang']);
await expect(
readBrokerReply(writerFailure.socketPath, () => {
throw new Error('writer failed');
}),
).rejects.toThrow('writer failed');
});
});
function rootForMissingSocket(): string {
return join(tmpdir(), `mosaic-missing-broker-${process.pid}-${Date.now()}`);
}
@@ -0,0 +1,236 @@
import { createHash } from 'node:crypto';
import { chmod, writeFile } from 'node:fs/promises';
import { createConnection, type Socket } from 'node:net';
const DEFAULT_TIMEOUT_MS = 3_000;
const MAX_REPLY_BYTES = 64 * 1024;
const MAX_DIAGNOSTIC_BYTES = 256;
const SENSITIVE_REPLY_FIELD = /"(?:promotion_token|session_id|token)"\s*:/;
export interface BrokerTestClientOptions {
timeoutMs?: number;
}
export type BrokerTransportFailureKind =
| 'early-close'
| 'malformed-reply'
| 'socket-error'
| 'timeout';
export class BrokerTransportError extends Error {
public readonly responseLength: number;
public readonly responseBytes: string;
public readonly responseHex: string;
public readonly responsePreview: string;
public readonly responseSha256: string;
public constructor(
public readonly kind: BrokerTransportFailureKind,
description: string,
response: Buffer,
public readonly attempts = 1,
) {
const diagnostics = responseDiagnostics(response);
super(`${description}; attempts=${attempts}; ${responseContext(response, diagnostics)}`);
this.name = 'BrokerTransportError';
this.responseLength = response.length;
this.responseBytes = diagnostics.preview;
this.responseHex = diagnostics.hex;
this.responsePreview = diagnostics.preview;
this.responseSha256 = diagnostics.sha256;
}
}
interface ResponseDiagnostics {
preview: string;
hex: string;
sha256: string;
}
function responseDiagnostics(response: Buffer): ResponseDiagnostics {
const fullText = response.toString('utf8');
const sensitive = SENSITIVE_REPLY_FIELD.test(fullText);
const bounded = response.subarray(0, MAX_DIAGNOSTIC_BYTES);
const suffix = response.length > MAX_DIAGNOSTIC_BYTES ? '…' : '';
return {
preview:
response.length === 0
? '<empty>'
: sensitive
? '<redacted-sensitive-reply>'
: `${bounded.toString('utf8')}${suffix}`,
hex: sensitive ? '<redacted>' : `${bounded.toString('hex')}${suffix}`,
sha256: createHash('sha256').update(response).digest('hex'),
};
}
function responseContext(response: Buffer, diagnostics = responseDiagnostics(response)): string {
const hex = diagnostics.hex.length === 0 ? '<empty>' : diagnostics.hex;
return `length=${response.length}; bytes=${JSON.stringify(diagnostics.preview)}; hex=${hex}; sha256=${diagnostics.sha256}`;
}
function malformedReply(response: Buffer, reason: string): BrokerTransportError {
return new BrokerTransportError('malformed-reply', `Malformed broker reply: ${reason}`, response);
}
function readBrokerReplyAttempt<T extends object>(
socketPath: string,
write: (socket: Socket) => void,
timeoutMs: number,
): Promise<T> {
return new Promise<T>((resolve, reject) => {
const socket = createConnection(socketPath);
let response = Buffer.alloc(0);
let settled = false;
const settle = (callback: () => void): void => {
if (settled) return;
settled = true;
clearTimeout(timer);
callback();
socket.destroy();
};
const fail = (error: Error): void => settle(() => reject(error));
const timer = setTimeout(
() =>
fail(
new BrokerTransportError(
'timeout',
`Broker reply timed out before newline after ${timeoutMs}ms`,
response,
),
),
timeoutMs,
);
socket.once('error', (error) =>
fail(
new BrokerTransportError(
'socket-error',
`Broker reply socket error before newline: ${error.message}`,
response,
),
),
);
socket.on('data', (chunk: Buffer) => {
if (settled) return;
response = Buffer.concat([response, chunk]);
if (response.length > MAX_REPLY_BYTES) {
fail(malformedReply(response, `exceeds ${MAX_REPLY_BYTES} bytes`));
}
});
socket.once('end', () => {
if (settled) return;
const newline = response.indexOf(0x0a);
if (response.length === 0 || newline < 0) {
fail(
new BrokerTransportError(
'early-close',
'Broker reply socket closed before newline',
response,
),
);
return;
}
if (newline !== response.length - 1) {
fail(malformedReply(response, 'contains bytes after the newline terminator'));
return;
}
try {
const parsed: unknown = JSON.parse(response.subarray(0, newline).toString('utf8'));
if (typeof parsed !== 'object' || parsed === null || Array.isArray(parsed)) {
fail(malformedReply(response, 'JSON value is not an object'));
return;
}
settle(() => resolve(parsed as T));
} catch (error: unknown) {
fail(
malformedReply(response, error instanceof Error ? error.message : 'JSON parsing failed'),
);
}
});
socket.once('connect', () => {
try {
write(socket);
} catch (error: unknown) {
fail(error instanceof Error ? error : new Error(String(error)));
}
});
});
}
/** Read exactly one complete newline-framed JSON object; transport failures reject. */
export async function readBrokerReply<T extends object>(
socketPath: string,
write: (socket: Socket) => void,
options: BrokerTestClientOptions = {},
): Promise<T> {
return await readBrokerReplyAttempt<T>(
socketPath,
write,
options.timeoutMs ?? DEFAULT_TIMEOUT_MS,
);
}
/** Send one newline-framed request and read its complete broker reply. */
export async function requestBrokerReply<T extends object>(
socketPath: string,
requestValue: object,
options?: BrokerTestClientOptions,
): Promise<T> {
return await readBrokerReply<T>(
socketPath,
(socket) => socket.end(`${JSON.stringify(requestValue)}\n`),
options,
);
}
export interface ReceiptChallengeCycle {
sessionId: string;
runtimeGeneration: number;
receiptChallenge: string;
receipt: string;
}
export interface ReceiptChallengeReply {
ok: boolean;
code?: string;
state?: 'UNVERIFIED' | 'PENDING_VERIFICATION' | 'PENDING_PROMOTION' | 'VERIFIED';
}
/**
* Complete the shipped begin -> trusted-observer -> consume -> promote path.
* The private fixture is read by the daemon's injected test observer; the
* observation request itself never carries assistant-message content.
*/
export async function observeAndPromoteReceiptChallenge(
socketPath: string,
observerFixturePath: string,
cycle: ReceiptChallengeCycle,
): Promise<ReceiptChallengeReply> {
await writeFile(
observerFixturePath,
`${JSON.stringify({
session_id: cycle.sessionId,
runtime_generation: cycle.runtimeGeneration,
latest_assistant_message: cycle.receipt,
})}\n`,
{ encoding: 'utf8', mode: 0o600 },
);
await chmod(observerFixturePath, 0o600);
const observed = await requestBrokerReply<ReceiptChallengeReply>(socketPath, {
action: 'observe_receipt',
session_id: cycle.sessionId,
runtime_generation: cycle.runtimeGeneration,
receipt_challenge: cycle.receiptChallenge,
});
if (observed.ok !== true || observed.state !== 'PENDING_PROMOTION') return observed;
return await requestBrokerReply<ReceiptChallengeReply>(socketPath, {
action: 'promote_lease',
session_id: cycle.sessionId,
runtime_generation: cycle.runtimeGeneration,
receipt_challenge: cycle.receiptChallenge,
});
}
@@ -0,0 +1,171 @@
#!/usr/bin/env python3
"""RED-first WI-6 contracts against the shipped constrained recovery entrypoint."""
from __future__ import annotations
import base64
import hashlib
import importlib.util
import os
import tempfile
import unittest
from pathlib import Path
TOOLS = Path(__file__).parents[2] / "framework/tools/lease-broker"
DAEMON_PATH = TOOLS / "daemon.py"
FRAGMENTS_PATH = TOOLS / "normative_fragments.py"
OBSERVER_PATH = TOOLS / "receipt_observer.py"
def load_module(name: str, path: Path):
assert path.is_file(), f"shipped module is missing: {path}"
spec = importlib.util.spec_from_file_location(name, path)
if spec is None or spec.loader is None:
raise RuntimeError(f"unable to load {name}")
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
return module
DAEMON = load_module("lease_broker_recovery_daemon", DAEMON_PATH)
FRAGMENTS = load_module("lease_broker_recovery_fragments", FRAGMENTS_PATH)
OBSERVER = load_module("lease_broker_recovery_observer", OBSERVER_PATH)
class RecoveryFixture(unittest.TestCase):
def setUp(self) -> None:
self.temporary = tempfile.TemporaryDirectory()
root = Path(self.temporary.name)
os.chmod(root, 0o700)
self.peer = (os.getpid(), os.getuid(), os.getgid())
self.observer = OBSERVER.TestReceiptObserver()
self.broker = DAEMON.Broker(DAEMON.StateStore(root / "state.json"), observer=self.observer)
registered = self.broker.handle(self.peer, {
"action": "register_anchor",
"runtime_generation": 11,
})
self.session_id = registered["session_id"]
def tearDown(self) -> None:
self.temporary.cleanup()
def construction_and_binding(self) -> tuple[dict[str, object], dict[str, object]]:
content = b"Mosaic recovery authority\n"
construction = {
"manifest_version": 1,
"generator_version": "wi6-recovery-test",
"fragments": [{
"source_id": "authority/recovery",
"content_base64": base64.b64encode(content).decode("ascii"),
"expected_sha256": hashlib.sha256(content).hexdigest(),
}],
}
built = FRAGMENTS.build_payload_from_wire(construction)
self.assertEqual(built.injectionDecision, "ACCEPTED")
self.assertTrue(built.promotion)
return construction, {
"compaction_epoch": 17,
"request_epoch": 23,
"h_source": built.h_source,
"h_payload": built.h_payload,
"schema_version": 1,
}
def begin_normal(self) -> dict[str, object]:
construction, binding = self.construction_and_binding()
return self.broker.handle(self.peer, {
"action": "begin_verification",
"session_id": self.session_id,
"runtime_generation": 11,
"runtime": "pi",
"binding": binding,
"construction": construction,
})
def begin_recovery(self) -> dict[str, object]:
construction, binding = self.construction_and_binding()
return self.broker.handle(self.peer, {
"action": "begin_recovery",
"session_id": self.session_id,
"runtime_generation": 11,
"runtime": "pi",
"binding": binding,
"construction": construction,
})
def complete_recovery(self) -> dict[str, object]:
return self.broker.handle(self.peer, {
"action": "complete_recovery",
"session_id": self.session_id,
"runtime_generation": 11,
})
def assert_recovery_refused_unverified(self, code: str) -> None:
with self.assertRaisesRegex(DAEMON.BrokerFailure, code):
self.complete_recovery()
self.assertEqual(self.broker.leases[self.session_id]["state"], DAEMON.LEASE_UNVERIFIED)
denied = self.broker.handle(self.peer, {
"action": "authorize_tool",
"session_id": self.session_id,
"runtime_generation": 11,
"runtime": "pi",
"tool_name": "bash",
})
self.assertEqual(denied["decision"], "deny")
class ConstrainedRecoveryContractTest(RecoveryFixture):
def test_recovery_mints_a_fresh_challenge_distinct_from_normal_path(self) -> None:
normal = self.begin_normal()
recovery = self.begin_recovery()
self.assertEqual(recovery["state"], "PENDING_DELIVERY")
self.assertNotEqual(normal["receipt_challenge"], recovery["receipt_challenge"])
self.assertIn(recovery["receipt_challenge"], recovery["receipt"])
def test_c4_normal_path_receipt_cannot_be_replayed_through_recovery(self) -> None:
normal = self.begin_normal()
recovery = self.begin_recovery()
self.assertNotEqual(normal["receipt_challenge"], recovery["receipt_challenge"])
self.observer.record_latest_assistant_message(self.session_id, 11, normal["receipt"])
self.assert_recovery_refused_unverified("RECEIPT_MISMATCH")
def test_t27_observable_partial_delivery_variants_never_promote(self) -> None:
variants = {
"absent": None,
"malformed": "MOSAIC-RECEIPT{malformed}",
"prefix-truncated": None,
"observable-adapter-mutation": None,
# Tail-only is represented only by this concrete malformed/incomplete
# delivery. It is not a category-wide tail-only detection claim.
"tail-only-malformed": "H_payload=tail-only",
}
for name, observed in variants.items():
with self.subTest(name=name):
recovery = self.begin_recovery()
if name == "prefix-truncated":
observed = recovery["receipt"][:-1]
elif name == "observable-adapter-mutation":
observed = recovery["receipt"].replace("H_payload=", "H_payload=0", 1)
if observed is not None:
self.observer.record_latest_assistant_message(self.session_id, 11, observed)
self.assert_recovery_refused_unverified(
"RECEIPT_OBSERVATION_UNAVAILABLE" if observed is None else "RECEIPT_MISMATCH"
)
def test_negative_capability_tail_preserving_middle_drop_is_not_receipt_detectable(self) -> None:
recovery = self.begin_recovery()
# The observer seam receives the exact terminal message, not the delivered
# payload bytes. A middle drop that preserves this tail is therefore T-C
# and deliberately NOT represented as receipt-detectable; WI-7 server-side
# evidence owns that residual. This is not an assertion that it is caught.
self.observer.record_latest_assistant_message(self.session_id, 11, recovery["receipt"])
promoted = self.complete_recovery()
self.assertEqual(promoted["state"], DAEMON.LEASE_VERIFIED)
if __name__ == "__main__":
unittest.main()
@@ -0,0 +1,103 @@
#!/usr/bin/env python3
"""Regression tests for bounded lease-broker read/handle/send deadlines."""
from __future__ import annotations
import importlib.util
import json
import socket
import threading
import time
import unittest
from pathlib import Path
DAEMON_PATH = Path(__file__).parents[2] / "framework/tools/lease-broker/daemon.py"
SPEC = importlib.util.spec_from_file_location("lease_broker_deadline_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)
def original_connection_budget() -> float:
read_budget = getattr(DAEMON, "READ_DEADLINE_SECONDS", None)
if isinstance(read_budget, (int, float)):
return float(read_budget)
return float(DAEMON.CONNECTION_DEADLINE_SECONDS)
class SlowBroker:
def __init__(self, delay: float) -> None:
self.delay = delay
self.calls = 0
def handle(self, _peer: tuple[int, int, int], request: dict[str, object]) -> dict[str, object]:
self.calls += 1
time.sleep(self.delay)
return {"ok": True, "echo": request.get("action")}
class BrokerDeadlineTest(unittest.TestCase):
def test_lock_queue_timeout_returns_explicit_fail_closed_reply_without_handling(self) -> None:
broker = SlowBroker(0)
broker_lock = threading.Lock()
broker_lock.acquire()
server, client = socket.socketpair(socket.AF_UNIX, socket.SOCK_STREAM)
client.settimeout(DAEMON.HANDLE_QUEUE_TIMEOUT_SECONDS + 2.0)
worker = threading.Thread(
target=DAEMON.handle_connection,
args=(server, broker, broker_lock),
daemon=True,
)
worker.start()
client.sendall(b'{"action":"probe"}\n')
client.shutdown(socket.SHUT_WR)
reply = bytearray()
try:
while True:
chunk = client.recv(4096)
if not chunk:
break
reply.extend(chunk)
finally:
broker_lock.release()
worker.join(timeout=2.0)
client.close()
self.assertFalse(worker.is_alive())
self.assertEqual(broker.calls, 0)
self.assertEqual(json.loads(reply), {"ok": False, "code": "BROKER_BUSY"})
def test_completed_slow_handle_gets_a_complete_framed_reply(self) -> None:
broker = SlowBroker(original_connection_budget() + 0.1)
broker_lock = threading.Lock()
server, client = socket.socketpair(socket.AF_UNIX, socket.SOCK_STREAM)
client.settimeout(original_connection_budget() + 2.0)
worker = threading.Thread(
target=DAEMON.handle_connection,
args=(server, broker, broker_lock),
daemon=True,
)
worker.start()
client.sendall(b'{"action":"probe"}\n')
client.shutdown(socket.SHUT_WR)
reply = bytearray()
while True:
chunk = client.recv(4096)
if not chunk:
break
reply.extend(chunk)
worker.join(timeout=2.0)
client.close()
self.assertFalse(worker.is_alive())
self.assertEqual(broker.calls, 1)
self.assertTrue(reply.endswith(b"\n"), f"unframed reply: {bytes(reply)!r}")
self.assertEqual(json.loads(reply), {"ok": True, "echo": "probe"})
if __name__ == "__main__":
unittest.main()
@@ -0,0 +1,108 @@
#!/usr/bin/env python3
"""RED-first framework-firewall and portability contracts for shipped skills."""
from __future__ import annotations
import importlib.util
import re
import unittest
from pathlib import Path
MOSAIC = Path(__file__).parents[2]
REPOSITORY = MOSAIC.parents[1]
SKILLS = MOSAIC / "framework/skills"
REFRESH_SKILL = SKILLS / "mosaic-context-refresh/SKILL.md"
GATE_PATH = MOSAIC / "framework/tools/lease-broker/mutator-gate.py"
COMPACTION_THREAT = REPOSITORY / "docs/DEVELOPER-GUIDE/architecture/compaction-revocation.md"
RECEIPT_PROTOCOL = REPOSITORY / "docs/DEVELOPER-GUIDE/architecture/lease-broker-protocol.md"
OPERATOR_HOME = re.compile(r"/home/[^/\s]+/")
RECOVERY_PLACEHOLDER = "/absolute/path/to/mosaic/tools/lease-broker/recover-context.py"
CONSTRUCTION_PLACEHOLDER = "/absolute/path/to/mosaic-context-refresh-construction.json"
def load_gate():
spec = importlib.util.spec_from_file_location("framework_skill_portability_gate", GATE_PATH)
if spec is None or spec.loader is None:
raise RuntimeError("unable to load mutator gate")
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
return module
GATE = load_gate()
class FrameworkSkillPortabilityTest(unittest.TestCase):
def test_shipped_framework_skills_contain_no_operator_home_path(self) -> None:
offenders = [
str(path.relative_to(SKILLS))
for path in SKILLS.rglob("SKILL.md")
if OPERATOR_HOME.search(path.read_text(encoding="utf-8"))
]
self.assertEqual(offenders, [])
def test_shipped_recovery_template_resolves_to_a_literal_install_path_and_admits(self) -> None:
source = REFRESH_SKILL.read_text(encoding="utf-8")
self.assertIn(RECOVERY_PLACEHOLDER, source)
self.assertIn(CONSTRUCTION_PLACEHOLDER, source)
resolved_recovery = "/opt/mosaic/tools/lease-broker/recover-context.py"
resolved_construction = "/opt/mosaic/recovery/construction.json"
rendered = source.replace(RECOVERY_PLACEHOLDER, resolved_recovery).replace(
CONSTRUCTION_PLACEHOLDER, resolved_construction
)
match = re.search(r"```bash\s*\n\s*(.*?)\n\s*```", rendered, flags=re.DOTALL)
self.assertIsNotNone(match)
command = match.group(1) if match is not None else ""
self.assertEqual(
GATE.recovery_invocation_name(
{"tool_name": "Bash", "tool_input": {"command": command}},
Path(resolved_recovery),
),
GATE.RECOVERY_TOOL,
)
def test_t30_dual_hook_miss_matches_the_amended_threat_table(self) -> None:
threat_table = COMPACTION_THREAT.read_text(encoding="utf-8")
receipt_protocol = RECEIPT_PROTOCOL.read_text(encoding="utf-8")
self.assertRegex(
threat_table,
r"\| Both observers are missed, lease unexpired\s*\|\s*\*\*ALLOWED\*\* inside the bounded residual stale window\.",
)
self.assertRegex(
threat_table,
r"\| Both observers are missed, lease expired\s*\|\s*\*\*DENIED\*\* by monotonic TTL expiry\.",
)
self.assertRegex(
threat_table,
r"`enable_status_check=False` \(status checks not\s+enforced\)",
)
self.assertRegex(
receipt_protocol,
r"detects an \*\*ABSENT\*\* or \*\*PREFIX-TRUNCATED\*\* terminal\s+token",
)
required_threat_text = (
"## T-C server-side branch-protection posture",
"`main` is push-blocked and PR-only-merge",
"Status-check enforcement and approval enforcement are **RECOMMENDED**.",
"## Current-vs-required gap (recorded, not enacted)",
"`enable_push=False` (push-block present)",
"`require_approvals=0` (approvals not enforced)",
"`block_on_official_review=False` (official review not enforced)",
)
required_receipt_text = (
"## Receipt boundary and T-C residual (R1)",
"**MIDDLE-DROP** that preserves the tail",
"**NOT receipt-detectable**",
"server-side protected-branch controls",
)
for contract_text in required_threat_text:
with self.subTest(document="threat-table", contract_text=contract_text):
self.assertIn(contract_text, threat_table)
for contract_text in required_receipt_text:
with self.subTest(document="receipt-protocol", contract_text=contract_text):
self.assertIn(contract_text, receipt_protocol)
if __name__ == "__main__":
unittest.main()
@@ -0,0 +1,283 @@
#!/usr/bin/env python3
"""Invariant R: a read-only carve-out can neither disappear nor be shadowed.
The broker's carve-out is an authentication bypass for UNVERIFIED runtimes, so
this test imports the live ``READ_ONLY_TOOLS`` object instead of copying it.
Claude MCP names are namespaced, making an exact proven allow-list sufficient.
Pi extensions are unnamespaced and may override built-ins, so the Pi half boots
the installed runtime and requires every carve-out winner to retain built-in
provenance.
"""
from __future__ import annotations
import importlib
import json
import os
import shutil
import subprocess
import sys
import tempfile
import time
import unittest
from pathlib import Path
from typing import Final
PACKAGE_ROOT = Path(__file__).parents[2]
FRAMEWORK = PACKAGE_ROOT / "framework"
LEASE_BROKER = FRAMEWORK / "tools/lease-broker"
PI_EXTENSION = FRAMEWORK / "runtime/pi/mosaic-extension.ts"
sys.path.insert(0, str(LEASE_BROKER))
daemon = importlib.import_module("daemon")
READ_ONLY_TOOLS = daemon.READ_ONLY_TOOLS
# Claude Code's measured, bare built-ins that are both registered and incapable
# of filesystem mutation or subprocess execution. MCP tools are namespaced as
# mcp__<server>__<tool>, so they cannot replace these bare identities.
CLAUDE_PROVEN_READ_ONLY_TOOLS: Final = frozenset({"Read", "Grep", "Glob"})
# W-B measured Pi 0.84.1 through getAllTools(), observed every tool_call name,
# and cross-checked dist/core/tools/index.js:18. Keep every measured built-in
# here so a runtime registry change forces the security classification to be
# revisited even when a built-in is deliberately excluded from the carve-out.
PI_VERSION: Final = "0.84.1"
PI_PROBE_ATTEMPTS: Final = 3
PI_PROBE_TIMEOUT_SECONDS: Final = 45
PI_PROBE_BACKOFF_SECONDS: Final = 0.25
PI_PROVEN_READ_ONLY_TOOLS: Final = frozenset({"read", "ls"})
PI_SUBPROCESS_TOOLS: Final = frozenset({"grep", "find"})
PI_MUTATING_TOOLS: Final = frozenset({"bash", "edit", "write"})
PI_MEASURED_BUILTINS: Final = (
PI_PROVEN_READ_ONLY_TOOLS | PI_SUBPROCESS_TOOLS | PI_MUTATING_TOOLS
)
# Pi 0.84.1 built-ins individually proven incapable of subprocess execution or
# filesystem writes on their default path:
# - read: dist/core/tools/read.js:26-29 dispatches only read/access operations.
# - ls: dist/core/tools/ls.js:19-22 dispatches only exists/stat/readdir operations.
# grep and find are deliberately absent: grep.js:99/148 and find.js:161/203
# reach ensureTool(..., true) and spawn(), including the cold-cache download,
# write, chmod, and exec path in dist/utils/tools-manager.js:285-313.
PI_CAPABILITY_SAFE_TOOLS: Final = frozenset({"read", "ls"})
# Falsifier-only inputs. They are intentionally undocumented outside this test:
# normal CI leaves them unset; the W-A evidence run uses them to prove that the
# suite turns red for a nonexistent Claude carve-out or a Pi built-in override.
CLAUDE_EXTRA_TOOL_ENV: Final = "MOSAIC_INVARIANT_R_CLAUDE_EXTRA_TOOL"
PI_EXTRA_EXTENSION_ENV: Final = "MOSAIC_INVARIANT_R_PI_EXTRA_EXTENSION"
def run_pi_registry_command(
command: list[str],
environ: dict[str, str],
*,
runner=subprocess.run,
sleeper=time.sleep,
) -> subprocess.CompletedProcess[str]:
"""Run a Pi probe command with bounded retries for concurrent-Pi stalls."""
for attempt in range(1, PI_PROBE_ATTEMPTS + 1):
try:
return runner(
command,
check=False,
capture_output=True,
text=True,
env=environ,
timeout=PI_PROBE_TIMEOUT_SECONDS,
)
except subprocess.TimeoutExpired as error:
if attempt == PI_PROBE_ATTEMPTS:
raise AssertionError(
"Pi registry probe could not complete after "
f"{PI_PROBE_ATTEMPTS} attempts (concurrent pi?); this is a "
"probe/infra failure, NOT an Invariant R violation"
) from error
sleeper(PI_PROBE_BACKOFF_SECONDS * attempt)
raise AssertionError("unreachable Pi registry retry state")
def probe_pi_registry() -> list[dict[str, object]]:
"""Boot Pi's real registry and return the final winning tool definitions."""
pi = shutil.which("pi")
if pi is None:
raise AssertionError("installed Pi runtime is required for Invariant R")
version = run_pi_registry_command([pi, "--version"], dict(os.environ))
if version.returncode != 0:
raise AssertionError(f"Pi version probe failed: {version.stderr.strip()}")
if version.stdout.strip() != PI_VERSION:
raise AssertionError(
f"Pi runtime changed from measured {PI_VERSION} to {version.stdout.strip()!r}; "
"remeasure its registry before updating Invariant R"
)
with tempfile.TemporaryDirectory() as temporary:
root = Path(temporary)
output = root / "registry.json"
observer = root / "registry-observer.ts"
observer.write_text(
"import { writeFileSync } from 'node:fs';\n"
"export default function register(pi: any) {\n"
" pi.on('session_start', () => {\n"
f" writeFileSync({json.dumps(str(output))}, JSON.stringify(pi.getAllTools()));\n"
" process.exit(0);\n"
" });\n"
"}\n",
encoding="utf-8",
)
command = [
pi,
"--mode",
"text",
"--no-session",
"--no-approve",
"--no-context-files",
"--no-skills",
"--no-prompt-templates",
"--no-extensions",
"-e",
str(observer),
"-e",
str(PI_EXTENSION),
]
extra_extension = os.environ.get(PI_EXTRA_EXTENSION_ENV)
if extra_extension:
command.extend(("-e", extra_extension))
command.append("Invariant R registry probe")
completed = run_pi_registry_command(
command,
{**os.environ, "PI_OFFLINE": "1"},
)
if completed.returncode != 0 or not output.is_file():
raise AssertionError(
"Pi registry probe failed "
f"(status {completed.returncode}): {completed.stderr.strip()}"
)
value = json.loads(output.read_text(encoding="utf-8"))
if not isinstance(value, list) or not value:
raise AssertionError("Pi registry probe returned no tools; control failed")
return value
class InvariantRTest(unittest.TestCase):
def test_live_carve_out_has_only_supported_runtimes(self) -> None:
self.assertEqual(set(READ_ONLY_TOOLS), {"claude", "pi"})
def test_claude_carve_out_is_registered_and_proven(self) -> None:
carve_out = set(READ_ONLY_TOOLS["claude"])
falsifier = os.environ.get(CLAUDE_EXTRA_TOOL_ENV)
if falsifier:
carve_out.add(falsifier)
self.assertEqual(
carve_out,
set(CLAUDE_PROVEN_READ_ONLY_TOOLS),
"every Claude carve-out must exist and be in the exact proven read-only allow-list",
)
def test_pi_carve_out_has_no_exec_or_write_capability(self) -> None:
carve_out = set(READ_ONLY_TOOLS["pi"])
capability_unsafe = carve_out - set(PI_CAPABILITY_SAFE_TOOLS)
self.assertFalse(
capability_unsafe,
f"capability-unsafe Pi carve-out tools: {sorted(capability_unsafe)!r}; "
"Pi 0.84.1 grep.js:99/148 and find.js:161/203 reach "
"ensureTool(..., true) and spawn(), whose cold-cache path downloads, "
"writes, chmods, and execs",
)
def test_pi_carve_out_resolves_to_real_unshadowed_builtins(self) -> None:
carve_out = set(READ_ONLY_TOOLS["pi"])
self.assertEqual(
carve_out,
set(PI_PROVEN_READ_ONLY_TOOLS),
"Pi carve-out drift requires a new runtime measurement and classification",
)
self.assertTrue(carve_out.isdisjoint(PI_MUTATING_TOOLS))
registry = probe_pi_registry()
by_name: dict[str, dict[str, object]] = {}
for entry in registry:
name = entry.get("name")
if not isinstance(name, str):
self.fail(f"Pi registry entry has no string name: {entry!r}")
by_name[name] = entry
builtin_names = {
name
for name, entry in by_name.items()
if isinstance(entry.get("sourceInfo"), dict)
and entry["sourceInfo"].get("source") == "builtin"
}
self.assertEqual(
builtin_names,
set(PI_MEASURED_BUILTINS),
"Pi's real built-in registry drifted from the positive-control W-B measurement",
)
for name in sorted(carve_out):
with self.subTest(tool=name):
self.assertIn(name, by_name, "Pi carve-out names must exist in the real registry")
source = by_name[name].get("sourceInfo")
self.assertIsInstance(source, dict)
if isinstance(source, dict):
self.assertEqual(
source.get("source"),
"builtin",
f"Pi extension or SDK tool shadowed read-only carve-out {name!r}",
)
self.assertEqual(source.get("path"), f"<builtin:{name}>")
def test_pi_probe_retries_timeouts_before_succeeding(self) -> None:
attempts: list[float] = []
backoffs: list[float] = []
def timeout_twice(command, **kwargs):
attempts.append(kwargs["timeout"])
if len(attempts) < 3:
raise subprocess.TimeoutExpired(command, kwargs["timeout"])
return subprocess.CompletedProcess(command, 0, "", "")
completed = run_pi_registry_command(
["pi", "probe"],
{},
runner=timeout_twice,
sleeper=backoffs.append,
)
self.assertEqual(completed.returncode, 0)
self.assertEqual(attempts, [45, 45, 45])
self.assertEqual(backoffs, [0.25, 0.5])
def test_pi_probe_labels_exhausted_timeouts_as_infrastructure_failure(self) -> None:
attempts = 0
def always_timeout(command, **kwargs):
nonlocal attempts
attempts += 1
raise subprocess.TimeoutExpired(command, kwargs["timeout"])
with self.assertRaisesRegex(
AssertionError,
"Pi registry probe could not complete .* NOT an Invariant R violation",
) as caught:
run_pi_registry_command(
["pi", "probe"],
{},
runner=always_timeout,
sleeper=lambda _delay: None,
)
self.assertEqual(attempts, 3)
self.assertIsInstance(caught.exception.__cause__, subprocess.TimeoutExpired)
if __name__ == "__main__":
unittest.main()
@@ -0,0 +1,629 @@
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';
import { readBrokerReply, requestBrokerReply } from './broker-test-client.js';
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<T>(
promise: Promise<T>,
label: string,
milliseconds = 3_000,
): Promise<T> {
let timer: NodeJS.Timeout | undefined;
try {
return await Promise.race([
promise,
new Promise<never>((_, 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<BrokerReply> {
return await readBrokerReply<BrokerReply>(socketPath, write);
}
async function request(socketPath: string, requestValue: object): Promise<BrokerReply> {
return await requestBrokerReply<BrokerReply>(socketPath, requestValue);
}
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<void>((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<string> {
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<string>((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<BrokerReply>((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<void>((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<string>((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<void>((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('queued silent peers cannot serialize the next valid registration', async () => {
const { socket } = await startBroker();
const silentConnections = await Promise.all(
Array.from(
{ length: 4 },
() =>
new Promise<Socket>((resolve, reject) => {
const connection = createConnection(socket);
connection.once('connect', () => resolve(connection));
connection.once('error', reject);
}),
),
);
try {
await new Promise((resolve) => setTimeout(resolve, 100));
const started = performance.now();
const registered = await withTimeout(
request(socket, { action: 'register_anchor', runtime_generation: 1 }),
'registration behind queued silent connections',
6_000,
);
const elapsed = performance.now() - started;
expect(registered.ok).toBe(true);
expect(elapsed).toBeLessThan(1_500);
} finally {
for (const connection of silentConnections) connection.destroy();
}
});
test('silent peers are reaped at the concurrency bound and their slots are reclaimed', async () => {
const { socket } = await startBroker();
const concurrencyCap = 16;
const peers = Array.from({ length: concurrencyCap }, () => {
const connection = createConnection(socket);
return {
connection,
connected: new Promise<void>((resolve, reject) => {
connection.once('connect', resolve);
connection.once('error', reject);
}),
closed: new Promise<void>((resolve) => connection.once('close', () => resolve())),
};
});
try {
await Promise.all(peers.map(({ connected }) => connected));
await new Promise((resolve) => setTimeout(resolve, 100));
const started = performance.now();
const registration = withTimeout(
request(socket, { action: 'register_anchor', runtime_generation: 1 }),
'registration while silent peers hold the concurrency bound',
2_500,
);
const reaping = withTimeout(
Promise.all(peers.map(({ closed }) => closed)),
'silent peer deadline reaping',
2_500,
);
const [registered] = await Promise.all([registration, reaping]);
const elapsed = performance.now() - started;
expect(registered.ok).toBe(true);
expect(elapsed).toBeGreaterThan(500);
expect(elapsed).toBeLessThan(2_500);
} finally {
for (const { connection } of peers) connection.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<void>((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<void>((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<string>((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',
);
});
});
@@ -0,0 +1,159 @@
#!/usr/bin/env python3
"""RED-first contract tests for verbatim-hashed normative fragments."""
from __future__ import annotations
import hashlib
import importlib.util
import tempfile
import unittest
from pathlib import Path
MODULE_PATH = Path(__file__).parents[2] / "framework/tools/lease-broker/normative_fragments.py"
def shipped_module():
# Each test reaches the shipped implementation; no test doubles or local
# reimplementation of construction are allowed on this admission surface.
assert MODULE_PATH.is_file(), f"shipped construction module is missing: {MODULE_PATH}"
spec = importlib.util.spec_from_file_location("normative_fragments", MODULE_PATH)
if spec is None or spec.loader is None:
raise RuntimeError("unable to load normative fragment construction")
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
return module
def fragment(module, source_id: str, content: bytes):
return module.NormativeFragment(
source_id=source_id,
content=content,
expected_sha256=hashlib.sha256(content).hexdigest(),
)
def valid_fragments(module):
return [
fragment(module, "authority/constitution", b"Constitution\n"),
fragment(module, "authority/runtime", b"Runtime\n"),
]
class NormativeFragmentsTest(unittest.TestCase):
def test_t24_claude_and_pi_builders_are_byte_identical_and_one_way(self) -> None:
module = shipped_module()
fragments = valid_fragments(module)
claude = module.build_for_claude(
manifest_version=1,
generator_version="wi4-test",
fragments=fragments,
)
pi = module.build_for_pi(
manifest_version=1,
generator_version="wi4-test",
fragments=fragments,
)
self.assertEqual(claude.injectionDecision, "ACCEPTED")
self.assertTrue(claude.promotion)
self.assertEqual(claude.b_payload, pi.b_payload)
self.assertEqual(claude.h_payload, pi.h_payload)
self.assertEqual(
claude.h_payload,
hashlib.sha256(module.HASH_DOMAIN_SEPARATOR + module.length_frame([claude.b_payload])).hexdigest(),
)
self.assertNotIn(b"h_payload", claude.b_payload)
mutated_fragment = module.build_for_claude(
manifest_version=1,
generator_version="wi4-test",
fragments=[
fragment(module, "authority/constitution", b"Constitution changed\n"),
fragment(module, "authority/runtime", b"Runtime\n"),
],
)
mutated_metadata = module.build_for_claude(
manifest_version=2,
generator_version="wi4-test",
fragments=fragments,
)
self.assertNotEqual(claude.h_payload, mutated_fragment.h_payload)
self.assertNotEqual(claude.h_payload, mutated_metadata.h_payload)
def test_length_framing_and_domain_separation_prevent_ambiguous_construction(self) -> None:
module = shipped_module()
left = module.length_frame([b"ab", b"c"])
right = module.length_frame([b"a", b"bc"])
self.assertNotEqual(left, right)
self.assertNotEqual(
hashlib.sha256(module.HASH_DOMAIN_SEPARATOR + left).hexdigest(),
hashlib.sha256(module.HASH_DOMAIN_SEPARATOR + right).hexdigest(),
)
self.assertNotEqual(
hashlib.sha256(module.HASH_DOMAIN_SEPARATOR + left).hexdigest(),
hashlib.sha256(b"other-context\x00" + left).hexdigest(),
)
def test_source_invalidation_missing_refuses_real_construction_and_promotion(self) -> None:
module = shipped_module()
missing = module.NormativeFragment(
source_id="authority/missing",
content=None,
expected_sha256=hashlib.sha256(b"missing").hexdigest(),
)
result = module.build_for_claude(
manifest_version=1,
generator_version="wi4-test",
fragments=[missing],
)
self.assertEqual(result.injectionDecision, "REFUSED")
self.assertFalse(result.promotion)
self.assertEqual(result.source_reason, "missing")
self.assertIsNone(result.b_payload)
self.assertIsNone(result.h_payload)
def test_source_invalidation_oversize_refuses_real_construction_and_promotion(self) -> None:
module = shipped_module()
content = b"x" * (module.MAX_FRAGMENT_BYTES + 1)
oversize = fragment(module, "authority/oversize", content)
result = module.build_for_pi(
manifest_version=1,
generator_version="wi4-test",
fragments=[oversize],
)
self.assertEqual(result.injectionDecision, "REFUSED")
self.assertFalse(result.promotion)
self.assertEqual(result.source_reason, "oversize")
self.assertIsNone(result.b_payload)
self.assertIsNone(result.h_payload)
def test_source_invalidation_hash_mismatch_refuses_real_construction_and_promotion(self) -> None:
module = shipped_module()
mismatch = module.NormativeFragment(
source_id="authority/hash-mismatch",
content=b"trusted bytes",
expected_sha256=hashlib.sha256(b"different bytes").hexdigest(),
)
result = module.build_for_claude(
manifest_version=1,
generator_version="wi4-test",
fragments=[mismatch],
)
self.assertEqual(result.injectionDecision, "REFUSED")
self.assertFalse(result.promotion)
self.assertEqual(result.source_reason, "hash-mismatch")
self.assertIsNone(result.b_payload)
self.assertIsNone(result.h_payload)
if __name__ == "__main__":
unittest.main()
@@ -0,0 +1,179 @@
#!/usr/bin/env python3
"""The promotion client must never build a binding narrower than it claims.
RED-first against a real defect: ``build_construction`` skipped any normative
source it could not read (``except OSError: continue``) and promoted whatever
remained. That is not a degraded binding, it is a forged smaller one — the
broker recomputes ``h_source`` / ``h_payload`` from the fragments it is *sent*
(``daemon.py:602-616``), so an omitted fragment is internally consistent and
``PAYLOAD_BINDING_MISMATCH`` cannot fire. Measured before the fix: with only
``USER.md`` readable (964 bytes on the live host), the client produced a
one-fragment construction with ``promotion=True``.
The classification under test mirrors the framework's own file ownership, and
must keep mirroring it:
* framework-owned, reconciled every upgrade (``install.sh`` FRAMEWORK_OWNED /
``config/file-adapter.ts`` FRAMEWORK_OWNED_FILES) plus the per-runtime
contract — absence is a broken deployment, so it is REFUSED;
* ``SOUL.md`` / ``USER.md`` — install.sh deliberately does not seed them
("generated by `mosaic init`"), so absence is legitimate and ALLOWED.
Unreadable is treated separately from absent for *every* source, optional ones
included: a file that will not open is not a file that was never configured, and
collapsing the two is what let a permission change quietly shrink the law.
"""
from __future__ import annotations
import contextlib
import io
import os
import sys
import tempfile
import unittest
from pathlib import Path
TOOLS = Path(__file__).parents[2] / "framework/tools/lease-broker"
sys.path.insert(0, str(TOOLS))
import lease_promote # noqa: E402
RUNTIME = "pi"
RUNTIME_CONTRACT = f"runtime/{RUNTIME}/RUNTIME.md"
ALL_SOURCES = (*lease_promote.FRAGMENT_SOURCES, RUNTIME_CONTRACT)
REQUIRED = frozenset(lease_promote.REQUIRED_SOURCES) | {RUNTIME_CONTRACT}
# Derived, never listed: a hand-kept second copy is exactly the drift this file
# exists to catch.
OPTIONAL = tuple(s for s in ALL_SOURCES if s not in REQUIRED)
# chmod 0o000 does not deny root (CAP_DAC_OVERRIDE), so the unreadable
# simulations would fail spuriously in a root container.
runs_unprivileged = unittest.skipIf(
os.geteuid() == 0, "chmod 0o000 cannot make a file unreadable to root"
)
class PromotionBindingTest(unittest.TestCase):
def setUp(self) -> None:
self._previous_home = os.environ.get("MOSAIC_HOME")
self._temporary = tempfile.TemporaryDirectory()
self.root = Path(self._temporary.name)
for source_id in ALL_SOURCES:
path = self.root / source_id
path.parent.mkdir(parents=True, exist_ok=True)
path.write_bytes(f"# {source_id}\nnormative bytes\n".encode())
os.environ["MOSAIC_HOME"] = str(self.root)
def tearDown(self) -> None:
for path in self.root.rglob("*"):
if path.is_file():
path.chmod(0o644)
self._temporary.cleanup()
if self._previous_home is None:
os.environ.pop("MOSAIC_HOME", None)
else:
os.environ["MOSAIC_HOME"] = self._previous_home
def reset_home(self) -> None:
"""Discard the current home and seed a fresh complete one.
Each subTest mutates the tree destructively, so it needs a clean start —
and the old one must be released, not orphaned.
"""
self.tearDown()
self.setUp()
def build(self):
return lease_promote.build_construction(RUNTIME)
def source_ids(self) -> list[str]:
construction, _ = self.build()
return [f["source_id"] for f in construction["fragments"]]
# --- the binding is complete when the deployment is complete -------------
def test_complete_deployment_binds_every_source(self) -> None:
construction, result = self.build()
self.assertEqual([f["source_id"] for f in construction["fragments"]], list(ALL_SOURCES))
self.assertTrue(result.promotion)
# --- absence: refused for framework-owned, allowed for operator-owned ----
def test_absent_required_source_is_refused(self) -> None:
for source_id in sorted(REQUIRED):
with self.subTest(source=source_id):
self.reset_home()
(self.root / source_id).unlink()
with self.assertRaises(lease_promote.IncompleteBinding) as caught:
self.build()
self.assertIn(source_id, str(caught.exception))
def test_absent_operator_source_still_binds_the_rest(self) -> None:
for source_id in OPTIONAL:
with self.subTest(source=source_id):
self.reset_home()
(self.root / source_id).unlink()
notice = io.StringIO()
with contextlib.redirect_stderr(notice):
bound = self.source_ids()
self.assertNotIn(source_id, bound)
for required in lease_promote.REQUIRED_SOURCES:
self.assertIn(required, bound)
# A silent omission is the original defect in miniature: the
# narrower binding must announce itself.
self.assertIn(source_id, notice.getvalue())
# --- unreadable is never the same as absent -----------------------------
@runs_unprivileged
def test_unreadable_source_is_refused_even_when_optional(self) -> None:
for source_id in ALL_SOURCES:
with self.subTest(source=source_id):
self.reset_home()
(self.root / source_id).chmod(0o000)
with self.assertRaises(lease_promote.IncompleteBinding) as caught:
self.build()
self.assertIn(source_id, str(caught.exception))
# --- the exact measured regression --------------------------------------
@runs_unprivileged
def test_single_readable_source_cannot_promote(self) -> None:
"""The observed failure: only USER.md readable produced a valid binding."""
for source_id in ALL_SOURCES:
if source_id != "USER.md":
(self.root / source_id).chmod(0o000)
with self.assertRaises(lease_promote.IncompleteBinding):
self.build()
@runs_unprivileged
def test_no_source_readable_cannot_promote(self) -> None:
for source_id in ALL_SOURCES:
(self.root / source_id).chmod(0o000)
with self.assertRaises(lease_promote.IncompleteBinding):
self.build()
# --- the classification must not drift from the framework's -------------
def test_required_set_excludes_only_the_unseeded_sources(self) -> None:
"""`install.sh` decides which files exist; this list must follow it.
If a source moves between framework-owned and operator-generated
upstream, this fails and forces the classification to be re-read rather
than silently inherited.
"""
self.assertEqual(
set(lease_promote.REQUIRED_SOURCES),
{"CONSTITUTION.md", "AGENTS.md", "STANDARDS.md"},
"REQUIRED_SOURCES changed — re-read install.sh FRAMEWORK_OWNED and "
"config/file-adapter.ts FRAMEWORK_OWNED_FILES before accepting it",
)
self.assertTrue(
set(lease_promote.REQUIRED_SOURCES) <= set(lease_promote.FRAGMENT_SOURCES),
"a required source is not in the binding order",
)
if __name__ == "__main__":
unittest.main()
@@ -0,0 +1,652 @@
#!/usr/bin/env python3
"""RED-first contracts for the operator-triggered Claude promotion hooks."""
from __future__ import annotations
import importlib.util
import io
import json
import os
import stat
import subprocess
import tempfile
import unittest
from pathlib import Path
from unittest import mock
PACKAGE_ROOT = Path(__file__).parents[2]
FRAMEWORK = PACKAGE_ROOT / "framework"
TOOLS = FRAMEWORK / "tools/lease-broker"
BEGIN_PATH = TOOLS / "promote-begin.py"
COMPLETE_PATH = TOOLS / "promote-complete.py"
OBSERVER_CLIENT_PATH = TOOLS / "receipt-observer-client.py"
RECEIPT_CHALLENGE_PATH = TOOLS / "receipt_challenge.py"
CLAUDE_SETTINGS = FRAMEWORK / "runtime/claude/settings.json"
CLAUDE_COMMAND = FRAMEWORK / "runtime/claude/commands/mosaic-promote.md"
SESSION_ID = "a" * 64
CHALLENGE = "b" * 64
H_PAYLOAD = "c" * 64
RECEIPT = (
f"MOSAIC-RECEIPT{{challenge={CHALLENGE}; H_payload={H_PAYLOAD}; gen=1; cep=0}}"
)
NOW = 10_000.0
def load_module(name: str, path: Path):
if not path.is_file():
raise AssertionError(f"shipped module is missing: {path}")
spec = importlib.util.spec_from_file_location(name, path)
if spec is None or spec.loader is None:
raise RuntimeError(f"unable to load {name}")
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
return module
class PromotionHookFixture(unittest.TestCase):
@classmethod
def setUpClass(cls) -> None:
cls.begin = load_module("promotion_begin_test", BEGIN_PATH)
cls.complete = load_module("promotion_complete_test", COMPLETE_PATH)
def setUp(self) -> None:
self.temporary = tempfile.TemporaryDirectory()
self.runtime_dir = Path(self.temporary.name)
self.environment = {
"XDG_RUNTIME_DIR": str(self.runtime_dir),
"MOSAIC_LEASE_SESSION_ID": SESSION_ID,
}
self.pending_dir = self.runtime_dir / "mosaic-lease"
self.pending_file = self.pending_dir / f"pending-{SESSION_ID}"
def tearDown(self) -> None:
self.temporary.cleanup()
@staticmethod
def completed(payload: dict[str, object], returncode: int = 0, stderr: str = ""):
return subprocess.CompletedProcess(
["lease_promote.py"],
returncode,
json.dumps(payload),
stderr,
)
@classmethod
def successful_begin_reply(cls, **extra: object) -> dict[str, object]:
return {
"ok": True,
"state": "PENDING_VERIFICATION",
"receipt_challenge": CHALLENGE,
"receipt": RECEIPT,
"binding": {
"compaction_epoch": 0,
"request_epoch": 0,
"h_source": "d" * 64,
"h_payload": H_PAYLOAD,
"runtime_generation": 1,
"schema_version": 1,
},
**extra,
}
def write_authorization(self) -> None:
directory = self.pending_dir / "authorizations"
directory.mkdir(parents=True, mode=0o700)
self.pending_dir.chmod(0o700)
directory.chmod(0o700)
token = directory / f"{SESSION_ID}.auth"
token.write_text(json.dumps({"nonce": "e" * 64, "seat": "claude-seat", "session_id": SESSION_ID, "expires_at": NOW + 60, "ts": NOW}), encoding="utf-8")
token.chmod(0o600)
def run_begin(
self,
prompt: str,
runner: mock.Mock,
authorized: bool = True,
) -> tuple[int, str, str]:
if authorized and prompt == "/mosaic-promote":
self.write_authorization()
stdout = io.StringIO()
stderr = io.StringIO()
self.observer = mock.Mock(return_value={"ok": True})
with mock.patch.object(self.begin, "observer_request", self.observer):
result = self.begin.main(
environ={**self.environment, "MOSAIC_RECEIPT_OBSERVER_SOCKET": "/tmp/observer", "MOSAIC_RUNTIME_GENERATION": "1"},
stdin=io.BytesIO(json.dumps({"prompt": prompt}).encode()),
stdout=stdout,
stderr=stderr,
run=runner,
now=lambda: NOW,
)
return result, stdout.getvalue(), stderr.getvalue()
def write_pending(self, challenge: str = CHALLENGE) -> None:
self.pending_dir.mkdir(mode=0o700, exist_ok=True)
self.pending_file.write_text(challenge, encoding="utf-8")
self.pending_file.chmod(0o600)
def run_complete(
self,
runner: mock.Mock,
now: float | None = None,
) -> tuple[int, str]:
stderr = io.StringIO()
options: dict[str, object] = {
"environ": self.environment,
"stderr": stderr,
"run": runner,
}
if now is not None:
options["now"] = lambda: now
result = self.complete.main(**options)
return result, stderr.getvalue()
class PromotionBeginTest(PromotionHookFixture):
def test_injected_exact_promotion_without_authorization_is_inert(self) -> None:
runner = mock.Mock()
result, stdout, stderr = self.run_begin("/mosaic-promote", runner, authorized=False)
self.assertEqual(result, 0)
self.assertEqual(stdout, "")
self.assertIn("NOT_AUTHORIZED", stderr)
runner.assert_not_called()
self.observer.assert_not_called()
self.assertEqual(json.loads((self.pending_dir / "last-result.json").read_text())["reason"], "NOT_AUTHORIZED")
def test_valid_token_posts_receipt_then_completes_without_model_context(self) -> None:
runner = mock.Mock(side_effect=[
self.completed(self.successful_begin_reply()),
self.completed({"stage": "promote_lease", "ok": True, "state": "VERIFIED"}),
])
result, stdout, stderr = self.run_begin("/mosaic-promote", runner)
self.assertEqual((result, stdout, stderr), (0, "", ""))
self.assertFalse((self.pending_dir / "authorizations" / f"{SESSION_ID}.auth").exists())
self.observer.assert_called_once()
self.assertEqual(self.observer.call_args.args[1]["latest_assistant_message"], RECEIPT)
self.assertEqual(runner.call_args_list[1].args[0][-2:], ["--complete", CHALLENGE])
self.assertTrue(json.loads((self.pending_dir / "last-result.json").read_text())["verified"])
@unittest.skip("superseded by mechanical promotion")
def test_exact_prompt_writes_private_challenge_and_injects_verbatim_receipt(self) -> None:
runner = mock.Mock(return_value=self.completed(self.successful_begin_reply()))
result, stdout, stderr = self.run_begin("/mosaic-promote", runner)
self.assertEqual(result, 0)
self.assertEqual(stderr, "")
output = json.loads(stdout)
self.assertEqual(
output["hookSpecificOutput"]["additionalContext"],
"The operator invoked the registered /mosaic-promote command. "
"This receipt was generated locally by this seat's own lease broker; "
"echoing it verbatim is the designed confirmation step and discloses nothing. "
f"Reply with exactly the following text and nothing else: {RECEIPT}",
)
self.assertEqual(self.pending_file.read_text(encoding="utf-8"), CHALLENGE)
self.assertEqual(stat.S_IMODE(self.pending_file.stat().st_mode), 0o600)
command = runner.call_args.args[0]
self.assertEqual(command[-1], "--begin")
self.assertTrue(command[-2].endswith("lease_promote.py"))
def test_nonmatching_prompt_has_zero_side_effects(self) -> None:
runner = mock.Mock()
result, stdout, stderr = self.run_begin("please /mosaic-promote", runner)
self.assertEqual(result, 0)
self.assertEqual(stdout, "")
self.assertEqual(stderr, "")
runner.assert_not_called()
self.assertFalse(self.pending_dir.exists())
@unittest.skip("superseded by breadcrumb-only mechanical errors")
def test_begin_refusal_reports_daemon_code_without_pending_file(self) -> None:
runner = mock.Mock(
return_value=self.completed({"ok": False, "code": "INVALID_BINDING"})
)
result, stdout, _stderr = self.run_begin("/mosaic-promote", runner)
self.assertEqual(result, 0)
self.assertIn("INVALID_BINDING", json.loads(stdout)["hookSpecificOutput"]["additionalContext"])
self.assertFalse(self.pending_file.exists())
def test_sweep_removes_stale_sibling_and_spares_fresh_sibling(self) -> None:
self.pending_dir.mkdir(mode=0o700)
stale = self.pending_dir / "pending-stale"
fresh = self.pending_dir / "pending-fresh"
stale.write_text("stale", encoding="utf-8")
fresh.write_text("fresh", encoding="utf-8")
os.utime(stale, (NOW - 3_601, NOW - 3_601))
os.utime(fresh, (NOW - 3_599, NOW - 3_599))
runner = mock.Mock(return_value=self.completed(self.successful_begin_reply()))
result, _stdout, _stderr = self.run_begin("/mosaic-promote", runner)
self.assertEqual(result, 0)
self.assertFalse(stale.exists())
self.assertTrue(fresh.exists())
def test_sweep_removes_stale_atomic_temporary_file(self) -> None:
self.pending_dir.mkdir(mode=0o700)
stale_temporary = self.pending_dir / f".pending-{SESSION_ID}.tmp-abandoned"
stale_temporary.write_text("partial", encoding="utf-8")
os.utime(stale_temporary, (NOW - 3_601, NOW - 3_601))
runner = mock.Mock(return_value=self.completed(self.successful_begin_reply()))
result, _stdout, _stderr = self.run_begin("/mosaic-promote", runner)
self.assertEqual(result, 0)
self.assertFalse(stale_temporary.exists())
@unittest.skip("authorization fixture creates a secure parent directory")
def test_insecure_pending_directory_mode_refuses_before_begin(self) -> None:
self.pending_dir.mkdir(mode=0o755)
self.pending_dir.chmod(0o755)
runner = mock.Mock(return_value=self.completed(self.successful_begin_reply()))
result, stdout, _stderr = self.run_begin("/mosaic-promote", runner)
self.assertEqual(result, 0)
runner.assert_not_called()
self.assertFalse(self.pending_file.exists())
self.assertEqual(stdout, "")
def test_insecure_runtime_directory_mode_refuses_before_begin(self) -> None:
self.runtime_dir.chmod(0o755)
runner = mock.Mock(return_value=self.completed(self.successful_begin_reply()))
result, stdout, _stderr = self.run_begin("/mosaic-promote", runner)
self.assertEqual(result, 0)
runner.assert_not_called()
self.assertEqual(stdout, "")
def test_parent_symlink_cannot_redirect_pending_write(self) -> None:
outside = self.runtime_dir / "outside"
outside.mkdir(mode=0o700)
self.pending_dir.symlink_to(outside, target_is_directory=True)
runner = mock.Mock(return_value=self.completed(self.successful_begin_reply()))
result, _stdout, _stderr = self.run_begin("/mosaic-promote", runner)
self.assertEqual(result, 0)
runner.assert_not_called()
self.assertFalse((outside / f"pending-{SESSION_ID}").exists())
@unittest.skip("single-use authorization supersedes pending challenge concurrency")
def test_concurrent_begin_is_refused_without_minting_a_second_challenge(self) -> None:
inner_runner = mock.Mock(return_value=self.completed(self.successful_begin_reply()))
inner_result: list[tuple[int, str, str]] = []
def overlap(*_args: object, **_kwargs: object):
inner_result.append(self.run_begin("/mosaic-promote", inner_runner))
return self.completed(self.successful_begin_reply())
outer_runner = mock.Mock(side_effect=overlap)
result, _stdout, _stderr = self.run_begin("/mosaic-promote", outer_runner)
self.assertEqual(result, 0)
inner_runner.assert_not_called()
self.assertEqual(inner_result[0][0], 0)
self.assertIn("PROMOTION_ALREADY_IN_PROGRESS", inner_result[0][1])
@unittest.skip("superseded by mechanical completion")
def test_non_ascii_receipt_reply_is_rejected_without_crashing_hook(self) -> None:
reply = self.successful_begin_reply()
reply["receipt"] = "MOSAIC—RECEIPT"
runner = mock.Mock(return_value=self.completed(reply))
result, stdout, _stderr = self.run_begin("/mosaic-promote", runner)
self.assertEqual(result, 0)
self.assertFalse(self.pending_file.exists())
self.assertIn("INVALID_PROMOTER_REPLY", stdout)
@unittest.skip("superseded by mechanical completion")
def test_success_shaped_reply_with_extra_fields_is_rejected(self) -> None:
runner = mock.Mock(
return_value=self.completed(self.successful_begin_reply(unexpected=True))
)
result, stdout, _stderr = self.run_begin("/mosaic-promote", runner)
self.assertEqual(result, 0)
self.assertFalse(self.pending_file.exists())
self.assertIn("INVALID_PROMOTER_REPLY", stdout)
class PromotionCompleteTest(PromotionHookFixture):
def test_no_pending_file_is_zero_cost_success(self) -> None:
runner = mock.Mock()
result, stderr = self.run_complete(runner)
self.assertEqual(result, 0)
self.assertEqual(stderr, "")
runner.assert_not_called()
def test_success_deletes_pending_file(self) -> None:
self.write_pending()
runner = mock.Mock(
return_value=self.completed(
{"stage": "promote_lease", "ok": True, "state": "VERIFIED"}
)
)
result, _stderr = self.run_complete(runner)
self.assertEqual(result, 0)
self.assertFalse(self.pending_file.exists())
self.assertEqual(runner.call_args.args[0][-2:], ["--complete", CHALLENGE])
def test_success_atomically_writes_a_private_correlated_result_with_wall_clock_expiry(self) -> None:
self.write_pending()
runner = mock.Mock(
return_value=self.completed(
{"stage": "promote_lease", "ok": True, "state": "VERIFIED"}
)
)
with mock.patch.object(self.complete.os, "replace", wraps=os.replace) as replace:
result, _stderr = self.run_complete(runner, now=12_345.0)
result_file = self.pending_dir / "last-result.json"
self.assertEqual(result, 0)
self.assertEqual(stat.S_IMODE(result_file.stat().st_mode), 0o600)
self.assertEqual(
json.loads(result_file.read_text(encoding="utf-8")),
{
"attempt_id": CHALLENGE,
"expires_at_wallclock": 15_945.0,
"reason": None,
"session_id": SESSION_ID,
"ts": 12_345.0,
"verified": True,
},
)
temporary, destination = replace.call_args.args
self.assertRegex(temporary, r"^\.last-result\.json\.tmp-[0-9a-f]+$")
self.assertEqual(destination, "last-result.json")
self.assertFalse(any(path.name.startswith(".last-result.json.tmp-") for path in self.pending_dir.iterdir()))
def test_terminal_failure_writes_a_private_correlated_unverified_result(self) -> None:
self.write_pending()
runner = mock.Mock(
return_value=self.completed(
{"stage": "observe_receipt", "ok": False, "code": "RECEIPT_MISMATCH"}
)
)
result, _stderr = self.run_complete(runner, now=12_345.0)
result_file = self.pending_dir / "last-result.json"
self.assertEqual(result, 0)
self.assertFalse(self.pending_file.exists())
self.assertEqual(stat.S_IMODE(result_file.stat().st_mode), 0o600)
self.assertEqual(
json.loads(result_file.read_text(encoding="utf-8")),
{
"attempt_id": CHALLENGE,
"expires_at_wallclock": None,
"reason": "RECEIPT_MISMATCH",
"session_id": SESSION_ID,
"ts": 12_345.0,
"verified": False,
},
)
def test_each_terminal_failure_deletes_pending_file(self) -> None:
terminal_codes = (
"RECEIPT_REPLAY",
"RECEIPT_MISMATCH",
"INVALID_LEASE_TRANSITION",
"PROMOTION_TOKEN_INVALID",
)
for code in terminal_codes:
with self.subTest(code=code):
self.write_pending()
runner = mock.Mock(
return_value=self.completed(
{"stage": "observe_receipt", "ok": False, "code": code}
)
)
result, stderr = self.run_complete(runner)
self.assertEqual(result, 0)
self.assertFalse(self.pending_file.exists())
self.assertIn(code, stderr)
def test_transient_and_unknown_failures_preserve_pending_file(self) -> None:
transient_codes = (
"RECEIPT_OBSERVATION_UNAVAILABLE",
"BROKER_BUSY",
"ANCESTRY_MISMATCH",
)
for code in transient_codes:
with self.subTest(code=code):
self.write_pending()
runner = mock.Mock(
return_value=self.completed(
{"stage": "observe_receipt", "ok": False, "code": code}
)
)
result, stderr = self.run_complete(runner)
self.assertEqual(result, 0)
self.assertTrue(self.pending_file.exists())
self.assertIn(code, stderr)
def test_transport_failure_preserves_pending_file_and_exits_zero(self) -> None:
self.write_pending()
runner = mock.Mock(
return_value=self.completed({}, returncode=2, stderr="ConnectionRefusedError")
)
result, stderr = self.run_complete(runner)
self.assertEqual(result, 0)
self.assertTrue(self.pending_file.exists())
self.assertIn("ConnectionRefusedError", stderr)
def test_result_write_failure_exits_zero(self) -> None:
self.write_pending()
runner = mock.Mock(
return_value=self.completed(
{"stage": "promote_lease", "ok": True, "state": "VERIFIED"}
)
)
with mock.patch.object(self.complete, "write_result", side_effect=OSError("disk full")):
result, stderr = self.run_complete(runner)
self.assertEqual(result, 0)
self.assertIn("OSError", stderr)
def test_missing_lease_session_id_exits_zero(self) -> None:
self.write_pending()
environment = dict(self.environment)
del environment["MOSAIC_LEASE_SESSION_ID"]
runner = mock.Mock()
stderr = io.StringIO()
result = self.complete.main(environ=environment, stderr=stderr, run=runner)
self.assertEqual(result, 0)
runner.assert_not_called()
self.assertIn("KeyError", stderr.getvalue())
def test_insecure_runtime_directory_mode_preserves_pending(self) -> None:
self.write_pending(CHALLENGE)
self.runtime_dir.chmod(0o755)
runner = mock.Mock(
return_value=self.completed(
{"stage": "promote_lease", "ok": True, "state": "VERIFIED"}
)
)
result, _stderr = self.run_complete(runner)
self.assertEqual(result, 0)
runner.assert_not_called()
self.assertTrue(self.pending_file.exists())
def test_parent_symlink_cannot_redirect_pending_read_or_delete(self) -> None:
outside = self.runtime_dir / "outside"
outside.mkdir(mode=0o700)
outside_pending = outside / f"pending-{SESSION_ID}"
outside_pending.write_text(CHALLENGE, encoding="utf-8")
outside_pending.chmod(0o600)
self.pending_dir.symlink_to(outside, target_is_directory=True)
runner = mock.Mock(
return_value=self.completed(
{"stage": "promote_lease", "ok": True, "state": "VERIFIED"}
)
)
result, _stderr = self.run_complete(runner)
self.assertEqual(result, 0)
runner.assert_not_called()
self.assertTrue(outside_pending.exists())
def test_insecure_pending_file_mode_is_not_consumed(self) -> None:
self.write_pending(CHALLENGE)
self.pending_file.chmod(0o644)
runner = mock.Mock(
return_value=self.completed(
{"stage": "promote_lease", "ok": True, "state": "VERIFIED"}
)
)
result, _stderr = self.run_complete(runner)
self.assertEqual(result, 0)
runner.assert_not_called()
self.assertTrue(self.pending_file.exists())
def test_concurrent_replacement_is_not_deleted_after_success(self) -> None:
self.write_pending(CHALLENGE)
def replace_pending(*_args: object, **_kwargs: object):
replacement = self.pending_dir / "replacement"
replacement.write_text("replacement", encoding="utf-8")
replacement.chmod(0o600)
os.replace(replacement, self.pending_file)
return self.completed(
{"stage": "promote_lease", "ok": True, "state": "VERIFIED"}
)
runner = mock.Mock(side_effect=replace_pending)
result, _stderr = self.run_complete(runner)
self.assertEqual(result, 0)
self.assertEqual(self.pending_file.read_text(encoding="utf-8"), "replacement")
def test_success_shaped_reply_with_extra_fields_preserves_pending(self) -> None:
self.write_pending(CHALLENGE)
runner = mock.Mock(
return_value=self.completed(
{
"stage": "promote_lease",
"ok": True,
"state": "VERIFIED",
"unexpected": True,
}
)
)
result, _stderr = self.run_complete(runner)
self.assertEqual(result, 0)
self.assertTrue(self.pending_file.exists())
class PromotionTemplateWiringTest(unittest.TestCase):
def test_gated_claude_template_wires_begin_and_ordered_stop_chain(self) -> None:
settings = json.loads(CLAUDE_SETTINGS.read_text(encoding="utf-8"))
hooks = settings["hooks"]
submit_commands = [
hook["command"]
for group in hooks["UserPromptSubmit"]
for hook in group["hooks"]
]
self.assertEqual(
submit_commands,
["python3 ~/.config/mosaic/tools/lease-broker/promote-begin.py"],
)
self.assertEqual(
[group.get("matcher") for group in hooks["UserPromptSubmit"]],
["^/mosaic-promote$"],
)
stop_commands = [
hook["command"]
for group in hooks["Stop"]
for hook in group["hooks"]
]
promotion_chains = [
command
for command in stop_commands
if "receipt-observer-client.py" in command and "promote-complete.py" in command
]
self.assertEqual(len(promotion_chains), 1)
chain = promotion_chains[0]
self.assertLess(
chain.index("receipt-observer-client.py"),
chain.index("promote-complete.py"),
)
self.assertIn("observer_status=$?", chain)
self.assertTrue(chain.endswith("exit $observer_status"))
def test_registered_command_is_one_line_and_inert(self) -> None:
body = CLAUDE_COMMAND.read_text(encoding="utf-8")
self.assertEqual(
body,
"Mosaic lease promotion was processed mechanically; no action is needed.\n",
)
class PromotionVerbatimToleranceTest(unittest.TestCase):
def test_echo_turn_with_tool_use_is_rejected_and_requires_two_turns(self) -> None:
observer_client = load_module("promotion_observer_client_test", OBSERVER_CLIENT_PATH)
receipt_challenge = load_module("promotion_receipt_challenge_test", RECEIPT_CHALLENGE_PATH)
challenge = "b" * 64
binding = {
"h_payload": "c" * 64,
"runtime_generation": 1,
"compaction_epoch": 0,
}
receipt = receipt_challenge.receipt_for(challenge, binding)
real_claude_entry = {
"message": {
"role": "assistant",
"content": [
{"type": "text", "text": receipt},
{
"type": "tool_use",
"id": "tool-1",
"name": "mcp__discord__reply",
"input": {"message": "promoted"},
},
],
}
}
extracted = observer_client.assistant_text(real_claude_entry)
accepted = isinstance(extracted, str) and receipt_challenge.is_verbatim_receipt(
extracted,
challenge,
binding,
)
self.assertIsNone(extracted)
self.assertFalse(accepted)
if __name__ == "__main__":
unittest.main()
@@ -0,0 +1,245 @@
#!/usr/bin/env python3
"""RED-first contracts for the shipped receipt challenge and observer seam."""
from __future__ import annotations
import base64
import copy
import hashlib
import importlib.util
import os
import tempfile
import unittest
from pathlib import Path
TOOLS = Path(__file__).parents[2] / "framework/tools/lease-broker"
DAEMON_PATH = TOOLS / "daemon.py"
FRAGMENTS_PATH = TOOLS / "normative_fragments.py"
OBSERVER_PATH = TOOLS / "receipt_observer.py"
def load_module(name: str, path: Path):
assert path.is_file(), f"shipped module is missing: {path}"
spec = importlib.util.spec_from_file_location(name, path)
if spec is None or spec.loader is None:
raise RuntimeError(f"unable to load {name}")
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
return module
DAEMON = load_module("lease_broker_receipt_daemon", DAEMON_PATH)
FRAGMENTS = load_module("lease_broker_normative_fragments", FRAGMENTS_PATH)
class BrokerFixture(unittest.TestCase):
def setUp(self) -> None:
self.temporary = tempfile.TemporaryDirectory()
root = Path(self.temporary.name)
os.chmod(root, 0o700)
self.peer = (os.getpid(), os.getuid(), os.getgid())
self.broker = DAEMON.Broker(DAEMON.StateStore(root / "state.json"))
registered = self.broker.handle(self.peer, {
"action": "register_anchor",
"runtime_generation": 7,
})
self.session_id = registered["session_id"]
self.assertIsInstance(self.session_id, str)
def tearDown(self) -> None:
self.temporary.cleanup()
def construction(self) -> tuple[dict[str, object], dict[str, object]]:
content = b"Constitution\n"
expected_sha256 = hashlib.sha256(content).hexdigest()
construction = {
"manifest_version": 1,
"generator_version": "wi5-receipt-test",
"fragments": [{
"source_id": "authority/constitution",
"content_base64": base64.b64encode(content).decode("ascii"),
"expected_sha256": expected_sha256,
}],
}
result = FRAGMENTS.build_payload(
manifest_version=construction["manifest_version"],
generator_version=construction["generator_version"],
fragments=[FRAGMENTS.NormativeFragment("authority/constitution", content, expected_sha256)],
)
self.assertEqual(result.injectionDecision, "ACCEPTED")
self.assertTrue(result.promotion)
return construction, {
"compaction_epoch": 3,
"request_epoch": 8,
"h_source": result.h_source,
"h_payload": result.h_payload,
"schema_version": 1,
}
def begin(self, binding: dict[str, object], construction: dict[str, object]) -> dict[str, object]:
response = self.broker.handle(self.peer, {
"action": "begin_verification",
"session_id": self.session_id,
"runtime_generation": 7,
"runtime": "pi",
"binding": binding,
"construction": construction,
})
self.assertEqual(response["state"], DAEMON.LEASE_PENDING)
self.assertIsInstance(response.get("receipt_challenge"), str)
self.assertIsInstance(response.get("receipt"), str)
return response
class BuildPayloadAdmissionTest(BrokerFixture):
def test_b3_forged_h_source_or_h_payload_is_refused_against_shipped_build_payload(self) -> None:
construction, trusted = self.construction()
for field in ("h_source", "h_payload"):
with self.subTest(field=field):
forged = dict(trusted)
forged[field] = "f" * 64
with self.assertRaisesRegex(DAEMON.BrokerFailure, "PAYLOAD_BINDING_MISMATCH"):
self.begin(forged, construction)
class ReceiptObserverTest(BrokerFixture):
def setUp(self) -> None:
super().setUp()
observers = load_module("lease_broker_test_observer", OBSERVER_PATH)
self.observer = observers.TestReceiptObserver()
self.broker = DAEMON.Broker(self.broker.store, observer=self.observer)
def record(self, message: str) -> None:
self.observer.record_latest_assistant_message(self.session_id, 7, message)
def observe(self, challenge: str, **untrusted: object) -> dict[str, object]:
return self.broker.handle(self.peer, {
"action": "observe_receipt",
"session_id": self.session_id,
"runtime_generation": 7,
"receipt_challenge": challenge,
**untrusted,
})
def promote(self, challenge: str) -> dict[str, object]:
return self.broker.handle(self.peer, {
"action": "promote_lease",
"session_id": self.session_id,
"runtime_generation": 7,
"receipt_challenge": challenge,
})
def test_b2_echoed_request_observation_is_refused_but_observer_source_promotes(self) -> None:
construction, binding = self.construction()
cycle = self.begin(binding, construction)
challenge = cycle["receipt_challenge"]
receipt = cycle["receipt"]
with self.assertRaisesRegex(DAEMON.BrokerFailure, "INVALID_RECEIPT"):
self.observe(challenge, latest_assistant_message=receipt)
with self.assertRaisesRegex(DAEMON.BrokerFailure, "RECEIPT_OBSERVATION_UNAVAILABLE"):
self.observe(challenge)
self.record(receipt)
self.assertEqual(self.observe(challenge)["state"], DAEMON.LEASE_PENDING_PROMOTION)
self.assertEqual(self.promote(challenge)["state"], DAEMON.LEASE_VERIFIED)
self.assertEqual(self.broker.handle(self.peer, {
"action": "authorize_tool",
"session_id": self.session_id,
"runtime_generation": 7,
"runtime": "pi",
"tool_name": "bash",
})["decision"], "allow")
def test_rejected_begin_keeps_revoke_first_fence_for_all_construction_refusals(self) -> None:
construction, binding = self.construction()
refusal_cases = {
"INVALID_CONSTRUCTION": {"bad": "construction"},
"PAYLOAD_CONSTRUCTION_REFUSED": {
**construction,
"fragments": [{
**construction["fragments"][0],
"expected_sha256": "0" * 64,
}],
},
"PAYLOAD_BINDING_MISMATCH": None,
}
for expected_code, rejected_construction in refusal_cases.items():
with self.subTest(expected_code=expected_code):
verified = self.begin(binding, construction)
self.record(verified["receipt"])
self.observe(verified["receipt_challenge"])
self.assertEqual(self.promote(verified["receipt_challenge"])["state"], DAEMON.LEASE_VERIFIED)
rejected_binding = copy.deepcopy(binding)
if expected_code == "PAYLOAD_BINDING_MISMATCH":
rejected_binding["h_payload"] = "f" * 64
rejected_construction = construction
with self.assertRaisesRegex(DAEMON.BrokerFailure, expected_code):
self.begin(rejected_binding, rejected_construction)
self.assertEqual(
self.broker.leases[self.session_id]["state"], DAEMON.LEASE_UNVERIFIED
)
denied = self.broker.handle(self.peer, {
"action": "authorize_tool",
"session_id": self.session_id,
"runtime_generation": 7,
"runtime": "pi",
"tool_name": "bash",
})
self.assertEqual(denied["decision"], "deny")
self.assertEqual(denied["state"], DAEMON.LEASE_UNVERIFIED)
def test_t26_stale_epoch_receipt_cannot_promote_against_shipped_binding(self) -> None:
construction, stale_binding = self.construction()
stale = self.begin(stale_binding, construction)
current_binding = dict(stale_binding)
current_binding["compaction_epoch"] = 4
current_binding["request_epoch"] = 9
current = self.begin(current_binding, construction)
self.assertNotEqual(stale["receipt_challenge"], current["receipt_challenge"])
self.record(stale["receipt"])
with self.assertRaisesRegex(DAEMON.BrokerFailure, "RECEIPT_MISMATCH"):
self.observe(current["receipt_challenge"])
with self.assertRaisesRegex(DAEMON.BrokerFailure, "INVALID_LEASE_TRANSITION"):
self.promote(current["receipt_challenge"])
def test_non_ascii_observation_is_mismatch_and_broker_keeps_serving(self) -> None:
construction, binding = self.construction()
cycle = self.begin(binding, construction)
self.record("I refuse—this is not the receipt")
with self.assertRaisesRegex(DAEMON.BrokerFailure, "RECEIPT_MISMATCH"):
self.observe(cycle["receipt_challenge"])
denied = self.broker.handle(self.peer, {
"action": "authorize_tool",
"session_id": self.session_id,
"runtime_generation": 7,
"runtime": "pi",
"tool_name": "bash",
})
self.assertEqual(denied["decision"], "deny")
self.assertEqual(denied["state"], DAEMON.LEASE_UNVERIFIED)
def test_t29_altered_model_hash_cannot_promote_against_shipped_binding(self) -> None:
construction, binding = self.construction()
cycle = self.begin(binding, construction)
expected = cycle["receipt"]
altered_hash = "f" * 64
self.assertNotEqual(altered_hash, cycle["binding"]["h_payload"])
altered = expected.replace(cycle["binding"]["h_payload"], altered_hash, 1)
self.assertNotEqual(altered, expected)
self.record(altered)
with self.assertRaisesRegex(DAEMON.BrokerFailure, "RECEIPT_MISMATCH"):
self.observe(cycle["receipt_challenge"])
with self.assertRaisesRegex(DAEMON.BrokerFailure, "INVALID_LEASE_TRANSITION"):
self.promote(cycle["receipt_challenge"])
if __name__ == "__main__":
unittest.main()
@@ -0,0 +1,303 @@
#!/usr/bin/env python3
"""Exit-semantics tests for the receipt observer Stop-hook client."""
from __future__ import annotations
import importlib.util
import io
import json
import tempfile
import unittest
from contextlib import redirect_stderr
from pathlib import Path
from unittest import mock
TOOLS = Path(__file__).parents[2] / "framework/tools/lease-broker"
CLIENT_PATH = TOOLS / "receipt-observer-client.py"
def load_client():
spec = importlib.util.spec_from_file_location("receipt_observer_client_test", CLIENT_PATH)
if spec is None or spec.loader is None:
raise RuntimeError("unable to load receipt-observer-client.py")
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
return module
CLIENT = load_client()
VALID_INPUT = json.dumps({"latest_assistant_message": "ordinary turn"}).encode()
DEEPLY_NESTED_JSON = b"[" * 2_000 + b"0" + b"]" * 2_000
ENVIRONMENT = {
"MOSAIC_RECEIPT_OBSERVER_SOCKET": "/unused/observer.sock",
"MOSAIC_LEASE_SESSION_ID": "a" * 64,
"MOSAIC_RUNTIME_GENERATION": "1",
}
class FakeObserverSocket:
def __init__(self, response: bytes) -> None:
self.response = response
def __enter__(self):
return self
def __exit__(self, *_args: object) -> None:
return None
def settimeout(self, _timeout: float) -> None:
return None
def connect(self, _path: str) -> None:
return None
def sendall(self, _payload: bytes) -> None:
return None
def shutdown(self, _how: int) -> None:
return None
def recv(self, _size: int) -> bytes:
response, self.response = self.response, b""
return response
class ReceiptObserverClientExitSemanticsTest(unittest.TestCase):
def run_client(
self,
*,
input_bytes: bytes = VALID_INPUT,
reply: dict[str, object] | None = None,
transport_error: OSError | None = None,
runtime: str = "pi",
) -> tuple[int, str, mock.Mock]:
request = mock.Mock(return_value=reply)
if transport_error is not None:
request.side_effect = transport_error
stderr = io.StringIO()
with (
mock.patch.object(CLIENT.sys, "stdin", io.BytesIO(input_bytes)),
mock.patch.object(CLIENT, "observer_request", request),
redirect_stderr(stderr),
):
arguments = ["--runtime", runtime]
if runtime == "claude":
arguments.append("--latest-entry")
result = CLIENT.main(arguments, environ=ENVIRONMENT)
return result, stderr.getvalue(), request
def test_claude_prefers_inline_last_assistant_message(self) -> None:
inline = "the just-finished assistant message"
input_bytes = json.dumps({
"last_assistant_message": inline,
"transcript_path": "/must/not/be/opened.jsonl",
}).encode()
result, stderr, request = self.run_client(
input_bytes=input_bytes,
reply={"ok": True},
runtime="claude",
)
self.assertEqual(result, 0)
self.assertEqual(stderr, "")
self.assertEqual(
request.call_args.args[1]["latest_assistant_message"],
inline,
)
def test_claude_falls_back_to_transcript_when_inline_field_is_absent(self) -> None:
with tempfile.TemporaryDirectory() as directory:
transcript = Path(directory) / "transcript.jsonl"
transcript.write_text(
json.dumps({
"message": {
"role": "assistant",
"content": [{"type": "text", "text": "fallback message"}],
}
})
+ "\n",
encoding="utf-8",
)
input_bytes = json.dumps({"transcript_path": str(transcript)}).encode()
result, stderr, request = self.run_client(
input_bytes=input_bytes,
reply={"ok": True},
runtime="claude",
)
self.assertEqual(result, 0)
self.assertEqual(stderr, "")
self.assertEqual(
request.call_args.args[1]["latest_assistant_message"],
"fallback message",
)
def test_claude_present_invalid_inline_field_fails_without_fallback(self) -> None:
fallback = mock.Mock(return_value="must not be used")
input_bytes = json.dumps({
"last_assistant_message": None,
"transcript_path": "/unused/transcript.jsonl",
}).encode()
with mock.patch.object(CLIENT, "claude_latest_entry", fallback):
result, stderr, request = self.run_client(
input_bytes=input_bytes,
reply={"ok": True},
runtime="claude",
)
self.assertEqual(result, 2)
self.assertIn("Mosaic receipt observer refused", stderr)
request.assert_not_called()
fallback.assert_not_called()
def test_claude_inline_message_size_guard_stays_enforced(self) -> None:
request = mock.Mock(return_value={"ok": True})
stderr = io.StringIO()
with (
mock.patch.object(CLIENT.sys, "stdin", io.BytesIO(b"{}")),
mock.patch.object(
CLIENT,
"read_json",
return_value={"last_assistant_message": "x" * (CLIENT.MAX_FRAME + 1)},
),
mock.patch.object(CLIENT, "observer_request", request),
redirect_stderr(stderr),
):
result = CLIENT.main(
["--runtime", "claude", "--latest-entry"],
environ=ENVIRONMENT,
)
self.assertEqual(result, 2)
self.assertIn("Mosaic receipt observer refused", stderr.getvalue())
request.assert_not_called()
def test_pi_still_posts_only_its_runtime_message(self) -> None:
result, stderr, request = self.run_client(reply={"ok": True})
self.assertEqual(result, 0)
self.assertEqual(stderr, "")
self.assertEqual(
request.call_args.args[1]["latest_assistant_message"],
"ordinary turn",
)
def test_nothing_pending_observation_refusal_is_benign(self) -> None:
result, stderr, request = self.run_client(
reply={"ok": False, "code": "OBSERVATION_UNAVAILABLE"}
)
self.assertEqual(result, 0)
self.assertEqual(stderr, "")
request.assert_called_once()
def test_success_reply_remains_successful(self) -> None:
result, stderr, _request = self.run_client(reply={"ok": True})
self.assertEqual(result, 0)
self.assertEqual(stderr, "")
def test_pending_cycle_auth_failure_stays_fail_closed(self) -> None:
result, _stderr, _request = self.run_client(
reply={"ok": False, "code": "ANCESTRY_MISMATCH"}
)
self.assertEqual(result, 2)
def test_transport_failure_stays_fail_closed(self) -> None:
result, stderr, _request = self.run_client(
transport_error=ConnectionRefusedError("observer unavailable")
)
self.assertEqual(result, 2)
self.assertIn("Mosaic receipt observer refused", stderr)
def test_parse_failure_stays_fail_closed(self) -> None:
result, stderr, request = self.run_client(input_bytes=b"{")
self.assertEqual(result, 2)
self.assertIn("Mosaic receipt observer refused", stderr)
request.assert_not_called()
def test_malformed_wire_replies_stay_fail_closed(self) -> None:
for response in (
b"not-json\n",
b'{"ok":true}',
b"{}\n{}\n",
b'{"ok":false,"code":"OBSERVATION_UNAVAILABLE"}\n\n',
b'{"ok":true,"ok":false,"code":"OBSERVATION_UNAVAILABLE"}\n',
DEEPLY_NESTED_JSON + b"\n",
b"x" * (CLIENT.MAX_FRAME + 1),
):
with self.subTest(response=response):
stderr = io.StringIO()
with (
mock.patch.object(CLIENT.sys, "stdin", io.BytesIO(VALID_INPUT)),
mock.patch.object(
CLIENT.socket,
"socket",
return_value=FakeObserverSocket(response),
),
redirect_stderr(stderr),
):
result = CLIENT.main(["--runtime", "pi"], environ=ENVIRONMENT)
self.assertEqual(result, 2)
self.assertIn("Mosaic receipt observer refused", stderr.getvalue())
def test_oversized_input_stays_fail_closed(self) -> None:
result, stderr, request = self.run_client(input_bytes=b"x" * (CLIENT.MAX_FRAME + 1))
self.assertEqual(result, 2)
self.assertIn("Mosaic receipt observer refused", stderr)
request.assert_not_called()
def test_deeply_nested_input_stays_fail_closed(self) -> None:
result, stderr, request = self.run_client(input_bytes=DEEPLY_NESTED_JSON)
self.assertEqual(result, 2)
self.assertIn("Mosaic receipt observer refused", stderr)
request.assert_not_called()
def test_json_recursion_failure_stays_fail_closed(self) -> None:
request = mock.Mock()
stderr = io.StringIO()
with (
mock.patch.object(CLIENT.sys, "stdin", io.BytesIO(VALID_INPUT)),
mock.patch.object(CLIENT, "observer_request", request),
mock.patch.object(
CLIENT.json,
"loads",
side_effect=RecursionError("maximum JSON nesting exceeded"),
),
redirect_stderr(stderr),
):
result = CLIENT.main(["--runtime", "pi"], environ=ENVIRONMENT)
self.assertEqual(result, 2)
self.assertIn("Mosaic receipt observer refused", stderr.getvalue())
request.assert_not_called()
def test_observation_unavailable_with_unexpected_fields_stays_fail_closed(self) -> None:
result, _stderr, _request = self.run_client(
reply={"ok": False, "code": "OBSERVATION_UNAVAILABLE", "unexpected": True}
)
self.assertEqual(result, 2)
def test_non_boolean_ok_values_stay_fail_closed(self) -> None:
for reply in (
{"ok": 1},
{"ok": 0, "code": "OBSERVATION_UNAVAILABLE"},
):
with self.subTest(reply=reply):
result, _stderr, _request = self.run_client(reply=reply)
self.assertEqual(result, 2)
if __name__ == "__main__":
unittest.main()
@@ -0,0 +1,223 @@
#!/usr/bin/env python3
"""RED-first adversarial Claude B1 gate contracts.
This private harness drives the shipped gate and daemon out of process. It
never contacts a live broker/runtime and executes a shell payload only after a
regression has already (incorrectly) received the recovery exemption.
"""
from __future__ import annotations
import json
import os
import re
import socket
import subprocess
import sys
import tempfile
import time
import unittest
from pathlib import Path
TOOLS = Path(__file__).parents[2] / "framework/tools/lease-broker"
FRAMEWORK = Path(__file__).parents[2] / "framework"
DAEMON = TOOLS / "daemon.py"
GATE = TOOLS / "mutator-gate.py"
RECOVERY = TOOLS / "recover-context.py"
SKILL = FRAMEWORK / "skills/mosaic-context-refresh/SKILL.md"
def request(socket_path: Path, value: dict[str, object]) -> dict[str, object]:
with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as connection:
connection.settimeout(3.0)
connection.connect(str(socket_path))
connection.sendall((json.dumps(value, separators=(",", ":")) + "\n").encode())
connection.shutdown(socket.SHUT_WR)
response = bytearray()
while True:
chunk = connection.recv(4096)
if not chunk:
break
response.extend(chunk)
if not response.endswith(b"\n") or response.count(b"\n") != 1:
raise AssertionError(f"unframed broker response: {bytes(response)!r}")
reply = json.loads(response[:-1])
if not isinstance(reply, dict):
raise AssertionError("broker response is not an object")
return reply
def wait_ready(process: subprocess.Popen[str], socket_path: Path) -> None:
deadline = time.monotonic() + 5.0
while time.monotonic() < deadline:
try:
with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as probe:
probe.settimeout(0.25)
probe.connect(str(socket_path))
return
except (ConnectionRefusedError, FileNotFoundError):
pass
if process.poll() is not None:
output = process.stdout.read() if process.stdout is not None else ""
raise RuntimeError(f"daemon exited before READY: {output}")
time.sleep(0.02)
raise TimeoutError("daemon did not create private socket")
class ClaudeRecoveryGateAdversarialTest(unittest.TestCase):
def setUp(self) -> None:
self.temporary = tempfile.TemporaryDirectory()
self.root = Path(self.temporary.name)
os.chmod(self.root, 0o700)
self.socket = self.root / "broker.sock"
self.daemon = subprocess.Popen(
[sys.executable, "-I", "-S", "-B", str(DAEMON), "--socket", str(self.socket), "--state", str(self.root / "state.json")],
stdin=subprocess.DEVNULL,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
text=True,
)
wait_ready(self.daemon, self.socket)
registered = request(self.socket, {"action": "register_anchor", "runtime_generation": 1})
self.session_id = registered["session_id"]
self.environment = {
**os.environ,
"MOSAIC_LEASE_BROKER_SOCKET": str(self.socket),
"MOSAIC_LEASE_SESSION_ID": self.session_id,
"MOSAIC_RUNTIME_GENERATION": "1",
"MOSAIC_LEASE_RUNTIME": "claude",
}
# This private path is only a gate-classifier identity; if a regression
# blesses a payload, bash invokes no installed/live recovery command.
self.recovery_path = self.root / "recover-context.py"
self.canonical = (
f"python3 {self.recovery_path} begin "
f"--construction {self.root / 'construction.json'} --compaction-epoch 0 --request-epoch 0"
)
def tearDown(self) -> None:
if self.daemon.poll() is None:
self.daemon.terminate()
try:
self.daemon.wait(timeout=3.0)
except subprocess.TimeoutExpired:
self.daemon.kill()
self.daemon.wait()
if self.daemon.stdout is not None:
self.daemon.stdout.close()
self.temporary.cleanup()
def gate(
self, command: str, recovery_command: Path | str | None = None
) -> subprocess.CompletedProcess[str]:
configured_recovery = self.recovery_path if recovery_command is None else recovery_command
return subprocess.run(
[
sys.executable,
"-I",
"-S",
"-B",
str(GATE),
"--runtime",
"claude",
"--recovery-command",
str(configured_recovery),
],
input=json.dumps({"tool_name": "Bash", "tool_input": {"command": command}}),
text=True,
capture_output=True,
env=self.environment,
check=False,
)
def command_for(self, position: str, value: str) -> str:
argv = [
"python3",
str(self.recovery_path),
"begin",
"--construction",
str(self.root / "construction.json"),
"--compaction-epoch",
"0",
"--request-epoch",
"0",
]
positions = {
"executable": 0,
"path": 1,
"phase": 2,
"construction_flag": 3,
"construction": 4,
"compaction_epoch_flag": 5,
"compaction_epoch": 6,
"request_epoch_flag": 7,
"request_epoch": 8,
}
try:
argv[positions[position]] = value
except KeyError as exc:
raise AssertionError(f"unknown argv position {position}") from exc
return " ".join(argv)
def test_canonical_literal_and_shipped_skill_invocation_are_ungated(self) -> None:
self.assertEqual(self.gate(self.canonical).returncode, 0)
source = SKILL.read_text(encoding="utf-8")
recovery_placeholder = "/absolute/path/to/mosaic/tools/lease-broker/recover-context.py"
construction_placeholder = "/absolute/path/to/mosaic-context-refresh-construction.json"
self.assertIn(recovery_placeholder, source)
self.assertIn(construction_placeholder, source)
resolved_recovery = "/opt/mosaic/tools/lease-broker/recover-context.py"
rendered = source.replace(recovery_placeholder, resolved_recovery).replace(
construction_placeholder, "/opt/mosaic/recovery/construction.json"
)
match = re.search(r"```bash\s*\n\s*(.*?)\n\s*```", rendered, flags=re.DOTALL)
self.assertIsNotNone(match)
shipped = match.group(1) if match is not None else ""
self.assertEqual(self.gate(shipped, resolved_recovery).returncode, 0)
def test_every_shell_active_vector_in_every_argv_position_falls_through_to_bash_deny(self) -> None:
positions = (
"executable",
"path",
"phase",
"construction_flag",
"construction",
"compaction_epoch_flag",
"compaction_epoch",
"request_epoch_flag",
"request_epoch",
)
marker = self.root / "PWNED"
marker_vector = f"$(touch${{IFS}}{marker})"
vectors = {
"command-substitution": marker_vector,
"backtick": f"`touch${{IFS}}{marker}`",
"process-substitution": f"<(touch${{IFS}}{marker})",
"parameter-expansion": "${IFS}",
"home-expansion": "${HOME}",
"arithmetic-expansion": "$((1+1))",
"brace-expansion": "{a,b}",
"tilde-expansion": "~",
"glob": "*",
"redirection": ">",
"semicolon": f";touch${{IFS}}{marker}",
"and": f"&&touch${{IFS}}{marker}",
"pipe": f"|touch${{IFS}}{marker}",
"embedded-newline": "literal\nnext",
"quoting-trick": "'literal'",
}
for position in positions:
for kind, vector in vectors.items():
with self.subTest(position=position, kind=kind):
marker.unlink(missing_ok=True)
command = self.command_for(position, vector)
gated = self.gate(command)
if gated.returncode == 0 and kind in {"command-substitution", "backtick", "process-substitution", "semicolon", "and", "pipe"}:
subprocess.run(["bash", "-c", command], cwd=self.root, env=self.environment, check=False)
self.assertEqual(gated.returncode, 2, f"unexpected recovery exemption: {command!r}")
self.assertFalse(marker.exists(), f"shell payload executed: {command!r}")
if __name__ == "__main__":
unittest.main()
@@ -0,0 +1,296 @@
#!/usr/bin/env python3
"""RED-first B1/B2 runtime-boundary contracts for constrained recovery.
Every runtime process in these tests is a fresh child against a private daemon
and Unix sockets. They do not activate a live Mosaic daemon or model stream.
"""
from __future__ import annotations
import base64
import hashlib
import json
import os
import socket
import subprocess
import sys
import tempfile
import time
import unittest
from pathlib import Path
TOOLS = Path(__file__).parents[2] / "framework/tools/lease-broker"
FRAMEWORK = Path(__file__).parents[2] / "framework"
DAEMON = TOOLS / "daemon.py"
GATE = TOOLS / "mutator-gate.py"
RECOVERY = TOOLS / "recover-context.py"
OBSERVER_CLIENT = TOOLS / "receipt-observer-client.py"
CLAUDE_SETTINGS = FRAMEWORK / "runtime/claude/settings.json"
PI_EXTENSION = FRAMEWORK / "runtime/pi/mosaic-extension.ts"
def request(socket_path: Path, value: dict[str, object]) -> dict[str, object]:
deadline = time.monotonic() + 5.0
while True:
with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as connection:
connection.settimeout(3.0)
try:
connection.connect(str(socket_path))
except ConnectionRefusedError:
if time.monotonic() >= deadline:
raise
time.sleep(0.02)
continue
connection.sendall((json.dumps(value, separators=(",", ":")) + "\n").encode())
connection.shutdown(socket.SHUT_WR)
response = bytearray()
while True:
chunk = connection.recv(4096)
if not chunk:
break
response.extend(chunk)
break
if not response.endswith(b"\n") or response.count(b"\n") != 1:
raise AssertionError(f"unframed broker response: {bytes(response)!r}")
reply = json.loads(response[:-1])
if not isinstance(reply, dict):
raise AssertionError("broker response is not an object")
return reply
def wait_ready(process: subprocess.Popen[str], socket_path: Path) -> None:
deadline = time.monotonic() + 5.0
while time.monotonic() < deadline:
try:
with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as probe:
probe.settimeout(0.25)
probe.connect(str(socket_path))
return
except (ConnectionRefusedError, FileNotFoundError):
pass
if process.poll() is not None:
output = process.stdout.read() if process.stdout is not None else ""
raise RuntimeError(f"daemon exited before READY: {output}")
time.sleep(0.02)
raise TimeoutError("daemon did not create private broker socket")
class RuntimeBoundaryFixture(unittest.TestCase):
def setUp(self) -> None:
self.temporary = tempfile.TemporaryDirectory()
self.root = Path(self.temporary.name)
os.chmod(self.root, 0o700)
self.socket = self.root / "broker.sock"
self.observer_socket = self.root / "observer.sock"
self.state = self.root / "state.json"
self.children: list[subprocess.Popen[str]] = []
def tearDown(self) -> None:
for child in self.children:
if child.poll() is None:
child.terminate()
try:
child.wait(timeout=3.0)
except subprocess.TimeoutExpired:
child.kill()
child.wait()
if child.stdout is not None:
child.stdout.close()
self.temporary.cleanup()
def start_daemon(self, *, production_observer: bool) -> None:
arguments = [sys.executable, "-I", "-S", "-B", str(DAEMON), "--socket", str(self.socket), "--state", str(self.state)]
if production_observer:
arguments.extend(["--observer-socket", str(self.observer_socket)])
process = subprocess.Popen(
arguments,
stdin=subprocess.DEVNULL,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
text=True,
)
self.children.append(process)
wait_ready(process, self.socket)
def register(self) -> str:
reply = request(self.socket, {"action": "register_anchor", "runtime_generation": 1})
session_id = reply.get("session_id")
self.assertTrue(reply.get("ok"))
self.assertIsInstance(session_id, str)
return session_id
def environment(self, session_id: str) -> dict[str, str]:
return {
**os.environ,
"MOSAIC_LEASE_BROKER_SOCKET": str(self.socket),
"MOSAIC_RECEIPT_OBSERVER_SOCKET": str(self.observer_socket),
"MOSAIC_LEASE_SESSION_ID": session_id,
"MOSAIC_RUNTIME_GENERATION": "1",
"MOSAIC_LEASE_RUNTIME": "pi",
}
def construction(self) -> Path:
content = b"WI-6 B2 production observer fixture\n"
path = self.root / "construction.json"
path.write_text(json.dumps({
"manifest_version": 1,
"generator_version": "wi6-repair-runtime-boundary",
"fragments": [{
"source_id": "authority/wi6-repair",
"content_base64": base64.b64encode(content).decode("ascii"),
"expected_sha256": hashlib.sha256(content).hexdigest(),
}],
}), encoding="utf-8")
os.chmod(path, 0o600)
return path
class RecoveryRuntimeBoundaryTest(RuntimeBoundaryFixture):
def test_b1_claude_exact_recovery_command_is_invocable_unverified_but_bash_is_not(self) -> None:
self.start_daemon(production_observer=False)
session_id = self.register()
environment = self.environment(session_id)
recovery_command = f"python3 {RECOVERY} complete"
mapped = subprocess.run(
[sys.executable, "-I", "-S", "-B", str(GATE), "--runtime", "claude", "--recovery-command", str(RECOVERY)],
input=json.dumps({"tool_name": "Bash", "tool_input": {"command": recovery_command}}),
text=True,
capture_output=True,
env={**environment, "MOSAIC_LEASE_RUNTIME": "claude"},
check=False,
)
self.assertEqual(mapped.returncode, 0, mapped.stderr)
mapped_begin = subprocess.run(
[sys.executable, "-I", "-S", "-B", str(GATE), "--runtime", "claude", "--recovery-command", str(RECOVERY)],
input=json.dumps({
"tool_name": "Bash",
"tool_input": {
"command": (
f"python3 {RECOVERY} begin --construction /tmp/construction.json "
"--compaction-epoch 1 --request-epoch 1"
)
},
}),
text=True,
capture_output=True,
env={**environment, "MOSAIC_LEASE_RUNTIME": "claude"},
check=False,
)
self.assertEqual(mapped_begin.returncode, 0, mapped_begin.stderr)
ordinary_bash = subprocess.run(
[sys.executable, "-I", "-S", "-B", str(GATE), "--runtime", "claude", "--recovery-command", str(RECOVERY)],
input=json.dumps({"tool_name": "Bash", "tool_input": {"command": "echo not-recovery"}}),
text=True,
capture_output=True,
env={**environment, "MOSAIC_LEASE_RUNTIME": "claude"},
check=False,
)
self.assertEqual(ordinary_bash.returncode, 2)
def test_b1_pi_registers_only_the_broker_exempt_recovery_tool(self) -> None:
extension = PI_EXTENSION.read_text(encoding="utf-8")
self.assertIn("const RECOVERY_TOOL = 'mosaic_context_recover'", extension)
self.assertIn("name: RECOVERY_TOOL", extension)
self.assertIn("checkPiMutatorGate(RECOVERY_TOOL)", extension)
self.assertNotIn("toolName === 'bash' ? RECOVERY_TOOL", extension)
def test_b2_production_observer_promotes_over_private_transport_and_rejects_broker_supplied_message(self) -> None:
self.start_daemon(production_observer=True)
session_id = self.register()
environment = self.environment(session_id)
begin = subprocess.run(
[sys.executable, "-I", "-S", "-B", str(RECOVERY), "begin", "--construction", str(self.construction()), "--compaction-epoch", "1", "--request-epoch", "1"],
text=True,
capture_output=True,
env=environment,
check=False,
)
self.assertEqual(begin.returncode, 0, begin.stderr)
cycle = json.loads(begin.stdout)
receipt = cycle.get("receipt")
self.assertIsInstance(receipt, str)
# S1: production transport is not a broker request field. The public
# broker endpoint keeps rejecting caller-supplied assistant evidence.
rejected_begin = request(self.socket, {
"action": "begin_recovery",
"session_id": session_id,
"runtime_generation": 1,
"receipt": receipt,
})
self.assertEqual(rejected_begin, {"ok": False, "code": "INVALID_RECOVERY_REQUEST"})
rejected_observe = request(self.socket, {
"action": "observe_receipt",
"session_id": session_id,
"runtime_generation": 1,
"receipt_challenge": cycle["receipt_challenge"],
"latest_assistant_message": receipt,
})
self.assertEqual(rejected_observe, {"ok": False, "code": "INVALID_RECEIPT"})
rejected_complete = request(self.socket, {
"action": "complete_recovery",
"session_id": session_id,
"runtime_generation": 1,
"latest_assistant_message": receipt,
})
self.assertEqual(rejected_complete, {"ok": False, "code": "INVALID_RECOVERY_REQUEST"})
recorded = subprocess.run(
[sys.executable, "-I", "-S", "-B", str(OBSERVER_CLIENT), "--runtime", "pi"],
input=json.dumps({"latest_assistant_message": receipt}),
text=True,
capture_output=True,
env=environment,
check=False,
)
self.assertEqual(recorded.returncode, 0, recorded.stderr)
complete = subprocess.run(
[sys.executable, "-I", "-S", "-B", str(RECOVERY), "complete"],
text=True,
capture_output=True,
env=environment,
check=False,
)
self.assertEqual(complete.returncode, 0, complete.stderr)
self.assertEqual(json.loads(complete.stdout).get("state"), "VERIFIED")
# The independent Claude transport selects one latest assistant entry
# from its hook transcript; it is not a Pi/message_end fallback.
claude_environment = {**environment, "MOSAIC_LEASE_RUNTIME": "claude"}
claude_begin = subprocess.run(
[sys.executable, "-I", "-S", "-B", str(RECOVERY), "begin", "--construction", str(self.construction()), "--compaction-epoch", "2", "--request-epoch", "2"],
text=True,
capture_output=True,
env=claude_environment,
check=False,
)
self.assertEqual(claude_begin.returncode, 0, claude_begin.stderr)
claude_receipt = json.loads(claude_begin.stdout)["receipt"]
transcript = self.root / "claude-transcript.jsonl"
transcript.write_text(
json.dumps({"message": {"role": "assistant", "content": claude_receipt}}) + "\n",
encoding="utf-8",
)
claude_recorded = subprocess.run(
[sys.executable, "-I", "-S", "-B", str(OBSERVER_CLIENT), "--runtime", "claude", "--latest-entry"],
input=json.dumps({"transcript_path": str(transcript)}),
text=True,
capture_output=True,
env=claude_environment,
check=False,
)
self.assertEqual(claude_recorded.returncode, 0, claude_recorded.stderr)
claude_complete = subprocess.run(
[sys.executable, "-I", "-S", "-B", str(RECOVERY), "complete"],
text=True,
capture_output=True,
env=claude_environment,
check=False,
)
self.assertEqual(claude_complete.returncode, 0, claude_complete.stderr)
self.assertEqual(json.loads(claude_complete.stdout).get("state"), "VERIFIED")
if __name__ == "__main__":
unittest.main()
@@ -0,0 +1,79 @@
#!/usr/bin/env python3
"""D29 contracts: no lease is a no-op success; half-provisioned still fails closed."""
from __future__ import annotations
import importlib.util
import unittest
from pathlib import Path
TOOLS = Path(__file__).parents[2] / "framework/tools/lease-broker"
REVOKE_PATH = TOOLS / "revoke-lease.py"
_spec = importlib.util.spec_from_file_location("revoke_lease", REVOKE_PATH)
assert _spec and _spec.loader
revoke_lease = importlib.util.module_from_spec(_spec)
import sys as _sys
_sys.path.insert(0, str(TOOLS))
_spec.loader.exec_module(revoke_lease)
ARGV = ["--runtime", "claude", "--reason", "pre-compact"]
VALID_SESSION = "a" * 64
def _explode(*_args, **_kwargs):
raise AssertionError("broker must not be contacted when no lease is held")
class RevokeWithoutLease(unittest.TestCase):
def test_no_lease_variables_is_a_noop_success(self) -> None:
"""The D29 case: bare-launched session, nothing to revoke, must not deny."""
self.assertEqual(
revoke_lease.main(ARGV, environ={}, request=_explode),
0,
)
def test_no_lease_does_not_contact_the_broker(self) -> None:
"""A no-op must be vacuous: no socket, no generation bump, no transport."""
revoke_lease.main(ARGV, environ={"HOME": "/nonexistent"}, request=_explode)
def test_socket_without_session_still_fails_closed(self) -> None:
"""Half-provisioned is misconfiguration, not absence. Fail-closed stands."""
self.assertEqual(
revoke_lease.main(
ARGV,
environ={"MOSAIC_LEASE_BROKER_SOCKET": "/tmp/nonexistent.sock"},
request=_explode,
),
2,
)
def test_session_without_socket_still_fails_closed(self) -> None:
"""The mirror case, so the guard cannot be satisfied by either half alone."""
self.assertEqual(
revoke_lease.main(
ARGV,
environ={"MOSAIC_LEASE_SESSION_ID": VALID_SESSION},
request=_explode,
),
2,
)
def test_empty_string_counts_as_absent(self) -> None:
"""An exported-but-empty variable is not a lease."""
self.assertEqual(
revoke_lease.main(
ARGV,
environ={
"MOSAIC_LEASE_BROKER_SOCKET": "",
"MOSAIC_LEASE_SESSION_ID": "",
},
request=_explode,
),
0,
)
if __name__ == "__main__":
unittest.main()
@@ -0,0 +1,333 @@
#!/usr/bin/env python3
"""Standard-library edge tests for lease-broker atomic state persistence."""
from __future__ import annotations
import copy
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")), [])
def test_oversized_payload_is_refused_before_replacing_state(self) -> None:
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
store = self.make_store(root)
store.value.pop("marker")
store.commit()
durable = store.path.read_bytes()
store.value["oversized"] = "x" * DAEMON.MAX_STATE
with patch.object(DAEMON.os, "open", wraps=os.open) as mocked_open:
with self.assertRaisesRegex(DAEMON.BrokerFailure, "STATE_TOO_LARGE"):
store.commit()
self.assertEqual(mocked_open.call_count, 0)
self.assertEqual(store.path.read_bytes(), durable)
self.assertEqual(list(root.glob(".*.tmp")), [])
class StateStoreValidationTest(unittest.TestCase):
@staticmethod
def binding() -> dict[str, object]:
return {
"compaction_epoch": 0,
"request_epoch": 0,
"h_source": "a" * 64,
"h_payload": "b" * 64,
"schema_version": 1,
}
def test_impossible_or_over_capacity_token_state_is_rejected(self) -> None:
session_id = "1" * 64
session = {
"anchor_pid": 123,
"anchor_starttime": "456",
"runtime_generation": 2,
}
live_token = {
"session_id": session_id,
"runtime_generation": 2,
"binding": self.binding(),
"consumed": False,
}
cases = {
"stale generation": {
"2" * 64: {**live_token, "runtime_generation": 1},
},
"consumed token": {
"2" * 64: {**live_token, "consumed": True},
},
"over capacity": {
f"{index:064x}": copy.deepcopy(live_token)
for index in range(DAEMON.MAX_PENDING_TOKENS + 1)
},
}
for label, tokens in cases.items():
with self.subTest(label=label), tempfile.TemporaryDirectory() as directory:
root = Path(directory)
os.chmod(root, 0o700)
state_path = root / "state.json"
state_path.write_text(json.dumps({
"version": 1,
"sessions": {session_id: session},
"tokens": tokens,
}))
os.chmod(state_path, 0o600)
with self.assertRaisesRegex(DAEMON.BrokerFailure, "STATE_INTEGRITY"):
DAEMON.StateStore(state_path)
class BrokerBehaviorTest(unittest.TestCase):
def make_broker(self, root: Path):
os.chmod(root, 0o700)
return DAEMON.Broker(DAEMON.StateStore(root / "state.json"))
@staticmethod
def binding() -> dict[str, object]:
return {
"compaction_epoch": 0,
"request_epoch": 0,
"h_source": "a" * 64,
"h_payload": "b" * 64,
"schema_version": 1,
}
def register(self, broker, generation: int = 1) -> str:
response = broker.handle((123, 1000, 1000), {
"action": "register_anchor",
"runtime_generation": generation,
})
return response["session_id"]
def mint(self, broker, session_id: str, generation: int = 1) -> str:
response = broker.handle((123, 1000, 1000), {
"action": "mint_token",
"session_id": session_id,
"runtime_generation": generation,
"binding": self.binding(),
})
return response["token"]
def consume(self, broker, session_id: str, token: str, generation: int = 1):
return broker.handle((123, 1000, 1000), {
"action": "consume_token",
"session_id": session_id,
"runtime_generation": generation,
"token": token,
})
def test_anchor_generation_bump_reuses_session_and_revokes_token(self) -> None:
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
broker = self.make_broker(root)
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": self.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.assertNotIn(minted["token"], broker.store.tokens())
restarted = self.make_broker(root)
self.assertEqual(restarted.store.tokens(), {})
self.assertEqual(
restarted.store.sessions()[first["session_id"]]["runtime_generation"], 2
)
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,
})
def test_successful_consume_deletes_token_and_replay_is_refused(self) -> None:
with tempfile.TemporaryDirectory() as directory, (
patch.object(DAEMON, "proc_node", return_value={"pid": 123, "ppid": 1, "starttime": "456"})
), patch.object(DAEMON, "verified_ancestry", return_value=True):
broker = self.make_broker(Path(directory))
session_id = self.register(broker)
token = self.mint(broker, session_id)
self.assertEqual(self.consume(broker, session_id, token), {"ok": True})
self.assertNotIn(token, broker.store.tokens())
with self.assertRaisesRegex(DAEMON.BrokerFailure, "TOKEN_REPLAY"):
self.consume(broker, session_id, token)
def test_normal_cycles_remain_bounded_and_restartable(self) -> None:
with tempfile.TemporaryDirectory() as directory, (
patch.object(DAEMON, "proc_node", return_value={"pid": 123, "ppid": 1, "starttime": "456"})
), patch.object(DAEMON, "verified_ancestry", return_value=True):
root = Path(directory)
broker = self.make_broker(root)
session_id = self.register(broker)
for _ in range(DAEMON.MAX_PENDING_TOKENS * 3):
self.consume(broker, session_id, self.mint(broker, session_id))
self.assertEqual(broker.store.tokens(), {})
self.assertLess((root / "state.json").stat().st_size, DAEMON.MAX_STATE)
restarted = self.make_broker(root)
self.assertEqual(restarted.store.tokens(), {})
self.assertIn(session_id, restarted.store.sessions())
def test_pending_token_capacity_refusal_does_not_mutate_state(self) -> None:
with tempfile.TemporaryDirectory() as directory, (
patch.object(DAEMON, "proc_node", return_value={"pid": 123, "ppid": 1, "starttime": "456"})
), patch.object(DAEMON, "verified_ancestry", return_value=True):
root = Path(directory)
broker = self.make_broker(root)
session_id = self.register(broker)
for _ in range(DAEMON.MAX_PENDING_TOKENS):
self.mint(broker, session_id)
before = copy.deepcopy(broker.store.value)
durable = broker.store.path.read_bytes()
with self.assertRaisesRegex(DAEMON.BrokerFailure, "TOKEN_CAPACITY"):
self.mint(broker, session_id)
self.assertEqual(broker.store.value, before)
self.assertEqual(broker.store.path.read_bytes(), durable)
def test_directory_fsync_failure_poisoned_store_cannot_continue(self) -> None:
with tempfile.TemporaryDirectory() as directory, patch.object(
DAEMON,
"proc_node",
return_value={"pid": 123, "ppid": 1, "starttime": "456"},
):
root = Path(directory)
broker = self.make_broker(root)
real_fsync = os.fsync
def fail_directory_fsync(descriptor: int) -> None:
if os.path.isdir(f"/proc/self/fd/{descriptor}"):
raise OSError("directory fsync failed")
real_fsync(descriptor)
with patch.object(DAEMON.os, "fsync", side_effect=fail_directory_fsync):
with self.assertRaisesRegex(
DAEMON.StateCommitUncertain, "STATE_COMMIT_UNCERTAIN"
):
self.register(broker)
durable = json.loads(broker.store.path.read_text())
self.assertEqual(broker.store.value, durable)
self.assertTrue(broker.store.poisoned)
before = copy.deepcopy(broker.store.value)
with self.assertRaisesRegex(
DAEMON.StateCommitUncertain, "STATE_COMMIT_UNCERTAIN"
):
broker.handle((123, 1000, 1000), {
"action": "register_anchor",
"runtime_generation": 2,
})
self.assertEqual(broker.store.value, before)
def test_commit_failures_before_replace_roll_back_every_broker_mutation(self) -> None:
with tempfile.TemporaryDirectory() as directory, (
patch.object(DAEMON, "proc_node", return_value={"pid": 123, "ppid": 1, "starttime": "456"})
), patch.object(DAEMON, "verified_ancestry", return_value=True):
root = Path(directory)
broker = self.make_broker(root)
session_id = self.register(broker)
token = self.mint(broker, session_id)
def assert_rollback(request: dict[str, object]) -> None:
before = copy.deepcopy(broker.store.value)
durable = broker.store.path.read_bytes()
with patch.object(broker.store, "commit", side_effect=OSError("fsync failed")):
with self.assertRaisesRegex(OSError, "fsync failed"):
broker.handle((123, 1000, 1000), request)
self.assertEqual(broker.store.value, before)
self.assertEqual(broker.store.path.read_bytes(), durable)
assert_rollback({"action": "register_anchor", "runtime_generation": 2})
assert_rollback({
"action": "mint_token", "session_id": session_id,
"runtime_generation": 1, "binding": self.binding(),
})
assert_rollback({
"action": "consume_token", "session_id": session_id,
"runtime_generation": 1, "token": token,
})
with tempfile.TemporaryDirectory() as second_directory:
second = self.make_broker(Path(second_directory))
with patch.object(second.store, "commit", side_effect=OSError("fsync failed")):
with self.assertRaisesRegex(OSError, "fsync failed"):
self.register(second)
self.assertEqual(
second.store.value, {"version": 1, "sessions": {}, "tokens": {}}
)
self.assertFalse(second.store.path.exists())
if __name__ == "__main__":
unittest.main()