263 lines
22 KiB
TypeScript
263 lines
22 KiB
TypeScript
import { MIGRATION_MANIFEST } from "../../db/migration-manifest.js"
|
|
import assert from "node:assert/strict"
|
|
import { randomUUID } from "node:crypto"
|
|
import { readFile, realpath, mkdir } from "node:fs/promises"
|
|
import { basename, dirname, resolve } from "node:path"
|
|
import type { PoolClient, QueryConfig } from "pg"
|
|
|
|
// Destructive acceptance is restricted to the runner's private temporary PG.
|
|
// All checks precede importing the configured pool or the approved host.
|
|
assert.equal(process.env.MMAPP_OWN_DIAGNOSTIC_CANCEL_HOST, "true")
|
|
const url = new URL(process.env.DATABASE_URL ?? "")
|
|
assert.equal(url.hostname, "127.0.0.1"); assert.equal(decodeURIComponent(url.username), "mcp_test")
|
|
assert.equal(url.pathname, "/postgres"); assert.equal(url.pathname.slice(1), process.env.MMAPP_OWN_SQL_DB)
|
|
assert(process.env.MCP_STATE_DIR)
|
|
const ownedState = resolve(process.env.MCP_STATE_DIR)
|
|
assert.equal(basename(ownedState), "state"); assert.match(basename(dirname(ownedState)), /^mm-mcp-test-[A-Za-z0-9-]+$/)
|
|
assert.equal(resolve(await realpath(dirname(ownedState))).toLowerCase(), dirname(ownedState).toLowerCase())
|
|
await mkdir(ownedState, { recursive: true, mode: 0o700 })
|
|
assert.equal(resolve(await realpath(ownedState)).toLowerCase(), ownedState.toLowerCase())
|
|
const { pool, closePool } = await import("../../db/index.js")
|
|
const { applySqlMigrations } = await import("../../db/migrate.js")
|
|
const security = await import("./security.js")
|
|
const { requestStorageDiagnosticCancellation, createStorageDiagnosticCancellationRequest, PROBE_STORAGE_LIMITS } = await import("./diagnostic-cancellation-storage.js")
|
|
const { createProbeJournalStorage, ProbeStorageUnknown } = await import("../../services/server-probe-journal-maintenance.js")
|
|
const { SERVER_PROBE_OPERATION } = await import("../../services/server-probe-authority.js")
|
|
const cases: string[] = []
|
|
const pass = (name: string) => { cases.push(name); console.log("✓ Owned cancellation host: " + name) }
|
|
let serverId: number
|
|
|
|
async function fixture(running = false) {
|
|
const ownerId = "cancel-host-" + randomUUID(), connectionId = randomUUID(), credentialId = randomUUID(), originalCredentialId = randomUUID()
|
|
const operationId = randomUUID(), planId = randomUUID(), receiptId = randomUUID(), at = Date.now()
|
|
const until = new Date(at + 3600000).toISOString(), originalUntil = new Date(at + 1800000).toISOString()
|
|
const principal = { connectionId, credentialId, credentialExpiresAt: until }
|
|
const policy = { id: connectionId, version: 1, status: "active", mode: "approval", permissions: ["mm:servers:write"], expiresAt: until,
|
|
scope: { serverIds: [serverId], operationIds: [SERVER_PROBE_OPERATION], allowGlobal: false } }
|
|
const binding = { version: 1, operation: SERVER_PROBE_OPERATION, principal: { connectionId, credentialId: originalCredentialId, credentialExpiresAt: originalUntil },
|
|
ownerId, issuedAt: at - 1000, authorityExpiresAt: Date.parse(originalUntil), prepared: { receiptId, serverId, tool: "ping", expiresAt: originalUntil,
|
|
bindingCiphertext: await security.encryptSecret("owned synthetic request; no provider exists") }, bindingProof: "a".repeat(64), stageBudget: 2 }
|
|
const input = { receiptId, serverId, inputDigest: await security.privateSecretFingerprint(security.fingerprint(binding), "server-probe-mcp-binding-v1"),
|
|
bindingCiphertext: await security.encryptSecret(JSON.stringify(binding)) }
|
|
const step = { kind: "service", service: SERVER_PROBE_OPERATION, input }
|
|
const evidence = { steps: [step], target: { scope: "servers", serverIds: [serverId] }, requiresHuman: true, verificationScope: "acknowledgement", mutationCost: 2, validUntil: originalUntil }
|
|
const sealed = { sealed: await security.encryptSecret(JSON.stringify(evidence)) }, generation = await security.accessGeneration()
|
|
await pool.query("INSERT INTO mcp_connections(id,name,owner_id,owner_permissions,policy,generation,owner_valid_until,owner_claims_issued_at) VALUES($1,'owned cancellation host',$2,$3::jsonb,$4::jsonb,$5,$6,clock_timestamp())", [connectionId, ownerId, JSON.stringify(policy.permissions), JSON.stringify(policy), generation, until])
|
|
for (const [id, expiry] of [[credentialId, until], [originalCredentialId, originalUntil]]) await pool.query("INSERT INTO mcp_credentials(id,connection_id,digest,kind,expires_at) VALUES($1,$2,$3,'bearer',$4)", [id, connectionId, randomUUID(), expiry])
|
|
await pool.query("INSERT INTO mcp_plans(id,connection_id,operation_id,operation_version,policy_version,generation,payload,evidence,state,expires_at,approved_by) VALUES($1,$2,$3,1,1,$4,'{}',$5::jsonb,'executing',$6,$7)", [planId, connectionId, SERVER_PROBE_OPERATION, generation, JSON.stringify(sealed), originalUntil, ownerId])
|
|
await pool.query("INSERT INTO mcp_operations(id,plan_id,connection_id,idempotency_key,state) VALUES($1,$2,$3,$4,$5)", [operationId, planId, connectionId, randomUUID(), running ? "running" : "pending"])
|
|
if (running) await pool.query("INSERT INTO mcp_operation_steps(operation_id,step_index,state,intended_state) VALUES($1,0,'intent',$2::jsonb)", [operationId, JSON.stringify(security.redact(step))])
|
|
const scope = { ownerId, receiptId, identityKind: "mcp" as const, executionId: operationId, bindingProof: binding.bindingProof, authorityExpiresAt: originalUntil }
|
|
return { ownerId, connectionId, credentialId, originalCredentialId, operationId, planId, receiptId, principal, policy, scope, binding, at }
|
|
}
|
|
type Fixture = Awaited<ReturnType<typeof fixture>>
|
|
const quota = async (ownerId: string) => (await pool.query("SELECT scope,admissions::text,reserved_files::text,reserved_bytes::text FROM server_probe_storage_usage WHERE scope=ANY($1::text[]) ORDER BY scope", [["global", "owner:" + ownerId]])).rows
|
|
async function counts(f: Fixture) {
|
|
return (await pool.query(`SELECT (SELECT count(*)::int FROM mcp_diagnostic_cancellations WHERE operation_id=$1) AS cancellations,
|
|
(SELECT count(*)::int FROM mcp_audit WHERE action='diagnostic.cancel-requested' AND payload->>'operationId'=$1::text) AS audits,
|
|
(SELECT count(*)::int FROM server_probe_storage_admissions WHERE receipt_id=$2) AS admissions`, [f.operationId, f.receiptId])).rows[0]
|
|
}
|
|
async function absent(f: Fixture, before: Awaited<ReturnType<typeof quota>>) {
|
|
assert.deepEqual(await counts(f), { cancellations: 0, audits: 0, admissions: 0 }); assert.deepEqual(await quota(f.ownerId), before)
|
|
}
|
|
type Hook = (text: string, client: PoolClient) => Promise<void>
|
|
async function proxyHost<T>(work: () => Promise<T>, hooks: { before?: Hook; after?: Hook } = {}) {
|
|
const saved = pool.connect, original = pool.connect.bind(pool), backends = new Set<number>(), texts: string[] = []
|
|
let checkouts = 0, nestedCallbacks = 0
|
|
// Wrapping checked-out objects never mutates a raw client's query method.
|
|
// Pool.query's callback overload passes straight through to the real pool.
|
|
pool.connect = ((callback?: unknown) => {
|
|
if (typeof callback === "function") { nestedCallbacks++; return (original as (...args: unknown[]) => unknown)(callback) }
|
|
return original().then(client => {
|
|
checkouts++; backends.add((client as PoolClient & { processID: number }).processID)
|
|
const rawQuery = client.query.bind(client)
|
|
return new Proxy(client, { get(target, key) {
|
|
if (key === "query") return (...args: unknown[]) => {
|
|
if (typeof args.at(-1) === "function") return (rawQuery as (...args: unknown[]) => unknown)(...args)
|
|
return (async () => {
|
|
const text = typeof args[0] === "string" ? args[0] : (args[0] as QueryConfig).text
|
|
texts.push(text); await hooks.before?.(text, client)
|
|
const result = await (rawQuery as (...args: unknown[]) => Promise<unknown>)(...args)
|
|
await hooks.after?.(text, client); return result
|
|
})()
|
|
}
|
|
const value = Reflect.get(target, key, target)
|
|
return typeof value === "function" ? value.bind(target) : value
|
|
} })
|
|
})
|
|
}) as typeof pool.connect
|
|
try { return { result: await work(), checkouts, backends, texts, nestedCallbacks } }
|
|
finally { pool.connect = saved }
|
|
}
|
|
async function activeClaim(f: Fixture) {
|
|
const storage = createProbeJournalStorage({ policy: { ...PROBE_STORAGE_LIMITS, retentionDays: 30 }, safetyWindowMs: 60000,
|
|
connect: async () => { const client = await pool.connect(); return { query: (text, values) => client.query(text, values), release: destroy => client.release(destroy) } }, authorize: async () => "c".repeat(64) })
|
|
await storage.claim(f.scope, { deadlineAt: Date.now() + 120000 }, async sql => {
|
|
await sql.query(`INSERT INTO server_probe_executions(receipt_id,binding_proof,identity_kind,execution_id,actor_id,authentication,mcp_connection_id,mcp_credential_id,authority_issued_at,authority_expires_at,source_server_id,tool)
|
|
VALUES($1,$2,'mcp',$3,$4,'mcp',$5,$6,$7,$8,$9,'ping')`, [f.receiptId, f.binding.bindingProof, f.operationId, f.ownerId, f.connectionId, f.originalCredentialId, f.binding.issuedAt, f.scope.authorityExpiresAt, serverId])
|
|
})
|
|
return storage
|
|
}
|
|
|
|
try {
|
|
await applySqlMigrations(pool)
|
|
assert.equal((await pool.query("SELECT count(*)::int AS n FROM schema_migrations")).rows[0].n, MIGRATION_MANIFEST.length)
|
|
await pool.query("UPDATE traffic_settings SET enabled=false; UPDATE uptime_settings SET enabled=false,resources_enabled=false,ping_enabled=false,speed_enabled=false; UPDATE mcp_settings SET enabled=true,requests_retention_days=30")
|
|
const installed = (await pool.query("SELECT to_regclass('public.server_probe_storage_admissions') AS name")).rows[0].name
|
|
if (!installed) {
|
|
const ddl = await pool.connect()
|
|
try { await ddl.query(await readFile(new URL("../../db/proposals/server-probe-storage-v2.sql", import.meta.url), "utf8")) } finally { ddl.release(true) }
|
|
}
|
|
serverId = Number((await pool.query("INSERT INTO servers(name,host,port,enabled) VALUES('owned-cancel-host','127.0.0.1',9,true) RETURNING id")).rows[0].id)
|
|
{
|
|
const f = await fixture(true); const storage = await activeClaim(f), before = await quota(f.ownerId)
|
|
const observed = await proxyHost(() => requestStorageDiagnosticCancellation(f.principal, f.operationId))
|
|
assert.deepEqual(observed.result, { operationId: f.operationId, cancelRequested: true, terminal: false, automaticReplay: false })
|
|
assert.equal(observed.checkouts, 1); assert.equal(observed.backends.size, 1); assert.equal(observed.nestedCallbacks, 0)
|
|
assert(observed.texts.findIndex(text => text.includes("probe-storage:lock")) < observed.texts.findIndex(text => text.includes("diagnostic-cancellation:source")))
|
|
assert.deepEqual(await counts(f), { cancellations: 1, audits: 1, admissions: 1 }); assert.deepEqual(await quota(f.ownerId), before)
|
|
await requestStorageDiagnosticCancellation(f.principal, f.operationId)
|
|
assert.deepEqual(await counts(f), { cancellations: 1, audits: 1, admissions: 1 }); assert.deepEqual(await quota(f.ownerId), before)
|
|
const row = (await pool.query("SELECT claim_token,cancelled_at FROM server_probe_storage_admissions WHERE receipt_id=$1", [f.receiptId])).rows[0]
|
|
assert.equal(row.claim_token, null); assert(row.cancelled_at)
|
|
await assert.rejects(storage.claim(f.scope, { deadlineAt: Date.now() + 120000 }, async () => assert.fail("Cancelled identity cannot execute again")))
|
|
assert.deepEqual(await quota(f.ownerId), before)
|
|
pass("actual active v2 claim cancels once on one backend after storage lock; duplicate shares quota without debit/refund or replay")
|
|
}
|
|
{
|
|
const f = await fixture(); const before = await quota(f.ownerId)
|
|
await requestStorageDiagnosticCancellation(f.principal, f.operationId)
|
|
assert.deepEqual(await counts(f), { cancellations: 1, audits: 1, admissions: 1 })
|
|
const after = await quota(f.ownerId), owner = after.find(row => row.scope === "owner:" + f.ownerId)
|
|
assert(owner); assert.equal(owner.admissions, "1"); assert.equal(owner.reserved_bytes, "294912"); assert.equal(owner.reserved_files, "1")
|
|
assert.equal(BigInt(after.find(row => row.scope === "global")!.reserved_bytes) - BigInt(before.find(row => row.scope === "global")!.reserved_bytes), 294912n)
|
|
await requestStorageDiagnosticCancellation(f.principal, f.operationId); assert.deepEqual(await quota(f.ownerId), after)
|
|
pass("pending source cancellation creates exactly one charged admission and one audit; duplicate changes nothing")
|
|
}
|
|
{
|
|
const f = await fixture(); await pool.query("UPDATE mcp_operations SET state='applied' WHERE id=$1;", [f.operationId])
|
|
await pool.query("UPDATE mcp_plans SET state='completed' WHERE id=$1", [f.planId]); const before = await quota(f.ownerId)
|
|
assert.deepEqual(await requestStorageDiagnosticCancellation(f.principal, f.operationId), { operationId: f.operationId, cancelRequested: false, terminal: true, automaticReplay: false })
|
|
await absent(f, before); pass("terminal source returns a no-op without cancellation, audit or quota mutation")
|
|
}
|
|
for (const reason of ["wrong-owner", "scope", "credential", "source", "settings"] as const) {
|
|
const f = await fixture(), before = await quota(f.ownerId)
|
|
if (reason === "wrong-owner") await pool.query("UPDATE mcp_connections SET owner_id='different-owned-user' WHERE id=$1", [f.connectionId])
|
|
if (reason === "scope") await pool.query("UPDATE mcp_connections SET policy=jsonb_set(policy,'{scope,serverIds}','[]'::jsonb) WHERE id=$1", [f.connectionId])
|
|
if (reason === "credential") await pool.query("UPDATE mcp_credentials SET revoked_at=clock_timestamp() WHERE id=$1", [f.credentialId])
|
|
if (reason === "source") await pool.query("UPDATE mcp_plans SET operation_version=2 WHERE id=$1", [f.planId])
|
|
if (reason === "settings") await pool.query("UPDATE mcp_settings SET enabled=false WHERE id=1")
|
|
try { await assert.rejects(requestStorageDiagnosticCancellation(f.principal, f.operationId)); await absent(f, before) }
|
|
finally { if (reason === "settings") await pool.query("UPDATE mcp_settings SET enabled=true WHERE id=1") }
|
|
pass(reason + " refusal writes no cancellation/audit/admission and changes no quota")
|
|
}
|
|
for (const reason of ["retention", "credential", "source"] as const) {
|
|
const f = await fixture(), before = await quota(f.ownerId), mutator = await pool.connect()
|
|
let changed = false
|
|
try {
|
|
await assert.rejects(proxyHost(() => requestStorageDiagnosticCancellation(f.principal, f.operationId), { before: async text => {
|
|
if (changed || !text.includes("probe-storage:lock")) return
|
|
changed = true
|
|
if (reason === "retention") await mutator.query("UPDATE mcp_settings SET requests_retention_days=31 WHERE id=1")
|
|
if (reason === "credential") await mutator.query("UPDATE mcp_credentials SET revoked_at=clock_timestamp() WHERE id=$1", [f.credentialId])
|
|
if (reason === "source") await mutator.query("UPDATE mcp_plans SET operation_version=2 WHERE id=$1", [f.planId])
|
|
} }))
|
|
assert(changed); await absent(f, before)
|
|
} finally { mutator.release(true); if (reason === "retention") await pool.query("UPDATE mcp_settings SET requests_retention_days=30 WHERE id=1") }
|
|
pass(reason + " drift committed on another backend before global lock is reauthorized and leaves atomic zero writes")
|
|
}
|
|
{
|
|
const f = await fixture(), before = await quota(f.ownerId)
|
|
let failed = false
|
|
await assert.rejects(proxyHost(() => requestStorageDiagnosticCancellation(f.principal, f.operationId), { after: async text => {
|
|
if (!failed && text.startsWith("INSERT INTO mcp_audit")) { failed = true; throw Error("Owned audit acknowledgement failure before COMMIT") }
|
|
} }))
|
|
assert(failed); await absent(f, before)
|
|
pass("failure after actual cancellation/audit INSERT before COMMIT rolls back cancellation, audit, admission and all quota")
|
|
}
|
|
{
|
|
const f = await fixture(); let lost = false
|
|
await assert.rejects(proxyHost(() => createStorageDiagnosticCancellationRequest()(f.principal, f.operationId), { after: async text => {
|
|
if (!lost && text === "COMMIT") { lost = true; throw Error("Owned lost actual COMMIT acknowledgement") }
|
|
} }), ProbeStorageUnknown)
|
|
assert(lost); assert.deepEqual(await counts(f), { cancellations: 1, audits: 1, admissions: 1 }); const after = await quota(f.ownerId)
|
|
await requestStorageDiagnosticCancellation(f.principal, f.operationId)
|
|
assert.deepEqual(await counts(f), { cancellations: 1, audits: 1, admissions: 1 }); assert.deepEqual(await quota(f.ownerId), after)
|
|
pass("lost actual cancellation COMMIT ACK reports unknown; independent repeat observes one spent admission and no second audit/debit")
|
|
}
|
|
{
|
|
const f = await fixture(true); await activeClaim(f); const before = await quota(f.ownerId)
|
|
const observed = await proxyHost(() => Promise.all([
|
|
requestStorageDiagnosticCancellation(f.principal, f.operationId),
|
|
requestStorageDiagnosticCancellation(f.principal, f.operationId),
|
|
]))
|
|
assert(observed.result.every(result => result.cancelRequested === true))
|
|
assert.equal(observed.checkouts, 2); assert.equal(observed.nestedCallbacks, 0)
|
|
assert.deepEqual(await counts(f), { cancellations: 1, audits: 1, admissions: 1 }); assert.deepEqual(await quota(f.ownerId), before)
|
|
pass("two concurrent actual host requests serialize without deadlock and commit one cancellation/audit against one shared admission")
|
|
}
|
|
{
|
|
const f = await fixture(), observer = await pool.connect()
|
|
let reached!: () => void, release!: () => void
|
|
let actualCommitObserved = false
|
|
const committed = new Promise<void>(resolve => { reached = resolve })
|
|
const resume = new Promise<void>(resolve => { release = resolve })
|
|
const pending = proxyHost(() => requestStorageDiagnosticCancellation(f.principal, f.operationId), { after: async text => {
|
|
if (text === "COMMIT") { actualCommitObserved = true; reached(); await resume }
|
|
} })
|
|
const timer = setTimeout(() => { reached(); release() }, 30000)
|
|
try {
|
|
await committed
|
|
assert(actualCommitObserved, "Host must reach the actual COMMIT barrier before timeout")
|
|
const durable = (await observer.query("SELECT count(*)::int AS n FROM mcp_diagnostic_cancellations WHERE operation_id=$1", [f.operationId])).rows[0].n
|
|
assert.equal(durable, 1, "Independent backend sees actual COMMIT before the host receives its acknowledgement")
|
|
await observer.query("UPDATE mcp_operations SET state='applied' WHERE id=$1", [f.operationId])
|
|
await observer.query("UPDATE mcp_plans SET state='completed' WHERE id=$1", [f.planId])
|
|
release()
|
|
const observed = await pending
|
|
assert.equal(observed.result.cancelRequested, true); assert.equal(observed.checkouts, 1); assert.equal(observed.nestedCallbacks, 0)
|
|
assert.deepEqual(await counts(f), { cancellations: 1, audits: 1, admissions: 1 })
|
|
pass("terminal transition after actual cancellation COMMIT but before host acknowledgement preserves the committed stop record")
|
|
} finally { clearTimeout(timer); release(); await pending.catch(() => {}); observer.release(true) }
|
|
}
|
|
{
|
|
const f = await fixture()
|
|
let reached!: () => void, release!: () => void, reservationCommitted = false, anchorCallbacks = 0
|
|
const reachedReservation = new Promise<void>(resolve => { reached = resolve })
|
|
const resume = new Promise<void>(resolve => { release = resolve })
|
|
const storage = createProbeJournalStorage({ policy: { ...PROBE_STORAGE_LIMITS, retentionDays: 30 }, safetyWindowMs: 60000,
|
|
authorize: async () => "c".repeat(64), connect: async () => {
|
|
const client = await pool.connect()
|
|
return { release: destroy => client.release(destroy), query: async (text, values) => {
|
|
const result = await client.query(text, values)
|
|
if (text === "COMMIT" && !reservationCommitted) { reservationCommitted = true; reached(); await resume }
|
|
return result
|
|
} }
|
|
} })
|
|
const claim = storage.claim(f.scope, { deadlineAt: Date.now() + 120000 }, async () => { anchorCallbacks++; assert.fail("Cancellation won before the anchor could execute") })
|
|
.then(() => false, () => true)
|
|
const timer = setTimeout(() => { reached(); release() }, 30000)
|
|
try {
|
|
await reachedReservation; assert(reservationCommitted, "Claimer must actually commit its reservation before the timeout")
|
|
assert.equal((await requestStorageDiagnosticCancellation(f.principal, f.operationId)).cancelRequested, true)
|
|
const charged = await quota(f.ownerId)
|
|
release(); assert.equal(await claim, true); assert.equal(anchorCallbacks, 0)
|
|
assert.deepEqual(await counts(f), { cancellations: 1, audits: 1, admissions: 1 }); assert.deepEqual(await quota(f.ownerId), charged)
|
|
assert.equal((await pool.query("SELECT count(*)::int AS n FROM server_probe_executions WHERE receipt_id=$1", [f.receiptId])).rows[0].n, 0)
|
|
pass("actual cancellation between reservation COMMIT and claim consumption spends the existing admission and prevents the anchor callback without extra quota")
|
|
} finally { clearTimeout(timer); release(); await claim }
|
|
}
|
|
{
|
|
const f = await fixture(true); await activeClaim(f)
|
|
const before = await quota(f.ownerId), protectionId = randomUUID()
|
|
await pool.query("INSERT INTO mcp_protections(id,object_ref,created_by) VALUES($1,$2::jsonb,$3)", [protectionId, JSON.stringify({ kind: "server", id: String(serverId), serverId, protection: "deny" }), f.ownerId])
|
|
try {
|
|
assert.equal((await requestStorageDiagnosticCancellation(f.principal, f.operationId)).cancelRequested, true)
|
|
assert.deepEqual(await counts(f), { cancellations: 1, audits: 1, admissions: 1 }); assert.deepEqual(await quota(f.ownerId), before)
|
|
}
|
|
finally { await pool.query("DELETE FROM mcp_protections WHERE id=$1", [protectionId]) }
|
|
pass("server deny protection preserves the owner's stop capability for an already active claim; no native write and no additional quota charge")
|
|
}
|
|
console.log(JSON.stringify({ suite: "approved owned diagnostic cancellation host", actualSql: true, productionHostExecuted: true, cases, nativeCalls: 0, apiCalls: 0 }))
|
|
} finally { await closePool() }
|