362 lines
15 KiB
JavaScript
362 lines
15 KiB
JavaScript
import assert from 'node:assert/strict';
|
|
import { spawn, spawnSync } from 'node:child_process';
|
|
import dgram from 'node:dgram';
|
|
import fs from 'node:fs';
|
|
import net from 'node:net';
|
|
import os from 'node:os';
|
|
import path from 'node:path';
|
|
import test from 'node:test';
|
|
|
|
import { createFailoverService } from '../../dist/server/features/failover/failoverService.js';
|
|
import { createDomainTrafficService } from '../../dist/server/services/domainTrafficService.js';
|
|
import { DEFAULT_FAILOVER_POLICY, normalizeFailoverPolicy } from '../../dist/shared/failover.js';
|
|
|
|
const image = process.env.HARBOR_SINGBOX_IMAGE;
|
|
|
|
function listen(server, host = '127.0.0.1') {
|
|
return new Promise((resolve, reject) => {
|
|
server.once('error', reject);
|
|
server.listen(0, host, () => resolve(server.address().port));
|
|
});
|
|
}
|
|
|
|
function listenUdp(socket, host = '127.0.0.1') {
|
|
return new Promise((resolve, reject) => {
|
|
socket.once('error', reject);
|
|
socket.bind(0, host, () => resolve(socket.address().port));
|
|
});
|
|
}
|
|
|
|
function readExactly(socket, size) {
|
|
return new Promise((resolve, reject) => {
|
|
let value = Buffer.alloc(0);
|
|
const onData = (chunk) => {
|
|
value = Buffer.concat([value, chunk]);
|
|
if (value.length < size) return;
|
|
cleanup();
|
|
if (value.length > size) {
|
|
socket.pause();
|
|
socket.unshift(value.subarray(size));
|
|
}
|
|
resolve(value.subarray(0, size));
|
|
};
|
|
const cleanup = () => {
|
|
socket.off('data', onData);
|
|
socket.off('error', reject);
|
|
socket.off('end', onEnd);
|
|
};
|
|
const onEnd = () => {
|
|
cleanup();
|
|
reject(new Error('Socket ended early'));
|
|
};
|
|
socket.on('data', onData);
|
|
socket.once('error', reject);
|
|
socket.once('end', onEnd);
|
|
socket.resume();
|
|
});
|
|
}
|
|
|
|
async function openSocksConnection(proxyPort, targetPort, diagnostic = () => {}) {
|
|
const socket = net.connect(proxyPort, '127.0.0.1');
|
|
socket.setTimeout(3_000, () => socket.destroy(new Error('SOCKS fixture timed out')));
|
|
await new Promise((resolve, reject) => {
|
|
socket.once('connect', resolve);
|
|
socket.once('error', reject);
|
|
});
|
|
diagnostic('tcp connected');
|
|
socket.write(Buffer.from([5, 1, 0]));
|
|
assert.deepEqual(await readExactly(socket, 2), Buffer.from([5, 0]));
|
|
diagnostic('socks greeting accepted');
|
|
const host = Buffer.from('host.docker.internal');
|
|
socket.write(Buffer.from([5, 1, 0, 3, host.length, ...host, targetPort >> 8, targetPort & 255]));
|
|
const response = await readExactly(socket, 4);
|
|
diagnostic(`socks connect response ${response.toString('hex')}`);
|
|
assert.equal(response[1], 0);
|
|
const addressLength = response[3] === 0
|
|
? 0
|
|
: response[3] === 1
|
|
? 4
|
|
: response[3] === 4
|
|
? 16
|
|
: (await readExactly(socket, 1))[0];
|
|
await readExactly(socket, addressLength + 2);
|
|
return socket;
|
|
}
|
|
|
|
async function echo(socket, value) {
|
|
socket.write(value);
|
|
assert.equal((await readExactly(socket, Buffer.byteLength(value))).toString(), value);
|
|
}
|
|
|
|
async function openSocksUdpAssociation(proxyPort, targetPort, resolveRelayPort) {
|
|
const control = net.connect(proxyPort, '127.0.0.1');
|
|
control.setTimeout(3_000, () => control.destroy(new Error('SOCKS UDP fixture timed out')));
|
|
await new Promise((resolve, reject) => {
|
|
control.once('connect', resolve);
|
|
control.once('error', reject);
|
|
});
|
|
control.write(Buffer.from([5, 1, 0]));
|
|
assert.deepEqual(await readExactly(control, 2), Buffer.from([5, 0]));
|
|
control.write(Buffer.from([5, 3, 0, 1, 0, 0, 0, 0, 0, 0]));
|
|
const response = await readExactly(control, 4);
|
|
assert.equal(response[1], 0);
|
|
const addressLength = response[3] === 1 ? 4 : response[3] === 4 ? 16 : (await readExactly(control, 1))[0];
|
|
const addressAndPort = await readExactly(control, addressLength + 2);
|
|
const boundPort = addressAndPort.readUInt16BE(addressAndPort.length - 2);
|
|
const relayPort = resolveRelayPort(boundPort);
|
|
|
|
const socket = dgram.createSocket('udp4');
|
|
await listenUdp(socket);
|
|
return {
|
|
close() { socket.close(); control.destroy(); },
|
|
async echo(value) {
|
|
const host = Buffer.from('host.docker.internal');
|
|
const payload = Buffer.from(value);
|
|
const packet = Buffer.from([0, 0, 0, 3, host.length, ...host, targetPort >> 8, targetPort & 255, ...payload]);
|
|
const reply = new Promise((resolve, reject) => {
|
|
const timeout = setTimeout(() => reject(new Error(`SOCKS UDP echo timed out; relay ${boundPort}`)), 3_000);
|
|
socket.once('message', (message) => {
|
|
clearTimeout(timeout);
|
|
const headerLength = message[3] === 1 ? 10 : message[3] === 4 ? 22 : 7 + message[4];
|
|
resolve(message.subarray(headerLength).toString());
|
|
});
|
|
});
|
|
socket.send(packet, relayPort, '127.0.0.1');
|
|
assert.equal(await reply, value);
|
|
},
|
|
};
|
|
}
|
|
|
|
async function waitForApi(port) {
|
|
for (let attempt = 0; attempt < 80; attempt += 1) {
|
|
try {
|
|
const response = await fetch(`http://127.0.0.1:${port}/proxies/channel-selector`);
|
|
if (response.ok) return;
|
|
} catch {}
|
|
await new Promise((resolve) => setTimeout(resolve, 100));
|
|
}
|
|
throw new Error('Clash API did not start');
|
|
}
|
|
|
|
test('sing-box 1.13 selector preserves inbound TCP and UDP connections and routes new connections', {
|
|
skip: image ? false : 'Set HARBOR_SINGBOX_IMAGE to run the local Docker capability proof',
|
|
timeout: 30_000,
|
|
}, async (t) => {
|
|
const echoServer = net.createServer((socket) => socket.pipe(socket));
|
|
const echoPort = await listen(echoServer, '0.0.0.0');
|
|
const udpEchoServer = dgram.createSocket('udp4');
|
|
udpEchoServer.on('message', (message, remote) => udpEchoServer.send(message, remote.port, remote.address));
|
|
const udpEchoPort = await listenUdp(udpEchoServer, '0.0.0.0');
|
|
const proxyReservation = net.createServer();
|
|
const proxyPort = await listen(proxyReservation);
|
|
await new Promise((resolve) => proxyReservation.close(resolve));
|
|
const apiReservation = net.createServer();
|
|
const apiPort = await listen(apiReservation);
|
|
await new Promise((resolve) => apiReservation.close(resolve));
|
|
const fixtureDir = fs.mkdtempSync(path.join(os.tmpdir(), 'harbor-selector-'));
|
|
const configPath = path.join(fixtureDir, 'config.json');
|
|
const config = {
|
|
log: { level: 'error' },
|
|
experimental: {
|
|
cache_file: { enabled: true, path: '/config/cache.db' },
|
|
clash_api: { external_controller: '0.0.0.0:19091' },
|
|
},
|
|
inbounds: [
|
|
{ type: 'mixed', tag: 'mixed-in', listen: '0.0.0.0', listen_port: 18081 },
|
|
{ type: 'mixed', tag: 'diagnostics-primary-in', listen: '0.0.0.0', listen_port: 18082 },
|
|
{ type: 'mixed', tag: 'diagnostics-reserve-in', listen: '0.0.0.0', listen_port: 18083 },
|
|
],
|
|
outbounds: [
|
|
{ type: 'direct', tag: 'channel-primary' },
|
|
{ type: 'direct', tag: 'channel-reserve' },
|
|
{
|
|
type: 'selector',
|
|
tag: 'channel-selector',
|
|
outbounds: ['channel-primary', 'channel-reserve'],
|
|
default: 'channel-primary',
|
|
interrupt_exist_connections: false,
|
|
},
|
|
],
|
|
route: {
|
|
rules: [
|
|
{ inbound: ['diagnostics-primary-in'], outbound: 'channel-primary' },
|
|
{ inbound: ['diagnostics-reserve-in'], outbound: 'channel-reserve' },
|
|
{ inbound: ['mixed-in'], outbound: 'channel-selector' },
|
|
],
|
|
final: 'channel-selector',
|
|
},
|
|
};
|
|
fs.writeFileSync(configPath, JSON.stringify(config));
|
|
|
|
const mount = `${fixtureDir}:/config`;
|
|
const check = spawnSync('docker', [
|
|
'run', '--rm', '-v', mount, '--entrypoint', 'sing-box', image,
|
|
'check', '-c', '/config/config.json',
|
|
], { encoding: 'utf8' });
|
|
assert.equal(check.status, 0, check.stderr || check.stdout);
|
|
t.diagnostic('config accepted');
|
|
|
|
const containerName = `harbor-selector-${process.pid}-${Date.now()}`;
|
|
const udpRelayContainerPorts = Array.from({ length: 32 }, (_, index) => 18084 + index);
|
|
const runtime = spawn('docker', [
|
|
'run', '--rm', '--name', containerName,
|
|
'--sysctl', `net.ipv4.ip_local_port_range=${udpRelayContainerPorts[0]} ${udpRelayContainerPorts.at(-1)}`,
|
|
'-p', `${proxyPort}:18081/tcp`,
|
|
...udpRelayContainerPorts.flatMap((port) => ['-p', `127.0.0.1::${port}/udp`]),
|
|
'-p', `${apiPort}:19091`,
|
|
'-v', mount, '--entrypoint', 'sing-box', image,
|
|
'run', '-c', '/config/config.json',
|
|
], { stdio: ['ignore', 'pipe', 'pipe'] });
|
|
let runtimeOutput = '';
|
|
runtime.stdout.on('data', (chunk) => { runtimeOutput += chunk; });
|
|
runtime.stderr.on('data', (chunk) => { runtimeOutput += chunk; });
|
|
t.after(async () => {
|
|
if (runtime.exitCode == null) {
|
|
runtime.kill('SIGTERM');
|
|
await new Promise((resolve) => runtime.once('close', resolve));
|
|
}
|
|
echoServer.close();
|
|
udpEchoServer.close();
|
|
fs.rmSync(fixtureDir, { recursive: true, force: true });
|
|
});
|
|
|
|
try {
|
|
await waitForApi(apiPort);
|
|
const runtimePid = runtime.pid;
|
|
t.diagnostic('api ready');
|
|
const resolveUdpRelayPort = (containerPort) => {
|
|
const mapping = spawnSync('docker', ['port', containerName, `${containerPort}/udp`], { encoding: 'utf8' });
|
|
assert.equal(mapping.status, 0, mapping.stderr || mapping.stdout);
|
|
const port = Number(mapping.stdout.trim().split(':').at(-1));
|
|
assert.ok(Number.isSafeInteger(port) && port > 0, mapping.stdout);
|
|
return port;
|
|
};
|
|
const before = await (await fetch(`http://127.0.0.1:${apiPort}/proxies/channel-selector`)).json();
|
|
assert.equal(before.now, 'channel-primary');
|
|
|
|
const existing = await openSocksConnection(proxyPort, echoPort, (message) => t.diagnostic(message));
|
|
t.diagnostic('primary connection open');
|
|
t.after(() => existing.destroy());
|
|
await echo(existing, 'before-switch');
|
|
const existingUdp = await openSocksUdpAssociation(proxyPort, udpEchoPort, resolveUdpRelayPort);
|
|
t.after(() => existingUdp.close());
|
|
await existingUdp.echo('before-switch-udp');
|
|
|
|
const policy = normalizeFailoverPolicy({
|
|
...DEFAULT_FAILOVER_POLICY,
|
|
enabled: true,
|
|
primary: { profileId: 'primary-profile', serverId: 'primary-server' },
|
|
reserve: { profileId: 'reserve-profile', serverId: 'reserve-server' },
|
|
intervalMs: 15_000,
|
|
failureWindowMs: 30_000,
|
|
recoveryWindowMs: 60_000,
|
|
trafficGuard: { enabled: true, thresholdBytesPerSecond: 1024, quietWindowMs: 5_000 },
|
|
minimumReserveMs: 60_000,
|
|
});
|
|
const applied = {
|
|
primary: policy.primary,
|
|
reserve: policy.reserve,
|
|
primaryConfigFingerprint: 'a'.repeat(64),
|
|
reserveConfigFingerprint: 'b'.repeat(64),
|
|
};
|
|
let clock = 0;
|
|
let state = {
|
|
revision: 1,
|
|
failoverPolicy: policy,
|
|
failoverRuntimeState: { lastSwitchAt: null, holdUntil: null, primaryQuarantineUntil: null, failoverHistory: [], reasonCode: null },
|
|
appliedFailoverPolicy: applied,
|
|
appliedProfileId: policy.primary.profileId,
|
|
appliedServerId: policy.primary.serverId,
|
|
appliedServerSnapshot: { id: policy.primary.serverId, label: 'Primary' },
|
|
profiles: [
|
|
{ id: policy.primary.profileId, servers: [{ id: policy.primary.serverId, label: 'Primary' }] },
|
|
{ id: policy.reserve.profileId, servers: [{ id: policy.reserve.serverId, label: 'Reserve' }] },
|
|
],
|
|
diagnostics: { customServices: [] },
|
|
};
|
|
const readSelector = async () => {
|
|
const value = await (await fetch(`http://127.0.0.1:${apiPort}/proxies/channel-selector`)).json();
|
|
return { role: value.now === 'channel-primary' ? 'primary' : value.now === 'channel-reserve' ? 'reserve' : 'other' };
|
|
};
|
|
const selectRole = async (role) => {
|
|
const response = await fetch(`http://127.0.0.1:${apiPort}/proxies/channel-selector`, {
|
|
method: 'PUT',
|
|
headers: { 'content-type': 'application/json' },
|
|
body: JSON.stringify({ name: role === 'primary' ? 'channel-primary' : 'channel-reserve' }),
|
|
});
|
|
assert.equal(response.status, 204);
|
|
const selected = await readSelector();
|
|
assert.equal(selected.role, role);
|
|
return selected;
|
|
};
|
|
const traffic = createDomainTrafficService({
|
|
observe: async () => (await fetch(`http://127.0.0.1:${apiPort}/connections`)).json(),
|
|
devices: () => [],
|
|
now: () => new Date(clock),
|
|
});
|
|
await traffic.refresh();
|
|
const failover = createFailoverService({
|
|
state: {
|
|
read: () => state,
|
|
update: (mutator) => {
|
|
state = { ...mutator(state), revision: state.revision + 1 };
|
|
return state;
|
|
},
|
|
},
|
|
runtime: { isRunning: async () => true },
|
|
dataplane: {
|
|
checkConfig: async () => ({}),
|
|
runFailoverProbe: async (role) => ({ vpn: { sites: [{ status: role === 'primary' ? 'unavailable' : 'available' }] } }),
|
|
readFailoverSelector: readSelector,
|
|
selectFailoverRole: selectRole,
|
|
setFailoverActivityEnabled: async (value) => value ? traffic.enableActivity() : traffic.disableActivity(),
|
|
readFailoverActivity: async (threshold) => {
|
|
await traffic.refresh();
|
|
return { activity: traffic.activitySnapshot(threshold) };
|
|
},
|
|
},
|
|
buildCandidate: () => ({ config: {}, applied }),
|
|
serialize: (operation) => operation(),
|
|
scheduler: { setTimeout: () => ({ unref() {} }), clearTimeout: () => {} },
|
|
now: () => new Date(clock),
|
|
});
|
|
t.after(() => failover.shutdown());
|
|
await failover.reconcile();
|
|
for (clock of [0, 30_000, 35_000]) {
|
|
await echo(existing, `active-${clock}-${'x'.repeat(64 * 1024)}`);
|
|
await failover.runRound();
|
|
}
|
|
assert.equal((await readSelector()).role, 'primary');
|
|
assert.equal(failover.snapshot().status, 'waiting-for-idle');
|
|
t.diagnostic('active traffic delayed selector switch');
|
|
|
|
clock = 46_000;
|
|
await failover.runRound();
|
|
assert.equal((await readSelector()).role, 'primary');
|
|
clock = 51_000;
|
|
await failover.runRound();
|
|
const after = await (await fetch(`http://127.0.0.1:${apiPort}/proxies/channel-selector`)).json();
|
|
assert.equal(after.now, 'channel-reserve');
|
|
t.diagnostic('quiet window elapsed; selector switched and read back');
|
|
|
|
await echo(existing, 'after-switch-existing');
|
|
await existingUdp.echo('after-switch-existing-udp');
|
|
const createdAfterSwitch = await openSocksConnection(proxyPort, echoPort, (message) => t.diagnostic(message));
|
|
t.diagnostic('reserve connection open');
|
|
t.after(() => createdAfterSwitch.destroy());
|
|
await echo(createdAfterSwitch, 'after-switch-new');
|
|
const udpCreatedAfterSwitch = await openSocksUdpAssociation(proxyPort, udpEchoPort, resolveUdpRelayPort);
|
|
t.after(() => udpCreatedAfterSwitch.close());
|
|
await udpCreatedAfterSwitch.echo('after-switch-new-udp');
|
|
|
|
const connections = await (await fetch(`http://127.0.0.1:${apiPort}/connections`)).json();
|
|
const chains = connections.connections.map((connection) => connection.chains);
|
|
assert.ok(chains.some((chain) => chain.includes('channel-primary')), JSON.stringify(chains));
|
|
assert.ok(chains.some((chain) => chain.includes('channel-reserve')), JSON.stringify(chains));
|
|
assert.equal(runtime.pid, runtimePid);
|
|
assert.equal(runtime.exitCode, null);
|
|
} catch (cause) {
|
|
assert.fail(`${cause.stack || cause}\n${runtimeOutput}`);
|
|
}
|
|
});
|