Refactor VPN proxy client implementation

2026-08-09 00:41:52 +03:00
parent 32be4380a3
commit 34d8b681ad
190 changed files with 20064 additions and 9761 deletions
@@ -0,0 +1,460 @@
+import crypto from 'node:crypto';
+import net from 'node:net';
+import { spawn, type ChildProcessWithoutNullStreams } from 'node:child_process';
+import { isDeviceInterface, type NeighborObservation } from '../adapters/neighbors.js';
+
+interface CommandOptions {
+  encoding: BufferEncoding;
+  timeout: number;
+  killSignal: NodeJS.Signals;
+  input?: string;
+}
+
+interface CommandResult {
+  status: number | null;
+  stdout: string;
+  stderr: string;
+  error?: unknown;
+}
+
+interface TrafficDevice {
+  ip: string;
+  mac: string;
+  interface: string;
+  key: string;
+}
+
+type CounterKind = 'upload' | 'download' | 'proxy-upload' | 'proxy-download';
+type CounterField = 'upload' | 'download' | 'proxyUpload' | 'proxyDownload';
+type CounterOutput = 'uploadBytes' | 'downloadBytes' | 'proxyUploadBytes' | 'proxyDownloadBytes';
+type CounterValues = Record<CounterField, bigint>;
+
+interface RetiredCounters {
+  slot: 'A' | 'B';
+  devices: TrafficDevice[];
+  counters: Map<string, string>;
+}
+
+interface TrafficSnapshot {
+  epoch: string;
+  generation: string;
+  observedAt: string | null;
+  source: { error: string | null };
+  devices: Record<string, unknown>[];
+}
+
+type RunCommand = (command: string, args: string[], options?: CommandOptions) => Promise<CommandResult>;
+
+const COMMAND_OPTIONS: CommandOptions = { encoding: 'utf8', timeout: 2_000, killSignal: 'SIGKILL' };
+const MAC_PATTERN = /^[0-9a-f]{2}(?::[0-9a-f]{2}){5}$/i;
+const CHAIN_PATTERN = /^[a-z0-9_]{1,24}$/i;
+const COUNTERS = [
+  ['upload', 'upload', 'uploadBytes'],
+  ['download', 'download', 'downloadBytes'],
+  ['proxy-upload', 'proxyUpload', 'proxyUploadBytes'],
+  ['proxy-download', 'proxyDownload', 'proxyDownloadBytes'],
+] as const satisfies readonly (readonly [CounterKind, CounterField, CounterOutput])[];
+
+const childChain = (chain: string, slot: string) => `${chain}_${slot}`;
+const proxyChildChain = (chain: string, slot: string) => `${childChain(chain, slot)}_P`;
+const counterKey = ({ ip, mac, interface: deviceInterface }: Omit<TrafficDevice, 'key'>) => crypto
+  .createHash('sha256')
+  .update(`${ip}|${mac}|${deviceInterface}`)
+  .digest('hex')
+  .slice(0, 16);
+
+function commandError(command: string, result: CommandResult) {
+  const cause = result.error instanceof Error ? result.error.message : result.error;
+  return new Error(String(
+    result.stderr || result.stdout || cause || `${command} завершился с ошибкой`,
+  ).trim());
+}
+
+function runCommand(command: string, args: string[], options: CommandOptions = COMMAND_OPTIONS): Promise<CommandResult> {
+  return new Promise((resolve) => {
+    let child: ChildProcessWithoutNullStreams;
+    try {
+      child = spawn(command, args, { stdio: ['pipe', 'pipe', 'pipe'] });
+    } catch (error) {
+      resolve({ status: null, stdout: '', stderr: '', error });
+      return;
+    }
+    const stdout: Buffer[] = [];
+    const stderr: Buffer[] = [];
+    let settled = false;
+    let timedOut = false;
+    let timer: ReturnType<typeof setTimeout> | undefined;
+    const finish = (result: Pick<CommandResult, 'status'> & { error?: unknown }) => {
+      if (settled) return;
+      settled = true;
+      clearTimeout(timer);
+      resolve({
+        stdout: Buffer.concat(stdout).toString(options.encoding || 'utf8'),
+        stderr: Buffer.concat(stderr).toString(options.encoding || 'utf8'),
+        ...result,
+      });
+    };
+    child.stdout.on('data', (chunk: Buffer) => stdout.push(chunk));
+    child.stderr.on('data', (chunk: Buffer) => stderr.push(chunk));
+    child.on('error', (error) => finish({ status: null, error }));
+    child.on('close', (status) => finish({
+      status,
+      error: timedOut ? new Error(`${command} превысил ${options.timeout} мс`) : null,
+    }));
+    timer = setTimeout(() => {
+      timedOut = true;
+      child.kill(options.killSignal || 'SIGKILL');
+    }, options.timeout);
+    child.stdin.on('error', () => {});
+    child.stdin.end(options.input == null ? undefined : options.input);
+  });
+}
+
+function isIpv4Cidr(value: unknown) {
+  const [address, prefix, extra] = String(value).split('/');
+  const size = Number(prefix);
+  return extra === undefined && net.isIPv4(address)
+    && Number.isInteger(size) && size >= 0 && size <= 32;
+}
+
+const zeroCounters = (): CounterValues => ({ upload: 0n, download: 0n, proxyUpload: 0n, proxyDownload: 0n });
+
+function record(value: unknown): Record<string, unknown> {
+  return value && typeof value === 'object' && !Array.isArray(value)
+    ? value as Record<string, unknown>
+    : {};
+}
+
+export function selectTrafficDevices(observations: unknown): TrafficDevice[] {
+  const candidates = new Map<string, Omit<TrafficDevice, 'key'>>();
+  const ipsByMac = new Map<string, Set<string>>();
+  const locationsByIp = new Map<string, Set<string>>();
+
+  for (const value of Array.isArray(observations) ? observations : []) {
+    const observation = record(value);
+    const ip = String(observation.ip || '');
+    const mac = String(observation.mac || '').toLowerCase();
+    const deviceInterface = String(observation.interface || '');
+    if (!net.isIPv4(ip) || !MAC_PATTERN.test(mac) || !isDeviceInterface(deviceInterface)) 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 buildTrafficRestore({
+  devices,
+  bypassCidrs,
+  uploadChain,
+  downloadChain,
+  slot,
+  proxyPort,
+}: {
+  devices: readonly TrafficDevice[];
+  bypassCidrs: readonly string[];
+  uploadChain: string;
+  downloadChain: string;
+  slot: string;
+  proxyPort: number;
+}) {
+  if (!CHAIN_PATTERN.test(uploadChain) || !CHAIN_PATTERN.test(downloadChain)
+    || !['A', 'B'].includes(slot) || !Number.isInteger(proxyPort)
+    || proxyPort < 1 || proxyPort > 65_535
+    || !Array.isArray(bypassCidrs) || bypassCidrs.some((cidr) => !isIpv4Cidr(cidr))) {
+    throw new Error('Некорректная конфигурация traffic accounting');
+  }
+  const uploadChild = childChain(uploadChain, slot);
+  const downloadChild = childChain(downloadChain, slot);
+  const proxyUploadChild = proxyChildChain(uploadChain, slot);
+  const proxyDownloadChild = proxyChildChain(downloadChain, slot);
+  const raw = [
+    '*raw',
+    `-F ${uploadChild}`,
+    `-F ${proxyUploadChild}`,
+    `-A ${uploadChild} -i br-+ -j RETURN`,
+    `-A ${uploadChild} -p tcp --dport ${proxyPort} -m addrtype --dst-type LOCAL -j ${proxyUploadChild}`,
+    `-A ${uploadChild} -p tcp --dport ${proxyPort} -m addrtype --dst-type LOCAL -j RETURN`,
+    `-A ${uploadChild} -p udp --dport ${proxyPort} -m addrtype --dst-type LOCAL -j ${proxyUploadChild}`,
+    `-A ${uploadChild} -p udp --dport ${proxyPort} -m addrtype --dst-type LOCAL -j RETURN`,
+  ];
+  const mangle = [
+    '*mangle',
+    `-F ${downloadChild}`,
+    `-F ${proxyDownloadChild}`,
+    `-A ${downloadChild} -p tcp --sport ${proxyPort} -m addrtype --src-type LOCAL -j ${proxyDownloadChild}`,
+    `-A ${downloadChild} -p tcp --sport ${proxyPort} -m addrtype --src-type LOCAL -j RETURN`,
+    `-A ${downloadChild} -p udp --sport ${proxyPort} -m addrtype --src-type LOCAL -j ${proxyDownloadChild}`,
+    `-A ${downloadChild} -p udp --sport ${proxyPort} -m addrtype --src-type LOCAL -j RETURN`,
+  ];
+
+  for (const device of devices) {
+    for (const protocol of ['tcp', 'udp']) {
+      raw.push(`-A ${proxyUploadChild} -i ${device.interface} -s ${device.ip} -m mac --mac-source ${device.mac} -p ${protocol} -m comment --comment harbor-traffic:${device.key}:proxy-upload -j RETURN`);
+      mangle.push(`-A ${proxyDownloadChild} -o ${device.interface} -d ${device.ip} -p ${protocol} -m comment --comment harbor-traffic:${device.key}:proxy-download -j RETURN`);
+    }
+  }
+  for (const cidr of bypassCidrs) {
+    raw.push(`-A ${uploadChild} -d ${cidr} -j RETURN`);
+    mangle.push(`-A ${downloadChild} -s ${cidr} -j RETURN`);
+  }
+  for (const device of devices) {
+    raw.push(`-A ${uploadChild} -i ${device.interface} -s ${device.ip} -m mac --mac-source ${device.mac} -m comment --comment harbor-traffic:${device.key}:upload -j RETURN`);
+    mangle.push(`-A ${downloadChild} -o ${device.interface} -d ${device.ip} -m comment --comment harbor-traffic:${device.key}:download -j RETURN`);
+  }
+  return [...raw, 'COMMIT', ...mangle, 'COMMIT', ''].join('\n');
+}
+
+export function parseTrafficCounters(text: unknown, chain: string): Map<string, string> {
+  const escapedChain = chain.replace(/[.*+?^${}()|[\]\\]/g, '\\$&');
+  const linePattern = new RegExp(
+    `^\\[(\\d+):(\\d+)\\] -A ${escapedChain} .*--comment "?harbor-traffic:([a-f0-9]{16}):(upload|download|proxy-upload|proxy-download)"?`,
+  );
+  const counters = new Map<string, string>();
+  for (const line of String(text || '').split(/\r?\n/)) {
+    const match = line.match(linePattern);
+    if (!match) continue;
+    const key = `${match[3]}:${match[4]}`;
+    counters.set(key, (BigInt(counters.get(key) || '0') + BigInt(match[2])).toString());
+  }
+  return counters;
+}
+
+export function createDeviceTrafficService({
+  observe,
+  uploadChain,
+  downloadChain,
+  bypassCidrs,
+  proxyPort,
+  run = runCommand,
+  nextGeneration = () => crypto.randomUUID(),
+}: {
+  observe: () => Promise<unknown> | unknown;
+  uploadChain: string;
+  downloadChain: string;
+  bypassCidrs: string[];
+  proxyPort: number;
+  run?: RunCommand;
+  nextGeneration?: () => string;
+}) {
+  const epoch = nextGeneration();
+  let activeSlot: 'A' | 'B' | null = null;
+  let activeDevices: TrafficDevice[] = [];
+  let activeSignature = '';
+  let activeCounters = new Map<string, string>();
+  let pendingRetired: RetiredCounters | null = null;
+  let refreshPromise: Promise<TrafficSnapshot> | null = null;
+  const finalized = new Map<string, CounterValues>();
+  const devicesByKey = new Map<string, TrafficDevice>();
+  let current: TrafficSnapshot = {
+    epoch,
+    generation: epoch,
+    observedAt: null,
+    source: { error: null },
+    devices: [],
+  };
+
+  async function execute(command: string, args: string[], options: CommandOptions = COMMAND_OPTIONS) {
+    const result = await run(command, args, options);
+    if (result.error || result.status !== 0) throw commandError(command, result);
+    return String(result.stdout || '');
+  }
+
+  async function prepare(slot: 'A' | 'B', devices: TrafficDevice[]) {
+    const input = buildTrafficRestore({
+      devices,
+      bypassCidrs,
+      uploadChain,
+      downloadChain,
+      slot,
+      proxyPort,
+    });
+    await execute('iptables-restore', ['-w', '1', '--noflush'], { ...COMMAND_OPTIONS, input });
+  }
+
+  async function switchTo(slot: 'A' | 'B') {
+    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];
+
+    await execute('iptables', uploadArgs);
+    try {
+      await 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];
+      await execute('iptables', rollbackArgs);
+      throw error;
+    }
+  }
+
+  async function readCounters(devices: TrafficDevice[], slot: 'A' | 'B' | null): Promise<Map<string, string>> {
+    if (!slot) return new Map<string, string>();
+    const [raw, mangle] = await Promise.all([
+      execute('iptables-save', ['-c', '-t', 'raw']),
+      execute('iptables-save', ['-c', '-t', 'mangle']),
+    ]);
+    const parsed = {
+      upload: parseTrafficCounters(raw, childChain(uploadChain, slot)),
+      download: parseTrafficCounters(mangle, childChain(downloadChain, slot)),
+      proxyUpload: parseTrafficCounters(raw, proxyChildChain(uploadChain, slot)),
+      proxyDownload: parseTrafficCounters(mangle, proxyChildChain(downloadChain, slot)),
+    };
+    const counters = new Map<string, string>();
+    for (const { key } of devices) {
+      for (const [kind, field] of COUNTERS) {
+        counters.set(`${key}:${kind}`, parsed[field].get(`${key}:${kind}`) || '0');
+      }
+    }
+    return counters;
+  }
+
+  function counter(counters: Map<string, string>, key: string, direction: CounterKind) {
+    return BigInt(counters.get(`${key}:${direction}`) || '0');
+  }
+
+  function remember(devices: TrafficDevice[]) {
+    for (const device of devices) devicesByKey.set(device.key, device);
+  }
+
+  async function finalizeRetired() {
+    if (!pendingRetired) return false;
+    const counters = await readCounters(pendingRetired.devices, pendingRetired.slot);
+    for (const { key } of pendingRetired.devices) {
+      const previous = finalized.get(key) || zeroCounters();
+      const next: CounterValues = { ...previous };
+      for (const [kind, field] of COUNTERS) next[field] += counter(counters, key, kind);
+      finalized.set(key, next);
+    }
+    pendingRetired = null;
+    return true;
+  }
+
+  function processTotals() {
+    const activeByMac = new Map(activeDevices.map((device) => [device.mac, device]));
+    const totalsByMac = new Map<string, TrafficDevice & CounterValues>();
+    for (const [key, remembered] of devicesByKey) {
+      const base = finalized.get(key) || zeroCounters();
+      const pending = pendingRetired?.counters || new Map();
+      const previous = totalsByMac.get(remembered.mac) || { ...remembered, ...zeroCounters() };
+      const total: TrafficDevice & CounterValues = {
+        ...(activeByMac.get(remembered.mac) || remembered),
+        ...zeroCounters(),
+      };
+      for (const [kind, field] of COUNTERS) {
+        total[field] = previous[field] + base[field]
+          + counter(pending, key, kind) + counter(activeCounters, key, kind);
+      }
+      totalsByMac.set(remembered.mac, total);
+    }
+    return [...totalsByMac.values()]
+      .map((total) => {
+        const { key: _key, upload, download, proxyUpload, proxyDownload, ...device } = total;
+        return {
+          ...device,
+          uploadBytes: upload.toString(),
+          downloadBytes: download.toString(),
+          proxyUploadBytes: proxyUpload.toString(),
+          proxyDownloadBytes: proxyDownload.toString(),
+        };
+      })
+      .sort((left, right) => left.mac.localeCompare(right.mac));
+  }
+
+  async function performRefresh() {
+    let observed: Record<string, unknown>;
+    try {
+      observed = record(await observe());
+    } catch (error) {
+      observed = { observedAt: new Date().toISOString(), observations: [], error: error instanceof Error ? error.message : String(error) };
+    }
+
+    let sourceError = observed.error ? String(observed.error) : null;
+    const nextDevices = sourceError
+      ? activeDevices
+      : selectTrafficDevices(observed?.observations);
+    const nextSignature = JSON.stringify(nextDevices);
+    let countersRead = false;
+
+    if (pendingRetired) {
+      try {
+        countersRead = await finalizeRetired() || countersRead;
+      } catch (error) {
+        sourceError = sourceError || (error instanceof Error ? error.message : String(error));
+      }
+    }
+
+    if (!pendingRetired && !sourceError && nextSignature !== activeSignature) {
+      const nextSlot = activeSlot === 'A' ? 'B' : 'A';
+      try {
+        await prepare(nextSlot, nextDevices);
+        await switchTo(nextSlot);
+        const retired = activeSlot ? {
+          slot: activeSlot,
+          devices: activeDevices,
+          counters: activeCounters,
+        } : null;
+        activeSlot = nextSlot;
+        activeDevices = nextDevices;
+        activeSignature = nextSignature;
+        activeCounters = new Map();
+        pendingRetired = retired;
+        remember(nextDevices);
+        current.generation = nextGeneration();
+        if (pendingRetired) countersRead = await finalizeRetired() || countersRead;
+      } catch (error) {
+        sourceError = error instanceof Error ? error.message : String(error);
+      }
+    }
+
+    try {
+      activeCounters = await readCounters(activeDevices, activeSlot);
+      countersRead = true;
+    } catch (error) {
+      sourceError = sourceError || (error instanceof Error ? error.message : String(error));
+    }
+    current = {
+      epoch,
+      generation: current.generation,
+      observedAt: countersRead && typeof observed.observedAt === 'string'
+        ? observed.observedAt
+        : current.observedAt,
+      source: { error: sourceError },
+      devices: countersRead ? processTotals() : current.devices,
+    };
+    return structuredClone(current);
+  }
+
+  function refresh() {
+    if (!refreshPromise) {
+      refreshPromise = performRefresh().finally(() => {
+        refreshPromise = null;
+      });
+    }
+    return refreshPromise;
+  }
+
+  return {
+    refresh,
+    snapshot: () => structuredClone(current),
+  };
+}