401 lines
35 KiB
TypeScript
401 lines
35 KiB
TypeScript
import assert from 'node:assert/strict';
|
|
import { createECDH, generateKeyPairSync, randomUUID, verify } from 'node:crypto';
|
|
import { EventEmitter, once } from 'node:events';
|
|
import { createServer } from 'node:http';
|
|
import { execFileSync } from 'node:child_process';
|
|
import { mkdtempSync, rmSync } from 'node:fs';
|
|
import { tmpdir } from 'node:os';
|
|
import { dirname, join, resolve } from 'node:path';
|
|
import test from 'node:test';
|
|
import { fileURLToPath } from 'node:url';
|
|
import Fastify from 'fastify';
|
|
import sensible from '@fastify/sensible';
|
|
import webpush from 'web-push';
|
|
import type { Context, Registration } from '../src/agent-push-contract.js';
|
|
import type { PushConfig } from '../src/agent-push-config.js';
|
|
import type { SendResult } from '../src/agent-push-transports.js';
|
|
|
|
const root = resolve(dirname(fileURLToPath(import.meta.url)), '..');
|
|
const directory = mkdtempSync(join(tmpdir(), 'sybil-push-test-'));
|
|
process.env.DATABASE_URL = `file:${join(directory, 'push.db')}`;
|
|
process.env.ADMIN_TOKEN = 'push-test-bearer-0123456789';
|
|
process.env.AGENT_PUSH_ENABLED = 'false';
|
|
for (const key of ['AGENT_PUSH_APNS_KEY_FILE','AGENT_PUSH_APNS_KEY_ID','AGENT_PUSH_APNS_TEAM_ID','AGENT_PUSH_VAPID_FILE']) delete process.env[key];
|
|
execFileSync(process.execPath, [join(root, 'node_modules/prisma/build/index.js'), 'migrate', 'deploy'], { cwd: root, env: process.env, stdio: 'pipe' });
|
|
const [{ prisma }, { AgentPushCoordinator, registerPushRoutes }, { createPushSender, isPublicAddress, createAPNsRequest, createWebRequest }, { validatePushEndpoint }, { env }] = await Promise.all([
|
|
import('../src/db.js'), import('../src/agent-push.js'), import('../src/agent-push-transports.js'), import('../src/agent-push-contract.js'), import('../src/env.js'),
|
|
]);
|
|
const apns: Registration = { transport: 'apns', enabled: true, token: 'a'.repeat(64), environment: 'sandbox', topic: 'net.buzzert.sybil2' };
|
|
const config: PushConfig = { enabled: true, adminToken: process.env.ADMIN_TOKEN,
|
|
apns: { key: 'fixture-unused-by-injected-sender', keyId: 'ABCDEFGHIJ', teamId: 'DQQH5H6GBD', topic: apns.topic, environment: 'sandbox' },
|
|
web: { publicKey: 'fixture-unused', privateKey: 'fixture-unused', subject: 'mailto:tests@example.com' } };
|
|
function subscription(): Registration {
|
|
const curve = createECDH('prime256v1'); curve.generateKeys();
|
|
return { transport: 'web', enabled: true, subscription: { endpoint: `https://web.push.apple.com/test/${randomUUID()}`, expirationTime: null,
|
|
keys: { p256dh: curve.getPublicKey().toString('base64url'), auth: Buffer.alloc(16, 1).toString('base64url') } } };
|
|
}
|
|
async function fixture(options: { warmupMs?: number; config?: PushConfig } = {}) {
|
|
let now = Date.now();
|
|
const source = randomUUID();
|
|
let context: Context = { version: 1, sourceId: source, conversationId: '1', cursor: '10', hasMore: false,
|
|
latestUserMessageAt: new Date(now - 30_000).toISOString(), nativeReadCursor: '0', uncertain: false,
|
|
webPresence: { active: false, unknown: false, observedAt: new Date(now).toISOString() }, candidates: [] };
|
|
const sent: { registration: Registration; identity: any }[] = [], reads: any[] = [];
|
|
let result: SendResult = { outcome: 'accepted' }, failure = false;
|
|
const manager = new AgentPushCoordinator({ db: prisma, config: options.config || config, now: () => now,
|
|
warmupMs: options.warmupMs ?? 0, context: async after => { if (failure) throw new Error('upstream secret must not leak');
|
|
return { ...context, webPresence: { ...context.webPresence, observedAt: new Date(now).toISOString() }, candidates: after === undefined ? [] : context.candidates.filter(c => Number(c.cursor) > Number(after)) }; },
|
|
send: async (registration, identity) => { sent.push({ registration, identity }); return result; }, broadcast: value => reads.push(value) });
|
|
await manager.initialize();
|
|
const installationId = randomUUID(); await manager.register(installationId, apns); await manager.tick();
|
|
return { manager, installationId, source, sent, reads, get now() { return now; }, advance(ms: number) { now += ms; },
|
|
result(value: SendResult) { result = value; }, failure(value: boolean) { failure = value; },
|
|
context(value: Partial<Context>) { context = { ...context, ...value }; },
|
|
candidate(cursor = '11', trigger: Context['candidates'][number]['trigger'] = 'user') { context = { ...context, cursor,
|
|
candidates: [...context.candidates, { cursor, messageId: `outbound:${cursor}`, createdAt: new Date(now).toISOString(), trigger }] }; },
|
|
};
|
|
}
|
|
test.beforeEach(async () => {
|
|
for (const table of ['AgentPushDelivery','AgentPushConnection','AgentPushPresence','AgentPushInstallation','AgentPushCoordinator','AgentReadState']) await prisma.$executeRawUnsafe(`DELETE FROM "${table}"`);
|
|
});
|
|
test.after(async () => { await prisma.$disconnect(); rmSync(directory, { recursive: true, force: true }); });
|
|
|
|
test('first source bootstrap never alerts history; later durable candidates debounce, survive restart and deduplicate', async () => {
|
|
const f = await fixture(); assert.equal((await f.manager.config()).sourceId, f.source); assert.equal(f.sent.length, 0);
|
|
f.candidate(); await f.manager.tick(); assert.equal(f.sent.length, 0);
|
|
f.advance(2_000); await f.manager.tick(); assert.equal(f.sent.length, 1);
|
|
await f.manager.tick(); assert.equal(f.sent.length, 1);
|
|
assert.deepEqual(f.sent[0].identity, { version: 1, sourceId: f.source, kind: 'message', messageId: 'outbound:11', cursor: '11' });
|
|
assert.equal((await prisma.agentPushDelivery.findFirst())?.status, 'accepted');
|
|
const restarted = new AgentPushCoordinator({ ...f.manager.options, warmupMs: 0 }); await restarted.tick(); assert.equal(f.sent.length, 1);
|
|
});
|
|
|
|
test('all tabs veto while focused; hidden heartbeats and stale sequences cannot extend short transition grace', async () => {
|
|
const f = await fixture(), tab1 = randomUUID(), tab2 = randomUUID();
|
|
await f.manager.presence(tab1, { sequence: 1, visibility: 'hidden', focused: false, interacted: false });
|
|
await f.manager.presence(tab2, { sequence: 1, visibility: 'visible', focused: true, interacted: false });
|
|
f.candidate(); await f.manager.tick(); f.advance(2_000); await f.manager.tick(); assert.equal(f.sent.length, 0, 'focused but idle still vetoes');
|
|
let receipt = await prisma.agentPushDelivery.findFirstOrThrow();
|
|
assert.equal(receipt.lastSuppressionReason, 'foreground_focused'); assert.equal(receipt.lastSuppressionClientKind, 'web');
|
|
await f.manager.presence(tab2, { sequence: 2, visibility: 'hidden', focused: false, interacted: true });
|
|
f.advance(2_000); await f.manager.tick(); assert.equal(f.sent.length, 0, 'recent input vetoes blur');
|
|
await f.manager.presence(tab2, { sequence: 1, visibility: 'visible', focused: true, interacted: false });
|
|
assert.equal((await prisma.agentPushPresence.findUniqueOrThrow({ where: { id: tab2 } })).sequence, 2);
|
|
await f.manager.presence(tab2, { sequence: 3, visibility: 'hidden', focused: false, interacted: true });
|
|
f.advance(4_000); await f.manager.tick(); assert.equal(f.sent.length, 1, 'hidden receipts do not renew attention');
|
|
receipt = await prisma.agentPushDelivery.findFirstOrThrow();
|
|
assert.equal(receipt.lastSuppressionReason, 'inactive_grace'); assert.ok(receipt.lastSuppressedAt);
|
|
});
|
|
|
|
test('blur and visible recent interaction have short grace; fresh focus renews a live lease', async () => {
|
|
const f = await fixture(), tab = randomUUID();
|
|
await f.manager.presence(tab, { sequence: 1, visibility: 'visible', focused: true, interacted: false });
|
|
f.candidate(); await f.manager.tick(); f.advance(2_000);
|
|
await f.manager.presence(tab, { sequence: 2, visibility: 'visible', focused: false, interacted: true });
|
|
await f.manager.tick(); assert.equal(f.sent.length, 0);
|
|
assert.equal((await prisma.agentPushDelivery.findFirstOrThrow()).lastSuppressionReason, 'recent_interaction');
|
|
f.advance(3_000); await f.manager.presence(tab, { sequence: 3, visibility: 'visible', focused: false, interacted: false });
|
|
f.advance(3_000); await f.manager.tick(); assert.equal(f.sent.length, 1);
|
|
await f.manager.presence(tab, { sequence: 4, visibility: 'visible', focused: true, interacted: false });
|
|
f.candidate('12'); await f.manager.tick(); f.advance(45_000);
|
|
await f.manager.presence(tab, { sequence: 5, visibility: 'visible', focused: true, interacted: false });
|
|
f.advance(20_000); await f.manager.tick(); assert.equal(f.sent.length, 1, 'fresh foreground heartbeat extends attention');
|
|
f.advance(41_000); await f.manager.tick(); assert.equal(f.sent.length, 2, 'stale focus expires');
|
|
});
|
|
|
|
test('anonymous introduction grace survives heartbeat, new UUID reconnects and coordinator restart without renewing', async () => {
|
|
const f = await fixture(), connection = randomUUID();
|
|
await f.manager.connect(connection); f.candidate(); await f.manager.tick(); f.advance(2_000); await f.manager.tick();
|
|
assert.equal(f.sent.length, 0);
|
|
const first = await prisma.agentPushCoordinator.findUniqueOrThrow({ where: { id: 'root' } });
|
|
assert.equal((await prisma.agentPushDelivery.findFirstOrThrow()).lastSuppressionReason, 'unknown_client_grace');
|
|
await f.manager.heartbeat(connection); await f.manager.disconnect(connection);
|
|
await f.manager.connect(randomUUID(), 'web', randomUUID()); f.advance(4_000); await f.manager.tick(); assert.equal(f.sent.length, 1);
|
|
f.candidate('12'); await f.manager.tick(); f.advance(2_000);
|
|
const restarted = new AgentPushCoordinator({ ...f.manager.options, warmupMs: 0 });
|
|
await restarted.connect(randomUUID()); await restarted.tick(); assert.equal(f.sent.length, 2);
|
|
assert.deepEqual((await prisma.agentPushCoordinator.findUniqueOrThrow({ where: { id: 'root' } })).unknownSince, first.unknownSince);
|
|
// Spending the anonymous budget never disables an independent authenticated browser lease.
|
|
await f.manager.presence(randomUUID(), { sequence: 1, visibility: 'visible', focused: true, interacted: false });
|
|
f.candidate('13'); await f.manager.tick(); f.advance(2_000); await f.manager.tick(); assert.equal(f.sent.length, 2);
|
|
});
|
|
|
|
test('expired observed web lease is inactive with a live socket; transport heartbeat preserves attribution and never recreates it', async () => {
|
|
const f = await fixture(), tab = randomUUID(), connection = randomUUID(), native = randomUUID();
|
|
await f.manager.presence(tab, { sequence: 1, visibility: 'visible', focused: true, interacted: false });
|
|
await f.manager.connect(connection, 'web', tab); await f.manager.connect(native, 'native');
|
|
f.candidate(); await f.manager.tick(); f.advance(61_000);
|
|
await f.manager.heartbeat(connection); await f.manager.heartbeat(native); await f.manager.tick(); assert.equal(f.sent.length, 1);
|
|
assert.equal((await prisma.agentPushCoordinator.findUniqueOrThrow({ where: { id: 'root' } })).unknownSince, null);
|
|
const row = await prisma.agentPushConnection.findUniqueOrThrow({ where: { id: connection } });
|
|
assert.equal(row.kind, 'web'); assert.equal(row.sessionId, tab);
|
|
assert.equal((await prisma.agentPushConnection.findUniqueOrThrow({ where: { id: native } })).kind, 'native');
|
|
await f.manager.disconnect(native); await f.manager.heartbeat(native);
|
|
assert.equal(await prisma.agentPushConnection.findUnique({ where: { id: native } }), null);
|
|
await f.manager.presence(tab, { sequence: 2, visibility: 'hidden', focused: false, interacted: false });
|
|
assert.equal((await prisma.agentPushPresence.findUniqueOrThrow({ where: { id: tab } })).inactiveUntil, null, 'stale focus cannot start fresh grace');
|
|
});
|
|
|
|
test('presence classification cannot borrow a session from a different credential generation', async () => {
|
|
const f = await fixture(), tab = randomUUID();
|
|
await f.manager.presence(tab, { sequence: 1, visibility: 'hidden', focused: false, interacted: false });
|
|
await prisma.agentPushPresence.update({ where: { id: tab }, data: { ownerEpoch: 'another-generation' } });
|
|
await f.manager.connect(randomUUID(), 'web', tab); f.candidate(); await f.manager.tick(); f.advance(2_000); await f.manager.tick();
|
|
assert.equal(f.sent.length, 0); assert.equal((await prisma.agentPushDelivery.findFirstOrThrow()).lastSuppressionReason, 'unknown_client_grace');
|
|
});
|
|
|
|
test('read at final context refresh cancels without a send and preserves the prior suppression evidence', async () => {
|
|
const f = await fixture(); await f.manager.connect(randomUUID()); f.candidate(); await f.manager.tick(); f.advance(2_000); await f.manager.tick();
|
|
const prior = await prisma.agentPushDelivery.findFirstOrThrow(), original = f.manager.options.context; let calls = 0;
|
|
f.manager.options.context = async (...args) => { if (++calls === 2) await f.manager.advanceRead('11', f.source); return original(...args); };
|
|
f.advance(4_000); await f.manager.tick();
|
|
const receipt = await prisma.agentPushDelivery.findFirstOrThrow();
|
|
assert.equal(f.sent.length, 0); assert.equal(receipt.attempts, 0); assert.equal(receipt.outcome, 'read');
|
|
assert.equal(receipt.lastSuppressionReason, prior.lastSuppressionReason); assert.deepEqual(receipt.lastSuppressedAt, prior.lastSuppressedAt);
|
|
});
|
|
|
|
test('instrumented hidden web socket is inactive; missing lease and native-runtime unknown veto', async () => {
|
|
const f = await fixture(), tab = randomUUID(), connection = randomUUID();
|
|
await f.manager.connect(connection, 'web', tab); f.candidate(); await f.manager.tick(); f.advance(2_000); await f.manager.tick(); assert.equal(f.sent.length, 0);
|
|
await f.manager.presence(tab, { sequence: 0, visibility: 'hidden', focused: false, interacted: false });
|
|
f.context({ webPresence: { active: false, unknown: true, observedAt: '' } }); f.advance(2_000); await f.manager.tick(); assert.equal(f.sent.length, 0);
|
|
f.context({ webPresence: { active: false, unknown: false, observedAt: '' } }); f.advance(2_000); await f.manager.tick(); assert.equal(f.sent.length, 1);
|
|
});
|
|
|
|
test('prompt grace resets on latest accepted prompt, root bursts coalesce, unknown provenance never sends', async () => {
|
|
const f = await fixture(); f.context({ latestUserMessageAt: new Date(f.now).toISOString() });
|
|
f.candidate(); await f.manager.tick(); f.advance(3_000); f.candidate('12', 'timer'); await f.manager.tick();
|
|
f.context({ latestUserMessageAt: new Date(f.now).toISOString() }); f.advance(2_000); await f.manager.tick(); assert.equal(f.sent.length, 0);
|
|
f.advance(3_000); await f.manager.tick(); assert.equal(f.sent.length, 1); assert.equal(f.sent[0].identity.cursor, '12');
|
|
f.candidate('13', 'unknown'); await f.manager.tick(); f.advance(10_000); await f.manager.tick(); assert.equal(f.sent.length, 1);
|
|
});
|
|
|
|
test('canonical read and native read cancel pending sends and broadcast source-bound state', async () => {
|
|
const f = await fixture(); f.candidate(); await f.manager.tick(); await f.manager.advanceRead('11', f.source); f.advance(2_000); await f.manager.tick(); assert.equal(f.sent.length, 0);
|
|
f.candidate('12'); await f.manager.tick(); f.context({ nativeReadCursor: '12' }); f.advance(2_000); await f.manager.tick(); assert.equal(f.sent.length, 0);
|
|
assert.equal((await f.manager.readState()).cursor, '12'); assert.equal(f.reads.at(-1).sourceId, f.source);
|
|
assert.equal((await f.manager.advanceRead('1')).cursor, '12');
|
|
await assert.rejects(f.manager.advanceRead('13', randomUUID()), { statusCode: 409 });
|
|
});
|
|
|
|
test('read clears target APNs only, coalesce and rate-limit; Web Push is never invisible', async () => {
|
|
const f = await fixture(); await f.manager.register(randomUUID(), subscription()); f.candidate(); await f.manager.tick(); f.advance(2_000); await f.manager.tick(); assert.equal(f.sent.length, 2);
|
|
await f.manager.advanceRead('11', f.source); await f.manager.tick();
|
|
const clears = f.sent.filter(v => v.identity.kind === 'read_state'); assert.equal(clears.length, 1); assert.equal(clears[0].registration.transport, 'apns');
|
|
await f.manager.advanceRead('12', f.source); await f.manager.tick(); assert.equal(f.sent.length, 3);
|
|
f.advance(60_000); await f.manager.tick(); assert.equal(f.sent.filter(v => v.identity.kind === 'read_state').length, 2);
|
|
});
|
|
|
|
test('unavailable source, uncertainty, startup warm-up and source replacement suppress safely', async () => {
|
|
const f = await fixture({ warmupMs: 60_000 }); f.candidate(); await f.manager.tick(); f.advance(2_000); await f.manager.tick(); assert.equal(f.sent.length, 0);
|
|
f.advance(60_000); f.failure(true); await f.manager.tick(); assert.equal(f.sent.length, 0);
|
|
f.failure(false); f.context({ uncertain: true }); await f.manager.tick(); assert.equal(f.sent.length, 0);
|
|
f.context({ uncertain: false, sourceId: randomUUID(), latestUserMessageAt: null }); await f.manager.tick(); assert.equal(f.sent.length, 0);
|
|
assert.equal((await f.manager.config()).quarantined, true); assert.equal((await f.manager.config()).sourceId, f.source);
|
|
assert.equal((await f.manager.readState()).sourceId, f.source);
|
|
});
|
|
|
|
test('known transient rejections retry boundedly; interrupted intent and ambiguous I/O never replay', async () => {
|
|
const f = await fixture(); f.result({ outcome: 'retry', retryAfterMs: 1_000 }); f.candidate(); await f.manager.tick(); f.advance(2_000);
|
|
for (let i = 0; i < 6; i++) { await f.manager.tick(); f.advance(2_000); }
|
|
assert.equal(f.sent.length, 4); assert.equal((await prisma.agentPushDelivery.findFirst())?.status, 'failed');
|
|
f.candidate('12'); f.result({ outcome: 'ambiguous' }); await f.manager.tick(); f.advance(2_000); await f.manager.tick();
|
|
f.advance(60_000); await f.manager.tick(); assert.equal(f.sent.length, 5);
|
|
f.candidate('13'); await f.manager.tick(); await prisma.agentPushDelivery.updateMany({ where: { status: 'pending' }, data: { status: 'intent', updatedAt: new Date(f.now - 40_000) } });
|
|
await f.manager.tick(); assert.equal(f.sent.length, 5); assert.equal(await prisma.agentPushDelivery.count({ where: { status: 'ambiguous' } }), 2);
|
|
});
|
|
|
|
test('revoke and rotation prevent pending sends; permanent rejection cannot invalidate a newer endpoint version', async () => {
|
|
const f = await fixture(); const same = await f.manager.register(f.installationId, apns); assert.equal(same.version, 1);
|
|
f.candidate(); await f.manager.tick(); await f.manager.revoke(f.installationId); f.advance(2_000); await f.manager.tick(); assert.equal(f.sent.length, 0);
|
|
const newId = randomUUID(); await f.manager.register(newId, apns); f.candidate('12'); await f.manager.tick(); f.advance(2_000);
|
|
f.manager.options.send = async () => { await f.manager.register(newId, { ...apns, token: 'b'.repeat(64) }); return { outcome: 'invalid' }; };
|
|
await f.manager.tick(); const latest = await prisma.agentPushInstallation.findUniqueOrThrow({ where: { id: newId } }); assert.equal(latest.enabled, true); assert.equal(latest.version, 2);
|
|
});
|
|
|
|
test('credential generation change disables opt-ins; two coordinators do not send one delivery twice', async () => {
|
|
const f = await fixture(); f.candidate(); await f.manager.tick(); f.advance(2_000);
|
|
const second = new AgentPushCoordinator({ ...f.manager.options, warmupMs: 0 }); await second.initialize();
|
|
await Promise.all([f.manager.tick(), second.tick()]); assert.equal(f.sent.length, 1);
|
|
const rotated = new AgentPushCoordinator({ ...f.manager.options, config: { ...config, adminToken: 'rotated-fixture-token' }, warmupMs: 0 }); await rotated.initialize();
|
|
assert.equal((await prisma.agentPushInstallation.findUniqueOrThrow({ where: { id: f.installationId } })).enabled, false);
|
|
});
|
|
|
|
test('authenticated routes reject open mode, wrong owner/body, malformed endpoints and secrets are never reflected', async () => {
|
|
const app = Fastify({ logger: false }); await app.register(sensible);
|
|
await registerPushRoutes(app, prisma, { baseUrl: 'http://unused.invalid' }, () => {}, { config, context: async () => { throw new Error(); }, send: async () => { throw new Error('must never send'); } });
|
|
const headers = { authorization: `Bearer ${process.env.ADMIN_TOKEN}` }, url = `/v1/agent/push/installations/${randomUUID()}`;
|
|
assert.equal((await app.inject({ url: '/v1/agent/push/config' })).statusCode, 401);
|
|
assert.equal((await app.inject({ url: '/v1/agent/push/config', headers: { authorization: 'Bearer wrong' } })).statusCode, 403);
|
|
const valid = await app.inject({ method: 'PUT', url, headers, payload: apns }); assert.equal(valid.statusCode, 200); assert.ok(!valid.body.includes(apns.token));
|
|
assert.equal((await app.inject({ method: 'PUT', url, headers, payload: { ...apns, ownerId: 'someone-else' } })).statusCode, 400);
|
|
const invalid = subscription(); if (invalid.transport === 'web') invalid.subscription.endpoint = 'https://127.0.0.1/private-secret';
|
|
const response = await app.inject({ method: 'PUT', url, headers, payload: invalid }); assert.equal(response.statusCode, 400); assert.ok(!response.body.includes('private-secret'));
|
|
const saved = env.ADMIN_TOKEN; env.ADMIN_TOKEN = undefined;
|
|
try { assert.equal((await app.inject({ url: '/v1/agent/push/config', headers })).statusCode, 503); } finally { env.ADMIN_TOKEN = saved; }
|
|
assert.equal((await app.inject({ method: 'DELETE', url, headers })).statusCode, 204);
|
|
await app.close();
|
|
});
|
|
|
|
test('provider endpoint validation rejects SSRF targets and private/reserved DNS addresses', () => {
|
|
for (const value of ['http://fcm.googleapis.com/x','https://user:pass@fcm.googleapis.com/x','https://fcm.googleapis.com.evil.example/x', 'https://fcm.googleapis.com:8443/x', 'https://[::1]/x','https://web.push.apple.com/x#fragment']) assert.throws(() => validatePushEndpoint(value));
|
|
for (const address of ['127.0.0.1','10.0.0.1','169.254.1.1','192.168.1.1','172.16.0.1','::1','::ffff:127.0.0.1','fd00::1','fe80::1','2001:db8::1']) assert.equal(isPublicAddress(address), false, address);
|
|
assert.equal(isPublicAddress('8.8.8.8'), true); assert.equal(isPublicAddress('2606:4700::1111'), true);
|
|
});
|
|
|
|
test('real APNs ES256 request construction uses generic alerts, explicit environment and background clear headers through injected HTTP2 boundary', async () => {
|
|
const key = generateKeyPairSync('ec', { namedCurve: 'prime256v1' });
|
|
const requests: any[] = []; const now = Date.now(), sourceId = randomUUID();
|
|
const sender = createPushSender({ ...config, apns: { ...config.apns!, key: key.privateKey.export({ format: 'pem', type: 'pkcs8' }).toString() } },
|
|
{ apns: async request => { requests.push(request); return { status: 200 }; } });
|
|
const identity = { version: 1 as const, sourceId, kind: 'message' as const, messageId: 'outbound:12', cursor: '12' };
|
|
assert.equal((await sender(apns, identity, now + 300_000, now)).outcome, 'accepted');
|
|
assert.equal(requests[0].endpoint, `https://api.sandbox.push.apple.com/3/device/${apns.token}`);
|
|
assert.equal(requests[0].headers['apns-push-type'], 'alert'); assert.equal(requests[0].headers['apns-topic'], apns.topic);
|
|
const [header, claims, signature] = requests[0].headers.authorization.slice(7).split('.');
|
|
assert.equal(JSON.parse(Buffer.from(header, 'base64url').toString()).kid, 'ABCDEFGHIJ');
|
|
assert.equal(JSON.parse(Buffer.from(claims, 'base64url').toString()).iss, 'DQQH5H6GBD');
|
|
assert.equal(verify('sha256', Buffer.from(`${header}.${claims}`), { key: key.publicKey, dsaEncoding: 'ieee-p1363' }, Buffer.from(signature, 'base64url')), true);
|
|
assert.deepEqual(JSON.parse(requests[0].body).aps.alert, { title: 'Sybil Agent', body: 'You have a new message.' });
|
|
await sender(apns, { version: 1, sourceId, kind: 'read_state', readCursor: '12' }, now + 60_000, now);
|
|
assert.equal(requests[1].headers['apns-push-type'], 'background'); assert.equal(requests[1].headers['apns-priority'], '5');
|
|
assert.deepEqual(JSON.parse(requests[1].body).aps, { 'content-available': 1 });
|
|
});
|
|
|
|
test('standard encrypted aes128gcm + VAPID Web Push uses library request construction and never sends read clears', async () => {
|
|
const vapid = webpush.generateVAPIDKeys(), registrations = subscription(), requests: any[] = [];
|
|
const sender = createPushSender({ ...config, web: { ...vapid, subject: 'mailto:tests@example.com' } }, { web: async request => { requests.push(request); return { status: 201 }; } });
|
|
const now = Date.now(), sourceId = randomUUID();
|
|
assert.equal((await sender(registrations, { version: 1, sourceId, kind: 'message', messageId: 'outbound:1', cursor: '1' }, now + 300_000, now)).outcome, 'accepted');
|
|
assert.equal(requests.length, 1); assert.equal(requests[0].headers['Content-Encoding'], 'aes128gcm'); assert.equal(requests[0].headers.TTL, 60);
|
|
assert.match(requests[0].headers.Authorization, /^vapid t=/); assert.ok(Buffer.isBuffer(requests[0].body)); assert.ok(!requests[0].body.includes('You have a new message.'));
|
|
assert.equal((await sender(registrations, { version: 1, sourceId, kind: 'read_state', readCursor: '1' }, now + 60_000, now)).outcome, 'failed'); assert.equal(requests.length, 1);
|
|
const gone = createPushSender({ ...config, web: { ...vapid, subject: 'mailto:tests@example.com' } }, { web: async () => ({ status: 410 }) });
|
|
assert.equal((await gone(registrations, { version: 1, sourceId, kind: 'message', messageId: 'outbound:2', cursor: '2' }, now + 60_000, now)).outcome, 'invalid');
|
|
});
|
|
|
|
|
|
test('production-scoped APNs configuration refuses sandbox registration and cannot send it', async () => {
|
|
const production = { ...config, apns: { ...config.apns!, environment: 'production' as const } };
|
|
const manager = new AgentPushCoordinator({ db: prisma, config: production, context: async () => { throw new Error(); }, send: async () => { throw new Error('must not send'); } });
|
|
await assert.rejects(manager.register(randomUUID(), apns), { statusCode: 400 });
|
|
let requests = 0;
|
|
const sender = createPushSender(production, { apns: async () => { requests++; throw new Error('must not send'); } });
|
|
const now = Date.now();
|
|
assert.equal((await sender(apns, { version: 1, sourceId: randomUUID(), kind: 'message', messageId: 'outbound:1', cursor: '1' }, now + 60_000, now)).outcome, 'failed');
|
|
assert.equal(requests, 0);
|
|
});
|
|
|
|
test('raw mocked HTTP2 handles request throws, goaway and close; raw HTTPS has absolute timeout even without socket inactivity', async () => {
|
|
const input = { endpoint: 'https://api.push.apple.com/3/device/fixture', headers: {}, body: Buffer.from('{}') };
|
|
for (const event of ['connect', 'goaway', 'close', 'never']) {
|
|
const session = new EventEmitter() as any; let destroyed = 0;
|
|
session.destroy = () => { destroyed++; }; session.request = () => { throw new Error('fixture request error'); };
|
|
const transport = createAPNsRequest((() => { if (event !== 'never') queueMicrotask(() => session.emit(event)); return session; }) as any, 5);
|
|
await assert.rejects(transport(input), /APNs transport unavailable/); assert.equal(destroyed, 1);
|
|
}
|
|
const req = new EventEmitter() as any; let destroyed = false; req.destroy = () => { destroyed = true; }; req.end = () => {};
|
|
const web = createWebRequest((() => req) as any, 5);
|
|
await assert.rejects(web({ ...input, endpoint: 'https://web.push.apple.com/fixture' }), /Web Push transport unavailable/); assert.equal(destroyed, true);
|
|
const throwing = createWebRequest((() => { throw new Error('fixture constructor error'); }) as any, 5);
|
|
await assert.rejects(throwing({ ...input, endpoint: 'https://web.push.apple.com/fixture' }), /Web Push transport unavailable/);
|
|
});
|
|
|
|
|
|
test('DELETE tombstones unknown and known UUIDs across restart; late PUT and rebound endpoints cannot re-enable retired IDs', async () => {
|
|
const f = await fixture(), unknown = randomUUID();
|
|
await f.manager.revoke(unknown);
|
|
await assert.rejects(f.manager.register(unknown, apns), { statusCode: 409 });
|
|
await f.manager.revoke(f.installationId); await f.manager.revoke(f.installationId);
|
|
const restarted = new AgentPushCoordinator({ ...f.manager.options, warmupMs: 0 }); await restarted.initialize();
|
|
await assert.rejects(restarted.register(f.installationId, apns), { statusCode: 409 });
|
|
const first = randomUUID(), second = randomUUID();
|
|
await restarted.register(first, apns); await restarted.register(second, apns);
|
|
await assert.rejects(restarted.register(first, apns), { statusCode: 409 });
|
|
assert.equal((await prisma.agentPushInstallation.findUniqueOrThrow({ where: { id: first } })).payload, '');
|
|
assert.equal((await prisma.agentPushInstallation.findUniqueOrThrow({ where: { id: second } })).enabled, true);
|
|
});
|
|
|
|
|
|
test('concurrent backend prompt admission starts durable grace even before runtime snapshot catches up', async () => {
|
|
const f = await fixture(); f.candidate(); await f.manager.tick(); f.advance(2_000);
|
|
await f.manager.notePrompt(); await f.manager.tick(); assert.equal(f.sent.length, 0);
|
|
f.advance(4_000); await f.manager.tick(); assert.equal(f.sent.length, 0);
|
|
f.advance(2_000); await f.manager.tick(); assert.equal(f.sent.length, 1);
|
|
});
|
|
|
|
|
|
async function httpContextFixture(f: Awaited<ReturnType<typeof fixture>>, snapshot: unknown, statuses = { incremental: 409, bootstrap: 200 }) {
|
|
const requests: (string | null)[] = [], sent: unknown[] = [];
|
|
const source = createServer((req, res) => {
|
|
assert.equal(req.headers.authorization, 'Bearer fixture-runtime-secret');
|
|
assert.equal(req.headers['x-sybil-client'], 'backend-proxy-v1');
|
|
const url = new URL(req.url!, 'http://fixture.invalid');
|
|
assert.equal(url.pathname, '/api/notification-context'); assert.equal(url.searchParams.get('limit'), '200');
|
|
const after = url.searchParams.get('after'); requests.push(after);
|
|
res.statusCode = after === null ? statuses.bootstrap : statuses.incremental;
|
|
res.setHeader('Content-Type', 'application/json');
|
|
res.end(JSON.stringify(after === null && statuses.bootstrap === 200 ? snapshot : { error: 'cursor unavailable' }));
|
|
});
|
|
source.listen(0, '127.0.0.1'); await once(source, 'listening');
|
|
const app = Fastify({ logger: false }); await app.register(sensible);
|
|
const manager = await registerPushRoutes(app, prisma, { baseUrl: `http://127.0.0.1:${(source.address() as { port: number }).port}`, apiKey: 'fixture-runtime-secret' }, () => {}, {
|
|
config, warmupMs: 0, now: () => f.now,
|
|
send: async (_registration, identity) => { sent.push(identity); return { outcome: 'accepted' }; },
|
|
});
|
|
return { manager, requests, sent, close: async () => { await manager.stop(); await app.close(); source.closeAllConnections(); await new Promise<void>(resolve => source.close(() => resolve())); } };
|
|
}
|
|
function highWaterSnapshot(f: Awaited<ReturnType<typeof fixture>>, overrides: Partial<Context> = {}): Context {
|
|
return { version: 1, sourceId: randomUUID(), conversationId: '1', cursor: '3', hasMore: false, latestUserMessageAt: null,
|
|
nativeReadCursor: '0', uncertain: false, webPresence: { active: false, unknown: false, observedAt: new Date(f.now).toISOString() }, candidates: [], ...overrides };
|
|
}
|
|
for (const identityChanged of [true, false]) test(`actual HTTP409 probes once and quarantines ${identityChanged ? 'replacement source' : 'same-source cursor regression'} without rebasing reads/checkpoint`, async () => {
|
|
const f = await fixture(); await f.manager.advanceRead('4', f.source); f.candidate(); await f.manager.tick(); f.advance(2_000);
|
|
const oldRead = await f.manager.readState();
|
|
const http = await httpContextFixture(f, highWaterSnapshot(f, identityChanged ? {} : { sourceId: f.source }));
|
|
try {
|
|
await http.manager.tick(); assert.deepEqual(http.requests, ['11', null]); assert.equal(http.sent.length, 0);
|
|
const state = await prisma.agentPushCoordinator.findUniqueOrThrow({ where: { id: 'root' } });
|
|
assert.equal(state.quarantined, true); assert.equal(state.sourceId, f.source); assert.equal(state.cursor, '11');
|
|
assert.deepEqual(await http.manager.readState(), oldRead);
|
|
const delivery = await prisma.agentPushDelivery.findFirstOrThrow(); assert.equal(delivery.status, 'cancelled'); assert.equal(delivery.outcome, 'source_changed');
|
|
assert.equal(await prisma.agentPushDelivery.count(), 1, 'probe never enumerates/replays historical candidates');
|
|
} finally { await http.close(); }
|
|
});
|
|
for (const failure of ['rejected', 'conflict', 'malformed', 'candidates', 'paginated', 'same-source-equal', 'same-source-higher', 'non-conflict']) test(`HTTP cursor recovery stays fail closed for ${failure}`, async () => {
|
|
const f = await fixture(); f.candidate(); await f.manager.tick(); f.advance(2_000); const oldRead = await f.manager.readState();
|
|
const snapshot = failure === 'malformed' ? { invalid: true } : highWaterSnapshot(f, {
|
|
...(failure === 'candidates' ? { candidates: [{ messageId: 'outbound:3', cursor: '3', createdAt: new Date(f.now).toISOString(), trigger: 'timer' as const }] } : {}),
|
|
...(failure === 'paginated' ? { hasMore: true } : {}),
|
|
...(failure.startsWith('same-source') ? { sourceId: f.source, cursor: failure === 'same-source-higher' ? '12' : '11', nativeReadCursor: '11' } : {}),
|
|
});
|
|
const http = await httpContextFixture(f, snapshot, { incremental: failure === 'non-conflict' ? 503 : 409,
|
|
bootstrap: failure === 'rejected' ? 503 : failure === 'conflict' ? 409 : 200 });
|
|
try {
|
|
await http.manager.tick(); assert.deepEqual(http.requests, failure === 'non-conflict' ? ['11'] : ['11', null]); assert.equal(http.sent.length, 0);
|
|
const state = await prisma.agentPushCoordinator.findUniqueOrThrow({ where: { id: 'root' } });
|
|
assert.equal(state.uncertain, true); assert.equal(state.quarantined, false); assert.equal(state.sourceId, f.source); assert.equal(state.cursor, '11');
|
|
assert.deepEqual(await http.manager.readState(), oldRead); assert.equal((await prisma.agentPushDelivery.findFirstOrThrow()).status, 'pending');
|
|
} finally { await http.close(); }
|
|
});
|
|
|
|
test('endpoint rebind retires a provider-disabled UUID, rejects its delayed PUT, and leaves no live endpoint after new owner revoke', async () => {
|
|
const f = await fixture(); f.result({ outcome: 'invalid' }); f.candidate(); await f.manager.tick(); f.advance(2_000); await f.manager.tick();
|
|
const invalid = await prisma.agentPushInstallation.findUniqueOrThrow({ where: { id: f.installationId } });
|
|
assert.equal(invalid.enabled, false); assert.equal(invalid.revokedAt, null); assert.notEqual(invalid.fingerprint, '');
|
|
const newer = randomUUID(); await f.manager.register(newer, apns);
|
|
assert.notEqual((await prisma.agentPushInstallation.findUniqueOrThrow({ where: { id: f.installationId } })).revokedAt, null);
|
|
await assert.rejects(f.manager.register(f.installationId, apns), { statusCode: 409 });
|
|
await f.manager.revoke(newer); assert.equal(await prisma.agentPushInstallation.count({ where: { enabled: true } }), 0);
|
|
});
|
|
|
|
test('endpoint rebind also retires disabled registrations from a prior credential generation', async () => {
|
|
const f = await fixture();
|
|
const current = new AgentPushCoordinator({ ...f.manager.options, config: { ...config, adminToken: 'new-fixture-bearer-generation' }, warmupMs: 0 });
|
|
await current.initialize(); assert.equal((await prisma.agentPushInstallation.findUniqueOrThrow({ where: { id: f.installationId } })).enabled, false);
|
|
const newer = randomUUID(); await current.register(newer, apns);
|
|
await assert.rejects(f.manager.register(f.installationId, apns), { statusCode: 409 });
|
|
await assert.rejects(current.register(f.installationId, apns), { statusCode: 409 });
|
|
await current.revoke(newer); assert.equal(await prisma.agentPushInstallation.count({ where: { enabled: true } }), 0);
|
|
});
|