harbor-net/src/server/dataplane.js

import fs from 'node:fs';
import http from 'node:http';
import path from 'node:path';
import { settings } from './config.js';
import { createSingboxRuntime } from './singboxRuntime.js';
import { buildVersionInfo } from './version.js';
import { readNeighborSnapshot } from './adapters/neighbors.js';
import { createDeviceTrafficService } from './services/deviceTrafficService.js';
import { createDevicePolicyService } from './services/devicePolicyService.js';
import { createConnectivityDiagnosticsService } from './services/connectivityDiagnosticsService.js';

const socketPath = settings.dataplaneSocket;
const runtime = createSingboxRuntime({
  configPath: settings.configPath,
  gateway: true,
  tproxyChain: settings.tproxyChain,
});
const versionInfo = buildVersionInfo('gateway');
const traffic = createDeviceTrafficService({
  observe: () => readNeighborSnapshot(),
  uploadChain: settings.trafficUploadChain,
  downloadChain: settings.trafficDownloadChain,
  bypassCidrs: settings.bypassCidrs,
  proxyPort: settings.proxyPort,
});
const devicePolicy = createDevicePolicyService({
  chain: settings.devicePolicyChain,
  tproxyPort: settings.tproxyPort,
  tproxyMark: settings.tproxyMark,
});
const connectivityDiagnostics = createConnectivityDiagnosticsService({
  proxyPort: settings.diagnosticsProxyPort,
});
let ready = false;
let trafficTimer = null;
const MAX_POLICY_BODY_BYTES = 256 * 1024;

function readJson(req) {
  return new Promise((resolve, reject) => {
    const chunks = [];
    let size = 0;
    let tooLarge = false;
    req.on('data', (chunk) => {
      size += chunk.length;
      if (!tooLarge && size > MAX_POLICY_BODY_BYTES) {
        tooLarge = true;
        reject(new Error('Device policy request слишком большой'));
        return;
      }
      if (!tooLarge) chunks.push(chunk);
    });
    req.on('end', () => {
      if (tooLarge) return;
      try {
        resolve(JSON.parse(Buffer.concat(chunks).toString('utf8') || '{}'));
      } catch {
        reject(new Error('Device policy request содержит невалидный JSON'));
      }
    });
    req.on('error', reject);
  });
}

function sendJson(res, statusCode, payload) {
  res.writeHead(statusCode, { 'content-type': 'application/json; charset=utf-8' });
  res.end(JSON.stringify(payload));
}

const server = http.createServer(async (req, res) => {
  try {
    if (req.method === 'GET' && req.url === '/status') {
      return sendJson(res, ready ? 200 : 503, {
        ...await runtime.refresh(),
        gatewayBackendVersion: versionInfo.components.gatewayBackend,
        singBoxVersion: versionInfo.runtime.singBox,
        devicePolicy: devicePolicy.snapshot(),
        ready,
      });
    }
    if (req.method === 'GET' && req.url === '/devices') {
      return sendJson(res, 200, readNeighborSnapshot());
    }
    if (req.method === 'GET' && req.url === '/device-traffic') {
      return sendJson(res, 200, traffic.snapshot());
    }
    if (req.method === 'GET' && req.url === '/device-policy') {
      return sendJson(res, 200, devicePolicy.snapshot());
    }
    if (req.method === 'PUT' && req.url === '/device-policy') {
      const body = await readJson(req);
      return sendJson(res, 200, await devicePolicy.apply(body.devices));
    }
    if (req.method === 'POST' && req.url === '/diagnostics/connectivity') {
      const { services = [] } = await readJson(req);
      return sendJson(res, 200, await connectivityDiagnostics.run({
        vpnAvailable: runtime.running,
        services,
      }));
    }
    if (req.method === 'POST' && req.url === '/apply') {
      return sendJson(res, 200, await runtime.apply());
    }
    if (req.method === 'POST' && req.url === '/restart') {
      return sendJson(res, 200, await runtime.restart());
    }
    if (req.method === 'POST' && req.url === '/stop') {
      return sendJson(res, 200, await runtime.stop());
    }
    return sendJson(res, 404, { error: 'Не найдено' });
  } catch (error) {
    return sendJson(res, 500, { error: error.message || String(error) });
  }
});

fs.mkdirSync(path.dirname(socketPath), { recursive: true });
fs.rmSync(socketPath, { force: true });
server.listen(socketPath, async () => {
  fs.chmodSync(socketPath, 0o660);
  try {
    await runtime.apply();
  } catch (error) {
    console.warn(`[dataplane] sing-box не запущен: ${error.message}`);
  } finally {
    ready = true;
    setImmediate(() => {
      traffic.refresh()
        .catch((error) => console.warn(`[dataplane] traffic counters не запущены: ${error.message}`));
    });
    trafficTimer = setInterval(() => {
      traffic.refresh().catch((error) => console.warn(`[dataplane] traffic counters не обновлены: ${error.message}`));
    }, 15_000);
    trafficTimer.unref();
    console.log(`[dataplane] control socket: ${socketPath}`);
  }
});

let shuttingDown = false;
async function shutdown() {
  if (shuttingDown) return;
  shuttingDown = true;
  ready = false;
  if (trafficTimer) clearInterval(trafficTimer);
  await runtime.shutdown();
  server.close(() => {
    fs.rmSync(socketPath, { force: true });
    process.exit(0);
  });
}

process.on('SIGTERM', shutdown);
process.on('SIGINT', shutdown);