diff --git a/backend/src/modules/mcp/diagnostic-cancellation-source.test.ts b/backend/src/modules/mcp/diagnostic-cancellation-source.test.ts index 2ab01a3a..4ff7cf95 100644 --- a/backend/src/modules/mcp/diagnostic-cancellation-source.test.ts +++ b/backend/src/modules/mcp/diagnostic-cancellation-source.test.ts @@ -1,6 +1,9 @@ import assert from "node:assert/strict" import { createHash, randomUUID } from "node:crypto" -import { createDiagnosticCancellationSourceForTests, createReadServerProbeExecutionSourceForTests } from "./diagnostic-cancellation-source.js" +import { mkdtemp, readdir, rm, rmdir } from "node:fs/promises" +import { tmpdir } from "node:os" +import { join } from "node:path" +import { createDiagnosticCancellationSourceForTests, createReadServerProbeExecutionSourceForTests, readServerProbeExecutionSource } from "./diagnostic-cancellation-source.js" import { fingerprint, redact } from "./security.js" const proof = async (value: unknown) => createHash("sha256").update(fingerprint(value)).digest("hex") @@ -25,6 +28,7 @@ function fixture() { return { rows } } } return { load: () => load(client, operationId, owner), read: () => read(snapshot, operationId, owner), readSql, + readWithExistingKey: () => readServerProbeExecutionSource(snapshot, operationId, owner), refresh, binding, input, evidence, row, sealed, setRows: (value: typeof rows) => { rows=value }, calls: () => calls, operationId, credentialId } } { @@ -100,4 +104,22 @@ for (const mutate of mutations) { assert.equal(source.originalContext.tool, "bandwidth"); assert.equal(source.stageBudget, 1) assert.equal(source.originalContext.identity.credentialId, f.credentialId) } -console.log(`diagnostic-cancellation-source: locked/read-only source, expired original contexts and ${mutations.length} shared tamper checks passed`) +{ + const state = await mkdtemp(join(tmpdir(), "mm-legacy-source-test-")) + const previousState = process.env.MCP_STATE_DIR + process.env.MCP_STATE_DIR = state + try { + const f = fixture() + await f.refresh() + f.row.evidence = { sealed: Buffer.alloc(32).toString("base64") } + await assert.rejects(f.readWithExistingKey, { code: "STATE_CONFLICT" }) + assert.equal(f.readSql.length, 1) + assert.deepEqual(await readdir(state), [], "Historical source reads must leave missing encryption state untouched") + } finally { + if (previousState === undefined) delete process.env.MCP_STATE_DIR + else process.env.MCP_STATE_DIR = previousState + await rm(join(state, "encryption-key"), { force: true }) + await rmdir(state) + } +} +console.log(`diagnostic-cancellation-source: locked/read-only source, missing key, expired original contexts and ${mutations.length} shared tamper checks passed`) diff --git a/backend/src/modules/mcp/diagnostic-cancellation-source.ts b/backend/src/modules/mcp/diagnostic-cancellation-source.ts index 8feb2d70..885dc87f 100644 --- a/backend/src/modules/mcp/diagnostic-cancellation-source.ts +++ b/backend/src/modules/mcp/diagnostic-cancellation-source.ts @@ -3,7 +3,7 @@ import type { ProbeStorageScope, ProbeStorageSql } from "../../services/server-p import type { ServerProbeJournalContext } from "../../services/server-probe-journal.js" import { SERVER_PROBE_OPERATION } from "../../services/server-probe-authority.js" import { openServerProbeMcpBinding } from "./server-probe-actions.js" -import { decryptSecret, fingerprint, privateSecretFingerprint, redact, McpError } from "./security.js" +import { decryptSecret, decryptSecretExisting, fingerprint, privateFingerprinterExisting, privateSecretFingerprint, redact, McpError } from "./security.js" const MAX_BYTES = 2097152 const uuid = z.string().uuid() @@ -113,7 +113,10 @@ export const loadDiagnosticCancellationSource = sourceLoader(productionCrypto) /** Test-only crypto seam. Never accept these dependencies from tool JSON. */ export const createDiagnosticCancellationSourceForTests = sourceLoader /** Historical authentication only; confers no execution/cancellation grant. */ -export const readServerProbeExecutionSource = readSourceLoader(productionCrypto) +export const readServerProbeExecutionSource = readSourceLoader({ + open: decryptSecretExisting, + proof: async value => (await privateFingerprinterExisting("server-probe-mcp-binding-v1"))(fingerprint(value)), +}) export const createReadServerProbeExecutionSourceForTests = readSourceLoader /** Exact previously locked source projection for private history conversion. */ export function authenticateServerProbeExecutionSourceRow(row: Record, operationId: string, diff --git a/backend/src/services/server-probe-legacy-verifier.test.ts b/backend/src/services/server-probe-legacy-verifier.test.ts index 3142c154..76c5f676 100644 --- a/backend/src/services/server-probe-legacy-verifier.test.ts +++ b/backend/src/services/server-probe-legacy-verifier.test.ts @@ -1,5 +1,8 @@ import assert from "node:assert/strict" import { createHash, randomUUID } from "node:crypto" +import { mkdtemp, readdir, rm, rmdir } from "node:fs/promises" +import { tmpdir } from "node:os" +import { join } from "node:path" import type { ServerProbeExecutionSource } from "../modules/mcp/diagnostic-cancellation-source.js" import { fingerprint } from "../modules/mcp/security.js" import type { ServerProbeJournalContext } from "./server-probe-journal.js" @@ -207,4 +210,25 @@ for (const field of ["operation_id", "connection_id", "actor_id"]) { const f = await fixture(), verify = createLegacyServerProbeVerifierForTests(f.deps, { ...f.crypto, proof: async () => { throw Error(secret) } }) const result = await verify(f.bundle); assert.equal(result.status, "unresolved"); assert.equal(JSON.stringify(result).includes(secret), false); checks++ } -console.log(`server-probe-legacy-verifier: ${checks} inert historical authentication checks passed; migration readiness always false`) +{ + const state = await mkdtemp(join(tmpdir(), "mm-legacy-verifier-test-")) + const previousState = process.env.MCP_STATE_DIR + process.env.MCP_STATE_DIR = state + try { + const receipt = await fixture() + const quarantine = await fixture("rest", false) + await quarantine.quarantine() + const verify = createLegacyServerProbeVerifier(receipt.deps) + for (const bundle of [receipt.bundle, quarantine.bundle]) { + assert.equal((await verify(bundle)).status, "unresolved") + assert.deepEqual(await readdir(state), [], "Historical reads must leave missing encryption state untouched") + checks++ + } + } finally { + if (previousState === undefined) delete process.env.MCP_STATE_DIR + else process.env.MCP_STATE_DIR = previousState + await rm(join(state, "encryption-key"), { force: true }) + await rmdir(state) + } +} +console.log(`server-probe-legacy-verifier: ${checks} historical authentication checks passed; migration readiness always false`) diff --git a/backend/src/services/server-probe-legacy-verifier.ts b/backend/src/services/server-probe-legacy-verifier.ts index e895fde6..c52b8eeb 100644 --- a/backend/src/services/server-probe-legacy-verifier.ts +++ b/backend/src/services/server-probe-legacy-verifier.ts @@ -3,7 +3,7 @@ import { isDeepStrictEqual } from "node:util" import { z } from "zod" import type { ServerProbeExecutionSource } from "../modules/mcp/diagnostic-cancellation-source.js" -import { decryptSecret, fingerprint, privateSecretFingerprint } from "../modules/mcp/security.js" +import { decryptSecretExisting, fingerprint, privateFingerprinterExisting } from "../modules/mcp/security.js" import { parseServerProbeJournalContext, validateServerProbeJournalReceipt, type ServerProbeJournalContext } from "./server-probe-journal.js" import type { ServerProbeReceipt } from "./server-probe-execution.js" import type { ProbeStorageScope } from "./server-probe-journal-maintenance.js" @@ -89,8 +89,9 @@ function stages(rows: Record[], ctx: ServerProbeJournalContext, } return received } -const productionCrypto: LegacyServerProbeVerifierCrypto = { open: decryptSecret, - proof: (domain, value) => privateSecretFingerprint(fingerprint(value), "server-probe-journal-v1:" + domain) } +// Inspection must never recreate missing application key state. +const productionCrypto: LegacyServerProbeVerifierCrypto = { open: decryptSecretExisting, + proof: async (domain, value) => (await privateFingerprinterExisting("server-probe-journal-v1:" + domain))(fingerprint(value)) } function verifier(dependencies: LegacyServerProbeVerifierDependencies, crypto: LegacyServerProbeVerifierCrypto) { return async (raw: LegacyServerProbeBundle): Promise => { diff --git a/backend/src/services/traffic-flow-engine-owned.test.ts b/backend/src/services/traffic-flow-engine-owned.test.ts index 85984861..5f1bb040 100644 --- a/backend/src/services/traffic-flow-engine-owned.test.ts +++ b/backend/src/services/traffic-flow-engine-owned.test.ts @@ -5,6 +5,7 @@ import * as crypto from "node:crypto" import { isDeepStrictEqual } from "node:util" import { createFlowIngestQuality, type FlowIngestQuality } from "./traffic-flow-ingest-quality.js" import ts from "typescript" +import { DatabaseBackgroundWork } from "./database-background-work.js" import * as owned from "./traffic-flow-owned-batch.js" import * as proof from "./traffic-flow-batch-proof.js" import * as accumulator from "./traffic-flow-fact-accumulator.js" @@ -264,3 +265,60 @@ await explicitPrune.flushPending({force:true,prune:true}) assert.equal(commits,pruneAttempts+1) assert.ok(dispatches>pruneDispatches,"explicit retention/prune still dispatches on a fresh engine") console.log("maintenance no-op: pristine/reset skip only; error-only unknown batch remains owned without replay; explicit prune preserved") + + +// A paused cold cache must not enqueue one pair of SQL loaders per datagram. +const coldWork = new DatabaseBackgroundWork() +let coldClock = 0 +let topologyLoads = 0 +let catalogLoads = 0 +let catalogReady = false +let topologyReady = false +let rejectTopology!: (error: Error) => void +let releaseCatalog!: () => void +const firstTopology = new Promise((_resolve, reject) => { rejectTopology = reject }) +const firstCatalog = new Promise(resolve => { releaseCatalog = resolve }) +const coldDependencies = { + ...dependencies, + "./database-background-work.js": { databaseBackgroundWork: coldWork }, + "./traffic-flow-topology.js": { + peekFlowTopology: () => topologyReady ? {} : null, + peekServerCatalog: () => catalogReady ? { byId: new Map() } : null, + loadFlowTopology: async () => { + topologyLoads++ + if (topologyLoads === 1) await firstTopology + topologyReady = true + }, + getServerCatalog: async () => { + catalogLoads++ + await firstCatalog + catalogReady = true + }, + }, +} +const coldExports: Record = {} +runInNewContext(output, { + exports: coldExports, + require: (name: keyof typeof coldDependencies) => coldDependencies[name], + performance: { now: () => coldClock }, + Buffer, console, setTimeout, clearTimeout, +}) +const coldEngine = coldExports as unknown as Engine +for (let i = 0; i < 128; i++) coldEngine.queueParsedFlows(1, []) +assert.deepEqual([topologyLoads, catalogLoads], [1, 1]) +let drained = false +const drain = coldWork.drain().then(() => { drained = true }) +await Promise.resolve() +assert.equal(drained, false) +rejectTopology(new Error("Owned topology load refused")) +releaseCatalog() +await drain +assert.equal(drained, true) +coldClock = 999 +for (let i = 0; i < 128; i++) coldEngine.queueParsedFlows(1, []) +assert.deepEqual([topologyLoads, catalogLoads], [1, 1]) +coldClock = 1000 +coldEngine.queueParsedFlows(1, []) +await coldWork.drain() +assert.deepEqual([topologyLoads, catalogLoads], [2, 1]) +assert.equal(topologyReady, true) diff --git a/backend/src/services/traffic-flow-engine.ts b/backend/src/services/traffic-flow-engine.ts index 4d3890b1..35e6b786 100644 --- a/backend/src/services/traffic-flow-engine.ts +++ b/backend/src/services/traffic-flow-engine.ts @@ -404,12 +404,28 @@ export function getEngineStats(): EngineStats { } } +function coldSourceRefresh(load: () => Promise): () => void { + let loading = false + let retryAfter = 0 + return () => { + if (loading || performance.now() < retryAfter) return + loading = true + void databaseBackgroundWork.track(load()).then( + () => { loading = false }, + () => { retryAfter = performance.now() + 1000; loading = false }, + ) + } +} + +const refreshColdTopology = coldSourceRefresh(loadFlowTopology) +const refreshColdCatalog = coldSourceRefresh(getServerCatalog) + export function queueParsedFlows(serverId: number, flows: ParsedFlowInput[]): void { if (flows.length) bumpDataEpoch() const at=Date.now(),bucketAt=minuteBucketIso(at),hourAt=hourBucketIso(at),observedAt=new Date(at).toISOString(),ripeMisses:string[]=[] const topo = peekFlowTopology(), catalog = peekServerCatalog() - if (!topo) void databaseBackgroundWork.track(loadFlowTopology()).catch(() => {}) - if (!catalog) void databaseBackgroundWork.track(getServerCatalog()).catch(() => {}) + if (!topo) refreshColdTopology() + if (!catalog) refreshColdCatalog() const serverType = catalog?.byId.get(serverId)?.type for (const raw of flows) { const flow = normalizeParsedFlow(raw)