Add native traffic inspection to Harbor Connect and Gateway

2026-08-31 05:19:15 +03:00
parent 116686a138
commit 4d066cb879
62 changed files with 10975 additions and 220 deletions
@@ -1,12 +1,28 @@
 import assert from 'node:assert/strict';
+import { create } from '@bufbuild/protobuf';
+import { connectNodeAdapter } from '@connectrpc/connect-node';
+import { spawn } from 'node:child_process';
 import fs from 'node:fs';
+import http from 'node:http';
+import http2 from 'node:http2';
 import os from 'node:os';
 import path from 'node:path';
 import test from 'node:test';

+import {
+  ConnectionEventsSchema,
+  StartedAtSchema,
+  StartedService,
+  StatusSchema,
+  VersionSchema,
+} from '../../dist/server/generated/daemon/started_service_pb.js';
+
+const root = path.resolve(import.meta.dirname, '../..');
+
 process.env.APP_MODE = 'gateway';
 process.env.DATA_DIR = fs.mkdtempSync(path.join(os.tmpdir(), 'vpn-proxy-gateway-test-'));
 process.env.SING_BOX_CACHE = path.join(process.env.DATA_DIR, 'cache.db');
+process.env.SING_BOX_TRAFFIC_SOURCE = 'snapshot';

 const {
  buildDualChannelGatewayConfig,
@@ -124,3 +140,257 @@ test('cached dual-channel outbounds must match the applied provider fingerprints
  config.outbounds[2].outbounds = ['channel-primary'];
  assert.equal(dualChannelConfigMatchesApplied(config, applied, 'primary'), false);
 });
