Add device traffic counters and ambiguous MAC detection

2026-08-07 14:29:51 +03:00
parent 3157e9e8f7
commit e774486b99
15 changed files with 528 additions and 7 deletions
@@ -16,6 +16,11 @@ export const settings = {
  proxyPort,
  tproxyPort: parsePort(process.env.TPROXY_PORT, 7895),
  tproxyChain: process.env.TPROXY_CHAIN || "VPN_PROXY_TPROXY",
+  trafficUploadChain: process.env.TRAFFIC_UPLOAD_CHAIN || "VPN_PROXY_TRAFFIC_UP",
+  trafficDownloadChain: process.env.TRAFFIC_DOWNLOAD_CHAIN || "VPN_PROXY_TRAFFIC_DOWN",
+  bypassCidrs: (process.env.BYPASS_CIDRS
+    || "0.0.0.0/8 10.0.0.0/8 100.64.0.0/10 127.0.0.0/8 169.254.0.0/16 172.16.0.0/12 192.168.0.0/16 224.0.0.0/4 240.0.0.0/4")
+    .trim().split(/\s+/).filter(Boolean),
  dataplaneSocket: process.env.DATAPLANE_SOCKET || "/run/vpn-proxy/dataplane.sock",
  bindIp: process.env.PROXY_BIND_IP || "0.0.0.0",
  dataDir,
@@ -5,6 +5,7 @@ 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';

 const socketPath = settings.dataplaneSocket;
 const runtime = createSingboxRuntime({
@@ -13,7 +14,14 @@ const runtime = createSingboxRuntime({
  tproxyChain: settings.tproxyChain,
 });
 const versionInfo = buildVersionInfo('gateway');
+const traffic = createDeviceTrafficService({
+  observe: () => readNeighborSnapshot(),
+  uploadChain: settings.trafficUploadChain,
+  downloadChain: settings.trafficDownloadChain,
+  bypassCidrs: settings.bypassCidrs,
+});
 let ready = false;
+let trafficTimer = null;

 function sendJson(res, statusCode, payload) {
  res.writeHead(statusCode, { 'content-type': 'application/json; charset=utf-8' });
@@ -33,6 +41,9 @@ const server = http.createServer(async (req, res) => {
    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 === 'POST' && req.url === '/apply') {
      return sendJson(res, 200, await runtime.apply());
    }
@@ -58,6 +69,14 @@ server.listen(socketPath, async () => {
    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}`);
  }
 });
@@ -67,6 +86,7 @@ async function shutdown() {
  if (shuttingDown) return;
  shuttingDown = true;
  ready = false;
+  if (trafficTimer) clearInterval(trafficTimer);
  await runtime.shutdown();
  server.close(() => {
    fs.rmSync(socketPath, { force: true });
@@ -44,6 +44,7 @@ export function createDataplaneClient(socketPath, send = request) {
    get startedAt() { return current.startedAt || null; },
    refresh: () => update('/status', 'GET'),
    observeDevices: () => send(socketPath, '/devices', 'GET'),
+    observeTraffic: () => send(socketPath, '/device-traffic', 'GET'),
    apply: () => update('/apply', 'POST'),
    restart: () => update('/restart', 'POST'),
    stop: () => update('/stop', 'POST'),
@@ -1,5 +1,6 @@
 import crypto from 'node:crypto';
 import fs from 'node:fs';
+import net from 'node:net';
 import { HarborError } from '../../shared/errors.js';

 const SCHEMA_VERSION = 1;
@@ -100,6 +101,13 @@ export function createDeviceInventoryService({ store, observe, vendor = () => nu
    }
    const observedAt = result?.observedAt || now().toISOString();
    const observations = Array.isArray(result?.observations) ? result.observations : [];
+    const ipsByMac = new Map();
+    for (const observation of observations) {
+      const mac = normalizeMac(observation.mac);
+      if (!mac || !net.isIPv4(String(observation.ip || ''))) continue;
+      if (!ipsByMac.has(mac)) ipsByMac.set(mac, new Set());
+      ipsByMac.get(mac).add(String(observation.ip));
+    }
    store.update((stored) => {
      const state = migrate(stored);
      const byMac = new Map(state.devices.map((device) => [device.mac, device]));
@@ -122,7 +130,9 @@ export function createDeviceInventoryService({ store, observe, vendor = () => nu
          firstSeenAt: previous?.firstSeenAt || observation.observedAt || observedAt,
          lastSeenAt,
          source: 'neighbor',
-          confidence: isPrivateMac(mac) ? 'medium' : 'high',
+          confidence: ipsByMac.get(mac)?.size > 1
+            ? 'ambiguous'
+            : isPrivateMac(mac) ? 'medium' : 'high',
        });
      }
      const cutoff = new Date(observedAt).getTime() - RETENTION_MS;
@@ -0,0 +1,240 @@
+import crypto from 'node:crypto';
+import net from 'node:net';
+import { spawnSync } from 'node:child_process';
+
+const COMMAND_OPTIONS = { encoding: 'utf8', timeout: 2_000, killSignal: 'SIGKILL' };
+const MAC_PATTERN = /^[0-9a-f]{2}(?::[0-9a-f]{2}){5}$/i;
+const INTERFACE_PATTERN = /^[a-z0-9_.:+-]{1,15}$/i;
+
+const childChain = (chain, slot) => `${chain}_${slot}`;
+const counterKey = ({ ip, mac, interface: deviceInterface }) => crypto
+  .createHash('sha256')
+  .update(`${ip}|${mac}|${deviceInterface}`)
+  .digest('hex')
+  .slice(0, 16);
+
+function commandError(command, result) {
+  return new Error(String(
+    result.stderr || result.stdout || result.error?.message || `${command} завершился с ошибкой`,
+  ).trim());
+}
+
+export function selectTrafficDevices(observations) {
+  const candidates = new Map();
+  const ipsByMac = new Map();
+  const locationsByIp = new Map();
+
+  for (const observation of Array.isArray(observations) ? observations : []) {
+    const ip = String(observation?.ip || '');
+    const mac = String(observation?.mac || '').toLowerCase();
+    const deviceInterface = String(observation?.interface || '');
+    if (!net.isIPv4(ip) || !MAC_PATTERN.test(mac) || !INTERFACE_PATTERN.test(deviceInterface)
+      || deviceInterface.startsWith('br-')) continue;
+
+    const location = `${mac}|${deviceInterface}`;
+    candidates.set(`${ip}|${location}`, { ip, mac, interface: deviceInterface });
+    if (!ipsByMac.has(mac)) ipsByMac.set(mac, new Set());
+    ipsByMac.get(mac).add(ip);
+    if (!locationsByIp.has(ip)) locationsByIp.set(ip, new Set());
+    locationsByIp.get(ip).add(location);
+  }
+
+  return [...candidates.values()]
+    .filter(({ ip, mac }) => ipsByMac.get(mac).size === 1 && locationsByIp.get(ip).size === 1)
+    .map((device) => ({ ...device, key: counterKey(device) }))
+    .sort((left, right) => (
+      left.ip.localeCompare(right.ip)
+      || left.mac.localeCompare(right.mac)
+      || left.interface.localeCompare(right.interface)
+    ));
+}
+
+export function buildTrafficRuleCommands({
+  devices,
+  bypassCidrs,
+  uploadChain,
+  downloadChain,
+  slot,
+}) {
+  const uploadChild = childChain(uploadChain, slot);
+  const downloadChild = childChain(downloadChain, slot);
+  const commands = [
+    ['iptables', ['-w', '1', '-t', 'raw', '-F', uploadChild]],
+    ['iptables', ['-w', '1', '-t', 'mangle', '-F', downloadChild]],
+    ['iptables', ['-w', '1', '-t', 'raw', '-A', uploadChild, '-i', 'br-+', '-j', 'RETURN']],
+  ];
+
+  for (const cidr of bypassCidrs) {
+    commands.push(
+      ['iptables', ['-w', '1', '-t', 'raw', '-A', uploadChild, '-d', cidr, '-j', 'RETURN']],
+      ['iptables', ['-w', '1', '-t', 'mangle', '-A', downloadChild, '-s', cidr, '-j', 'RETURN']],
+    );
+  }
+  for (const device of devices) {
+    commands.push(
+      ['iptables', [
+        '-w', '1', '-t', 'raw', '-A', uploadChild,
+        '-i', device.interface, '-s', device.ip,
+        '-m', 'mac', '--mac-source', device.mac,
+        '-m', 'comment', '--comment', `harbor-traffic:${device.key}:upload`,
+        '-j', 'RETURN',
+      ]],
+      ['iptables', [
+        '-w', '1', '-t', 'mangle', '-A', downloadChild,
+        '-o', device.interface, '-d', device.ip,
+        '-m', 'comment', '--comment', `harbor-traffic:${device.key}:download`,
+        '-j', 'RETURN',
+      ]],
+    );
+  }
+  return commands;
+}
+
+export function parseTrafficCounters(text, chain) {
+  const escapedChain = chain.replace(/[.*+?^${}()|[\]\\]/g, '\\$&');
+  const linePattern = new RegExp(
+    `^\\[(\\d+):(\\d+)\\] -A ${escapedChain} .*--comment "?harbor-traffic:([a-f0-9]{16}):(upload|download)"?`,
+  );
+  const counters = new Map();
+  for (const line of String(text || '').split(/\r?\n/)) {
+    const match = line.match(linePattern);
+    if (!match) continue;
+    counters.set(`${match[3]}:${match[4]}`, match[2]);
+  }
+  return counters;
+}
+
+export function createDeviceTrafficService({
+  observe,
+  uploadChain,
+  downloadChain,
+  bypassCidrs,
+  run = spawnSync,
+  nextGeneration = () => crypto.randomUUID(),
+}) {
+  let activeSlot = null;
+  let activeDevices = [];
+  let activeSignature = '';
+  let refreshPromise = null;
+  let current = {
+    generation: nextGeneration(),
+    observedAt: null,
+    source: { error: null },
+    devices: [],
+  };
+
+  function execute(command, args) {
+    const result = run(command, args, COMMAND_OPTIONS);
+    if (result.error || result.status !== 0) throw commandError(command, result);
+    return String(result.stdout || '');
+  }
+
+  function prepare(slot, devices) {
+    for (const [command, args] of buildTrafficRuleCommands({
+      devices,
+      bypassCidrs,
+      uploadChain,
+      downloadChain,
+      slot,
+    })) execute(command, args);
+  }
+
+  function switchTo(slot) {
+    const uploadChild = childChain(uploadChain, slot);
+    const downloadChild = childChain(downloadChain, slot);
+    const replace = activeSlot ? '-R' : '-A';
+    const uploadArgs = activeSlot
+      ? ['-w', '1', '-t', 'raw', replace, uploadChain, '1', '-j', uploadChild]
+      : ['-w', '1', '-t', 'raw', replace, uploadChain, '-j', uploadChild];
+    const downloadArgs = activeSlot
+      ? ['-w', '1', '-t', 'mangle', replace, downloadChain, '1', '-j', downloadChild]
+      : ['-w', '1', '-t', 'mangle', replace, downloadChain, '-j', downloadChild];
+
+    execute('iptables', uploadArgs);
+    try {
+      execute('iptables', downloadArgs);
+    } catch (error) {
+      const rollbackArgs = activeSlot
+        ? ['-w', '1', '-t', 'raw', '-R', uploadChain, '1', '-j', childChain(uploadChain, activeSlot)]
+        : ['-w', '1', '-t', 'raw', '-F', uploadChain];
+      execute('iptables', rollbackArgs);
+      throw error;
+    }
+  }
+
+  function readCounters(devices) {
+    if (!activeSlot) return [];
+    const upload = parseTrafficCounters(
+      execute('iptables-save', ['-c', '-t', 'raw']),
+      childChain(uploadChain, activeSlot),
+    );
+    const download = parseTrafficCounters(
+      execute('iptables-save', ['-c', '-t', 'mangle']),
+      childChain(downloadChain, activeSlot),
+    );
+    return devices.map(({ key, ...device }) => ({
+      ...device,
+      uploadBytes: upload.get(`${key}:upload`) || '0',
+      downloadBytes: download.get(`${key}:download`) || '0',
+    }));
+  }
+
+  async function performRefresh() {
+    let observed;
+    try {
+      observed = await observe();
+    } catch (error) {
+      observed = { observedAt: new Date().toISOString(), observations: [], error: error.message || String(error) };
+    }
+
+    let sourceError = observed?.error || null;
+    const nextDevices = sourceError
+      ? activeDevices
+      : selectTrafficDevices(observed?.observations);
+    const nextSignature = JSON.stringify(nextDevices);
+
+    if (!sourceError && nextSignature !== activeSignature) {
+      const nextSlot = activeSlot === 'A' ? 'B' : 'A';
+      try {
+        prepare(nextSlot, nextDevices);
+        switchTo(nextSlot);
+        activeSlot = nextSlot;
+        activeDevices = nextDevices;
+        activeSignature = nextSignature;
+        current.generation = nextGeneration();
+      } catch (error) {
+        sourceError = error.message || String(error);
+      }
+    }
+
+    let devices = current.devices;
+    let countersRead = false;
+    try {
+      devices = readCounters(activeDevices);
+      countersRead = true;
+    } catch (error) {
+      sourceError = sourceError || error.message || String(error);
+    }
+    current = {
+      generation: current.generation,
+      observedAt: countersRead ? observed?.observedAt || current.observedAt : current.observedAt,
+      source: { error: sourceError },
+      devices,
+    };
+    return structuredClone(current);
+  }
+
+  function refresh() {
+    if (!refreshPromise) {
+      refreshPromise = performRefresh().finally(() => {
+        refreshPromise = null;
+      });
+    }
+    return refreshPromise;
+  }
+
+  return {
+    refresh,
+    snapshot: () => structuredClone(current),
+  };
+}