diff --git a/.env.example b/.env.example index 04e9d58..5bbc983 100644 --- a/.env.example +++ b/.env.example @@ -36,6 +36,8 @@ NETWORK_AGENT_MQTT_CLIENT_ID=kt-template-online-api-network-nas-main NETWORK_AGENT_MQTT_USERNAME= NETWORK_AGENT_MQTT_PASSWORD= NETWORK_AGENT_MQTT_RETRY_MS=5000 +NETWORK_TCP_NATMAP_RELEASE_MODE=off +NETWORK_TCP_NATMAP_CANARY_PORTS= # Optional Tencent Cloud DNS DDNS; credentials must come from a private runtime Secret. NETWORK_DDNS_DNSPOD_ENABLED=false diff --git a/src/modules/admin/platform-config/admin-platform-config.module.ts b/src/modules/admin/platform-config/admin-platform-config.module.ts index bd6ff25..f846025 100644 --- a/src/modules/admin/platform-config/admin-platform-config.module.ts +++ b/src/modules/admin/platform-config/admin-platform-config.module.ts @@ -20,6 +20,7 @@ import { NetworkPortForward } from '@/modules/admin/platform-config/network-mana import { NetworkPortForwardGroup } from '@/modules/admin/platform-config/network-management/network-port-forward-group.entity'; import { NetworkManagementService } from '@/modules/admin/platform-config/network-management/network-management.service'; import { NetworkStunMessageSourceAdapter } from '@/modules/admin/platform-config/network-management/network-stun-message-source.adapter'; +import { NetworkTcpReleasePolicyService } from '@/modules/admin/platform-config/network-management/network-tcp-release-policy.service'; import { SystemLogController } from '@/modules/admin/platform-config/system-log/system-log.controller'; import { SystemLogService } from '@/modules/admin/platform-config/system-log/system-log.service'; import { AdminTimezoneController } from '@/modules/admin/platform-config/timezone/admin-timezone.controller'; @@ -91,6 +92,7 @@ export const ADMIN_PLATFORM_CONFIG_PROVIDERS = [ NetworkDnsPodClient, NetworkDdnsService, NetworkStunMessageSourceAdapter, + NetworkTcpReleasePolicyService, NetworkAgentMqttService, ]; diff --git a/src/modules/admin/platform-config/network-management/network-agent-mqtt.service.ts b/src/modules/admin/platform-config/network-management/network-agent-mqtt.service.ts index c039c9d..1a7f356 100644 --- a/src/modules/admin/platform-config/network-management/network-agent-mqtt.service.ts +++ b/src/modules/admin/platform-config/network-management/network-agent-mqtt.service.ts @@ -22,6 +22,13 @@ import { NetworkEndpointHistory } from './network-endpoint-history.entity'; import { NetworkDdnsService } from './network-ddns.service'; import { NetworkManagementEventStreamService } from './network-management-event-stream.service'; import { NetworkPortForward } from './network-management.entity'; +import { NetworkPortForwardGroup } from './network-port-forward-group.entity'; +import { NetworkTcpReleasePolicyService } from './network-tcp-release-policy.service'; +import { + buildDesiredSnapshotV2, + parseStatusSnapshotV2, + type NetworkStatusSnapshotV2, +} from './network-agent-v2.types'; import { buildDesiredSnapshot, desiredSnapshotDigest, @@ -74,6 +81,8 @@ export class NetworkAgentMqttService implements OnModuleInit, OnModuleDestroy { private readonly clientFactory?: NetworkMqttClientFactory, @Optional() private readonly ddnsService?: NetworkDdnsService, + @Optional() + private readonly releasePolicy?: NetworkTcpReleasePolicyService, ) {} onModuleInit(): void { @@ -128,28 +137,57 @@ export class NetworkAgentMqttService implements OnModuleInit, OnModuleDestroy { async publishLatestDesired(force = false): Promise { if (!this.client?.connected) return false; - const snapshot = await this.dataSource.transaction(async (manager) => { + const publication = await this.dataSource.transaction(async (manager) => { const state = await manager.getRepository(NetworkAgentState).findOne({ where: { agentId: this.agentId() }, }); if (!state || BigInt(state.desiredRevision) === 0n) return null; + const desiredSchemaVersion: 1 | 2 = + state.desiredSchemaVersion === 2 ? 2 : 1; + const publishedSchemaVersion: 1 | 2 = + state.publishedSchemaVersion === 2 ? 2 : 1; if ( !force && - BigInt(state.publishedRevision) >= BigInt(state.desiredRevision) + BigInt(state.publishedRevision) >= BigInt(state.desiredRevision) && + publishedSchemaVersion === desiredSchemaVersion ) { return null; } const mappings = await manager.getRepository(NetworkPortForward).find({ where: { isDeleted: false }, }); - return buildDesiredSnapshot(state, mappings); + if (desiredSchemaVersion === 2) { + const snapshot = buildDesiredSnapshotV2(state, mappings); + return { + payload: Buffer.from(JSON.stringify(snapshot), 'utf8'), + publishedSchemaVersion, + revision: String(snapshot.snapshotRevision), + schemaVersion: 2 as const, + }; + } + const snapshot = buildDesiredSnapshot(state, mappings); + return { + payload: desiredSnapshotBytes(snapshot), + publishedSchemaVersion, + revision: String(snapshot.revision), + schemaVersion: 1 as const, + }; }); - if (!snapshot) return false; + if (!publication) return false; + if (publication.publishedSchemaVersion !== publication.schemaVersion) { + await this.publishWithPuback( + this.topic(publication.publishedSchemaVersion, 'desired'), + Buffer.alloc(0), + ); + } await this.publishWithPuback( - this.topic('desired'), - desiredSnapshotBytes(snapshot), + this.topic(publication.schemaVersion, 'desired'), + publication.payload, + ); + await this.markRevisionPublished( + publication.revision, + publication.schemaVersion, ); - await this.markRevisionPublished(String(snapshot.revision)); return true; } @@ -167,17 +205,22 @@ export class NetworkAgentMqttService implements OnModuleInit, OnModuleDestroy { let changed = false; let ddnsSourceChanged = false; let source: NetworkStateChangeSource; - if (topic === this.topic('reported')) { + if (topic === this.topic(1, 'reported')) { source = 'reported'; const result = await this.applyReported(parseReportedSnapshot(parsed)); changed = result.visibleStateChanged; ddnsSourceChanged = result.ddnsSourceChanged; - } else if (topic === this.topic('status')) { + } else if (topic === this.topic(1, 'status')) { source = 'status'; const result = await this.applyStatus(parseStatusSnapshot(parsed)); changed = result.visibleStateChanged; ddnsSourceChanged = result.ddnsSourceChanged; - } else if (topic === this.topic('events')) { + } else if (topic === this.topic(2, 'status')) { + source = 'status'; + const result = await this.applyV2Status(parseStatusSnapshotV2(payload)); + changed = result.visibleStateChanged; + ddnsSourceChanged = result.ddnsSourceChanged; + } else if (topic === this.topic(1, 'events')) { source = 'events'; const endpointResult = await this.appendEndpointEvent( parseEndpointEvent(parsed), @@ -203,9 +246,10 @@ export class NetworkAgentMqttService implements OnModuleInit, OnModuleDestroy { this.recoveryInProgress = false; client.subscribe( { - [this.topic('reported')]: { qos: 1 }, - [this.topic('status')]: { qos: 1 }, - [this.topic('events')]: { qos: 1 }, + [this.topic(1, 'reported')]: { qos: 1 }, + [this.topic(1, 'status')]: { qos: 1 }, + [this.topic(1, 'events')]: { qos: 1 }, + [this.topic(2, 'status')]: { qos: 1 }, }, (error) => { if (this.shuttingDown || this.client !== client) return; @@ -280,7 +324,10 @@ export class NetworkAgentMqttService implements OnModuleInit, OnModuleDestroy { }); } - private async markRevisionPublished(revision: string): Promise { + private async markRevisionPublished( + revision: string, + schemaVersion: 1 | 2, + ): Promise { await this.dataSource.transaction(async (manager) => { const repository = manager.getRepository(NetworkAgentState); const state = await repository.findOne({ @@ -289,11 +336,16 @@ export class NetworkAgentMqttService implements OnModuleInit, OnModuleDestroy { }); if (!state) return; const confirmed = BigInt(revision); + const desiredSchemaVersion = + state.desiredSchemaVersion === 2 ? 2 : 1; if ( - confirmed > BigInt(state.publishedRevision) && - confirmed <= BigInt(state.desiredRevision) + confirmed <= BigInt(state.desiredRevision) && + desiredSchemaVersion === schemaVersion && + (confirmed > BigInt(state.publishedRevision) || + state.publishedSchemaVersion !== schemaVersion) ) { state.publishedRevision = revision; + state.publishedSchemaVersion = schemaVersion; await repository.save(state); } }); @@ -522,6 +574,54 @@ export class NetworkAgentMqttService implements OnModuleInit, OnModuleDestroy { visibleStateChanged: boolean; }> { this.assertAgentId(status.agentId); + return await this.applyStatusSnapshot(status, true); + } + + private async applyV2Status(status: NetworkStatusSnapshotV2): Promise<{ + ddnsSourceChanged: boolean; + visibleStateChanged: boolean; + }> { + this.assertAgentId(status.agentId); + await this.dataSource.transaction(async (manager) => { + const repository = manager.getRepository(NetworkAgentState); + const state = await repository.findOne({ + lock: { mode: 'pessimistic_write' }, + where: { agentId: status.agentId }, + }); + if (!state) { + throw new NetworkMessageValidationError('Unknown network Agent'); + } + if ( + state.maxSupportedSchemaVersion !== 2 || + state.tcpNatmapCapable !== status.tcpNatmapCapable + ) { + state.maxSupportedSchemaVersion = 2; + state.tcpNatmapCapable = status.tcpNatmapCapable; + await repository.save(state); + } + }); + const result = await this.applyStatusSnapshot(status, false); + if (await this.activateV2DesiredSchema()) this.requestDesiredPublish(); + return result; + } + + private async applyStatusSnapshot( + status: Pick< + NetworkStatusSnapshot, + | 'agentId' + | 'errorCode' + | 'errorMessage' + | 'observedAt' + | 'online' + | 'publicIpv6' + | 'startedAt' + | 'version' + >, + ignoreAfterV2Capability: boolean, + ): Promise<{ + ddnsSourceChanged: boolean; + visibleStateChanged: boolean; + }> { return await this.dataSource.transaction(async (manager) => { const repository = manager.getRepository(NetworkAgentState); const state = await repository.findOne({ @@ -531,6 +631,12 @@ export class NetworkAgentMqttService implements OnModuleInit, OnModuleDestroy { if (!state) { throw new NetworkMessageValidationError('Unknown network Agent'); } + if ( + ignoreAfterV2Capability && + state.maxSupportedSchemaVersion >= 2 + ) { + return { ddnsSourceChanged: false, visibleStateChanged: false }; + } const observedAt = new Date(status.observedAt); const incomingStartedAt = status.startedAt ? new Date(status.startedAt) @@ -588,6 +694,48 @@ export class NetworkAgentMqttService implements OnModuleInit, OnModuleDestroy { }); } + async requestV2Downgrade(): Promise { + if (!this.tcpReleasePolicy().mayExplicitlyDowngradeToV1()) return false; + const changed = await this.dataSource.transaction(async (manager) => { + const stateRepository = manager.getRepository(NetworkAgentState); + const state = await stateRepository.findOne({ + lock: { mode: 'pessimistic_write' }, + where: { agentId: this.agentId() }, + }); + if ( + !state || + state.desiredSchemaVersion !== 2 || + state.publishedSchemaVersion !== 2 || + BigInt(state.publishedRevision) !== BigInt(state.desiredRevision) || + BigInt(state.appliedRevision) !== BigInt(state.desiredRevision) + ) { + return false; + } + const groups = await manager.getRepository(NetworkPortForwardGroup).find({ + lock: { mode: 'pessimistic_read' }, + where: { isDeleted: false }, + }); + if ( + groups.some( + (group) => + group.protocolMode === 'tcp' || group.protocolMode === 'tcp_udp', + ) + ) { + return false; + } + const tcpChannels = await manager.getRepository(NetworkPortForward).find({ + lock: { mode: 'pessimistic_read' }, + where: { protocol: 'tcp' }, + }); + if (tcpChannels.some((channel) => !channel.isDeleted)) return false; + state.desiredSchemaVersion = 1; + await stateRepository.save(state); + return true; + }); + if (changed) this.requestDesiredPublish(); + return changed; + } + private async appendEndpointEvent( event: NetworkEndpointEvent, ): Promise<{ changed: boolean; deliveryAccepted: boolean }> { @@ -762,8 +910,37 @@ export class NetworkAgentMqttService implements OnModuleInit, OnModuleDestroy { ); } - private topic(kind: 'desired' | 'events' | 'reported' | 'status'): string { - return `kt/network/v1/agents/${this.agentId()}/${kind}`; + private async activateV2DesiredSchema(): Promise { + if (!this.tcpReleasePolicy().mayAutomaticallyActivateV2()) return false; + return await this.dataSource.transaction(async (manager) => { + const repository = manager.getRepository(NetworkAgentState); + const state = await repository.findOne({ + lock: { mode: 'pessimistic_write' }, + where: { agentId: this.agentId() }, + }); + if ( + !state || + state.desiredSchemaVersion === 2 || + state.maxSupportedSchemaVersion < 2 || + !state.tcpNatmapCapable + ) { + return false; + } + state.desiredSchemaVersion = 2; + await repository.save(state); + return true; + }); + } + + private tcpReleasePolicy(): NetworkTcpReleasePolicyService { + return this.releasePolicy || new NetworkTcpReleasePolicyService(this.configService); + } + + private topic( + schemaVersion: 1 | 2, + kind: 'desired' | 'events' | 'reported' | 'status', + ): string { + return `kt/network/v${schemaVersion}/agents/${this.agentId()}/${kind}`; } private retryMs(): number { diff --git a/src/modules/admin/platform-config/network-management/network-agent-v2.types.ts b/src/modules/admin/platform-config/network-management/network-agent-v2.types.ts index 7e023eb..794d1e6 100644 --- a/src/modules/admin/platform-config/network-management/network-agent-v2.types.ts +++ b/src/modules/admin/platform-config/network-management/network-agent-v2.types.ts @@ -1,6 +1,8 @@ import { createHash } from 'node:crypto'; import { isIP } from 'node:net'; import { TextDecoder } from 'node:util'; +import type { NetworkAgentState } from './network-agent-state.entity'; +import type { NetworkPortForward } from './network-management.entity'; export const NETWORK_AGENT_V2_SCHEMA_VERSION = 2 as const; export const NETWORK_AGENT_V2_MAX_CHANNELS = 64; @@ -66,6 +68,21 @@ export type NetworkDesiredSnapshotV2 = { snapshotRevision: number; }; +type NetworkDesiredChannelSourceV2 = Pick< + NetworkPortForward, + | 'desiredPresence' + | 'desiredRevision' + | 'externalPort' + | 'groupId' + | 'id' + | 'internalPort' + | 'keeperDesiredEnabled' + | 'name' + | 'natmapDesiredEnabled' + | 'probeRequestId' + | 'protocol' +>; + export type NetworkEndpointLeaseV2 = { mechanism: NetworkV2EndpointMechanism; observedAt: string; @@ -177,6 +194,58 @@ export function canonicalDesiredSnapshotDigestV2( }); } +export function buildDesiredSnapshotV2( + state: Pick, + channels: NetworkDesiredChannelSourceV2[], +): NetworkDesiredSnapshotV2 { + if (channels.length > NETWORK_AGENT_V2_MAX_CHANNELS) invalid('desired.channels'); + const desiredChannels = channels.map((channel) => { + const base = { + channelDesiredDigest: '', + channelDesiredRevision: safeRevisionFromString( + channel.desiredRevision, + 'channelDesiredRevision', + ), + channelId: String(channel.id), + desiredPresence: channel.desiredPresence, + externalPort: channel.externalPort, + groupId: String(channel.groupId), + internalPort: channel.internalPort, + name: channel.name, + }; + const desired: NetworkDesiredChannelV2 = + channel.protocol === 'tcp' + ? { + ...base, + natmapDesiredEnabled: channel.natmapDesiredEnabled, + protocol: 'tcp', + } + : { + ...base, + keeperDesiredEnabled: channel.keeperDesiredEnabled, + ...(channel.probeRequestId + ? { probeRequestId: channel.probeRequestId } + : {}), + protocol: 'udp', + }; + desired.channelDesiredDigest = canonicalDesiredChannelDigestV2(desired); + return desired; + }); + const snapshot: NetworkDesiredSnapshotV2 = { + agentId: state.agentId, + channels: desiredChannels, + issuedAt: isoFromDateTime(state.desiredIssuedAt, 'issuedAt'), + schemaVersion: NETWORK_AGENT_V2_SCHEMA_VERSION, + snapshotDigest: '', + snapshotRevision: safeRevisionFromString( + state.desiredRevision, + 'snapshotRevision', + ), + }; + snapshot.snapshotDigest = canonicalDesiredSnapshotDigestV2(snapshot); + return snapshot; +} + export function parseDesiredSnapshotV2( payload: NetworkV2WirePayload, ): NetworkDesiredSnapshotV2 { @@ -350,6 +419,8 @@ function errorCode(value: unknown, name: string): string { const result = string function boundedString(value: unknown, name: string, max: number): string { const result = stringValue(value, name, max); if (result.length === 0) invalid(name); return result; } function stringValue(value: unknown, name: string, max: number): string { if (typeof value !== 'string' || Buffer.byteLength(value, 'utf8') > max) invalid(name); return value; } function positiveRevision(value: unknown, name: string): number { if (typeof value !== 'number' || !Number.isSafeInteger(value) || value <= 0) invalid(name); return value; } +function safeRevisionFromString(value: string, name: string): number { if (!/^\d+$/.test(value)) invalid(name); const parsed = BigInt(value); if (parsed <= 0n || parsed > BigInt(Number.MAX_SAFE_INTEGER)) invalid(name); return Number(parsed); } +function isoFromDateTime(value: unknown, name: string): string { const date = new Date(value as string | number | Date); if (Number.isNaN(date.getTime())) invalid(name); return date.toISOString(); } function port(value: unknown, name: string): number { if (typeof value !== 'number' || !Number.isInteger(value) || value < 1 || value > 65535) invalid(name); return value; } function booleanValue(value: unknown, name: string): boolean { if (typeof value !== 'boolean') invalid(name); return value; } function enumValue(value: unknown, allowed: readonly T[], name: string): T { if (typeof value !== 'string' || !allowed.includes(value as T)) invalid(name); return value as T; } diff --git a/src/modules/admin/platform-config/network-management/network-tcp-release-policy.service.ts b/src/modules/admin/platform-config/network-management/network-tcp-release-policy.service.ts new file mode 100644 index 0000000..805519c --- /dev/null +++ b/src/modules/admin/platform-config/network-management/network-tcp-release-policy.service.ts @@ -0,0 +1,115 @@ +import { Injectable } from '@nestjs/common'; +import { ConfigService } from '@nestjs/config'; + +export type TcpReleaseMode = 'canary' | 'draining' | 'off' | 'on'; +export type TcpProtocolMode = 'tcp' | 'tcp_udp' | 'udp'; + +export type TcpReleaseState = { + externalPort: number; + natmapDesiredEnabled: boolean; + protocolMode: TcpProtocolMode; +}; + +export type TcpReleaseMutation = { + after?: TcpReleaseState; + before?: TcpReleaseState; +}; + +export class NetworkTcpReleasePolicyError extends Error {} + +@Injectable() +export class NetworkTcpReleasePolicyService { + constructor(private readonly configService: ConfigService) {} + + readMode(): TcpReleaseMode { + const configured = this.configService.get( + 'NETWORK_TCP_NATMAP_RELEASE_MODE', + ); + if (!configured) return 'off'; + if ( + configured === 'canary' || + configured === 'draining' || + configured === 'off' || + configured === 'on' + ) { + return configured; + } + throw new NetworkTcpReleasePolicyError('Invalid TCP NATMap release mode'); + } + + readCanaryPorts(): ReadonlySet { + const configured = this.configService.get( + 'NETWORK_TCP_NATMAP_CANARY_PORTS', + ); + if (!configured) return new Set(); + const ports = configured.split(','); + const result = new Set(); + for (const port of ports) { + if (!/^[1-9]\d*$/.test(port)) this.invalidCanaryPorts(); + const value = Number(port); + if (value > 65_535 || value === 8213 || result.has(value)) { + this.invalidCanaryPorts(); + } + result.add(value); + } + return result; + } + + isTcpVisibleToAdmin(): boolean { + return this.readMode() === 'on'; + } + + mayAutomaticallyActivateV2(): boolean { + const mode = this.readMode(); + return mode === 'canary' || mode === 'on'; + } + + mayExplicitlyDowngradeToV1(): boolean { + const mode = this.readMode(); + return mode === 'draining' || mode === 'off'; + } + + assertMutationAllowed(mutation: TcpReleaseMutation): void { + const before = mutation.before; + const after = mutation.after; + if (!this.hasTcp(before) && !this.hasTcp(after)) return; + + const mode = this.readMode(); + if (mode === 'on') return; + if (mode === 'canary') { + const port = after?.externalPort ?? before?.externalPort; + if (port !== undefined && this.readCanaryPorts().has(port)) return; + throw new NetworkTcpReleasePolicyError('TCP NATMap release policy rejects this port'); + } + if (this.isSafeCleanup(before, after)) return; + throw new NetworkTcpReleasePolicyError('TCP NATMap release policy permits cleanup only'); + } + + private hasTcp(value?: TcpReleaseState): boolean { + return value?.protocolMode === 'tcp' || value?.protocolMode === 'tcp_udp'; + } + + private isSafeCleanup( + before?: TcpReleaseState, + after?: TcpReleaseState, + ): boolean { + if (!before) return false; + if (!after) return before.protocolMode === 'tcp'; + if ( + before.protocolMode === 'tcp_udp' && + after.protocolMode === 'udp' + ) { + return true; + } + return ( + before.protocolMode === after.protocolMode && + before.externalPort === after.externalPort && + before.natmapDesiredEnabled && + !after.natmapDesiredEnabled + ); + } + + private invalidCanaryPorts(): never { + throw new NetworkTcpReleasePolicyError('Invalid TCP NATMap canary ports'); + } +} diff --git a/test/admin/network-management/network-agent-mqtt.service.spec.ts b/test/admin/network-management/network-agent-mqtt.service.spec.ts index e375ebc..97bf25b 100644 --- a/test/admin/network-management/network-agent-mqtt.service.spec.ts +++ b/test/admin/network-management/network-agent-mqtt.service.spec.ts @@ -11,6 +11,7 @@ import { NetworkAgentState } from '../../../src/modules/admin/platform-config/ne import { NetworkEndpointHistory } from '../../../src/modules/admin/platform-config/network-management/network-endpoint-history.entity'; import type { NetworkManagementEventStreamService } from '../../../src/modules/admin/platform-config/network-management/network-management-event-stream.service'; import { NetworkPortForward } from '../../../src/modules/admin/platform-config/network-management/network-management.entity'; +import { NetworkPortForwardGroup } from '../../../src/modules/admin/platform-config/network-management/network-port-forward-group.entity'; import type { SystemMessageEventInput, SystemMessageEventStager, @@ -621,6 +622,7 @@ describe('NetworkAgentMqttService', () => { 'kt/network/v1/agents/nas-main/events': { qos: 1 }, 'kt/network/v1/agents/nas-main/reported': { qos: 1 }, 'kt/network/v1/agents/nas-main/status': { qos: 1 }, + 'kt/network/v2/agents/nas-main/status': { qos: 1 }, }; harness.service.onModuleInit(); @@ -1484,4 +1486,205 @@ describe('NetworkAgentMqttService', () => { expect(harness.publishCommitted).toHaveBeenCalledTimes(1); expect(harness.publishCommitted).toHaveBeenCalledWith('status'); }); + + it('subscribes to v2 status, persists capability before desired schema activation, and keeps v2 status authoritative', async () => { + const harness = createHarness(); + (harness.service as any).configService.get = (key: string) => + ({ + NETWORK_AGENT_ID: 'nas-main', + NETWORK_AGENT_MQTT_RETRY_MS: '60000', + NETWORK_AGENT_MQTT_URL: 'mqtt://broker.test:1883', + NETWORK_TCP_NATMAP_RELEASE_MODE: 'on', + })[key]; + harness.service.onModuleInit(); + harness.client.emit('connect'); + expect(harness.client.subscribe).toHaveBeenCalledWith( + expect.objectContaining({ + 'kt/network/v2/agents/nas-main/status': { qos: 1 }, + }), + expect.any(Function), + ); + + await harness.service.consumeMessage( + 'kt/network/v2/agents/nas-main/status', + v2Status({ online: false }), + ); + expect(harness.state.maxSupportedSchemaVersion).toBe(2); + expect(harness.state.tcpNatmapCapable).toBe(true); + expect(harness.state.desiredSchemaVersion).toBe(2); + expect(harness.stateSave.mock.calls.length).toBeGreaterThanOrEqual(2); + + await harness.service.consumeMessage( + 'kt/network/v1/agents/nas-main/status', + Buffer.from( + JSON.stringify({ + agentId: 'nas-main', + observedAt: '2026-07-26T00:02:00Z', + online: true, + schemaVersion: 1, + }), + ), + ); + expect(harness.state.online).toBe(false); + await harness.service.onModuleDestroy(); + }); + + it('clears the old retained owner before v2 publication, commits schema only after PUBACK, and retries toward v2', async () => { + const harness = createHarness(); + Object.assign(harness.state, { + desiredSchemaVersion: 2, + maxSupportedSchemaVersion: 2, + publishedRevision: '7', + publishedSchemaVersion: 1, + tcpNatmapCapable: true, + }); + Object.assign(harness.mapping, { + groupId: '1', + natmapDesiredEnabled: false, + }); + harness.service.onModuleInit(); + + const first = harness.service.publishLatestDesired(); + await flushPromises(); + expect(harness.client.publish).toHaveBeenNthCalledWith( + 1, + 'kt/network/v1/agents/nas-main/desired', + Buffer.alloc(0), + { qos: 1, retain: true }, + expect.any(Function), + ); + harness.publishCallback()(undefined); + await flushPromises(); + expect(harness.client.publish).toHaveBeenNthCalledWith( + 2, + 'kt/network/v2/agents/nas-main/desired', + expect.any(Buffer), + { qos: 1, retain: true }, + expect.any(Function), + ); + expect(harness.state.publishedSchemaVersion).toBe(1); + harness.publishCallback()(new Error('PUBACK failed')); + await expect(first).rejects.toThrow('PUBACK failed'); + + const retry = harness.service.publishLatestDesired(); + await flushPromises(); + expect(harness.client.publish).toHaveBeenNthCalledWith( + 3, + 'kt/network/v1/agents/nas-main/desired', + Buffer.alloc(0), + { qos: 1, retain: true }, + expect.any(Function), + ); + harness.publishCallback()(undefined); + await flushPromises(); + expect(harness.client.publish).toHaveBeenNthCalledWith( + 4, + 'kt/network/v2/agents/nas-main/desired', + expect.any(Buffer), + { qos: 1, retain: true }, + expect.any(Function), + ); + harness.publishCallback()(undefined); + await expect(retry).resolves.toBe(true); + expect(harness.state.publishedSchemaVersion).toBe(2); + }); + + it('allows explicit v2 downgrade only after every release, snapshot, group, and TCP tombstone gate passes', async () => { + const accepted = createDowngradeHarness(); + await expect(accepted.service.requestV2Downgrade()).resolves.toBe(true); + expect(accepted.state.desiredSchemaVersion).toBe(1); + expect(accepted.groupFind).toHaveBeenCalledWith( + expect.objectContaining({ lock: { mode: 'pessimistic_read' } }), + ); + expect(accepted.channelFind).toHaveBeenCalledWith( + expect.objectContaining({ lock: { mode: 'pessimistic_read' } }), + ); + + const releaseOn = createDowngradeHarness('on'); + await expect(releaseOn.service.requestV2Downgrade()).resolves.toBe(false); + const unpublished = createDowngradeHarness(); + unpublished.state.appliedRevision = '6'; + await expect(unpublished.service.requestV2Downgrade()).resolves.toBe(false); + const mixedGroup = createDowngradeHarness(); + mixedGroup.groups[0].protocolMode = 'tcp_udp'; + await expect(mixedGroup.service.requestV2Downgrade()).resolves.toBe(false); + const tcpChannel = createDowngradeHarness(); + tcpChannel.tcpChannels.push( + Object.assign(new NetworkPortForward(), { + desiredPresence: 'present', + isDeleted: false, + protocol: 'tcp', + }), + ); + await expect(tcpChannel.service.requestV2Downgrade()).resolves.toBe(false); + const tcpTombstone = createDowngradeHarness(); + tcpTombstone.tcpChannels.push( + Object.assign(new NetworkPortForward(), { + desiredPresence: 'absent', + isDeleted: false, + protocol: 'tcp', + }), + ); + await expect(tcpTombstone.service.requestV2Downgrade()).resolves.toBe(false); + }); }); + +function v2Status(overrides: Record = {}): Buffer { + return Buffer.from( + JSON.stringify({ + agentId: 'nas-main', + observedAt: '2026-07-26T00:01:10Z', + online: true, + schemaVersion: 2, + supportedSchemaVersions: [1, 2], + tcpNatmapCapable: true, + ...overrides, + }), + ); +} + +function createDowngradeHarness(mode = 'off') { + const state = Object.assign(new NetworkAgentState(), { + agentId: 'nas-main', + appliedRevision: '7', + desiredRevision: '7', + desiredSchemaVersion: 2, + publishedRevision: '7', + publishedSchemaVersion: 2, + }); + const groups = [ + Object.assign(new NetworkPortForwardGroup(), { + isDeleted: false, + protocolMode: 'udp', + }), + ]; + const tcpChannels: NetworkPortForward[] = []; + const groupFind = jest.fn(async () => groups); + const channelFind = jest.fn(async () => tcpChannels); + const stateRepository = { + findOne: async () => state, + save: jest.fn(async (value) => Object.assign(state, value)), + }; + const manager = { + getRepository: (entity) => { + if (entity === NetworkAgentState) return stateRepository; + if (entity === NetworkPortForwardGroup) return { find: groupFind }; + if (entity === NetworkPortForward) return { find: channelFind }; + throw new Error('unexpected downgrade repository'); + }, + } as unknown as EntityManager; + const service = new NetworkAgentMqttService( + { + get: (key: string) => + ({ + NETWORK_AGENT_ID: 'nas-main', + NETWORK_TCP_NATMAP_RELEASE_MODE: mode, + })[key], + } as ConfigService, + { transaction: async (work) => await work(manager) } as unknown as DataSource, + { publishCommitted: jest.fn() } as never, + { stage: jest.fn() } as never, + { requestDrain: jest.fn() } as never, + ); + return { channelFind, groupFind, groups, service, state, tcpChannels }; +} diff --git a/test/admin/network-management/network-agent-v2.types.spec.ts b/test/admin/network-management/network-agent-v2.types.spec.ts index 7995725..6b5f833 100644 --- a/test/admin/network-management/network-agent-v2.types.spec.ts +++ b/test/admin/network-management/network-agent-v2.types.spec.ts @@ -2,6 +2,7 @@ import { readFileSync } from 'node:fs'; import { join } from 'node:path'; import { NETWORK_AGENT_V2_MAX_MESSAGE_BYTES, + buildDesiredSnapshotV2, canonicalDesiredChannelDigestV2, canonicalDesiredSnapshotDigestV2, parseDesiredSnapshotV2, @@ -191,4 +192,33 @@ describe('network agent MQTT v2 contract', () => { ), ).toThrow(); }); + + it('builds v2 desired channels with canonical digests and rejects unsafe bigint revisions', () => { + const state = { + agentId: 'nas-main', + desiredIssuedAt: new Date('2026-07-26T00:01:10Z'), + desiredRevision: '7', + }; + const channels = [ + { + desiredPresence: 'present' as const, + desiredRevision: '7', + externalPort: 48213, + groupId: '1', + id: '10', + internalPort: 48213, + keeperDesiredEnabled: false, + name: 'TCP NATMap', + natmapDesiredEnabled: true, + protocol: 'tcp' as const, + }, + ]; + const desired = buildDesiredSnapshotV2(state, channels); + expect(desired).toMatchObject({ schemaVersion: 2, snapshotRevision: 7 }); + expect(desired.channels[0].channelDesiredDigest).toMatch(/^[0-9a-f]{64}$/); + expect(desired.snapshotDigest).toBe(canonicalDesiredSnapshotDigestV2(desired)); + expect(() => + buildDesiredSnapshotV2({ ...state, desiredRevision: '9007199254740992' }, channels), + ).toThrow(); + }); }); diff --git a/test/admin/network-management/network-management-deployment.spec.ts b/test/admin/network-management/network-management-deployment.spec.ts index 65cafc9..67c5600 100644 --- a/test/admin/network-management/network-management-deployment.spec.ts +++ b/test/admin/network-management/network-management-deployment.spec.ts @@ -59,4 +59,14 @@ describe('Network management production deployment contract', () => { expect(envExample).toMatch(/^NETWORK_DDNS_DNSPOD_SECRET_ID=$/m); expect(envExample).toMatch(/^NETWORK_DDNS_DNSPOD_SECRET_KEY=$/m); }); + + it('documents fail-closed TCP NATMap release defaults without promoting them to Jenkins secrets', () => { + const envExample = readDeploymentFile('.env.example'); + const jenkinsfile = readDeploymentFile('Jenkinsfile'); + + expect(envExample).toMatch(/^NETWORK_TCP_NATMAP_RELEASE_MODE=off$/m); + expect(envExample).toMatch(/^NETWORK_TCP_NATMAP_CANARY_PORTS=$/m); + expect(jenkinsfile).not.toContain('NETWORK_TCP_NATMAP_RELEASE_MODE'); + expect(jenkinsfile).not.toContain('NETWORK_TCP_NATMAP_CANARY_PORTS'); + }); }); diff --git a/test/admin/network-management/network-tcp-release-policy.service.spec.ts b/test/admin/network-management/network-tcp-release-policy.service.spec.ts new file mode 100644 index 0000000..9cae7fd --- /dev/null +++ b/test/admin/network-management/network-tcp-release-policy.service.spec.ts @@ -0,0 +1,113 @@ +import type { ConfigService } from '@nestjs/config'; +import { + NetworkTcpReleasePolicyError, + NetworkTcpReleasePolicyService, +} from '../../../src/modules/admin/platform-config/network-management/network-tcp-release-policy.service'; + +function policy(values: Record) { + return new NetworkTcpReleasePolicyService({ + get: (key: string) => values[key], + } as ConfigService); +} + +const tcp = (overrides = {}) => ({ + externalPort: 48213, + natmapDesiredEnabled: true, + protocolMode: 'tcp' as const, + ...overrides, +}); + +describe('NetworkTcpReleasePolicyService', () => { + it('fails closed by default and parses the four release modes', () => { + expect(policy({}).readMode()).toBe('off'); + for (const mode of ['canary', 'draining', 'off', 'on'] as const) { + expect(policy({ NETWORK_TCP_NATMAP_RELEASE_MODE: mode }).readMode()).toBe( + mode, + ); + } + expect(() => + policy({ NETWORK_TCP_NATMAP_RELEASE_MODE: 'secret-mode' }).readMode(), + ).toThrow(NetworkTcpReleasePolicyError); + expect(() => + policy({ NETWORK_TCP_NATMAP_RELEASE_MODE: 'secret-mode' }).readMode(), + ).not.toThrow('secret-mode'); + }); + + it('accepts only canonical canary ports and never includes the raw config in errors', () => { + expect( + [...policy({ NETWORK_TCP_NATMAP_CANARY_PORTS: '443,48213' }).readCanaryPorts()], + ).toEqual([443, 48213]); + for (const value of ['', '443,,48213', '0443', '+443', '0', '65536', '443,443', '8213', 'tcp']) { + const service = policy({ NETWORK_TCP_NATMAP_CANARY_PORTS: value }); + if (!value) { + expect(service.readCanaryPorts()).toEqual(new Set()); + } else { + expect(() => service.readCanaryPorts()).toThrow(NetworkTcpReleasePolicyError); + expect(() => service.readCanaryPorts()).not.toThrow(value); + } + } + }); + + it('limits canary TCP writes to the allowlist and hides TCP from ordinary Admin', () => { + const service = policy({ + NETWORK_TCP_NATMAP_CANARY_PORTS: '48213', + NETWORK_TCP_NATMAP_RELEASE_MODE: 'canary', + }); + expect(() => service.assertMutationAllowed({ after: tcp() })).not.toThrow(); + expect(() => + service.assertMutationAllowed({ after: tcp({ externalPort: 48214 }) }), + ).toThrow(NetworkTcpReleasePolicyError); + expect(service.isTcpVisibleToAdmin()).toBe(false); + expect( + policy({ NETWORK_TCP_NATMAP_RELEASE_MODE: 'on' }).isTcpVisibleToAdmin(), + ).toBe(true); + expect(service.mayAutomaticallyActivateV2()).toBe(true); + expect( + policy({ NETWORK_TCP_NATMAP_RELEASE_MODE: 'on' }).mayAutomaticallyActivateV2(), + ).toBe(true); + expect( + policy({ NETWORK_TCP_NATMAP_RELEASE_MODE: 'off' }).mayAutomaticallyActivateV2(), + ).toBe(false); + expect( + policy({ NETWORK_TCP_NATMAP_RELEASE_MODE: 'draining' }).mayAutomaticallyActivateV2(), + ).toBe(false); + }); + + it.each(['draining', 'off'] as const)( + '%s permits only safe TCP cleanup while preserving UDP-only writes', + (mode) => { + const service = policy({ NETWORK_TCP_NATMAP_RELEASE_MODE: mode }); + expect(() => + service.assertMutationAllowed({ + before: tcp(), + after: tcp({ natmapDesiredEnabled: false }), + }), + ).not.toThrow(); + expect(() => + service.assertMutationAllowed({ + before: { ...tcp(), protocolMode: 'tcp_udp' }, + after: { ...tcp(), natmapDesiredEnabled: false, protocolMode: 'udp' }, + }), + ).not.toThrow(); + expect(() => service.assertMutationAllowed({ before: tcp() })).not.toThrow(); + expect(() => + service.assertMutationAllowed({ before: tcp(), after: { ...tcp(), protocolMode: 'udp' } }), + ).toThrow(NetworkTcpReleasePolicyError); + expect(() => + service.assertMutationAllowed({ + before: { ...tcp(), protocolMode: 'tcp_udp' }, + after: tcp(), + }), + ).toThrow(NetworkTcpReleasePolicyError); + expect(() => + service.assertMutationAllowed({ after: tcp({ externalPort: 48214 }) }), + ).toThrow(NetworkTcpReleasePolicyError); + expect(() => + service.assertMutationAllowed({ + before: { externalPort: 8213, natmapDesiredEnabled: false, protocolMode: 'udp' }, + after: { externalPort: 8214, natmapDesiredEnabled: false, protocolMode: 'udp' }, + }), + ).not.toThrow(); + }, + ); +});