diff --git a/backend/package.json b/backend/package.json index 5d36917f..4f0bdc37 100644 --- a/backend/package.json +++ b/backend/package.json @@ -12,6 +12,7 @@ "test:traffic-flow:rebuild:pg": "node scripts/test-mcp-postgres.mjs --flow-rebuild", "test:traffic-flow:pg": "node scripts/test-mcp-postgres.mjs --flow-writes", "test:traffic-flow:quality:pg": "node scripts/test-mcp-postgres.mjs --flow-quality", + "test:traffic-flow:chr-evidence": "tsx --test src/test/traffic-flow-chr-capture-check.test.ts src/test/traffic-flow-chr-wire.test.ts src/test/traffic-flow-chr-target-harness.test.ts src/test/traffic-flow-chr-owned-sequence.test.ts src/test/traffic-flow-chr-reconcile.test.ts", "test:mcp:pg:all": "node scripts/test-mcp-postgres.mjs --all", "mcp:stdio": "node dist/scripts/mcp-stdio.js", "mcp:stdio:dev": "tsx src/scripts/mcp-stdio.ts", @@ -111,6 +112,7 @@ "test:mcp:routing-schema:pg": "node scripts/test-mcp-postgres.mjs --routing-reservation-schema", "test:mcp:reserved-step": "tsx --test src/modules/mcp/routing-reserved-step.test.ts", "test:mcp:routing-forward-rounds:pg": "node scripts/test-mcp-postgres.mjs --routing-network-forward-rounds", + "test:mcp:routing-original-effects:pg": "node scripts/test-mcp-postgres.mjs --routing-original-effects", "test:flow-active-services:pg": "node scripts/test-mcp-postgres.mjs --flow-active-services", "test:flow-active-rebuild:pg": "node scripts/test-mcp-postgres.mjs --flow-active-rebuild", "test:flow-worker-ipc:pg": "node scripts/test-mcp-postgres.mjs --flow-worker-ipc", diff --git a/backend/scripts/test-mcp-postgres.mjs b/backend/scripts/test-mcp-postgres.mjs index a684ece4..7dafc4b3 100644 --- a/backend/scripts/test-mcp-postgres.mjs +++ b/backend/scripts/test-mcp-postgres.mjs @@ -65,8 +65,9 @@ const telemetrySettingsConfiguration = process.argv.includes('--telemetry-settin if (telemetrySettingsConfiguration && process.argv.slice(2).some(arg => arg !== '--telemetry-settings-configuration')) throw new Error('Telemetry settings configuration requires its own disposable cluster') const routingReservationSchema = process.argv.includes('--routing-reservation-schema') if (routingReservationSchema && process.argv.slice(2).some(arg => arg !== '--routing-reservation-schema')) throw new Error('Routing reservation schema requires its own disposable cluster') -const routingForwardRounds = process.argv.includes('--routing-network-forward-rounds') -if (routingForwardRounds && process.argv.slice(2).some(arg => arg !== '--routing-network-forward-rounds')) throw new Error('Routing forward rounds requires its own disposable cluster') +const routingOriginalEffects = process.argv.includes('--routing-original-effects') +const routingForwardRounds = routingOriginalEffects || process.argv.includes('--routing-network-forward-rounds') +if (routingForwardRounds && process.argv.slice(2).some(arg => arg !== (routingOriginalEffects ? '--routing-original-effects' : '--routing-network-forward-rounds'))) throw new Error('Routing forward rounds requires its own disposable cluster') const flowActiveServices = process.argv.includes('--flow-active-services') if (flowActiveServices && process.argv.slice(2).some(arg => arg !== '--flow-active-services')) throw new Error('Flow active services requires its own disposable cluster') const flowActiveRebuild = process.argv.includes('--flow-active-rebuild') @@ -244,6 +245,7 @@ try { const result = regressionOnly || profilesOnly ? 0 : await new Promise((resolve, reject) => { const child = spawn(process.execPath, pollMemory ? ['--max-old-space-size=192', '--max-semi-space-size=16', fileURLToPath(new URL('../dist-mcp-memory/services/server-poll-memory.owned.test.js', import.meta.url))] : ['--import', 'tsx', principalSecuritySource ? 'src/modules/mcp/principal-security-source.owned.test.ts' : pollJournalSnapshot ? 'src/services/server-poll-journal-snapshot.owned.test.ts' : pollTerminalArchive ? 'src/services/server-poll-terminal-archive.owned.test.ts' : pollArchiveSchema ? 'src/db/server-poll-terminal-archives.sql.test.ts' : telemetrySettingsConfiguration ? 'src/modules/mcp/telemetry-settings-configuration.owned.test.ts' : routingReservationSchema ? 'src/db/routing-reservation-schema.owned.test.ts' : routingForwardRounds ? 'src/modules/mcp/routing-network-forward-rounds.owned.test.ts' : flowActiveServices ? 'src/services/traffic-flow-active-services.owned.test.ts' : flowActiveRebuild ? 'src/services/traffic-flow-active-rebuild.owned.test.ts' : flowQuality ? 'src/services/traffic-flow-quality.owned.test.ts' : flowWorkerIpc ? 'src/services/traffic-flow-worker-ipc.owned.test.ts' : manualUptime ? 'src/modules/mcp/manual-uptime-http.test.ts' : manualPingGroup ? 'src/modules/mcp/manual-ping-group-http.test.ts' : manualTraffic ? 'src/modules/mcp/manual-traffic-http.test.ts' : manualApiPing ? 'src/modules/mcp/manual-api-ping-http.test.ts' : simpleSpeedChr ? 'src/modules/mcp/simple-speed-chr.test.ts' : simpleSpeed ? 'src/modules/mcp/simple-speed-http.test.ts' : externalPeerMetadata ? 'src/services/bandwidth-external-peer-metadata.sql.test.ts' : operatorPeerMetadata ? 'src/services/bandwidth-operator-metadata.sql.test.ts' : publicRegistryMetadata ? 'src/services/bandwidth-public-registry.sql.test.ts' : operatorDockerSql ? 'src/services/bandwidth-operator-docker-stop.sql.test.ts' : mcpOperationOrigin ? 'src/modules/mcp/operation-execution-origin.sql.test.ts' : mcpSpeedOriginal ? 'src/modules/mcp/speed-bandwidth-original.sql.test.ts' : routerConfigurationGuard ? 'src/services/router-configuration-guard.sql.test.ts' : scheduledSpeedOriginal ? 'src/services/bandwidth-scheduled-speed-original.sql.test.ts' : bandwidthSpeedWorkflow ? 'src/services/bandwidth-speed-workflow.sql.test.ts' : restSpeedOriginal ? 'src/services/bandwidth-rest-speed-original.sql.test.ts' : operationLocks ? 'src/services/operation-config-lock-lease.sql.test.ts' : serviceExecutionOrigin ? 'src/modules/mcp/service-execution-origin.sql.test.ts' : bandwidthNodeSource ? 'src/services/bandwidth-node-source.sql.test.ts' : bandwidthExternalRegistry ? 'src/services/bandwidth-external-node-registry.sql.test.ts' : bandwidthNodeRegistry ? 'src/services/bandwidth-node-registry.sql.test.ts' : bandwidthControlSource ? 'src/services/bandwidth-control-source.sql.test.ts' : bandwidthAdmission ? 'src/services/bandwidth-admission.sql.test.ts' : serverPollStorageFresh ? 'src/services/server-poll-storage-fresh.sql.test.ts' : serverPollStorage ? 'src/services/server-poll-storage.sql.test.ts' : probeStorageFresh ? 'src/services/server-probe-storage-fresh.sql.test.ts' : probeLegacyOrphans ? 'src/services/server-probe-legacy-orphans.sql.test.ts' : diagnosticCancelHost ? 'src/modules/mcp/diagnostic-cancellation-storage.sql.test.ts' : ipsecExportHttp ? 'src/modules/mcp/ipsec-existing-certificate-export-http.sql.test.ts' : ipsecExportJournal ? 'src/services/ipsec-existing-certificate-export-rest-journal.sql.test.ts' : migrationDataInventory ? 'src/db/migration-data-inventory.sql.test.ts' : probeStorageHistory ? 'src/services/server-probe-storage-history.sql.test.ts' : probeStorageJournal ? 'src/services/server-probe-storage-journal.sql.test.ts' : controlledSchemaReference ? 'src/db/controlled-schema-reference.sql.test.ts' : controlledMigrations ? 'src/db/controlled-migrations.sql.test.ts' : databaseSchemaControl ? 'src/db/database-schema-control.sql.test.ts' : migrationSchemaInventory ? 'src/db/migration-schema-inventory.sql.test.ts' : routingNetworkNative ? 'src/modules/mcp/routing-network-native.sql.test.ts' : routingReservationRuntime ? 'src/modules/mcp/routing-reservation-runtime.sql.test.ts' : routingReservations ? 'src/modules/mcp/routing-reservation.sql.test.ts' : collectorPartitionWriters ? 'src/services/collector-partition-writes.sql.test.ts' : serverPollDurable ? 'src/services/server-poll-durable-session.sql.test.ts' : routingRecoveryAuthority ? 'src/modules/mcp/routing-recovery-authority.sql.test.ts' : legacyFlowRebuild ? 'src/test/legacy-flow-rebuild-sql.ts' : probeJournal ? 'src/services/server-probe-journal.sql.test.ts' : flowHostKeys ? 'src/modules/mcp/flow-host-keys-http.test.ts' : flowRebuildMcp ? 'src/modules/mcp/flow-facts-rebuild-http.test.ts' : flowRebuild ? 'src/test/flow-facts-rebuild-sql.ts' : flowWrites ? 'src/services/traffic-flow-write-transaction.pg.test.ts' : nativeConservation ? 'src/modules/mcp/native-conservation-sql.test.ts' : nativeCreateId ? 'src/modules/mcp/native-create-id-http.test.ts' : ipsecAdapters ? "src/modules/mcp/ipsec-adapters-http.test.ts" : recoveryMaintenance ? "src/modules/mcp/recovery-maintenance-http.test.ts" : scopedCheckpointAdapters ? "src/modules/mcp/scoped-checkpoint-adapters-http.test.ts" : wireguardTransfer ? 'src/modules/mcp/wireguard-transfer-http.test.ts' : interfaceEventsCompatibility ? 'src/modules/mcp/interface-events-compatibility-http.test.ts' : scopedRoutingDiagnostics ? 'src/modules/mcp/scoped-routing-diagnostics-http.test.ts' : applicationAdapters ? 'src/modules/mcp/application-adapters-http.test.ts' : greRevision ? 'src/modules/mcp/gre-revision-http.test.ts' : chrGreRevisionRecovery ? 'src/modules/mcp/chr-gre-revision-recovery.test.ts' : chrGreRevision ? 'src/modules/mcp/chr-gre-revision.test.ts' : chrLegacyOspf ? 'src/modules/mcp/chr-ospf-legacy.test.ts' : chrFilters ? 'src/modules/mcp/chr-filter-write.test.ts' : chrGreMutation ? 'src/modules/mcp/chr-gre-mutation.test.ts' : chrFirewallRestore ? 'src/modules/mcp/chr-firewall-restore.test.ts' : chrAddress ? 'src/modules/mcp/chr-address-list.test.ts' : chrFirewallWrite ? 'src/modules/mcp/chr-firewall-write.test.ts' : chrVpn ? 'src/modules/mcp/chr-vpn-catalog.test.ts' : chrFirewallMove ? 'src/modules/mcp/chr-firewall-move.test.ts' : chrRecursive ? 'src/modules/mcp/chr-recursive-restore.test.ts' : chrIdentity ? 'src/modules/mcp/chr-ospf-identity-readonly.test.ts' : chrRead ? 'src/modules/mcp/chr-readonly-acceptance.test.ts' : chr ? 'src/modules/mcp/chr-acceptance.test.ts' : 'src/modules/mcp/integration.test.ts'], { cwd: pollMemory ? directory : new URL('../', import.meta.url), stdio: 'inherit', env: { ...(pollMemory ? pollMemoryEnvironment : process.env), + ...(routingOriginalEffects ? { MMAPP_OWN_ROUTING_ORIGINAL_EFFECTS: 'true' } : {}), DATABASE_URL: `postgresql://mcp_test:${password}@127.0.0.1:${port}/postgres`, DATABASE_PATH: join(directory, 'no-legacy-database.sqlite'), SQLITE_IMPORT_FULL_HISTORY: 'false', MCP_STATE_DIR: join(directory, 'state'), AUTH_REQUIRED: 'true', AUTH_JWT_SECRET: 'mcp-integration-test-only', MMAPP_TEST_WG_EXPORT_ONLY: String(wireguardExport), @@ -268,6 +270,7 @@ try { else if (result === 0 && pollArchiveSchema) console.log('Poll archive schema: owned nonempty40 preservation, numbered41 and structural refusal guards; no positive MCP archive claim') else if (result === 0 && telemetrySettingsConfiguration) console.log('Telemetry settings: genuine MCP/SQL configuration binding, collector metadata tolerated; no native or external sources') else if (result === 0 && routingReservationSchema) console.log('Routing reservation numbered0040: ownSQL fresh/legacy/manual preservation; no READY, backfill, capabilities or public native activation') + else if (result === 0 && routingOriginalEffects) console.log('Routing original effects: approved manual revision 2 and original effects checked against disposable PostgreSQL and synthetic RouterOS HTTP') else if (result === 0 && routingForwardRounds) console.log('Routing forward rounds: genuine approved running CONFIG_APPLIED, shared cumulative native budget; logs/security/autoRecovery remain unaccepted') else if (result === 0 && flowActiveServices) console.log('Flow active settings/purge: production services and main Worker; disposable SQL/localhost only') else if (result === 0 && flowActiveRebuild) console.log('Flow active rebuild: production shared service and genuine Worker drain; disposable SQL/localhost only') diff --git a/backend/src/modules/mcp/routing-community-scoped-acceptance.ts b/backend/src/modules/mcp/routing-community-scoped-acceptance.ts index 699b0abf..2ff02082 100644 --- a/backend/src/modules/mcp/routing-community-scoped-acceptance.ts +++ b/backend/src/modules/mcp/routing-community-scoped-acceptance.ts @@ -188,7 +188,7 @@ export async function acceptScopedRouteCommunitySwitch(actor: AuthUser, options: } await pool.query("INSERT INTO filter_rules(server_id,sort_order,community,community_name,action,gateway,gateway_tunnel_id,description) VALUES($1,77,'65000:10','NEVER_SCOPED_SQL_NAME','route','203.0.113.99','NEVER_SCOPED_SQL_TUNNEL','NEVER_SCOPED_SQL_DESCRIPTION')", [sid]) const dbBefore = (await pool.query("SELECT * FROM filter_rules WHERE server_id=$1 ORDER BY id", [sid])).rows - const prepare = async (key = "a", persistent?: { id: string; version: number }, exactProfile?: ReturnType) => { + const prepare = async (key = "a", persistent?: { id: string; version: number }, exactProfile?: ReturnType, assertApprovalRequired = false) => { const current = exactProfile === undefined ? profile(key) : routeAiProfileSchema.parse(exactProfile) assert.equal((await invokeMcpTool(principal, "mm_route_path_verify", { profile: current, candidateId: "b" }) as { status: string }).status, "passed") const plan = await invokeMcpTool(principal, "mm_route_optimizer_plan", { profile: current, candidateId: "b" }, undefined, @@ -204,6 +204,11 @@ export async function acceptScopedRouteCommunitySwitch(actor: AuthUser, options: assert(!JSON.stringify(projection).includes("persistentProfileAuthority"), "Persistent authority remains private") } assert.equal(evidence.requiresHuman, true); assert.equal(evidence.mutationCost, 2); assert.ok(!JSON.stringify(input).includes("NEVER_")); assert.equal(input.compiledBody, undefined) + if (assertApprovalRequired) { + const beforeApproval = { writes, requests } + await assert.rejects(applyPlan(principal, plan.planId, randomUUID()), (error: unknown) => error instanceof McpError && error.code === "APPROVAL_REQUIRED") + assert.deepEqual({ writes, requests }, beforeApproval) + } await approvePlan(plan.planId, writer, true); return { plan, input } } const execute = async (planId: string, executingPrincipal: McpPrincipal = principal) => { @@ -238,21 +243,28 @@ export async function acceptScopedRouteCommunitySwitch(actor: AuthUser, options: const disablePersistent = (id: string) => pool.query("UPDATE mcp_routing_profiles SET enabled=false WHERE id=$1", [id]) if (options.originalEffectsOnly) { const cases: Array> = [] - for (const effectCase of ["transition", "no-op", "ambiguous"] as const) { + let publicManualRevision2: Record | undefined + for (const scenario of ["transition", "no-op", "ambiguous", "public-manual-rev2"] as const) { + const manual = scenario === "public-manual-rev2" + const effectCase = manual ? "transition" : scenario reset(); await refresh(); originalEffectCase = effectCase const config = profile("a") for (const candidate of config.candidates) candidate.controlProbe!.sourceAddress = "198.51.100.1" assert.equal(config.autoRecovery, false) // The baseline and persistent authority are issued by the SAME actual // fixed readers/CAS used by existing tests, not hand-written proofs. - const persistent = await persistentFixture(config) + const persistent = manual ? undefined : await persistentFixture(config) try { - const { plan, input } = await prepare("a", persistent, config) + const { plan, input } = await prepare("a", persistent, config, manual) + assert.equal(input.manualEffectPolicy, manual ? "observational-v1" : undefined) + if (manual) assert.equal(input.persistentProfileAuthority, undefined) const sealedSelection = originalEffectSelectionSchema.parse(input.originalEffectSelection) assert.equal(sealedSelection.status, effectCase === "ambiguous" ? "unknown" : "prepared") const storedPlan = await loadPlan(plan.planId) assert.equal(storedPlan.approved_by, writer.id) - assert(!JSON.stringify(await publicPlan(storedPlan)).includes("originalEffectSelection")) + const publicProjection = JSON.stringify(await publicPlan(storedPlan)) + assert(!publicProjection.includes("originalEffectSelection")) + assert(!publicProjection.includes("manualEffectPolicy")) const beforeWrites = writes, wireStart = originalEffectWire.length const operation = await execute(plan.planId) assert.equal(operation.state, "applied", "Optional effect unknown must preserve genuine manual configuration/probe acceptance") @@ -266,15 +278,19 @@ export async function acceptScopedRouteCommunitySwitch(actor: AuthUser, options: assert.equal(rows[0].result.configurationVerified, true); assert.equal(rows[0].result.controlProbeVerified, true) const report = originalEffectReportSchema.parse(rows[0].result.originalEffects) const requestSeal = await assertScopedCommunityRequest(input) + if (manual) { + const changedInput = { ...input, manualEffectPolicy: "unapproved-policy" } + await assert.rejects(assertScopedCommunityRequest(changedInput), (error: unknown) => error instanceof McpError && error.code === "STATE_CONFLICT") + } const configuration = await loadScopedConfigurationPacket(input, requestSeal, { operationId: operation.id, stepIndex: 0 }) assert(configuration); assert.equal(configuration.packet.operationId, operation.id) - assert.equal(configuration.packet.revision, 2); assert.equal(configuration.packet.phase, "CONFIG_APPLIED") + assert.equal(configuration.packet.revision, manual ? 1 : 2); assert.equal(configuration.packet.phase, "CONFIG_APPLIED") const network = await loadScopedNetworkCheckpoint(input, requestSeal) assert(network); assert.equal(network.status, "passed"); assert.deepEqual(network.originalEffects, report) const runningWire = originalEffectWire.slice(wireStart) const ribBefore = runningWire.filter(row => row.method === "GET" && row.path === "/routing/route" && !row.afterPatch).length const ribAfter = runningWire.filter(row => row.method === "GET" && row.path === "/routing/route" && row.afterPatch).length - console.log("Owned original effect diagnostic:", JSON.stringify({ case: effectCase, state: operation.state, + console.log("Owned original effect diagnostic:", JSON.stringify({ case: scenario, state: operation.state, status: report.status, reason: report.reason, rounds: report.rounds.length, budget: report.budget ?? null, ribBefore, ribAfter, runtimeNativeRequests: runningWire.length, lastNativePaths: runningWire.slice(-8).map(row => row.path) })) assert.equal(ribBefore, effectCase === "ambiguous" ? 0 : 1) @@ -293,6 +309,22 @@ export async function acceptScopedRouteCommunitySwitch(actor: AuthUser, options: assert.deepEqual(report.transition?.changedPrefixes, effectCase === "transition" ? ["203.0.113.0/24"] : []) assert.equal(report.transitionAttribution, "unproven") } + if (manual) { + const repeatReads = requests + const repeated = await applyPlan(principal, plan.planId, randomUUID()) + assert.equal(repeated.id, operation.id) + assert.equal(repeated.state, "applied") + assert.equal(requests, repeatReads) + assert.equal(writes - beforeWrites, 1) + assert.equal(Number((await pool.query("SELECT count(*) AS n FROM mcp_routing_profiles WHERE enabled")).rows[0].n), 0) + assert.equal(Number((await pool.query("SELECT count(*) AS n FROM mcp_scoped_recovery_intents WHERE source_operation_id=$1", [operation.id])).rows[0].n), 0) + publicManualRevision2 = { accepted: true, humanApprovalRequired: true, proofRevision: input.proofRevision, + configurationCheckpointRevision: configuration.packet.revision, effectStatus: report.status, + rounds: report.rounds.length, nativeRibBefore: ribBefore, nativeRibAfter: ribAfter, sharedBudget: report.budget, + autoRecoveryAllowed: report.autoRecoveryAllowed, scopedControllerEnabled: false, nativeWrites: 1 } + continue + } + assert(persistent) const row = (await pool.query("SELECT state,grant_expires_at FROM mcp_routing_profiles WHERE id=$1 AND version=1", [persistent.id])).rows[0] assert(row?.state.baseline) const advance = () => advanceRoutingControllerBaseline(row.state.baseline, persistent.id, 1, config, row.grant_expires_at.toISOString(), @@ -332,10 +364,11 @@ export async function acceptScopedRouteCommunitySwitch(actor: AuthUser, options: effectStatus: report.status, observedFibTransition: report.switchEffectObserved, nativeRibBefore: ribBefore, nativeRibAfter: ribAfter, sharedBudget: report.budget ?? null, effectBaselineAdvanceAccepted: effectCase === "transition", effectGateNativeReads: effectCase === "transition" ? "fresh-native-checked" : 0, actualTickPausedByExistingScopedGuard: true, tickEffectPauseAccepted: false, publicManualRevision2EffectOwnerAccepted: false }) - } finally { await disablePersistent(persistent.id) } + } finally { if (persistent) await disablePersistent(persistent.id) } } + assert(publicManualRevision2) assert.deepEqual((await pool.query("SELECT * FROM filter_rules WHERE server_id=$1 ORDER BY id", [sid])).rows, dbBefore) - const result = { accepted: true, cases, persistentAuthorityCases: 0, realMcpDispatcher: true, realSql: true, routerOsTransport: "declared-synthetic-localhost-http", + const result = { accepted: true, cases, publicManualRevision2, persistentAuthorityCases: 0, realMcpDispatcher: true, realSql: true, routerOsTransport: "declared-synthetic-localhost-http", genuineApprovedRunningOrigin: true, actualConfigurationCheckpoint: true, signedEffectTamperRejected: true, originalAutoRecoveryEnabled: false, fullRoutingAccepted: false, scopedControllerEnabled: false, nativeChrAccepted: false, forwardingVerified: false } console.log("Original effect SQL/runtime component:", JSON.stringify(result)) diff --git a/backend/src/modules/mcp/routing-community-scoped-service.ts b/backend/src/modules/mcp/routing-community-scoped-service.ts index 1d3f0c35..9ebe05a0 100644 --- a/backend/src/modules/mcp/routing-community-scoped-service.ts +++ b/backend/src/modules/mcp/routing-community-scoped-service.ts @@ -138,7 +138,9 @@ export async function prepareScopedCommunitySwitch(input: Record = { ...withPath, originalEffectSelection, nativeId: owned.nativeId, outIface, contextSources: value.context.before.sourceProofs, dependencyProof: value.context.before.dependencyProof, + const sealed: Record = { ...withPath, originalEffectSelection, + ...(!persistent && !profile.autoRecovery && originalEffectSelection.status === "prepared" ? { manualEffectPolicy: "observational-v1" } : {}), + nativeId: owned.nativeId, outIface, contextSources: value.context.before.sourceProofs, dependencyProof: value.context.before.dependencyProof, originalTargetProof: value.context.before.targetRuleProof, desiredTargetProof: owned.desiredTargetProof(compiled.compiledBody), receiptId: randomUUID(), proofRevision: persistent ? 3 : 2, originalCandidateId: value.context.before.currentCandidateId, originalRuleCiphertext: await encryptSecret(String(owned.rawRow.rule)), desiredRuleCiphertext: await encryptSecret(compiled.compiledBody), managementBindingCiphertext, rules: [publicRule(compiled.proposed)], beforeRules: [publicRule(compiled.before)] } @@ -174,7 +176,7 @@ export function registerScopedCommunityService() { try { // Optional original-effect vector is never an extra write/recovery grant. // Old manual configuration+probe acceptance remains authoritative. - if (context?.phase === "forward" && input.proofRevision === 3 && effectSelection.success && effectSelection.data.status === "prepared") { + if (context?.phase === "forward" && (input.proofRevision === 3 || input.proofRevision === 2 && input.manualEffectPolicy === "observational-v1") && effectSelection.success && effectSelection.data.status === "prepared") { try { const selected = effectSelection.data effectSession = await createRoutingNetworkSessionForCurrentExecution({ servers: scopedProfileIds(before.profile), diff --git a/backend/src/modules/mcp/routing-network-configuration-bridge.test.ts b/backend/src/modules/mcp/routing-network-configuration-bridge.test.ts index 6a3da884..bb7adea9 100644 --- a/backend/src/modules/mcp/routing-network-configuration-bridge.test.ts +++ b/backend/src/modules/mcp/routing-network-configuration-bridge.test.ts @@ -14,7 +14,7 @@ async function fixture() { let nativeReads = 0, profileChecks = 0, sourceReads = 0, configChecks = 0 let nativeBody = "desired", nativeDependency = "dependency", nativeTarget = "desired-target", nativeCandidate = "b", nativeId = "*1", generation = "generation" let afterNative: (() => void) | undefined, afterCheckpoint: (() => void) | undefined, afterProfile: (() => void) | undefined - let sharedReads = 0, sharedClosed = false + let sharedReads = 0, sharedClosed = false, manualOwner = false class ClockDate extends Date { static now() { return now } } class BridgeError extends Error { constructor(readonly code: string, message: string) { super(message) } } const fingerprint = (value: unknown) => JSON.stringify(value) @@ -37,6 +37,7 @@ async function fixture() { const sharedToken = Object.freeze({}), acquisition = Object.freeze({}), sharedBinding = { sourceOperationId: row.id, profile: structuredClone(profile), principal: structuredClone(principal), deadline: now + 10000, contextSources: structuredClone(input.contextSources), candidateId: "b" } const providers: Record = { + "./routing-network-execution-session.js": { isVerifiedManualEffectOwner: (owner: unknown) => owner === sharedToken && manualOwner }, zod: { z }, "@mmapp/contracts/mcp-routing": { routeAiProfileSchema: { parse: (value: unknown) => value } }, "../../db/index.js": { pool: { query: async () => { sourceReads++; return { rows: [{ ...row, plan_id: sourceChanged ? "changed" : row.plan_id }] } } } }, "../../lib/permissions.js": { hasPermission: (values: string[], permission: string) => values.includes(permission) }, @@ -93,7 +94,7 @@ async function fixture() { const reseal = () => { row.evidence.sealed = JSON.stringify(evidence) } return { resolve: () => resolve(row.id, action), resolveShared: (token = sharedToken) => resolveShared(token, action), sharedToken, sharedBinding, consume, close, resolveRound: () => resolveRound(sharedToken, action), readRound: (token: object, read = acquisition) => readRound(token, read), closeRound, acquisition, - row, input, profile, checkpoint, packet, action, authority, reseal, + row, input, profile, checkpoint, packet, action, authority, reseal, brandManual: () => { manualOwner = true }, stats: () => ({ nativeReads, profileChecks, sourceReads, configChecks, sharedReads, sharedClosed }), advance: (ms: number) => { now += ms }, revoke: () => { revoked = true }, sourceChanged: () => { sourceChanged = true }, checkpointMissing: () => { checkpointMissing = true }, checkpointChanged: () => { checkpointChanged = true }, guard: () => { guard = true }, @@ -110,6 +111,25 @@ test("fixed resolver captures opaque context and consume rechecks configuration assert.equal(f.stats().nativeReads, 2); assert(f.stats().profileChecks > 5) await assert.rejects(f.consume(token), /Unknown or consumed/); await assert.rejects(f.resolve(), /already issued/) }) + +for (const mode of ["positive", "unbranded", "wrong-checkpoint", "terminal-source", "standalone"] as const) { + test(`manual revision-2 round ${mode}`, async () => { + const f = await fixture() + f.input.proofRevision = 2; f.input.manualEffectPolicy = "observational-v1"; delete f.input.persistentProfileAuthority + f.row.state = mode === "terminal-source" ? "applied" : "running"; f.row.step_state = "intent" + f.packet.revision = mode === "wrong-checkpoint" ? 2 : 1 + if (mode !== "unbranded") f.brandManual() + f.reseal() + if (mode !== "positive") { + await assert.rejects(mode === "standalone" ? f.resolve() : f.resolveRound()) + assert.equal(f.stats().nativeReads, 0) + return + } + const round = await f.resolveRound(), observation = await f.readRound(round) + assert.equal(observation.sharedNetworkBudgetVerified, true) + assert.equal(observation.forwardingVerified, false) + }) +} test("caller objects, JSON, another resolver and closed contexts cannot be consumed", async () => { const f = await fixture(), other = await fixture(), token = await f.resolve() await assert.rejects(f.consume({}), /Unknown or consumed/); await assert.rejects(f.consume(JSON.parse(JSON.stringify(token))), /Unknown or consumed/) diff --git a/backend/src/modules/mcp/routing-network-configuration-bridge.ts b/backend/src/modules/mcp/routing-network-configuration-bridge.ts index d717e99d..65b69a62 100644 --- a/backend/src/modules/mcp/routing-network-configuration-bridge.ts +++ b/backend/src/modules/mcp/routing-network-configuration-bridge.ts @@ -8,6 +8,7 @@ import { findMcpOperation } from "./registry.js" import { accessGeneration, decryptSecret, fingerprint, McpError, redact } from "./security.js" import { assertScopedCommunityRequest } from "./routing-community-scoped-request.js" import { loadScopedConfigurationPacket } from "./routing-community-scoped-checkpoint.js" +import { isVerifiedManualEffectOwner } from "./routing-network-execution-session.js" import { assertPersistentScopedProfileAuthority, routingProfileAuthorityBindingSchema } from "./routing-profile-authority.js" import { readRouteCommunityContext, readRouteCommunityContextWithSharedNativeSession } from "./routing-community-context.js" import { readRoutingNetworkScopedCommunityContext } from "./routing-network-scoped-context.js" @@ -137,9 +138,16 @@ async function resolveConfigurationContext(sourceOperationId: string, action: Ac if (binding.sourceOperationId !== sourceOperationId || fingerprint(binding.profile) !== fingerprint(profile) || fingerprint(binding.contextSources) !== fingerprint(input.contextSources) || binding.candidateId !== input.candidateId) throw fail("Shared configuration source binding changed") } - if (input.proofRevision !== 3 || input.persistentProfileAuthority === undefined || profile.autoRecovery) throw fail("Persistent manual revision-3 source required; recovery guard remains closed") - const authority = routingProfileAuthorityBindingSchema.parse(input.persistentProfileAuthority) - deadline = Math.min(deadline, Date.parse(authority.packet.validUntil)) + const manual = !!sharedContext && deferredRoundReadback && isVerifiedManualEffectOwner(sharedContext) + if (profile.autoRecovery) throw fail("Recovery guard remains closed") + if (manual) { + if (row.state !== "running" || row.step_state !== "intent" || input.proofRevision !== 2 || + input.persistentProfileAuthority !== undefined || input.manualEffectPolicy !== "observational-v1") throw fail("Exact running manual effect source required") + } else { + if (input.proofRevision !== 3 || input.persistentProfileAuthority === undefined) throw fail("Persistent manual revision-3 source required") + const authority = routingProfileAuthorityBindingSchema.parse(input.persistentProfileAuthority) + deadline = Math.min(deadline, Date.parse(authority.packet.validUntil)) + } signal = AbortSignal.any([signal, AbortSignal.timeout(Math.max(1, deadline - Date.now()))]) active() const ids = [...new Set([profile.serverId, ...profile.candidates.flatMap(candidate => [candidate.jumpHostId, candidate.exitNodeId])])].sort((a,b) => a-b) @@ -160,7 +168,7 @@ async function resolveConfigurationContext(sourceOperationId: string, action: Ac const checkpoint = async () => { await authorize() const result = await guarded(() => loadScopedConfigurationPacket(input, requestSeal, { operationId: sourceOperationId, stepIndex: 0 })) - if (!result || result.packet.revision !== 2 || result.packet.operationId !== sourceOperationId || result.packet.stepIndex !== 0 || + if (!result || result.packet.revision !== (manual ? 1 : 2) || result.packet.operationId !== sourceOperationId || result.packet.stepIndex !== 0 || result.packet.phase !== "CONFIG_APPLIED" || Date.parse(result.observedAt) < row.created_at.getTime() || Date.parse(result.observedAt) > Date.now() || Date.parse(result.observedAt) >= deadline) throw fail("Genuine persistent CONFIG_APPLIED unavailable") await authorize() diff --git a/backend/src/modules/mcp/routing-network-execution-origin.test.mjs b/backend/src/modules/mcp/routing-network-execution-origin.test.mjs index da696348..c6cf94bb 100644 --- a/backend/src/modules/mcp/routing-network-execution-origin.test.mjs +++ b/backend/src/modules/mcp/routing-network-execution-origin.test.mjs @@ -21,7 +21,7 @@ function fixture() { const lifetime = { expiresAt: now + 60000, monotonicDeadline: performance.now() + 60000, signal: new AbortController().signal }; let calls = 0, inspections = 0, closed = 0, options, query, originAvailable = true, afterCreate = () => {}; const connection = { owner_valid_until: new Date(now + 110000), policy: { expiresAt: new Date(now + 120000).toISOString() } }; - const token = Object.freeze({}); + const token = Object.freeze({}), sharedContext = Object.freeze({}); const providers = { 'node:crypto': { createHash }, '../../db/index.js': { pool: { query: async value => { query = value; return { rows: [row] }; } } }, @@ -34,15 +34,35 @@ function fixture() { './registry.js': { findMcpOperation: () => ({ version: 7 }) }, './security.js': { fingerprint: JSON.stringify, redact: value => value, McpError: Error }, './routing-profile-authority.js': { routingProfileAuthorityBindingSchema: { parse: value => value } }, - './routing-network-native-session.js': { createRoutingNetworkNativeSession: async value => { calls++; options = value; afterCreate(); return { close() { closed++; } }; } } + './routing-network-native-session.js': { createRoutingNetworkNativeSession: async value => { calls++; options = value; afterCreate(); return { sharedContext, close() { closed++; } }; } } }; const exports = {}; new Function('exports', 'require', code)(exports, name => { assert.ok(providers[name], name); return providers[name]; }); const manifest = { servers: [1, 2], probes: [] }; - return { metadata, row, lifetime, connection, manifest, now, run: () => exports.createRoutingNetworkSessionForCurrentExecution(manifest), + return { metadata, row, lifetime, connection, manifest, now, sharedContext, isManual: exports.isVerifiedManualEffectOwner, run: () => exports.createRoutingNetworkSessionForCurrentExecution(manifest), revokeOrigin: () => { originAvailable = false; }, afterCreate: fn => { afterCreate = fn; }, stats: () => ({ calls, closed, inspections, options, query }) }; } +for (const mode of ['positive', 'missing-policy', 'missing-selection', 'recovery', 'late-revocation']) { + test(`manual effect owner origin ${mode}`, async () => { + const f = fixture(); + f.metadata.serviceInput = { proofRevision: 2, manualEffectPolicy: 'observational-v1', profile: { autoRecovery: mode === 'recovery' }, originalEffectSelection: { status: 'prepared' } }; + if (mode === 'missing-policy') delete f.metadata.serviceInput.manualEffectPolicy; + if (mode === 'missing-selection') delete f.metadata.serviceInput.originalEffectSelection; + if (mode === 'late-revocation') f.afterCreate(() => { f.metadata.approved = false; }); + f.row.intended_state.input = structuredClone(f.metadata.serviceInput); + if (mode === 'positive') { + await f.run(); + assert.equal(f.isManual(f.sharedContext), true); + assert.equal(f.isManual(JSON.parse(JSON.stringify(f.sharedContext))), false); + } else { + await assert.rejects(f.run()); + assert.equal(f.isManual(f.sharedContext), false); + assert.equal(f.stats().calls, mode === 'late-revocation' ? 1 : 0); + assert.equal(f.stats().closed, mode === 'late-revocation' ? 1 : 0); + } + }); +} test('real origin supplies operation, principal, signal and minimum immutable horizon', async () => { const f = fixture(); await f.run(); const { options, query } = f.stats(); assert.equal(options.sourceOperationId, f.metadata.operationId); assert.equal(options.context.signal, f.lifetime.signal); diff --git a/backend/src/modules/mcp/routing-network-execution-session.ts b/backend/src/modules/mcp/routing-network-execution-session.ts index 7d214cf6..7231a12a 100644 --- a/backend/src/modules/mcp/routing-network-execution-session.ts +++ b/backend/src/modules/mcp/routing-network-execution-session.ts @@ -6,7 +6,14 @@ import { effectiveMcpConnection } from "./store.js" import { findMcpOperation } from "./registry.js" import { fingerprint, McpError, redact } from "./security.js" import { routingProfileAuthorityBindingSchema } from "./routing-profile-authority.js" -import { createRoutingNetworkNativeSession, type RoutingNetworkNativeManifest } from "./routing-network-native-session.js" +import { createRoutingNetworkNativeSession, type RoutingNetworkNativeManifest, type RoutingNetworkSharedContext } from "./routing-network-native-session.js" + +const manualEffectOwners = new WeakSet() + +/** Only the verified execution factory can mark a genuine shared owner. */ +export function isVerifiedManualEffectOwner(owner: RoutingNetworkSharedContext): boolean { + return manualEffectOwners.has(owner) +} /** Private producer for the existing native/configuration-round reader. The * manifest selects reads, never identity or authority. Its exact scope, probe @@ -21,6 +28,10 @@ export async function createRoutingNetworkSessionForCurrentExecution(manifest: R if (!initial.approved || initial.phase !== "forward" || initial.stepIndex !== 0 || initial.operationName !== "routing.community.switch" || initial.operationVersion !== findMcpOperation(initial.operationName)?.version || initial.serviceName !== "routing.community.scoped") throw fail() + const input = initial.serviceInput, manual = input.proofRevision === 2 + if (manual && (input.manualEffectPolicy !== "observational-v1" || input.persistentProfileAuthority !== undefined || + (input.profile as { autoRecovery?: unknown } | undefined)?.autoRecovery !== false || + (input.originalEffectSelection as { status?: unknown } | undefined)?.status !== "prepared")) throw fail() const lifetime = await inspectVerifiedMcpServiceExecutionLifetime(origin) const authorize = async () => { const current = await inspectVerifiedMcpServiceExecution(origin) @@ -57,6 +68,10 @@ export async function createRoutingNetworkSessionForCurrentExecution(manifest: R if (!Number.isFinite(readDeadline) || Date.now() >= readDeadline || lifetime.signal.aborted || performance.now() >= lifetime.monotonicDeadline) throw fail() const session = await createRoutingNetworkNativeSession({ sourceOperationId: initial.operationId, context: { principal: initial.principal, connection, signal: lifetime.signal }, authorize, deadline: readDeadline, manifest }) - try { await authorize(); return session } + try { + await authorize() + if (manual) manualEffectOwners.add(session.sharedContext) + return session + } catch (error) { session.close(); throw error } } diff --git a/backend/src/modules/mcp/routing-original-effects-caller.test.mjs b/backend/src/modules/mcp/routing-original-effects-caller.test.mjs index b0495230..c341938a 100644 --- a/backend/src/modules/mcp/routing-original-effects-caller.test.mjs +++ b/backend/src/modules/mcp/routing-original-effects-caller.test.mjs @@ -43,6 +43,7 @@ for(const mode of ['legacy','unknown'])test(`existing ${mode} manual path does n for(const failure of ['open','baseline','round','round-closed','publication'])test(`optional ${failure} uncertainty preserves actual manual config/control receipt with effects unknown`,async()=>{const f=await fixture('prepared',failure),result=await f.run();assert.equal(result.configurationVerified,true);assert.equal(result.controlProbeVerified,true);assert.equal(result.originalEffects.status,'unknown');assert.equal(f.calls.filter(x=>x==='PATCH').length,1);assert(f.calls.includes('save-network'));}); test('foreign configuration drift after optional effect observations still refuses publication, never repeats PATCH',async()=>{const f=await fixture('prepared','foreign');await assert.rejects(f.run);assert.equal(f.calls.filter(x=>x==='PATCH').length,1);assert(!f.calls.includes('save-network'));assert.equal(f.stats().closed,1);}); test('revision2 remains on existing manual path even with a prepared-like field',async()=>{const f=await fixture();f.input.proofRevision=2;await f.run();assert(!f.calls.includes('open-session'));}); +test('sealed manual policy runs read-only rounds through the real scoped consumer',async()=>{const f=await fixture();f.input.proofRevision=2;f.input.manualEffectPolicy='observational-v1';const result=await f.run();assert.equal(f.calls.filter(x=>x==='PATCH').length,1);assert(f.calls.includes('round:2'));assert.equal(result.originalEffects.status,'observed-expected');}); test('actual preparation consumes the required real-context selection method and includes it in sealed service input',async()=>{const f=await fixture(),evidence=await f.prepare();assert(f.calls.includes('capture-prepared-selection'));assert.deepEqual(evidence.steps[0].input.originalEffectSelection,f.input.originalEffectSelection);assert.equal(evidence.steps[0].input.requestProof,'seal');assert(!f.calls.includes('PATCH'));}); test('closed effect owner falls back to exactly the old final context read without bypassing publication guard',async()=>{const legacy=await fixture('legacy');await legacy.run();const f=await fixture('prepared','round-closed');await f.run();assert(!f.calls.includes('shared-publication'));assert.equal(f.stats().reads,legacy.stats().reads);}); diff --git a/backend/src/routes/mcp-oauth-database-admission.test.ts b/backend/src/routes/mcp-oauth-database-admission.test.ts index 992dce92..ffe350dd 100644 --- a/backend/src/routes/mcp-oauth-database-admission.test.ts +++ b/backend/src/routes/mcp-oauth-database-admission.test.ts @@ -50,7 +50,7 @@ async function fixture() { }, "../modules/mcp/security.js": { McpError, accessGeneration: async () => generation, constantEqual: (a: string, b: string) => a === b, randomToken: (kind: string) => kind + "-owned-inert", tokenDigest: (token: string) => crypto.createHash("sha256").update(token).digest("hex") }, "./mcp-transport.js": { mcpPublicUrl: () => new URL(resource) }, - "../modules/mcp/portal-revocation.js": { assertPortalClaimsCurrent: async () => {} }, + "../modules/mcp/portal-revocation.js": { assertPortalClaimsCurrent: async () => { await query("SELECT invalid_before FROM mcp_owner_revocations") } }, }) const api = await load<{ installDatabaseRequestGate(app: FastifyInstance): void }>(new URL("../plugins/database-request-gate.ts", import.meta.url), { "../services/shared-database-gate.js": shared, @@ -118,7 +118,7 @@ test("both pure discovery endpoints remain200 with no SQL or admission while mai assert.deepEqual(f.counts(), before) } finally { await f.app.close() } }) -test("public OAuth and existing API admin handlers acquire exactly one gate admission", async () => { +test("OAuth admits public SQL once and protects both API admin SQL stages", async () => { const f = await fixture() try { const before = f.counts().admissions @@ -126,7 +126,13 @@ test("public OAuth and existing API admin handlers acquire exactly one gate admi assert.equal(f.counts().admissions - before, 1) const middle = f.counts().admissions assert.equal((await f.app.inject("/api/mcp/oauth/clients")).statusCode, 200) - assert.equal(f.counts().admissions - middle, 1) + assert.equal(f.counts().admissions - middle, 2) + await f.gate.exclusive(async () => { + const sqlBefore = f.counts().sqlCalls + const response = await f.app.inject("/api/mcp/oauth/clients") + assert.equal(response.statusCode, 503) + assert.equal(f.counts().sqlCalls, sqlBefore) + }) } finally { await f.app.close() } }) test("ordinary invalid OAuth input preserves its existing protocol error and no-store behavior", async () => { diff --git a/backend/src/routes/mcp-oauth.ts b/backend/src/routes/mcp-oauth.ts index ec6f326d..7168df46 100644 --- a/backend/src/routes/mcp-oauth.ts +++ b/backend/src/routes/mcp-oauth.ts @@ -109,7 +109,8 @@ export const mcpOauth: FastifyPluginAsync = async app => { if (!env.authRequired) return reply.code(503).send({ error: "Требуется авторизация портала" }) await requireAuth(request, reply) if (reply.sent) return - await assertPortalClaimsCurrent(request.authUser!) + // The generic API gate admits the handler, not this SQL-bearing hook. + await sharedDatabaseGate.request(() => assertPortalClaimsCurrent(request.authUser!)) if (request.headers.origin && request.headers.origin !== env.CORS_ORIGIN) return reply.code(403).send({ error: "Недопустимый Origin" }) }) admin.get("/api/mcp/oauth/clients", async () => ({ clients: (await pool.query("SELECT * FROM mcp_oauth_clients ORDER BY created_at DESC LIMIT 100")).rows })) diff --git a/backend/src/routes/mcp.test.ts b/backend/src/routes/mcp.test.ts index b81c727e..31daabc8 100644 --- a/backend/src/routes/mcp.test.ts +++ b/backend/src/routes/mcp.test.ts @@ -10,6 +10,7 @@ const { default: authPlugin } = await import("../plugins/auth.js") const { createMcpRoutes, mcpDevicePlanBodySchema, mcpDevicePlanArguments } = await import("./mcp.js") const { findAction } = await import("../modules/mcp/actions.js") const { pool } = await import("../db/index.js") +const { sharedDatabaseGate } = await import("../services/shared-database-gate.js") const { env } = await import("../config.js") const { mcpPolicyPreviewRequestSchema, mcpCatalogResponseSchema, mcpPolicyPreviewResponseSchema } = await import("@mmapp/contracts/mcp") const { listMcpOperations } = await import("../modules/mcp/registry.js") @@ -37,6 +38,12 @@ const payload = mcpPolicyPreviewRequestSchema.parse({ target: { scope: "servers", serverIds: [1] }, }) try { + await sharedDatabaseGate.exclusive(async () => { + const before = claimChecks + const response = await app.inject({ url: "/api/mcp/catalog", headers: admin }) + assert.equal(response.statusCode, 503) + assert.equal(claimChecks, before, "Maintenance stops the claims hook before SQL") + }) for (const request of [{ method: "GET" as const, url: "/api/mcp/catalog" }, { method: "POST" as const, url: "/api/mcp/policy/preview", payload }]) { assert.equal((await app.inject(request)).statusCode, 401) assert.equal((await app.inject({ ...request, headers: { authorization: "Bearer arbitrary-agent-token" } })).statusCode, 401) diff --git a/backend/src/routes/mcp.ts b/backend/src/routes/mcp.ts index 700265bf..a42df6b4 100644 --- a/backend/src/routes/mcp.ts +++ b/backend/src/routes/mcp.ts @@ -24,6 +24,7 @@ import { resourceById } from "../modules/mcp/resources.js" import { registerManagementChannel } from "../modules/mcp/management.js" import { MAX_RESTORE_ARCHIVE_BYTES, uploadRestoreArchive, readRestoreReceipt, readRestoreReceiptAuditStatus } from "../modules/mcp/database-restore.js" import { assertPortalClaimsCurrent } from "../modules/mcp/portal-revocation.js" +import { sharedDatabaseGate } from "../services/shared-database-gate.js" import { readFile } from "node:fs/promises" import { checkMcpProtocol, type ProtocolProbe } from "../modules/mcp/readiness.js" import { humanBackupDownloadMetadata, readHumanBackupDownloadArtifact } from "../modules/mcp/backup-download-artifact.js" @@ -96,7 +97,8 @@ export function createMcpRoutes(checkClaims = assertPortalClaimsCurrent, options // Explicit guard: these routes must stay protected if the global API hook changes. await requireAuth(request, reply) if (reply.sent) return - await checkClaims(actorOf(request.authUser)) + // This hook runs before the API handler's database admission. + await sharedDatabaseGate.request(() => checkClaims(actorOf(request.authUser))) if (request.headers.origin && request.headers.origin !== env.CORS_ORIGIN) return reply.code(403).send({ error: "Недопустимый Origin" }) }) diff --git a/backend/src/test/traffic-flow-chr-capture-check.test.ts b/backend/src/test/traffic-flow-chr-capture-check.test.ts new file mode 100644 index 00000000..ebff717c --- /dev/null +++ b/backend/src/test/traffic-flow-chr-capture-check.test.ts @@ -0,0 +1,68 @@ +import assert from "node:assert/strict" +import { test } from "node:test" +import { checkChrCapture } from "./traffic-flow-chr-capture-check.js" + +function fixture() { + const exporters = [ + { address: "192.0.2.10", port: 2055, version: 9, domain: 7, templateId: 256 }, + { address: "192.0.2.20", port: 2055, version: 10, domain: 7, templateId: 256 }, + ] + const expected = { src: "192.0.2.1", dst: "198.51.100.1", bytes: 1200, packets: 3, inIface: "11", outIface: "12" } + const fields = [8, 12, 1, 2, 10, 14] + const data = Buffer.alloc(24) + ;[0xc0000201, 0xc6336401, 1200, 3, 11, 12].forEach((value, index) => data.writeUInt32BE(value, index * 4)) + const frames = (["template", "data", "refresh", "data"] as const).flatMap((kind, phase) => exporters.map((source, exporter) => { + const isTemplate = kind === "template" || kind === "refresh" + const reversed = kind === "refresh" || (kind === "template" && exporter === 1) + const body = isTemplate ? Buffer.alloc(28) : data + if (isTemplate) { + body.writeUInt16BE(source.templateId) + body.writeUInt16BE(fields.length, 2) + const ordered = reversed ? [12, 8, ...fields.slice(2)] : fields + ordered.forEach((id, index) => { body.writeUInt16BE(id, 4 + index * 4); body.writeUInt16BE(4, 6 + index * 4) }) + } + const header = source.version === 9 ? 20 : 16 + const packet = Buffer.alloc(header + 4 + body.length) + packet.writeUInt16BE(source.version) + packet.writeUInt16BE(source.version === 9 ? 1 : packet.length, 2) + packet.writeUInt32BE(source.domain, header - 4) + packet.writeUInt16BE(isTemplate ? source.version === 9 ? 0 : 2 : source.templateId, header) + packet.writeUInt16BE(body.length + 4, header + 2) + body.copy(packet, header + 4) + return { + exporter, kind: kind as string, ordinal: phase * 2 + exporter + 1, receivedMs: phase + 1, payloadHex: packet.toString("hex"), + expected: isTemplate ? [] : [phase === 3 || exporter === 1 ? { ...expected, src: expected.dst, dst: expected.src } : { ...expected }], + } + })) + return { destination: "192.0.2.100", exporters, frames } +} + +test("interleaved v9 and IPFIX replay matches exact counters and ifIndex after same-port template reuse", () => { + const result = checkChrCapture(fixture()) + assert.equal(result.captureReplayMatched, true) + assert.deepEqual(result.exporters.map(row => [row.baselineRecords, row.refreshedRecords]), [[1, 1], [1, 1]]) + assert.equal(result.captureProvenanceVerified, false) + assert.equal(result.restartIsolationAccepted, false) +}) + +test("refuses swapped exporter identity, altered ifIndex, unordered frames and missing refreshed traffic", () => { + const cases = [ + (input: ReturnType) => { input.frames[2].exporter = 1 }, + (input: ReturnType) => { input.frames[2].expected[0].inIface = "99" }, + (input: ReturnType) => { input.frames[2].ordinal = input.frames[1].ordinal }, + (input: ReturnType) => { input.frames[7] = { ...input.frames[5], ordinal: 8, receivedMs: 8 } }, + ] + for (const mutate of cases) { + const input = fixture() + mutate(input) + assert.throws(() => checkChrCapture(input)) + } +}) + +test("same-tuple pre-refresh packet still decodes with previous template and refuses restart evidence", () => { + const input = fixture() + input.frames.splice(4, 0, { ...input.frames[2], kind: "pre-refresh-probe", expected: [] }) + input.frames.forEach((frame, index) => { frame.ordinal = index + 1; frame.receivedMs = index + 1 }) + assert.throws(() => checkChrCapture(input), /reused the previous template/) + assert.equal(checkChrCapture(fixture()).captureReplayMatched, true) +}) diff --git a/backend/src/test/traffic-flow-chr-capture-check.ts b/backend/src/test/traffic-flow-chr-capture-check.ts new file mode 100644 index 00000000..ce5eb53c --- /dev/null +++ b/backend/src/test/traffic-flow-chr-capture-check.ts @@ -0,0 +1,128 @@ +/** Offline replay of a reviewed UDP capture projection through the current parser. */ +import { createHash } from "node:crypto" +import { open } from "node:fs/promises" +import { isIP } from "node:net" +import { resolve } from "node:path" +import { fileURLToPath } from "node:url" +import { isDeepStrictEqual } from "node:util" +import { z } from "zod" +import { clearFlowTemplateCache, FlowTemplateRefused, parseFlowPacket } from "../services/traffic-flow-parse.js" + +const maxInput = 512 * 1024 +const ip = z.string().max(45).refine(value => isIP(value) !== 0) +const ipv4 = ip.refine(value => isIP(value) === 4) +const counter = z.number().int().nonnegative().safe() +const iface = z.string().regex(/^(?:0|[1-9][0-9]{0,9})$/) +const flow = z.object({ src: ip, dst: ip, bytes: counter, packets: counter, inIface: iface, outIface: iface }).strict() +const exporter = z.object({ + address: ipv4, port: z.number().int().min(1).max(65535), version: z.union([z.literal(9), z.literal(10)]), + domain: counter.max(0xffffffff), templateId: z.number().int().min(256).max(65535), +}).strict() +const capture = z.object({ + destination: ipv4, + exporters: z.tuple([exporter, exporter]), + frames: z.array(z.object({ + exporter: z.union([z.literal(0), z.literal(1)]), + ordinal: counter, + receivedMs: counter, + payloadHex: z.string().min(40).max(8192).regex(/^(?:[0-9a-f]{2})+$/), + kind: z.enum(["template", "data", "refresh", "pre-refresh-probe"]), + expected: z.array(flow).max(64), + }).strict()).min(8).max(64), +}).strict() + +function requireEvidence(condition: unknown, reason: string): asserts condition { + if (!condition) throw Error(reason) +} + +/** Synchronous, isolated replay. Call in an offline process with no live parser. */ +export function checkChrCapture(input: unknown) { + const parsed = capture.safeParse(input) + requireEvidence(parsed.success, "Invalid capture projection") + const { destination, exporters, frames } = parsed.data + requireEvidence(exporters[0].address !== exporters[1].address && exporters.every(row => row.address !== destination), "Expected two distinct exporter addresses and one destination") + const state = exporters.map(() => ({ phase: 0, baselineRecords: 0, refreshedRecords: 0, missingTemplateObserved: false })) + let lastOrdinal = -1, lastReceivedMs = -1 + clearFlowTemplateCache() + try { + for (const frame of frames) { + requireEvidence(frame.ordinal > lastOrdinal && frame.receivedMs >= lastReceivedMs, "Frames must have strict capture order") + lastOrdinal = frame.ordinal + lastReceivedMs = frame.receivedMs + const source = exporters[frame.exporter], current = state[frame.exporter] + const packet = Buffer.from(frame.payloadHex, "hex") + const header = source.version === 9 ? 20 : 16 + requireEvidence(packet.length >= header + 8 && packet.readUInt16BE(0) === source.version && + packet.readUInt32BE(header - 4) === source.domain && + (source.version === 9 || packet.readUInt16BE(2) === packet.length) && + packet.readUInt16BE(header + 2) === packet.length - header, "Packet does not match the declared exporter or single-set envelope") + const isTemplate = frame.kind === "template" || frame.kind === "refresh" + requireEvidence(packet.readUInt16BE(header) === (isTemplate ? source.version === 9 ? 0 : 2 : source.templateId), "Unexpected set ID") + if (isTemplate) { + requireEvidence(packet.readUInt16BE(header + 4) === source.templateId && frame.expected.length === 0, "Unexpected template ID or template records") + requireEvidence(frame.kind === "template" ? current.phase === 0 : current.phase === 1 && current.baselineRecords > 0, "Unexpected template transition") + } else { + requireEvidence(current.phase > 0, "Data preceding baseline template") + } + const transport = { sourcePort: source.port, collectorSocketId: "offline_capture_socket" } + if (frame.kind === "pre-refresh-probe") { + requireEvidence(current.phase === 1 && current.baselineRecords > 0 && !current.missingTemplateObserved && frame.expected.length === 0, "Unexpected pre-refresh probe") + try { + parseFlowPacket(packet, source.address, transport) + } catch (error) { + if (!(error instanceof FlowTemplateRefused) || error.reason !== "missing-template") throw Error("Probe did not yield a missing-template refusal") + current.missingTemplateObserved = true + } + requireEvidence(current.missingTemplateObserved, "Pre-refresh data reused the previous template") + continue + } + const actual = parseFlowPacket(packet, source.address, transport).map(({ src, dst, bytes, packets, inIface, outIface }) => ({ src, dst, bytes, packets, inIface, outIface })) + requireEvidence(isDeepStrictEqual(actual, frame.expected), "Decoded flow records differ from the expected capture projection") + if (isTemplate) current.phase++ + else { + requireEvidence(actual.length > 0 && actual.every(row => row.bytes > 0 && row.packets > 0), "Expected positive traffic after the template") + if (current.phase === 1) current.baselineRecords += actual.length + else current.refreshedRecords += actual.length + } + } + requireEvidence(state.every(row => row.baselineRecords > 0 && row.refreshedRecords > 0), "Both exporters need positive baseline and refreshed traffic") + return { + captureReplayMatched: true, + exporterCount: state.length, + frames: frames.length, + exporters: state.map(({ baselineRecords, refreshedRecords, missingTemplateObserved }) => ({ baselineRecords, refreshedRecords, missingTemplateObserved })), + captureProvenanceVerified: false, + liveCollectorAccepted: false, + restartIsolationAccepted: false, + } + } finally { + clearFlowTemplateCache() + } +} + +async function main() { + const [path, expectedSha, ...extra] = process.argv.slice(2) + requireEvidence(path && /^[a-f0-9]{64}$/.test(expectedSha ?? "") && !extra.length, "Pass a capture projection file and its SHA-256") + const handle = await open(path, "r") + try { + const info = await handle.stat() + requireEvidence(info.isFile() && info.size > 0 && info.size <= maxInput, "Capture projection file is outside its size bound") + const bytes = Buffer.alloc(maxInput + 1) + let count = 0 + while (count < bytes.length) { + const { bytesRead } = await handle.read(bytes, count, bytes.length - count, count) + if (!bytesRead) break + count += bytesRead + } + requireEvidence(count === info.size, "Capture projection changed while reading") + const body = bytes.subarray(0, count) + requireEvidence(createHash("sha256").update(body).digest("hex") === expectedSha, "Capture projection SHA-256 differs") + console.log(JSON.stringify({ projectionSha256: expectedSha, ...checkChrCapture(JSON.parse(body.toString("utf8"))) })) + } finally { + await handle.close() + } +} + +if (process.argv[1] && resolve(process.argv[1]) === fileURLToPath(import.meta.url)) { + main().catch(() => { console.error("CHR capture projection refused"); process.exitCode = 1 }) +} diff --git a/backend/src/test/traffic-flow-chr-owned-sequence.test.ts b/backend/src/test/traffic-flow-chr-owned-sequence.test.ts new file mode 100644 index 00000000..e3fa383d --- /dev/null +++ b/backend/src/test/traffic-flow-chr-owned-sequence.test.ts @@ -0,0 +1,95 @@ +import assert from "node:assert/strict" +import { createHash } from "node:crypto" +import { mkdtemp, readFile, rm } from "node:fs/promises" +import { tmpdir } from "node:os" +import { join } from "node:path" +import { test } from "node:test" +import { bindChrCapture, executeOwnedChrCapture } from "./traffic-flow-chr-owned-sequence.js" + +type Options = Parameters[0] +const scope = { sources: ["192.0.2.1", "192.0.2.2"] as const, destination: "192.0.2.100", port: 4739 } +const sha = (bytes: Buffer) => createHash("sha256").update(bytes).digest("hex") +function fixtures(at = 1700000000000, port = scope.port) { + const pcap = Buffer.alloc(24) + pcap.writeUInt32LE(0xa1b2c3d4); pcap.writeUInt16LE(2, 4); pcap.writeUInt16LE(4, 6) + pcap.writeUInt32LE(4124, 16); pcap.writeUInt32LE(101, 20) + const sourceRows = scope.sources.map(address => ({ address, port: 2055, version: 10, domain: 7, templateId: 256 })) + const records: Buffer[] = [] + const frames = (["template", "data", "refresh", "data"] as const).flatMap(kind => sourceRows.map((_source, exporter) => { + const template = kind !== "data" + const payload = Buffer.alloc(template ? 48 : 44) + payload.writeUInt16BE(10); payload.writeUInt16BE(payload.length, 2); payload.writeUInt32BE(7, 12) + payload.writeUInt16BE(template ? 2 : 256, 16); payload.writeUInt16BE(payload.length - 16, 18) + if (template) { + payload.writeUInt16BE(256, 20); payload.writeUInt16BE(6, 22) + ;[8, 12, 1, 2, 10, 14].forEach((field, index) => { payload.writeUInt16BE(field, 24 + index * 4); payload.writeUInt16BE(4, 26 + index * 4) }) + } else { + ;[0xc0000201, 0xc6336401, 1200, 3, 11, 12].forEach((value, index) => payload.writeUInt32BE(value, 20 + index * 4)) + } + const record = Buffer.alloc(44 + payload.length), ordinal = records.length + 1 + record.writeUInt32LE(Math.floor(at / 1000)); record.writeUInt32LE(at % 1000 * 1000, 4) + record.writeUInt32LE(record.length - 16, 8); record.writeUInt32LE(record.length - 16, 12) + const ip = record.subarray(16) + ip[0] = 0x45; ip.writeUInt16BE(ip.length, 2); ip[9] = 17 + Buffer.from([192, 0, 2, exporter + 1, 192, 0, 2, 100]).copy(ip, 12) + ip.writeUInt16BE(2055, 20); ip.writeUInt16BE(port, 22); ip.writeUInt16BE(ip.length - 20, 24) + payload.copy(ip, 28) + records.push(record) + return { kind, ordinal, receivedMs: at, exporter, payloadHex: payload.toString("hex"), + expected: template ? [] : [{ src: "192.0.2.1", dst: "198.51.100.1", bytes: 1200, packets: 3, inIface: "11", outIface: "12" }] } + })) + const bytes = Buffer.concat([pcap, ...records]) + return { bytes, sha256: sha(bytes), dropped: 0, complete: true, projection: { destination: scope.destination, exporters: sourceRows, frames } } +} + +test("all reviewed datagrams are bound to the original ordered pcap bytes before replay", () => { + const capture = fixtures() + assert.equal(bindChrCapture(capture, scope).wire.frames.length, 8) + assert.throws(() => bindChrCapture({ ...capture, dropped: 1 }, scope), /completeness/) + assert.throws(() => bindChrCapture({ ...capture, sha256: "0".repeat(64) }, scope), /pin/) + const changed = structuredClone(capture.projection) + changed.frames[2].payloadHex = "ff" + changed.frames[2].payloadHex.slice(2) + assert.throws(() => bindChrCapture({ ...capture, projection: changed }, scope), /differs from wire/) +}) + +test("owned journal is synchronized before target dispatch and captures target cleanup before reconciliation", async () => { + const directory = await mkdtemp(join(tmpdir(), "mm-chr-sequence-")) + const journalPath = join(directory, "owned.jsonl") + try { + const capture = fixtures(Date.now() - 1000, 4740), events: string[] = [] + let bound = true + const device = (deviceId: string): Options["targets"][0] => { + const baseline = { deviceId, flow: { enabled: "true" }, targets: [], + ipfix: Object.fromEntries(["bytes", "packets", "src-address", "dst-address", "protocol", "src-port", "dst-port", "in-interface", "out-interface"].map(key => [key, true])) } + let rows: Record[] = [] + const intentSaved = async (action: string) => { + const records = (await readFile(journalPath, "utf8")).trim().split("\n").map(line => JSON.parse(line)) + assert(records.some(row => row.action === action && row.phase === "intent" && row.deviceId === deviceId)) + } + return { baseline, read: async () => ({ ...baseline, targets: rows }), native: { + async put(_path: string, body: Record, _timeout?: number, fence?: () => Promise) { + await fence?.(); await intentSaved("create"); events.push("create:" + deviceId) + rows = [{ ...body, ".id": "*A", disabled: false }] + return { ".id": "*A" } as T + }, + async delete(_path, _timeout, fence) { await fence?.(); await intentSaved("delete"); rows = []; events.push("delete:" + deviceId) }, + } } + } + const options: Options = { + journalPath, deadlineAt: Date.now() + 5000, scope, ingress: { bytes: capture.bytes, sha256: capture.sha256, existingPort: 4740 }, + serverIds: [1, 2], targets: [device("A"), device("B")], + collector: { authorize: async () => {}, assertBound: async () => { assert(bound) }, restore: async () => { bound = false; events.push("restore") } }, + capture: async run => { events.push("start"); await run(); events.push("stop"); return fixtures(Date.now()) }, + exercise: async () => { events.push("exercise") }, + reconcile: async (wire, ids) => { assert.equal(bound, false); assert.equal(wire.frames.length, 8); assert.deepEqual(ids, [1, 2]); events.push("reconcile") }, + } + await assert.rejects(executeOwnedChrCapture({ ...options, ingress: { ...options.ingress, sha256: "0".repeat(64) } }), /ingress SHA/) + assert.deepEqual(events, []) + assert.equal((await executeOwnedChrCapture(options)).nativeAcceptance, false) + assert.deepEqual(events, ["start", "create:A", "create:B", "exercise", "delete:B", "delete:A", "stop", "restore", "reconcile"]) + await assert.rejects(executeOwnedChrCapture(options), { code: "EEXIST" }) + assert.equal(events.length, 9) + } finally { + await rm(directory, { recursive: true, force: true }) + } +}) diff --git a/backend/src/test/traffic-flow-chr-owned-sequence.ts b/backend/src/test/traffic-flow-chr-owned-sequence.ts new file mode 100644 index 00000000..273959c5 --- /dev/null +++ b/backend/src/test/traffic-flow-chr-owned-sequence.ts @@ -0,0 +1,128 @@ +/** Local composition for a future guarded runner. Callbacks are trusted runner + * boundaries, never populated from a request or evidence JSON. */ +import { open } from "node:fs/promises" +import { dirname } from "node:path" +import { createChrTargetHarness } from "./traffic-flow-chr-target-harness.js" +import { projectChrWire } from "./traffic-flow-chr-wire.js" +import { checkChrCapture } from "./traffic-flow-chr-capture-check.js" + +type Target = Parameters[0] +type Scope = Parameters[1] +type Wire = ReturnType +type Projection = { + destination: string + exporters: readonly { address: string; port: number; version: number; domain: number }[] + frames: readonly { exporter: number; ordinal: number; receivedMs: number; payloadHex: string }[] +} +type Capture = { bytes: Buffer; sha256: string; dropped: number; complete: boolean; projection: Projection } +function must(value: unknown, reason: string): asserts value { + if (!value) throw Error(reason) +} + +/** Bind every reviewed row to its original wire record and require complete + * capture coverage for conservation. Sample-only replay uses the separate CLI. */ +export function bindChrCapture(capture: Capture, scope: Scope) { + const wire = projectChrWire(capture.bytes, scope) + const { projection } = capture + must(wire.captureSha256 === capture.sha256 && capture.dropped === 0 && capture.complete === true, + "Capture provenance pin or completeness differs") + must(projection.destination === scope.destination && projection.exporters.length === 2 && + projection.frames.length === wire.frames.length, "Projection scope or coverage differs") + for (const [index, frame] of wire.frames.entries()) { + const row = projection.frames[index], exporter = projection.exporters[frame.exporter] + must(row && row.ordinal === frame.ordinal && row.receivedMs === frame.receivedMs && + row.exporter === frame.exporter && row.payloadHex === frame.payloadHex && + exporter?.address === scope.sources[frame.exporter] && exporter.port === frame.sourcePort && + exporter.version === frame.version && exporter.domain === frame.domain, "Projection differs from wire record") + } + return { wire, replay: checkChrCapture(projection) } +} + +type Options = { + journalPath: string + deadlineAt: number + scope: Scope + ingress: { bytes: Buffer; sha256: string; existingPort: number } + serverIds: readonly [number, number] + targets: readonly [Omit, Omit] + collector: { authorize: () => Promise; assertBound: () => Promise; restore: () => Promise } + capture: (exercise: () => Promise) => Promise + exercise: () => Promise + reconcile: (wire: Wire, serverIds: readonly [number, number]) => Promise +} + +export async function executeOwnedChrCapture(options: Options) { + const scope = structuredClone(options.scope), serverIds = Object.freeze([...options.serverIds]) as readonly [number, number] + const deadlineAt = options.deadlineAt + let active = true + const live = () => must(active && Number.isSafeInteger(deadlineAt) && Date.now() < deadlineAt, "Owned capture deadline expired") + live() + must(deadlineAt - Date.now() <= 10 * 60_000 && serverIds.length === 2 && new Set(serverIds).size === 2 && + serverIds.every(id => Number.isSafeInteger(id) && id > 0) && options.targets.length === 2 && + options.targets[0].baseline.deviceId !== options.targets[1].baseline.deviceId, "Invalid owned capture identity") + const ingress = projectChrWire(options.ingress.bytes, { ...scope, port: options.ingress.existingPort }) + must(ingress.captureSha256 === options.ingress.sha256, "Baseline ingress SHA-256 differs") + const admittedAt = Date.now() + must(ingress.frames.every(frame => frame.receivedMs >= admittedAt - 10 * 60_000 && frame.receivedMs <= admittedAt), + "Baseline ingress is outside the current evidence window") + const file = await open(options.journalPath, "wx", 0o600) + const append = async (record: object) => { + live() + await file.writeFile(JSON.stringify(record) + "\n") + await file.sync() + live() + } + const authorize = async () => { + live() + await options.collector.authorize() + await options.collector.assertBound() + await options.collector.authorize() + live() + } + try { + await authorize() + await append({ phase: "admitted", scope, serverIds, ingressSha256: ingress.captureSha256, devices: options.targets.map(target => target.baseline.deviceId) }) + if (process.platform !== "win32") { + const directory = await open(dirname(options.journalPath), "r") + try { await directory.sync() } finally { await directory.close() } + } + const targets = options.targets.map((target, index) => createChrTargetHarness({ + ...target, source: scope.sources[index], destination: scope.destination, port: scope.port, authorize, + checkpoint: entry => append({ ...entry, serverId: serverIds[index], deviceId: target.baseline.deviceId }), + })) + await authorize() + let started = false, exercised = false + const captureStartedAt = Date.now() + const capture = await options.capture(async () => { + must(!started, "Capture exercise was already attempted") + started = true + await authorize() + for (const target of targets) await target.create() + await authorize() + await options.exercise() + await authorize() + for (const target of [...targets].reverse()) await target.remove() + exercised = true + }) + const captureEndedAt = Date.now() + must(exercised, "Capture omitted or did not complete its exercise") + await authorize() + await append({ phase: "collector-restore-intent" }) + await options.collector.authorize() + live() + await options.collector.restore() + await append({ phase: "collector-restored" }) + const { wire, replay } = bindChrCapture(capture, scope) + must(wire.captureSha256 !== ingress.captureSha256 && wire.frames.every(frame => frame.receivedMs >= captureStartedAt && frame.receivedMs <= captureEndedAt), + "Capture does not belong to the owned exercise window") + await append({ phase: "wire-bound", captureSha256: wire.captureSha256, frames: wire.frames.length }) + await options.collector.authorize() + live() + await options.reconcile(wire, serverIds) + await append({ phase: "reconciled" }) + return { replay, captureSha256: wire.captureSha256, componentSequenceComplete: true, nativeAcceptance: false } + } finally { + active = false + await file.close() + } +} diff --git a/backend/src/test/traffic-flow-chr-reconcile.test.ts b/backend/src/test/traffic-flow-chr-reconcile.test.ts new file mode 100644 index 00000000..cc34a3f9 --- /dev/null +++ b/backend/src/test/traffic-flow-chr-reconcile.test.ts @@ -0,0 +1,75 @@ +import assert from "node:assert/strict" +import { test } from "node:test" +import { reconcileOwnedChrFlow } from "./traffic-flow-chr-reconcile.js" + +type Options = Parameters[0] +function fixture() { + const bucketAt = "2026-11-04T00:01:00.000Z", until = "2026-11-04T00:02:00.000Z" + const flow = { src: "192.0.2.1", dst: "198.51.100.1", proto: 17, srcPort: 9000, dstPort: 9001, inIface: "11", outIface: "12", bytes: "1200", packets: "3" } + const frames = [0, 1, 0].map((exporter, index) => ({ + ordinal: index + 1, exporter, receivedMs: Date.parse(bucketAt) + 10000 + index, receivedNs: "0", + sourcePort: 2055, version: 10, domain: 7, payloadHex: "", + })) + const wire = { captureSha256: "a".repeat(64), scope: { sources: ["192.0.2.20", "192.0.2.30"] as const, destination: "192.0.2.100", port: 4739 }, frames } + const raw = [1, 2].map(serverId => ({ ...flow, serverId, bucketAt: new Date(bucketAt), bytes: serverId === 1 ? "2400" : "1200", packets: serverId === 1 ? "6" : "3" })) + const minutes = raw.map(({ serverId, bucketAt, bytes, packets }) => ({ serverId, bucketAt, bytes, packets })) + const calls: string[] = [] + let drift = false, publicDrift = false, active = true + const options: Options = { + wire, decoded: { captureSha256: wire.captureSha256, frames: frames.map(frame => ({ ordinal: frame.ordinal, flows: [flow] })) }, + serverIds: [1, 2], from: bucketAt, until, minutes: 2, + interfaceLabel: (id, index) => "ether-" + id + "-" + index, + assertCurrent: async () => { if (!active) throw Error("retired scope") }, + readOnlySnapshot: async work => work(async (sql, parameters) => { + assert.deepEqual(parameters, [[1, 2], bucketAt, until]) + assert.match(sql, /^SELECT /) + if (sql.includes("flow_buckets")) { calls.push("raw"); return structuredClone(raw) } + calls.push("minute"); return structuredClone(minutes) + }), + callTool: async (name, input) => { + assert.equal(name, "mm_netflow_talkers_get") + assert.deepEqual(input, { serverId: input.serverId, minutes: 2, serverOffset: 0, serverLimit: 1, offset: 0, limit: 200 }) + calls.push("public:" + input.serverId) + const row = raw[input.serverId - 1] + const bytes = Number(row.bytes), packets = Number(row.packets) + const result = { + serverIds: [input.serverId], from: "2026-11-04T00:00:00.000Z", to: until, topConversationsLimit: 20, + items: [{ ...row, serverId: String(row.serverId), inIface: publicDrift ? "wrong-iface" : options.interfaceLabel(row.serverId, row.inIface), + inIfaceIndex: row.inIface, bytes, packets, bps: bytes * 8 / 120 }], + total: 1, totalConversations: 1, observedExporters: 1, nextOffset: null, retainedRowsBytes: bytes, + coverageVerified: false, rowSumIsNetworkTotal: false, deduplicatedAcrossInterfaces: false, forwardingVerified: false, + } + if (drift) raw[0].bytes = "999" + return result + }, + } + return { options, raw, minutes, calls, drift: () => { drift = true }, publicDrift: () => { publicDrift = true }, revoke: () => { active = false } } +} + +test("reconciles wire records to exact raw/minute totals and independently projected scoped talkers", async () => { + const f = fixture() + assert.deepEqual(await reconcileOwnedChrFlow(f.options), + { rawRows: 2, minuteRows: 2, publicScopes: 2, localReconciliationMatched: true, nativeAcceptance: false }) + assert.deepEqual(f.calls, ["raw", "minute", "public:1", "public:2", "raw", "minute"]) +}) + +test("raw egress drift, minute count drift and foreign SQL rows refuse before MCP reads", async () => { + for (const mode of ["egress", "count", "foreign"]) { + const f = fixture() + if (mode === "egress") f.raw[0].outIface = "13" + else if (mode === "count") f.minutes[0].packets = "7" + else f.raw[0].serverId = 3 + await assert.rejects(reconcileOwnedChrFlow(f.options), /persisted SQL/) + assert.deepEqual(f.calls, ["raw", "minute"]) + } +}) + +test("rejects altered public label, drifting persisted window and revoked current scope", async () => { + const publicDrift = fixture(); publicDrift.publicDrift() + await assert.rejects(reconcileOwnedChrFlow(publicDrift.options), /Public retained/) + const drift = fixture(); drift.drift() + await assert.rejects(reconcileOwnedChrFlow(drift.options), /persisted SQL/) + const revoked = fixture(); revoked.revoke() + await assert.rejects(reconcileOwnedChrFlow(revoked.options), /retired scope/) + assert.deepEqual(revoked.calls, []) +}) diff --git a/backend/src/test/traffic-flow-chr-reconcile.ts b/backend/src/test/traffic-flow-chr-reconcile.ts new file mode 100644 index 00000000..84af55fb --- /dev/null +++ b/backend/src/test/traffic-flow-chr-reconcile.ts @@ -0,0 +1,147 @@ +/** Independent conservation check for the owned, drained collector database. + * SQL and MCP transports must be scoped by the enclosing runner. */ +import { isIP } from "node:net" +import { isDeepStrictEqual } from "node:util" +import type { projectChrWire } from "./traffic-flow-chr-wire.js" + +type Wire = ReturnType +type Counters = { bytes: string; packets: string } +type Key = { serverId: number; bucketAt: string; src: string; dst: string; proto: number; srcPort: number; dstPort: number; inIface: string; outIface: string } +type Flow = Omit & Counters +type Raw = Key & Counters +type Minute = Pick & Counters +type Query = (text: string, parameters: readonly unknown[]) => Promise +type Options = { + wire: Wire + serverIds: readonly [number, number] + decoded: { captureSha256: string; frames: readonly { ordinal: number; flows: readonly Flow[] }[] } + from: string + until: string + minutes: number + interfaceLabel: (serverId: number, ifIndex: string) => string + readOnlySnapshot: (work: (query: Query) => Promise) => Promise + callTool: (name: "mm_netflow_talkers_get", input: { serverId: number; minutes: number; serverOffset: 0; serverLimit: 1; offset: 0; limit: 200 }) => Promise + assertCurrent: () => Promise +} +const rawSql = `SELECT server_id AS "serverId", bucket_at AS "bucketAt", host(src) AS src, host(dst) AS dst, proto, + src_port AS "srcPort", dst_port AS "dstPort", in_iface AS "inIface", out_iface AS "outIface", bytes::text, packets::text + FROM flow_buckets WHERE server_id = ANY($1::bigint[]) AND bucket_at >= $2 AND bucket_at < $3 LIMIT 201` +const minuteSql = `SELECT server_id AS "serverId", bucket_at AS "bucketAt", bytes::text, packets::text + FROM flow_minute_stats WHERE server_id = ANY($1::bigint[]) AND bucket_at >= $2 AND bucket_at < $3 LIMIT 21` +const stable = (rows: unknown[]) => rows.map(row => JSON.stringify(row)).sort() +const conversation = (row: Key) => JSON.stringify([row.serverId, row.src, row.dst, row.proto, row.srcPort, row.dstPort, row.inIface]) +const minute = (row: Pick) => JSON.stringify([row.serverId, row.bucketAt]) +const rawKey = (row: Key) => JSON.stringify([row.bucketAt, conversation(row)]) +const rowFields = (row: Raw) => [row.serverId, row.bucketAt, row.src, row.dst, row.proto, row.srcPort, row.dstPort, row.inIface, row.outIface, row.bytes, row.packets] +const minuteFields = (row: Minute) => [row.serverId, row.bucketAt, row.bytes, row.packets] +function must(condition: unknown, reason: string): asserts condition { + if (!condition) throw Error(reason) +} +function count(value: unknown) { + must(typeof value === "string" && /^(?:0|[1-9]\d{0,18})$/.test(value) && BigInt(value) <= 9223372036854775807n, "Invalid decimal flow counter") + return BigInt(value) +} +function number(value: bigint) { + must(value <= BigInt(Number.MAX_SAFE_INTEGER), "Public flow counter exceeds exact integer range") + return Number(value) +} +function time(value: unknown) { + const text = value instanceof Date && Number.isFinite(value.getTime()) ? value.toISOString() : value + must(typeof text === "string" && Number.isFinite(Date.parse(text)) && new Date(text).toISOString() === text, "Invalid canonical flow timestamp") + return text +} +function add(rows: Map, key: string, row: T) { + const prior = rows.get(key) + rows.set(key, { ...row, bytes: String(count(row.bytes) + count(prior?.bytes ?? "0")), packets: String(count(row.packets) + count(prior?.packets ?? "0")) }) +} +function object(value: unknown): Record { + must(value !== null && typeof value === "object" && !Array.isArray(value), "Invalid flow row") + return value as Record +} +function storedIdentity(value: unknown) { + const row = object(value) + const serverId = typeof row.serverId === "string" && /^[1-9]\d{0,15}$/.test(row.serverId) ? Number(row.serverId) : row.serverId + must(typeof serverId === "number" && Number.isSafeInteger(serverId) && serverId > 0, "Invalid stored flow identity") + return { ...row, serverId, bucketAt: time(row.bucketAt) } +} + +export async function reconcileOwnedChrFlow(options: Options) { + const ids = [...options.serverIds] + const from = time(options.from), until = time(options.until) + must(ids.length === 2 && new Set(ids).size === 2 && ids.every(id => Number.isSafeInteger(id) && id > 0) && + Date.parse(until) > Date.parse(from) && Date.parse(until) - Date.parse(from) <= 10 * 60_000 && + Number.isInteger(options.minutes) && options.minutes >= 1 && options.minutes <= 15, "Invalid reconciliation scope") + const { wire, decoded } = options + must(decoded.captureSha256 === wire.captureSha256 && wire.frames.length >= 2 && wire.frames.length <= 64 && + wire.frames.length === decoded.frames.length, "Decoded evidence is not bound to all wire frames") + const raw = new Map(), rollup = new Map() + let records = 0 + for (const [index, frame] of wire.frames.entries()) { + const independent = decoded.frames[index] + must(independent.ordinal === frame.ordinal && independent.flows.length <= 64, "Decoded frame ordinal or size differs") + for (const flow of independent.flows) { + const serverId = ids[frame.exporter], bucketAt = new Date(Math.floor(frame.receivedMs / 60000) * 60000).toISOString() + must(serverId !== undefined && bucketAt >= from && bucketAt < until && ++records <= 200 && + isIP(flow.src) && isIP(flow.dst) && Number.isInteger(flow.proto) && flow.proto >= 0 && flow.proto <= 255 && + [flow.srcPort, flow.dstPort].every(port => Number.isInteger(port) && port >= 0 && port <= 65535) && + [flow.inIface, flow.outIface].every(iface => /^(?:0|[1-9]\d{0,9})$/.test(iface)) && + count(flow.bytes) > 0n && count(flow.packets) > 0n, "Decoded flow is outside the owned reconciliation scope") + const row = { ...flow, serverId, bucketAt } + const key = rawKey(row), previous = raw.get(key) + must(!previous || previous.outIface === row.outIface, "Egress interface changed within an aggregate") + add(raw, key, row) + add(rollup, minute(row), { serverId, bucketAt, bytes: row.bytes, packets: row.packets }) + } + } + const expected = [...raw.values()] + must(ids.every(id => expected.some(row => row.serverId === id)), "Both exporters need positive decoded traffic") + const read = async () => { + await options.assertCurrent() + return options.readOnlySnapshot(async query => { + const parameters = [ids, from, until] + await options.assertCurrent() + const stored = await query(rawSql, parameters) + await options.assertCurrent() + const minutes = await query(minuteSql, parameters) + await options.assertCurrent() + must(stored.length <= 200 && minutes.length <= 20, "SQL reconciliation exceeded the row budget") + const rows = stored.map(row => storedIdentity(row) as Raw) + const totals = minutes.map(row => storedIdentity(row) as Minute) + must(isDeepStrictEqual(stable(rows.map(rowFields)), stable(expected.map(rowFields))) && + isDeepStrictEqual(stable(totals.map(minuteFields)), stable([...rollup.values()].map(minuteFields))), "Wire flow counters or identities differ from persisted SQL") + return rows + }) + } + const stored = await read() + for (const serverId of ids) { + await options.assertCurrent() + const result = object(await options.callTool("mm_netflow_talkers_get", { serverId, minutes: options.minutes, serverOffset: 0, serverLimit: 1, offset: 0, limit: 200 })) + await options.assertCurrent() + const since = time(result.from), end = time(result.to) + const rows = stored.filter(row => row.serverId === serverId) + must(isDeepStrictEqual(result.serverIds, [serverId]) && Date.parse(end) - Date.parse(since) === options.minutes * 60000 && + rows.every(row => row.bucketAt >= since && row.bucketAt <= end) && result.coverageVerified === false && + result.rowSumIsNetworkTotal === false && result.deduplicatedAcrossInterfaces === false && result.forwardingVerified === false, + "Public scope, window or evidence semantics differ") + const aggregate = new Map() + for (const row of rows) add(aggregate, conversation(row), row) + const limit = result.topConversationsLimit + must(typeof limit === "number" && Number.isSafeInteger(limit) && limit >= aggregate.size && limit <= 1000 && + result.total === aggregate.size && result.totalConversations === aggregate.size && + result.observedExporters === 1 && result.nextOffset === null && Array.isArray(result.items) && result.items.length === aggregate.size, + "Public result is truncated or has different conversation totals") + const ordered = [...aggregate.entries()].sort(([keyA, a], [keyB, b]) => number(count(b.bytes)) - number(count(a.bytes)) || keyA.localeCompare(keyB)) + const expectedItems = ordered.map(([, row]) => [ + String(serverId), row.src, row.dst, row.proto, row.srcPort, row.dstPort, row.inIface, + options.interfaceLabel(serverId, row.inIface), number(count(row.bytes)), number(count(row.packets)), number(count(row.bytes)) * 8 / (options.minutes * 60), + ]) + const items = result.items.map(value => { + const row = object(value) + return [row.serverId, row.src, row.dst, row.proto, row.srcPort, row.dstPort, row.inIfaceIndex, row.inIface, row.bytes, row.packets, row.bps] + }) + must(isDeepStrictEqual(items, expectedItems) && result.retainedRowsBytes === number(rows.reduce((sum, row) => sum + count(row.bytes), 0n)), + "Public retained conversations differ from independently aggregated SQL") + } + await read() + return { rawRows: raw.size, minuteRows: rollup.size, publicScopes: ids.length, localReconciliationMatched: true, nativeAcceptance: false } +} diff --git a/backend/src/test/traffic-flow-chr-target-harness.test.ts b/backend/src/test/traffic-flow-chr-target-harness.test.ts new file mode 100644 index 00000000..6ae2159d --- /dev/null +++ b/backend/src/test/traffic-flow-chr-target-harness.test.ts @@ -0,0 +1,78 @@ +import assert from "node:assert/strict" +import { test } from "node:test" +import { createChrTargetHarness } from "./traffic-flow-chr-target-harness.js" + +type Options = Parameters[0] +function fixture() { + const baseline: Options["baseline"] = { deviceId: "chr-fixture-a", flow: { enabled: "true", interfaces: "ether1" }, + ipfix: Object.fromEntries(["bytes", "packets", "src-address", "dst-address", "protocol", "src-port", "dst-port", "in-interface", "out-interface"].map(key => [key, "true"])), + targets: [{ ".id": "*3", "dst-address": "192.0.2.200", port: "4739", unknown: "retained" }] } + const live = structuredClone(baseline), effects: string[] = [] + let lostAck = false, wrongAck = false, extraChange = false, loseDeleteAck = false + let allowed = true, afterIntent: (() => void) | undefined + const options: Options = { + baseline, source: "192.0.2.1", destination: "192.0.2.100", port: 4739, + authorize: async () => { if (!allowed) throw Error("revoked") }, + read: async () => live, + checkpoint: async entry => { effects.push(entry.action + ":" + entry.phase + ":" + (entry.id ?? "")); if (entry.phase === "intent") afterIntent?.() }, + native: { + async put(path: string, body: Record, _timeout?: number, beforeDispatch?: () => Promise): Promise { + await beforeDispatch?.() + effects.push("PUT " + path) + live.targets.push({ ...body, ".id": "*A", disabled: "false", unknown: "owned" }) + if (extraChange) live.flow.interfaces = "all" + if (lostAck) throw Error("private native failure") + return { ".id": wrongAck ? "*B" : "*A" } as T + }, + async delete(path, _timeout, beforeDispatch) { + await beforeDispatch?.() + effects.push("DELETE " + path) + live.targets = live.targets.filter(row => row[".id"] !== "*A") + if (loseDeleteAck) throw Error("private native failure") + }, + }, + } + return { harness: createChrTargetHarness(options), options, live, effects, + loseAck: () => { lostAck = true }, wrongAck: () => { wrongAck = true }, extraChange: () => { extraChange = true }, + loseDeleteAck: () => { loseDeleteAck = true }, revoke: () => { allowed = false }, onIntent: (fn: () => void) => { afterIntent = fn } } +} + +test("creates one exact ACK-owned target and restores only its ID with durable checkpoints", async () => { + const f = fixture() + assert.deepEqual(await f.harness.create(), { id: "*A", verified: true, automaticReplay: false }) + assert.deepEqual(await f.harness.remove(), { id: "*A", restored: true, automaticReplay: false }) + assert.deepEqual(f.live, f.options.baseline) + assert.deepEqual(f.effects, ["create:intent:", "PUT /ip/traffic-flow/target", "create:acknowledged:*A", "create:verified:*A", + "delete:intent:*A", "DELETE /ip/traffic-flow/target/*A", "delete:acknowledged:*A", "delete:verified:*A"]) + await assert.rejects(f.harness.remove(), /No verified/) +}) + +test("lost or foreign ACK and unexpected changes refuse ownership without automatic deletion or replay", async () => { + for (const mode of ["loseAck", "wrongAck", "extraChange"] as const) { + const f = fixture(); f[mode]() + await assert.rejects(f.harness.create(), /outcome unknown/) + await assert.rejects(f.harness.create(), /already attempted/) + await assert.rejects(f.harness.remove(), /No verified/) + assert.equal(f.effects.filter(value => value.startsWith("PUT ")).length, 1) + assert.equal(f.effects.some(value => value.startsWith("DELETE ")), false) + } +}) + +test("dispatch rechecks authority, complete inventory and the exact owned row after the intent checkpoint", async () => { + const revoked = fixture(); revoked.onIntent(revoked.revoke) + await assert.rejects(revoked.harness.create(), /revoked/) + assert.deepEqual(revoked.effects, ["create:intent:"]) + const changed = fixture(); await changed.harness.create() + changed.onIntent(() => { changed.live.targets.find(row => row[".id"] === "*A")!.unknown = "replacement" }) + await assert.rejects(changed.harness.remove(), /inventory changed/) + assert.equal(changed.effects.some(value => value.startsWith("DELETE ")), false) +}) + +test("delete ACK loss remains unknown even when the fixture row was removed", async () => { + const f = fixture() + await f.harness.create(); f.loseDeleteAck() + await assert.rejects(f.harness.remove(), /outcome unknown/) + await assert.rejects(f.harness.remove(), /No verified/) + assert.equal(f.effects.filter(value => value.startsWith("DELETE ")).length, 1) + assert.equal(f.effects.includes("delete:verified:*A"), false) +}) diff --git a/backend/src/test/traffic-flow-chr-target-harness.ts b/backend/src/test/traffic-flow-chr-target-harness.ts new file mode 100644 index 00000000..95d2ca79 --- /dev/null +++ b/backend/src/test/traffic-flow-chr-target-harness.ts @@ -0,0 +1,120 @@ +/** Target-only component for a reviewed two-CHR fixture. Transport and durable + * checkpoint ownership belong to the enclosing native runner. */ +import { isIPv4 } from "node:net" +import { isDeepStrictEqual } from "node:util" +import { encodeRosId, type MikrotikClient } from "../services/mikrotik.js" + +type Row = Record +type Inventory = { deviceId: string; flow: Row; ipfix: Row; targets: Row[] } +type Checkpoint = { action: "create" | "delete"; phase: "intent" | "acknowledged" | "verified"; id?: string } +type Options = { + baseline: Inventory + source: string + destination: string + port: number + native: Pick + read: () => Promise + authorize: () => Promise + checkpoint: (entry: Checkpoint) => Promise +} +const targetPath = "/ip/traffic-flow/target" +const isId = (value: unknown): value is string => typeof value === "string" && /^\*[a-f0-9]{1,16}$/i.test(value) +const yes = (value: unknown) => value === true || value === "true" || value === "yes" +const no = (value: unknown) => value === false || value === "false" || value === "no" +const requiredFields = ["bytes", "packets", "src-address", "dst-address", "protocol", "src-port", "dst-port", "in-interface", "out-interface"] +function must(value: unknown, message: string): asserts value { + if (!value) throw Error(message) +} +function snapshot(inventory: Inventory): Inventory { + must(typeof inventory.deviceId === "string" && inventory.deviceId.length > 0 && inventory.deviceId.length <= 128 && + inventory.flow && inventory.ipfix && Array.isArray(inventory.targets) && inventory.targets.length <= 64 && + Buffer.byteLength(JSON.stringify(inventory)) <= 64 * 1024, "Invalid target inventory") + must(inventory.targets.every(row => isId(row?.[".id"])) && + new Set(inventory.targets.map(row => row[".id"])).size === inventory.targets.length, "Invalid or duplicate target identity") + const copy = structuredClone(inventory) + copy.targets.sort((left, right) => String(left[".id"]).localeCompare(String(right[".id"]))) + return copy +} + +export function createChrTargetHarness(options: Options) { + const { read, authorize, checkpoint, native, source, destination, port } = options + const baseline = snapshot(options.baseline) + must(isIPv4(source) && isIPv4(destination) && source !== destination && + Number.isInteger(port) && port > 0 && port <= 65535, "Invalid owned target tuple") + must(yes(baseline.flow.enabled) && requiredFields.every(key => yes(baseline.ipfix[key])), "Traffic Flow or required IPFIX fields are unavailable") + must(!baseline.targets.some(row => row["dst-address"] === destination && String(row.port) === String(port)), "The target tuple already exists") + const fields = Object.freeze({ "dst-address": destination, "src-address": source, port: String(port), version: "ipfix" }) + let phase: "new" | "creating" | "owned" | "deleting" | "done" | "refused" = "new" + let owned: Inventory | undefined + let ownedId: string | undefined + const current = async () => { + await authorize() + const result = snapshot(await read()) + must(result.deviceId === baseline.deviceId, "Target device identity changed") + await authorize() + return result + } + const expect = async (value: Inventory) => { + must(isDeepStrictEqual(await current(), value), "Target inventory changed") + } + return Object.freeze({ + async create() { + must(phase === "new", "Target creation already attempted") + phase = "creating" + let dispatched = false + try { + await expect(baseline) + await checkpoint({ action: "create", phase: "intent" }) + const ack = await native.put(targetPath, { ...fields }, 15000, async () => { + must(!dispatched, "Target creation already dispatched") + await expect(baseline) + must(!dispatched, "Target creation already dispatched") + dispatched = true + }) + must(dispatched, "Target creation skipped dispatch fence") + const id = ack && typeof ack === "object" ? (ack as Row)[".id"] : undefined + must(isId(id) && !baseline.targets.some(row => row[".id"] === id), "Target creation has no exact new ACK ID") + await checkpoint({ action: "create", phase: "acknowledged", id }) + const result = await current() + const row = result.targets.find(item => item[".id"] === id) + must(row && no(row.disabled) && Object.entries(fields).every(([key, value]) => String(row[key]) === value), "ACK target differs on read-back") + must(isDeepStrictEqual({ ...result, targets: result.targets.filter(item => item[".id"] !== id) }, baseline), "Unrelated target inventory changed") + await checkpoint({ action: "create", phase: "verified", id }) + owned = result + ownedId = id + phase = "owned" + return { id, verified: true as const, automaticReplay: false as const } + } catch (error) { + phase = "refused" + if (dispatched) throw Error("Target creation outcome unknown. Automatic replay forbidden.") + throw error + } + }, + async remove() { + must(phase === "owned" && owned && ownedId, "No verified owned target to delete") + const expected = owned, id = ownedId + phase = "deleting" + let dispatched = false + try { + await expect(expected) + await checkpoint({ action: "delete", phase: "intent", id }) + await native.delete(targetPath + "/" + encodeRosId(id), 10000, async () => { + must(!dispatched, "Target deletion already dispatched") + await expect(expected) + must(!dispatched, "Target deletion already dispatched") + dispatched = true + }) + must(dispatched, "Target deletion skipped dispatch fence") + await checkpoint({ action: "delete", phase: "acknowledged", id }) + await expect(baseline) + await checkpoint({ action: "delete", phase: "verified", id }) + phase = "done" + return { id, restored: true as const, automaticReplay: false as const } + } catch (error) { + phase = "refused" + if (dispatched) throw Error("Target deletion outcome unknown. Automatic replay forbidden.") + throw error + } + }, + }) +} diff --git a/backend/src/test/traffic-flow-chr-wire.test.ts b/backend/src/test/traffic-flow-chr-wire.test.ts new file mode 100644 index 00000000..30cf510c --- /dev/null +++ b/backend/src/test/traffic-flow-chr-wire.test.ts @@ -0,0 +1,57 @@ +import assert from "node:assert/strict" +import { test } from "node:test" +import { projectChrWire } from "./traffic-flow-chr-wire.js" + +const scope = { sources: ["192.0.2.1", "192.0.2.2"] as const, destination: "192.0.2.100", port: 4739 } +function capture() { + const header = Buffer.alloc(24) + header.writeUInt32LE(0xa1b2c3d4) + header.writeUInt16LE(2, 4) + header.writeUInt16LE(4, 6) + header.writeUInt32LE(4124, 16) + header.writeUInt32LE(101, 20) + const frames = [1, 2].map(lastOctet => { + const record = Buffer.alloc(16 + 20 + 8 + 16) + record.writeUInt32LE(1700000000) + record.writeUInt32LE(lastOctet, 4) + record.writeUInt32LE(44, 8) + record.writeUInt32LE(44, 12) + const ip = record.subarray(16) + ip[0] = 0x45 + ip.writeUInt16BE(44, 2) + ip[9] = 17 + Buffer.from([192, 0, 2, lastOctet, 192, 0, 2, 100]).copy(ip, 12) + ip.writeUInt16BE(2055, 20) + ip.writeUInt16BE(4739, 22) + ip.writeUInt16BE(24, 24) + ip.writeUInt16BE(10, 28) + ip.writeUInt16BE(16, 30) + ip.writeUInt32BE(lastOctet, 40) + return record + }) + return Buffer.concat([header, ...frames]) +} + +test("projects exact wire payload, exporter source port, time and domain from a two-source pcap", () => { + const pcap = capture() + const projected = projectChrWire(pcap, scope) + assert.deepEqual(projected.frames.map(({ exporter, sourcePort, domain, receivedNs }) => [exporter, sourcePort, domain, receivedNs]), + [[0, 2055, 1, "1700000000000001000"], [1, 2055, 2, "1700000000000002000"]]) + assert.equal(projected.frames[0].payloadHex, pcap.subarray(68, 84).toString("hex")) + assert.match(projected.captureSha256, /^[a-f0-9]{64}$/) +}) + +test("refuses wrong link type, incomplete wire bytes, fragments, wrong destinations and single-exporter capture", () => { + assert.throws(() => projectChrWire(capture().subarray(0, -1), scope), /Incomplete/) + assert.throws(() => projectChrWire(capture().subarray(0, 84), scope), /Both exporters/) + for (const mutate of [ + (bytes: Buffer) => bytes.writeUInt32LE(1, 20), + (bytes: Buffer) => bytes.writeUInt16BE(0x2000, 46), + (bytes: Buffer) => { bytes[59] = 101 }, + (bytes: Buffer) => bytes.writeUInt16BE(9999, 62), + ]) { + const bytes = capture() + mutate(bytes) + assert.throws(() => projectChrWire(bytes, scope)) + } +}) diff --git a/backend/src/test/traffic-flow-chr-wire.ts b/backend/src/test/traffic-flow-chr-wire.ts new file mode 100644 index 00000000..ba99780a --- /dev/null +++ b/backend/src/test/traffic-flow-chr-wire.ts @@ -0,0 +1,82 @@ +/** Offline projection of bounded classic RAW-linktype IPv4/UDP pcap files. */ +import { createHash } from "node:crypto" +import { open } from "node:fs/promises" +import { isIPv4 } from "node:net" +import { resolve } from "node:path" +import { fileURLToPath } from "node:url" + +const maxCaptureBytes = 512 * 1024 +type Scope = { sources: readonly [string, string]; destination: string; port: number } +const magic = new Map([["d4c3b2a1", [true, false]], ["a1b2c3d4", [false, false]], ["4d3cb2a1", [true, true]], ["a1b23c4d", [false, true]]]) +function must(value: unknown, error: string): asserts value { + if (!value) throw Error(error) +} + +export function projectChrWire(bytes: Buffer, scope: Scope) { + must(scope.sources.length === 2 && new Set([...scope.sources, scope.destination]).size === 3 && + [...scope.sources, scope.destination].every(isIPv4) && Number.isInteger(scope.port) && scope.port > 0 && scope.port <= 65535, "Invalid capture scope") + must(bytes.length >= 24 && bytes.length <= maxCaptureBytes, "Capture outside size bound") + const encoding = magic.get(bytes.subarray(0, 4).toString("hex")) + must(encoding, "Expected classic pcap") + const [little, nanos] = encoding + const uint = (at: number) => little ? bytes.readUInt32LE(at) : bytes.readUInt32BE(at) + const short = (at: number) => little ? bytes.readUInt16LE(at) : bytes.readUInt16BE(at) + must(short(4) === 2 && short(6) === 4 && uint(20) === 101, "Expected pcap v2.4 with LINKTYPE_RAW") + const frames = [] + let lastTimestamp = -1n + for (let at = 24; at < bytes.length;) { + must(frames.length < 64 && bytes.length - at >= 16, "Capture has too many frames or an incomplete frame header") + const seconds = uint(at), fraction = uint(at + 4), captured = uint(at + 8), original = uint(at + 12) + must(fraction < (nanos ? 1e9 : 1e6) && captured === original && captured >= 28 && captured <= 4124 && + captured <= uint(16) && bytes.length - at - 16 >= captured, "Incomplete or oversized UDP capture") + const timestamp = BigInt(seconds) * 1000000000n + BigInt(fraction) * (nanos ? 1n : 1000n) + must(timestamp >= lastTimestamp, "Capture timestamps went backwards") + lastTimestamp = timestamp + const packet = bytes.subarray(at + 16, at + 16 + captured) + at += 16 + captured + const header = (packet[0] & 15) * 4 + must(packet[0] >>> 4 === 4 && header >= 20 && packet.length >= header + 8 && + packet.readUInt16BE(2) === packet.length && (packet.readUInt16BE(6) & 0x3fff) === 0 && packet[9] === 17, "Expected complete unfragmented IPv4 UDP") + const source = [...packet.subarray(12, 16)].join(".") + const destination = [...packet.subarray(16, 20)].join(".") + const exporter = scope.sources.indexOf(source), sourcePort = packet.readUInt16BE(header) + const payload = packet.subarray(header + 8) + must(exporter >= 0 && destination === scope.destination && sourcePort > 0 && packet.readUInt16BE(header + 2) === scope.port && + packet.readUInt16BE(header + 4) === packet.length - header, "Packet outside admitted UDP tuple") + const version = payload.length >= 2 ? payload.readUInt16BE(0) : 0 + must(payload.length <= 4096 && (version === 9 && payload.length >= 20 || version === 10 && payload.length >= 16 && + payload.readUInt16BE(2) === payload.length), "Invalid or unsupported NetFlow payload") + frames.push({ + ordinal: frames.length + 1, receivedMs: Number(timestamp / 1000000n), receivedNs: timestamp.toString(), exporter, + sourcePort, version, domain: payload.readUInt32BE(version === 9 ? 16 : 12), payloadHex: payload.toString("hex"), + }) + } + must(frames.some(frame => frame.exporter === 0) && frames.some(frame => frame.exporter === 1), "Both exporters must be present") + return { captureSha256: createHash("sha256").update(bytes).digest("hex"), scope, frames } +} + +async function main() { + const [path, sha, first, second, destination, port, ...rest] = process.argv.slice(2) + must(path && /^[a-f0-9]{64}$/.test(sha ?? "") && first && second && destination && /^\d{1,5}$/.test(port ?? "") && !rest.length, "Expected pcap path, SHA-256 and exact scope") + const file = await open(path, "r") + try { + const stat = await file.stat() + must(stat.isFile() && stat.size > 0 && stat.size <= maxCaptureBytes, "Capture file outside size bound") + const bytes = Buffer.alloc(stat.size + 1) + let size = 0 + while (size < bytes.length) { + const { bytesRead } = await file.read(bytes, size, bytes.length - size, size) + if (!bytesRead) break + size += bytesRead + } + must(size === stat.size, "Capture changed while reading") + const projected = projectChrWire(bytes.subarray(0, size), { sources: [first, second], destination, port: Number(port) }) + must(projected.captureSha256 === sha, "Capture SHA-256 differs") + console.log(JSON.stringify(projected)) + } finally { + await file.close() + } +} +if (process.argv[1] && resolve(process.argv[1]) === fileURLToPath(import.meta.url)) { + main().catch(() => { console.error("CHR wire capture refused"); process.exitCode = 1 }) +} diff --git a/components/auth-guard.tsx b/components/auth-guard.tsx index f728982c..6616310f 100644 --- a/components/auth-guard.tsx +++ b/components/auth-guard.tsx @@ -1,17 +1,22 @@ "use client" import { useEffect, useState, type ReactNode } from "react" +import { usePathname } from "next/navigation" import { + can, ensureAuthConfig, getClaims, getToken, isAuthEnabled, + permissionForPath, redirectToPortalLogin, redirectToPortalLoginInteractive, } from "@/lib/auth" export function AuthGuard({ children }: { children: ReactNode }) { + const pathname = usePathname() const [ready, setReady] = useState(false) + const permission = pathname ? permissionForPath(pathname) : null useEffect(() => { let cancelled = false @@ -28,7 +33,7 @@ export function AuthGuard({ children }: { children: ReactNode }) { if (!ok) redirectToPortalLoginInteractive() return } - if (!claims.apps.includes("mm")) { + if (!claims.apps.includes("mm") || (permission && !can(permission))) { window.location.assign("/access-denied") return } @@ -37,9 +42,9 @@ export function AuthGuard({ children }: { children: ReactNode }) { return () => { cancelled = true } - }, []) + }, [pathname, permission]) - if (!ready) { + if (!ready || (isAuthEnabled() && (!getToken() || !getClaims()?.apps.includes("mm") || (permission && !can(permission))))) { return (
Проверка сессии… diff --git a/tests/e2e/mcp-auth-context.spec.ts b/tests/e2e/mcp-auth-context.spec.ts new file mode 100644 index 00000000..02b8c09d --- /dev/null +++ b/tests/e2e/mcp-auth-context.spec.ts @@ -0,0 +1,124 @@ +import { test, expect, type Page } from '@playwright/test' + +// Synthetic JWT and routed API responses. This exercises the shipped UI only; +// it does not verify backend enforcement, Portal/Sender or a live end-user route. +const id = 'fa58fa40-4bcb-4616-9316-cb69463b3150' +const at = '2026-10-29T00:00:00Z' +const read = { id: 'logs.query', toolName: 'mm_logs_query', title: 'Поиск логов', section: 'logs', version: 1, effect: 'read', risk: 'ordinary', scope: 'servers', requiredPermissions: ['mm:servers:read'], implementation: 'available', annotations: { readOnlyHint: true, destructiveHint: false, idempotentHint: true, openWorldHint: true } } +const forbidden = { ...read, id: 'users.delete', title: 'Запрещённая операция каталога', requiredPermissions: ['mm:users:admin'], effect: 'write' } +async function setup(page: Page, permissions: string[]) { + await page.addInitScript(permissions => { localStorage.setItem('mmapp_token', 'header.' + btoa(JSON.stringify({ sub: 'synthetic-owner', apps: ['mm'], permissions, exp: 4102444800 })) + '.signature') }, permissions) + const connection = { id, name: 'Синтетический AI клиент', ownerId: 'synthetic-owner', policy: { id, version: 1, status: 'active', mode: 'observer', expiresAt: '2099-01-01T00:00:00Z', permissions: ['mm:servers:read'], scope: { serverIds: [1], operationIds: ['logs.query'], allowGlobal: false }, dangerousPolicy: 'approval', autonomousDangerousOperationIds: [], limits: { callsPerMinute: 60, concurrentReads: 4, mutationsPerHour: 10, serversPerPlan: 5, objectsPerPlan: 100 } }, createdAt: at, updatedAt: at, lastUsedAt: null, generationValid: true } + const mutations: { path: string, body: Record }[] = [] + let enabled = true + await page.route('**/api/**', async route => { + const request = route.request(), path = new URL(request.url()).pathname, method = request.method() + const headers = { 'access-control-allow-origin': '*', 'access-control-allow-methods': 'GET,POST,PUT,DELETE,OPTIONS', 'access-control-allow-headers': 'Authorization,Content-Type' } + if (method === 'OPTIONS') return route.fulfill({ status: 204, headers }) + const json = (body: unknown) => route.fulfill({ headers, json: body }) + if (method !== 'GET') mutations.push({ path, body: request.postDataJSON() }) + if (path === '/api/auth/config') return json({ required: true }) + if (path === '/api/mcp/catalog') return json({ stage: 'runtime', executionAvailable: true, operations: [read, forbidden] }) + if (path === '/api/mcp/overview') return json({ connections: 1, calls: 3, errors: 1, approvals: 0, running: 0, enabled, observedAt: at }) + if (path === '/api/mcp/connections') return json(method === 'POST' ? { connection, token: 'synthetic-one-time-token' } : { connections: [connection] }) + if (path === `/api/mcp/connections/${id}/status`) { connection.policy.status = request.postDataJSON().status; return json({ ok: true }) } + if (path === '/api/mcp/stop-automation') { enabled = false; return json({ ok: true }) } + if (path === '/api/mcp/requests') return json({ items: [{ id: 'synthetic-call', tool: 'mm_logs_query', created_at: at, connection_id: id, status: connection.policy.status === 'active' ? 'completed' : 'denied', error_code: connection.policy.status === 'active' ? null : `connection_${connection.policy.status}`, detail: 'synthetic fixture outcome' }], total: 1, nextCursor: null }) + if (path.endsWith('/grants')) return json({ grants: [] }) + if (path === '/api/servers') return json([]) + return json({ items: [], total: 0, nextCursor: null }) + }) + return { mutations } +} + +test('synthetic: issue only selected server/read permission and hide one-time token', async ({ page }) => { + const state = await setup(page, ['mm:mcp:admin', 'mm:servers:read']) + await page.goto('/mcp?tab=connections') + await page.getByRole('button', { name: 'Выдать доступ', exact: true }).click() + const dialog = page.getByRole('dialog') + await dialog.getByLabel('Название', { exact: true }).fill('Ограниченная диагностика') + await dialog.getByRole('button', { name: 'Далее', exact: true }).click() + await expect(dialog.getByRole('checkbox', { name: forbidden.title, exact: true })).toHaveCount(0) + await dialog.getByLabel('ID разрешённых серверов через запятую').fill('1') + await dialog.getByRole('checkbox', { name: read.title, exact: true }).check() + await dialog.getByRole('button', { name: 'Далее', exact: true }).click() + await dialog.getByRole('button', { name: 'Далее', exact: true }).click() + await expect(dialog).toContainText('Глобальный доступ: нет') + await dialog.getByRole('button', { name: 'Сохранить доступ', exact: true }).click() + const token = page.getByLabel('Одноразовый токен MCP') + await expect(token).toHaveValue('synthetic-one-time-token') + await expect(token).toHaveAttribute('type', 'password') + expect(state.mutations).toHaveLength(1) + expect(state.mutations[0]).toMatchObject({ path: '/api/mcp/connections', body: { permissions: ['mm:servers:read'], scope: { serverIds: [1], operationIds: ['logs.query'], allowGlobal: false }, mode: 'observer' } }) + await page.getByRole('button', { name: 'Закрыть и скрыть токен' }).click() + await expect(token).toHaveCount(0) + await page.reload() + await expect(token).toHaveCount(0) +}) + +test('synthetic: pause, resume, cancelled revoke, revoke, monitor denied outcome, stop automation', async ({ page }) => { + const state = await setup(page, ['mm:mcp:admin', 'mm:mcp:pause', 'mm:servers:read']) + await page.goto('/mcp?tab=connections') + await page.getByRole('button', { name: 'Пауза', exact: true }).click() + await expect(page.getByRole('button', { name: 'Включить', exact: true })).toBeEnabled() + expect(state.mutations.at(-1)?.body).toEqual({ status: 'paused' }) + await page.getByRole('tab', { name: 'Обращения', exact: true }).click() + await expect(page.getByRole('cell', { name: 'connection_paused', exact: true })).toBeVisible() + await page.getByRole('button', { name: 'Открыть', exact: true }).click() + await expect(page.getByRole('dialog')).toContainText('synthetic fixture outcome') + await page.keyboard.press('Escape') + await page.getByRole('tab', { name: 'Подключения', exact: true }).click() + await page.getByRole('button', { name: 'Включить', exact: true }).click() + await expect(page.getByRole('button', { name: 'Пауза', exact: true })).toBeEnabled() + page.once('dialog', dialog => dialog.dismiss()) + await page.getByRole('button', { name: 'Отозвать', exact: true }).click() + expect(state.mutations.filter(x => x.body.status === 'revoked')).toHaveLength(0) + page.once('dialog', dialog => dialog.accept()) + await page.getByRole('button', { name: 'Отозвать', exact: true }).click() + await expect(page.getByRole('button', { name: 'Отозвать', exact: true })).toBeDisabled() + await expect(page.getByRole('button', { name: 'Включить', exact: true })).toBeDisabled() + await expect(page.getByRole('button', { name: 'Новый токен', exact: true })).toBeDisabled() + await page.getByRole('tab', { name: 'Обращения', exact: true }).click() + await expect(page.getByRole('cell', { name: 'connection_revoked', exact: true })).toBeVisible() + await expect(page.getByRole('cell', { name: 'denied', exact: true })).toBeVisible() + await page.getByRole('tab', { name: 'Обзор', exact: true }).click() + page.once('dialog', dialog => dialog.accept()) + await page.getByRole('button', { name: 'Остановить автоматические изменения', exact: true }).click() + await expect(page.getByText('MCP выключен', { exact: true })).toBeVisible() + expect(state.mutations.map(x => [x.path, x.body])).toEqual([ + [`/api/mcp/connections/${id}/status`, { status: 'paused' }], + [`/api/mcp/connections/${id}/status`, { status: 'active' }], + [`/api/mcp/connections/${id}/status`, { status: 'revoked' }], + ['/api/mcp/stop-automation', {}], + ]) +}) + +test('synthetic: read-only role cannot issue/revoke/pause or reach settings', async ({ page }) => { + const state = await setup(page, ['mm:mcp:read']) + await page.goto('/mcp?tab=connections') + await expect(page.getByText('Синтетический AI клиент', { exact: true })).toBeVisible() + await expect(page.getByRole('tab', { name: 'Настройки', exact: true })).toHaveCount(0) + for (const name of ['Выдать доступ', 'Пауза', 'Отозвать', 'Новый токен', 'Выдать временный допуск']) await expect(page.getByRole('button', { name, exact: true })).toHaveCount(0) + await page.getByRole('tab', { name: 'Обзор', exact: true }).click() + await expect(page.getByText('MCP включён', { exact: true })).toBeVisible() + await expect(page.getByRole('button', { name: 'Остановить автоматические изменения', exact: true })).toHaveCount(0) + await page.goto('/mcp?tab=settings') + await expect(page.getByRole('heading', { name: 'MCP', exact: true })).toBeVisible() + await expect(page.getByText('Настройки MCP', { exact: true })).toHaveCount(0) + expect(state.mutations).toEqual([]) +}) + +test('synthetic: direct application URLs enforce declared route permissions', async ({ page }) => { + await setup(page, ['mm:mcp:read', 'mm:servers:read']) + + await page.goto('/logs') + await expect(page.getByRole('heading', { name: 'Логи MikroTik', exact: true })).toBeVisible() + await page.goto('/mcp?tab=settings') + await expect(page.getByRole('heading', { name: 'MCP', exact: true })).toBeVisible() + + for (const path of ['/settings', '/traffic']) { + await page.goto(path) + await expect(page).toHaveURL(/\/access-denied$/) + await expect(page.getByRole('heading', { name: 'Нет доступа', exact: true })).toBeVisible() + } +})