379 lines
12 KiB
JavaScript
379 lines
12 KiB
JavaScript
import assert from 'node:assert/strict';
|
|
import { execFileSync } from 'node:child_process';
|
|
import fs from 'node:fs';
|
|
import os from 'node:os';
|
|
import path from 'node:path';
|
|
import { pathToFileURL } from 'node:url';
|
|
import test from 'node:test';
|
|
|
|
import { createLiveTrafficService } from '../../dist/server/services/liveTrafficService.js';
|
|
|
|
const subscriptionConfig = {
|
|
outbounds: [{
|
|
type: 'vless',
|
|
tag: 'test-vpn',
|
|
server: 'vpn.example.test',
|
|
server_port: 443,
|
|
uuid: '00000000-0000-4000-8000-000000000000',
|
|
tls: { enabled: true },
|
|
}],
|
|
};
|
|
|
|
function buildClientConfig(trafficSource) {
|
|
const dataDir = fs.mkdtempSync(path.join(os.tmpdir(), `harbor-native-${trafficSource}-`));
|
|
const singboxUrl = pathToFileURL(path.resolve('dist/server/singbox.js')).href;
|
|
const script = `
|
|
const { buildGatewayConfig } = await import(${JSON.stringify(singboxUrl)});
|
|
const config = buildGatewayConfig(${JSON.stringify(subscriptionConfig)}, 'test-vpn');
|
|
process.stdout.write(JSON.stringify(config));
|
|
`;
|
|
try {
|
|
return JSON.parse(execFileSync(process.execPath, ['--input-type=module', '--eval', script], {
|
|
cwd: path.resolve('.'),
|
|
encoding: 'utf8',
|
|
env: {
|
|
...process.env,
|
|
APP_MODE: 'client',
|
|
DATA_DIR: dataDir,
|
|
SING_BOX_CACHE: path.join(dataDir, 'cache.db'),
|
|
SING_BOX_TRAFFIC_SOURCE: trafficSource,
|
|
},
|
|
}));
|
|
} finally {
|
|
fs.rmSync(dataDir, { recursive: true, force: true });
|
|
}
|
|
}
|
|
|
|
function abortableStream(value, signal) {
|
|
return (async function* stream() {
|
|
yield value;
|
|
await new Promise((_, reject) => {
|
|
const abort = () => reject(signal.reason || new Error('aborted'));
|
|
if (signal.aborted) abort();
|
|
else signal.addEventListener('abort', abort, { once: true });
|
|
});
|
|
}());
|
|
}
|
|
|
|
async function waitFor(check, timeout = 1_000) {
|
|
const deadline = Date.now() + timeout;
|
|
while (!check()) {
|
|
if (Date.now() >= deadline) throw new Error('Timed out waiting for native traffic collector');
|
|
await new Promise((resolve) => setTimeout(resolve, 5));
|
|
}
|
|
}
|
|
|
|
test('client native mode adds the private API service and removes the retired DNS option', () => {
|
|
const config = buildClientConfig('native');
|
|
assert.deepEqual(config.services, [{
|
|
type: 'api',
|
|
listen: '127.0.0.1',
|
|
listen_port: 19091,
|
|
dashboard: false,
|
|
}]);
|
|
assert.deepEqual(config.dns, {});
|
|
assert.equal(config.experimental.clash_api, undefined);
|
|
});
|
|
|
|
test('client disabled mode omits the API service and keeps the 1.13-compatible DNS option', () => {
|
|
const config = buildClientConfig('disabled');
|
|
assert.equal(config.services, undefined);
|
|
assert.deepEqual(config.dns, { independent_cache: true });
|
|
assert.equal(config.experimental.clash_api, undefined);
|
|
});
|
|
|
|
test('native collector authenticates all RC5 lifecycle calls and uses one-second Go duration intervals', async () => {
|
|
const requests = [];
|
|
const connection = {
|
|
id: 'connection-1',
|
|
inbound: 'mixed-in',
|
|
inboundType: 'mixed',
|
|
network: 'tcp',
|
|
source: '127.0.0.1:54000',
|
|
destination: '203.0.113.10:443',
|
|
domain: 'example.test',
|
|
protocol: 'tls',
|
|
createdAt: 1_700_000_000_000n,
|
|
closedAt: 0n,
|
|
uplinkTotal: 12n,
|
|
downlinkTotal: 34n,
|
|
outbound: 'test-vpn',
|
|
outboundType: 'vless',
|
|
rule: 'final',
|
|
chainList: ['test-vpn'],
|
|
};
|
|
const clientFactory = (port) => {
|
|
requests.push(['factory', port]);
|
|
return {
|
|
async getVersion(_input, { signal, headers }) {
|
|
requests.push(['version', signal instanceof AbortSignal, headers?.authorization]);
|
|
return { version: '1.14.0-rc.5', apiVersion: 4 };
|
|
},
|
|
async getStartedAt(_input, { signal, headers }) {
|
|
requests.push(['started-at', signal instanceof AbortSignal, headers?.authorization]);
|
|
return { startedAt: 1_700_000_000_000n };
|
|
},
|
|
subscribeConnections({ interval }, { signal, headers }) {
|
|
requests.push(['connections', interval, headers?.authorization]);
|
|
return abortableStream({
|
|
reset: true,
|
|
events: [{ type: 0, id: connection.id, connection }],
|
|
}, signal);
|
|
},
|
|
subscribeStatus({ interval }, { signal, headers }) {
|
|
requests.push(['status', interval, headers?.authorization]);
|
|
return abortableStream({
|
|
connectionsIn: 1,
|
|
uplinkTotal: 12n,
|
|
downlinkTotal: 34n,
|
|
}, signal);
|
|
},
|
|
};
|
|
};
|
|
const service = createLiveTrafficService({
|
|
port: 19091,
|
|
enabled: true,
|
|
isRuntimeRunning: () => true,
|
|
authorization: () => 'test-secret',
|
|
clientFactory,
|
|
});
|
|
|
|
service.start();
|
|
await waitFor(() => service.snapshot().source.state === 'live');
|
|
const snapshot = service.snapshot();
|
|
assert.equal(snapshot.epoch, 'sing-box-1700000000000');
|
|
assert.equal(snapshot.source.singBoxVersion, '1.14.0-rc.5');
|
|
assert.equal(snapshot.source.singBoxApiVersion, 4);
|
|
assert.equal(snapshot.connections[0].destination.domain, 'example.test');
|
|
assert.deepEqual(requests, [
|
|
['factory', 19091],
|
|
['version', true, 'Bearer test-secret'],
|
|
['started-at', true, 'Bearer test-secret'],
|
|
['connections', 1_000_000_000n, 'Bearer test-secret'],
|
|
['status', 1_000_000_000n, 'Bearer test-secret'],
|
|
]);
|
|
await service.stop();
|
|
});
|
|
|
|
test('disabled collector stays cached and never constructs a native client', async () => {
|
|
let factoryCalls = 0;
|
|
const service = createLiveTrafficService({
|
|
port: 19091,
|
|
enabled: false,
|
|
isRuntimeRunning: () => true,
|
|
clientFactory: () => {
|
|
factoryCalls += 1;
|
|
throw new Error('must not connect');
|
|
},
|
|
});
|
|
|
|
service.start();
|
|
assert.equal(service.snapshot().source.state, 'disabled');
|
|
assert.equal(factoryCalls, 0);
|
|
await service.stop();
|
|
});
|
|
|
|
test('a native runtime materialization error is cached as incompatible without constructing a client', async () => {
|
|
let factoryCalls = 0;
|
|
const service = createLiveTrafficService({
|
|
port: 19091,
|
|
enabled: false,
|
|
unavailableError: 'secret file unavailable at https://private.example/path',
|
|
isRuntimeRunning: () => true,
|
|
clientFactory: () => {
|
|
factoryCalls += 1;
|
|
throw new Error('must not connect');
|
|
},
|
|
});
|
|
|
|
service.start();
|
|
assert.equal(service.snapshot().source.state, 'incompatible');
|
|
assert.equal(service.snapshot().source.error, 'secret file unavailable at [endpoint]');
|
|
assert.equal(factoryCalls, 0);
|
|
await service.stop();
|
|
});
|
|
|
|
for (const incompatible of [
|
|
{ version: '1.13.18', apiVersion: 4 },
|
|
{ version: '1.14.0-rc.5', apiVersion: 5 },
|
|
]) {
|
|
test(`collector rejects unqualified sing-box ${incompatible.version} API ${incompatible.apiVersion}`, async () => {
|
|
let startedAtCalls = 0;
|
|
const service = createLiveTrafficService({
|
|
port: 19091,
|
|
enabled: true,
|
|
isRuntimeRunning: () => true,
|
|
clientFactory: () => ({
|
|
async getVersion() { return incompatible; },
|
|
async getStartedAt() {
|
|
startedAtCalls += 1;
|
|
return { startedAt: 1n };
|
|
},
|
|
subscribeConnections() { throw new Error('must not subscribe'); },
|
|
subscribeStatus() { throw new Error('must not subscribe'); },
|
|
}),
|
|
});
|
|
|
|
service.start();
|
|
await waitFor(() => service.snapshot().source.state === 'incompatible');
|
|
assert.equal(service.snapshot().source.singBoxVersion, incompatible.version);
|
|
assert.equal(service.snapshot().source.singBoxApiVersion, incompatible.apiVersion);
|
|
assert.equal(startedAtCalls, 0);
|
|
await service.stop();
|
|
});
|
|
}
|
|
|
|
test('collector stop interrupts an active retry delay', async () => {
|
|
const service = createLiveTrafficService({
|
|
port: 19091,
|
|
enabled: true,
|
|
isRuntimeRunning: () => true,
|
|
clientFactory: () => ({
|
|
async getVersion() { return { version: '1.13.18', apiVersion: 4 }; },
|
|
async getStartedAt() { throw new Error('must not start'); },
|
|
subscribeConnections() { throw new Error('must not subscribe'); },
|
|
subscribeStatus() { throw new Error('must not subscribe'); },
|
|
}),
|
|
});
|
|
|
|
service.start();
|
|
await waitFor(() => service.snapshot().source.state === 'incompatible');
|
|
await new Promise((resolve) => setImmediate(resolve));
|
|
const stoppedBeforeNextTurn = await Promise.race([
|
|
service.stop().then(() => true),
|
|
new Promise((resolve) => setImmediate(() => resolve(false))),
|
|
]);
|
|
assert.equal(stoppedBeforeNextTurn, true);
|
|
});
|
|
|
|
test('collector reconnects when either RC5 stream ends and cancels its sibling', async () => {
|
|
let factoryCalls = 0;
|
|
let authorizationCalls = 0;
|
|
const attachedSecrets = [];
|
|
let firstStatusSignal;
|
|
const pending = (signal) => (async function* stream() {
|
|
await new Promise((_, reject) => {
|
|
const abort = () => reject(signal.reason || new Error('aborted'));
|
|
if (signal.aborted) abort();
|
|
else signal.addEventListener('abort', abort, { once: true });
|
|
});
|
|
}());
|
|
const clientFactory = () => {
|
|
factoryCalls += 1;
|
|
const attempt = factoryCalls;
|
|
return {
|
|
async getVersion(_input, { headers }) {
|
|
attachedSecrets.push(headers.authorization);
|
|
return { version: '1.14.0-rc.5', apiVersion: 4 };
|
|
},
|
|
async getStartedAt() {
|
|
return { startedAt: 1_700_000_000_000n };
|
|
},
|
|
subscribeConnections(_input, { signal }) {
|
|
return attempt === 1 ? (async function* ended() {})() : pending(signal);
|
|
},
|
|
subscribeStatus(_input, { signal }) {
|
|
if (attempt === 1) firstStatusSignal = signal;
|
|
return pending(signal);
|
|
},
|
|
};
|
|
};
|
|
const service = createLiveTrafficService({
|
|
port: 19091,
|
|
enabled: true,
|
|
isRuntimeRunning: () => true,
|
|
authorization: () => `secret-${++authorizationCalls}`,
|
|
clientFactory,
|
|
});
|
|
|
|
service.start();
|
|
await waitFor(() => factoryCalls >= 2, 2_000);
|
|
assert.equal(firstStatusSignal.aborted, true);
|
|
assert.equal(factoryCalls, 2);
|
|
assert.deepEqual(attachedSecrets, ['Bearer secret-1', 'Bearer secret-2']);
|
|
await service.stop();
|
|
});
|
|
|
|
test('a failed lifecycle projection is retained across one collector reconnect', async () => {
|
|
let factoryCalls = 0;
|
|
const projected = [];
|
|
const idleStream = (signal) => (async function* stream() {
|
|
await new Promise((_, reject) => {
|
|
const abort = () => reject(signal.reason || new Error('aborted'));
|
|
if (signal.aborted) abort();
|
|
else signal.addEventListener('abort', abort, { once: true });
|
|
});
|
|
}());
|
|
const connection = {
|
|
id: 'projection',
|
|
inbound: 'tproxy-in',
|
|
inboundType: 'tproxy',
|
|
network: 'tcp',
|
|
source: '192.168.50.7:54000',
|
|
destination: '203.0.113.10:443',
|
|
domain: 'example.test',
|
|
protocol: 'tls',
|
|
createdAt: 1_700_000_000_000n,
|
|
closedAt: 0n,
|
|
uplinkTotal: 1n,
|
|
downlinkTotal: 0n,
|
|
outbound: 'channel-selector',
|
|
outboundType: 'selector',
|
|
rule: 'final',
|
|
chainList: ['channel-primary', 'channel-selector'],
|
|
};
|
|
const clientFactory = () => {
|
|
factoryCalls += 1;
|
|
const attempt = factoryCalls;
|
|
return {
|
|
async getVersion() {
|
|
return { version: '1.14.0-rc.5', apiVersion: 4 };
|
|
},
|
|
async getStartedAt() {
|
|
return { startedAt: 1_700_000_000_000n };
|
|
},
|
|
subscribeConnections(_input, { signal }) {
|
|
return (async function* stream() {
|
|
yield { reset: true, events: [{ type: 0, id: connection.id, connection }] };
|
|
await new Promise((_, reject) => {
|
|
signal.addEventListener('abort', () => reject(signal.reason || new Error('aborted')), { once: true });
|
|
});
|
|
}());
|
|
},
|
|
subscribeStatus(_input, { signal }) {
|
|
return attempt === 1
|
|
? idleStream(signal)
|
|
: (async function* stream() {
|
|
while (!signal.aborted) {
|
|
yield { connectionsIn: 1, uplinkTotal: 1n, downlinkTotal: 0n };
|
|
await new Promise((resolve) => setTimeout(resolve, 20));
|
|
}
|
|
}());
|
|
},
|
|
};
|
|
};
|
|
const service = createLiveTrafficService({
|
|
port: 19091,
|
|
enabled: true,
|
|
gateway: true,
|
|
isRuntimeRunning: () => true,
|
|
clientFactory,
|
|
onProjection: async (batch) => {
|
|
projected.push(batch.connections[0]?.traffic.uploadBytes || 'heartbeat');
|
|
if (projected.length === 1) throw new Error('writer unavailable');
|
|
},
|
|
});
|
|
|
|
service.start();
|
|
await waitFor(() => factoryCalls >= 2
|
|
&& projected.includes('heartbeat')
|
|
&& service.snapshot().source.state === 'live', 2_000);
|
|
assert.equal(projected[0], '1');
|
|
assert.ok(projected.filter((value) => value === '1').length >= 2);
|
|
assert.ok(projected.includes('heartbeat'));
|
|
assert.equal(factoryCalls, 2);
|
|
assert.equal(service.snapshot().source.error, null);
|
|
assert.equal(service.snapshot().connections[0].route.kind, 'vpn');
|
|
await service.stop();
|
|
});
|