feat: 增加TCP NATMap发布门禁
This commit is contained in:
parent
25b68f0c15
commit
d18a489b4e
@ -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
|
||||
|
||||
@ -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,
|
||||
];
|
||||
|
||||
|
||||
@ -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<boolean> {
|
||||
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('desired'),
|
||||
desiredSnapshotBytes(snapshot),
|
||||
this.topic(publication.publishedSchemaVersion, 'desired'),
|
||||
Buffer.alloc(0),
|
||||
);
|
||||
}
|
||||
await this.publishWithPuback(
|
||||
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<void> {
|
||||
private async markRevisionPublished(
|
||||
revision: string,
|
||||
schemaVersion: 1 | 2,
|
||||
): Promise<void> {
|
||||
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<boolean> {
|
||||
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<boolean> {
|
||||
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 {
|
||||
|
||||
@ -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<NetworkAgentState, 'agentId' | 'desiredIssuedAt' | 'desiredRevision'>,
|
||||
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<T extends string>(value: unknown, allowed: readonly T[], name: string): T { if (typeof value !== 'string' || !allowed.includes(value as T)) invalid(name); return value as T; }
|
||||
|
||||
@ -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<string>(
|
||||
'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<number> {
|
||||
const configured = this.configService.get<string>(
|
||||
'NETWORK_TCP_NATMAP_CANARY_PORTS',
|
||||
);
|
||||
if (!configured) return new Set();
|
||||
const ports = configured.split(',');
|
||||
const result = new Set<number>();
|
||||
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');
|
||||
}
|
||||
}
|
||||
@ -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<string, unknown> = {}): 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 };
|
||||
}
|
||||
|
||||
@ -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();
|
||||
});
|
||||
});
|
||||
|
||||
@ -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');
|
||||
});
|
||||
});
|
||||
|
||||
@ -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<string, string | undefined>) {
|
||||
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();
|
||||
},
|
||||
);
|
||||
});
|
||||
Loading…
Reference in New Issue
Block a user