+
+function request(socketPath, pathname, method = 'GET', body) {
+  return new Promise((resolve, reject) => {
+    const payload = body === undefined ? null : JSON.stringify(body);
+    const value = http.request({
+      socketPath,
+      path: pathname,
+      method,
+      ...(payload ? { headers: { 'content-type': 'application/json', 'content-length': Buffer.byteLength(payload) } } : {}),
+    }, (response) => {
+      const chunks = [];
+      response.on('data', (chunk) => chunks.push(chunk));
+      response.on('end', () => resolve(JSON.parse(Buffer.concat(chunks).toString('utf8'))));
+    });
+    value.on('error', reject);
+    value.end(payload);
+  });
+}
+
+async function waitForSocket(socketPath, child, stderr) {
+  for (let attempt = 0; attempt < 100; attempt += 1) {
+    if (child.exitCode !== null) throw new Error(`dataplane exited: ${stderr()}`);
+    try {
+      const status = await request(socketPath, '/status');
+      if (status.ready) return;
+    } catch {}
+    await new Promise((resolve) => setTimeout(resolve, 20));
+  }
+  throw new Error(`dataplane did not start: ${stderr()}`);
+}
+
+async function startDataplane(mode, apiPort) {
+  const directory = fs.mkdtempSync(path.join(os.tmpdir(), `harbor-${mode}-`));
+  const socketPath = path.join(directory, 'dataplane.sock');
+  const configPath = path.join(directory, 'sing-box.json');
+  const binDirectory = path.join(directory, 'bin');
+  fs.mkdirSync(binDirectory);
+  fs.writeFileSync(configPath, JSON.stringify({
+    services: [{
+      type: 'api',
+      listen: '127.0.0.1',
+      listen_port: 19091,
+      dashboard: false,
+    }],
+    inbounds: [],
+    outbounds: [],
+  }));
+  for (const [name, source] of [
+    ['sing-box', `#!/bin/sh
+if [ "$1" = check ]; then exit 0; fi
+trap 'exit 0' TERM INT
+while :; do sleep 1; done
+`],
+    ['ip', `#!/bin/sh
+printf '%s\n' '[{"dst":"192.168.50.7","lladdr":"00:11:22:33:44:55","dev":"en0","state":["REACHABLE"]}]'
+`],
+    ['iptables', '#!/bin/sh\nexit 0\n'],
+    ['iptables-restore', '#!/bin/sh\nexit 0\n'],
+  ]) {
+    const executable = path.join(binDirectory, name);
+    fs.writeFileSync(executable, source);
+    fs.chmodSync(executable, 0o755);
+  }
+  const child = spawn(process.execPath, ['dist/server/main.js'], {
+    cwd: root,
+    env: {
+      ...process.env,
+      APP_MODE: 'gateway',
+      APP_COMPONENT: 'dataplane',
+      DATA_DIR: directory,
+      DATAPLANE_SOCKET: socketPath,
+      DEVICE_TRAFFIC_ACCOUNTING_ENABLED: 'false',
+      SING_BOX_API_PORT: String(apiPort),
+      SING_BOX_CACHE: path.join(directory, 'cache.db'),
+      SING_BOX_CONFIG: configPath,
+      SING_BOX_API_SECRET: path.join(directory, 'api.secret'),
+      SING_BOX_RUNTIME_CONFIG: path.join(directory, 'runtime-config.json'),
+      SING_BOX_TRAFFIC_SOURCE: mode,
+      PATH: `${binDirectory}:${process.env.PATH || ''}`,
+    },
+    stdio: ['ignore', 'ignore', 'pipe'],
+  });
+  let error = '';
+  child.stderr.on('data', (chunk) => { error += chunk; });
+  const stop = async () => {
+    if (child.exitCode === null) {
+      const exited = new Promise((resolve) => child.once('exit', resolve));
+      child.kill('SIGTERM');
+      await exited;
+    }
+    fs.rmSync(directory, { recursive: true, force: true });
+  };
+  try {
+    await waitForSocket(socketPath, child, () => error);
+    return { socketPath, child, stop };
+  } catch (reason) {
+    await stop();
+    throw reason;
+  }
+}
+
+function waitForAbort(signal) {
+  return new Promise((resolve) => {
+    if (signal.aborted) resolve();
+    else signal.addEventListener('abort', resolve, { once: true });
+  });
+}
+
+async function waitForSnapshot(socketPath, pathname, predicate, method = 'GET', body) {
+  const deadline = Date.now() + 3_000;
+  while (Date.now() < deadline) {
+    const snapshot = await request(socketPath, pathname, method, body);
+    if (predicate(snapshot)) return snapshot;
+    await new Promise((resolve) => setTimeout(resolve, 20));
+  }
+  throw new Error(`timed out waiting for ${pathname}`);
+}
+
+function nativeRoutes(router, source) {
+  router.service(StartedService, {
+    getVersion: () => create(VersionSchema, { version: '1.14.0-rc.5', apiVersion: 4 }),
+    getStartedAt: () => create(StartedAtSchema, { startedAt: 1_700_000_000_000n }),
+    async *subscribeConnections(_request, context) {
+      yield create(ConnectionEventsSchema, {
+        reset: true,
+        events: [{
+          type: 0,
+          id: 'native-1',
+          connection: {
+            id: 'native-1',
+            inbound: 'tproxy-in',
+            inboundType: 'tproxy',
+            network: 'tcp',
+            source: '192.168.50.7:54000',
+            destination: '203.0.113.10:443',
+            domain: 'native.example',
+            protocol: 'tls',
+            createdAt: 1_700_000_000_000n,
+            uplinkTotal: 101n,
+            downlinkTotal: 202n,
+            outbound: 'channel-selector',
+            outboundType: 'selector',
+            chainList: ['channel-primary', 'channel-selector'],
+          },
+        }],
+      });
+      await waitForAbort(context.signal);
+    },
+    async *subscribeStatus(_request, context) {
+      while (!context.signal.aborted) {
+        yield create(StatusSchema, {
+          connectionsIn: source.mismatch ? 2 : 1,
+          uplinkTotal: source.mismatch ? 999n : 101n,
+          downlinkTotal: source.mismatch ? 999n : 202n,
+        });
+        await Promise.race([
+          waitForAbort(context.signal),
+          new Promise((resolve) => setTimeout(resolve, 100)),
+        ]);
+      }
+    },
+  });
+}
+
+test('Gateway modes keep snapshot compare-only and make native the sole canonical writer', async (t) => {
+  let connectionReads = 0;
+  const nativeSource = { mismatch: false };
+  const clash = http.createServer((req, res) => {
+    if (req.url === '/connections') connectionReads += 1;
+    res.writeHead(200, { 'content-type': 'application/json' });
+    res.end(JSON.stringify({ connections: [{
+      id: 'legacy-1',
+      upload: 11,
+      download: 22,
+      metadata: {
+        type: 'tproxy/tproxy-in',
+        host: 'legacy.example',
+        sourceIP: '192.168.50.7',
+      },
+      chains: ['channel-primary'],
+    }] }));
+  });
+  await new Promise((resolve) => clash.listen(0, '127.0.0.1', resolve));
+  t.after(() => new Promise((resolve) => clash.close(resolve)));
+  const apiPort = clash.address().port;
+  const native = http2.createServer(connectNodeAdapter({
+    routes: (router) => nativeRoutes(router, nativeSource),
+  }));
+  await new Promise((resolve) => native.listen(19091, '127.0.0.1', resolve));
+  t.after(() => new Promise((resolve) => native.close(resolve)));
+
+  for (const mode of ['snapshot', 'shadow', 'native']) {
+    connectionReads = 0;
+    const dataplane = await startDataplane(mode, apiPort);
+    try {
+      const live = mode === 'snapshot'
+        ? await request(dataplane.socketPath, '/traffic/live')
+        : await waitForSnapshot(
+          dataplane.socketPath,
+          '/traffic/live',
+          (snapshot) => snapshot.source.state === 'live',
+        );
+      const domain = await waitForSnapshot(
+        dataplane.socketPath,
+        '/domain-traffic',
+        (snapshot) => snapshot.tracked[0]?.uploadBytes === (mode === 'native' ? '101' : '11'),
+      );
+      assert.equal(domain.source.mode, mode);
+      assert.equal(domain.source.writer, mode === 'native' ? 'native' : 'snapshot');
+      assert.equal(domain.source.native === null, mode === 'snapshot');
+      assert.equal(domain.source.shadow === null, mode !== 'shadow');
+      assert.equal(live.source.state, mode === 'snapshot' ? 'disabled' : 'live');
+      assert.equal(connectionReads > 0, mode !== 'native');
+      if (mode === 'shadow') {
+        assert.equal(domain.source.native.active, 1);
+        assert.equal(domain.source.shadow.uploadDifferenceBytes, '90');
+        assert.equal(domain.source.shadow.downloadDifferenceBytes, '180');
+      }
+      if (mode === 'native') {
+        assert.equal(connectionReads, 0);
+        assert.equal(live.connections[0].id, 'native-1');
+        assert.deepEqual(domain.tracked, [{
+          source: 'gateway',
+          outbound: 'vpn',
+          uploadBytes: '101',
+          downloadBytes: '202',
+        }]);
+        await request(dataplane.socketPath, '/failover/activity', 'PUT', { enabled: true });
+        await waitForSnapshot(
+          dataplane.socketPath,
+          '/failover/activity/read',
+          (response) => response.activity?.state === 'quiet',
+          'POST',
+          { thresholdBytesPerSecond: 0 },
+        );
+        nativeSource.mismatch = true;
+        await waitForSnapshot(
+          dataplane.socketPath,
+          '/traffic/live',
+          (snapshot) => snapshot.source.state === 'degraded',
+        );
+        const degradedActivity = await request(
+          dataplane.socketPath,
+          '/failover/activity/read',
+          'POST',
+          { thresholdBytesPerSecond: 0 },
+        );
+        assert.equal(degradedActivity.activity, null);
+      }
+    } finally {
+      await dataplane.stop();
+    }
+  }
+});