fix(traffic): записывать потоки при подмене UDP-источника в Docker

Пакеты IPFIX доходили, но разговоры отбрасывались, если Docker подменял адрес jump-host на 172.x. Co-authored-by: Cursor <cursoragent@cursor.com>
2026-09-07 00:02:50 +07:00
parent 13889005f8
commit 04f037c32c
12 changed files with 333 additions and 46 deletions
@@ -1114,7 +1114,15 @@ export default function TrafficPage() {
                </div>
              )}
              <DataPageCard>
-                <TrafficFlowsDataGrid rows={flowStats?.talkers ?? []} />
+                <TrafficFlowsDataGrid
+                  rows={flowStats?.talkers ?? []}
+                  emptyHint={
+                    flowStats?.packetsReceived
+                      ? (flowStats.lastError
+                        || `IPFIX приходит (${flowStats.lastExporterIp ?? "экспортёр"}), но разговоры ещё не записаны.`)
+                      : undefined
+                  }
+                />
              </DataPageCard>
              <FlowOverlaySheet
                open={overlayOpen}
@@ -14,7 +14,7 @@
    "test:auth": "tsx src/lib/permissions.test.ts && tsx src/plugins/auth.smoke.test.ts",
    "test:wireguard": "npx tsx src/services/wireguard-config.test.ts",
    "test:traffic-rate": "tsx src/services/traffic-rate.test.ts",
-    "test:traffic-flow": "tsx src/services/traffic-flow-parse.test.ts",
+    "test:traffic-flow": "tsx src/services/traffic-flow-parse.test.ts && tsx src/services/traffic-flow-map-exporter.test.ts",
    "test:users": "tsx src/modules/users/iface-type.test.ts && tsx src/modules/users/bindings.test.ts"
  },
  "dependencies": {
@@ -30,6 +30,7 @@ export function buildHostComposeOverride(): string {
    "# Не править docker-compose.yml. Traefik не трогать.",
    "# Сначала: wg-quick up wg-flow  (адрес " + row.collectorIp + ")",
    "# затем: docker compose up -d backend",
+    "# Docker userland-proxy может SNAT UDP source в 172.x — ingest сопоставит единственный JH.",
    "",
    "services:",
    "  backend:",
@@ -4,8 +4,10 @@ import { db } from "../db/index.js"
 import { flowBuckets, servers } from "../db/schema.js"
 import type { FlowStatsDto, FlowTalkerDto } from "@mmapp/contracts/traffic-flow"
 import { parseFlowPacket, protoName, type ParsedFlow } from "./traffic-flow-parse.js"
+import { pickServerIdForExporter } from "./traffic-flow-map-exporter.js"
 import {
  getTrafficFlowSettingsRow,
+  listHostPeers,
  recordFlowListenerError,
  recordFlowPacket,
 } from "./traffic-flow-settings.js"
@@ -37,13 +39,30 @@ function minuteBucketIso(at = Date.now()): string {
 }

 function resolveServerId(exporterIp: string): number | null {
-  const exact = db.select().from(servers).where(eq(servers.mgmtTunnelIp, exporterIp)).limit(1).all()[0]
-  return exact ? exact.id : null
+  const settings = getTrafficFlowSettingsRow()
+  const rows = db.select({
+    id: servers.id,
+    host: servers.host,
+    mgmtTunnelIp: servers.mgmtTunnelIp,
+  }).from(servers).all()
+  const byTunnelIp = new Map<string, number>()
+  const hostIps = new Map<string, number>()
+  for (const row of rows) {
+    if (row.mgmtTunnelIp) byTunnelIp.set(row.mgmtTunnelIp, row.id)
+    if (/^\d{1,3}(?:\.\d{1,3}){3}$/.test(row.host)) hostIps.set(row.host, row.id)
+  }
+  return pickServerIdForExporter({
+    exporterIp,
+    overlayPrefix: settings.prefix,
+    byTunnelIp,
+    peers: listHostPeers(),
+    hostIps,
+  })
 }

-function queueFlows(exporterIp: string, flows: ParsedFlow[]) {
+function queueFlows(exporterIp: string, flows: ParsedFlow[]): boolean {
  const serverId = resolveServerId(exporterIp)
-  if (serverId == null) return
+  if (serverId == null) return false
  const bucketAt = minuteBucketIso()
  for (const flow of flows) {
    const key = `${serverId}\0${bucketAt}\0${flow.src}\0${flow.dst}\0${flow.proto}\0${flow.srcPort}\0${flow.dstPort}`
@@ -61,6 +80,7 @@ function queueFlows(exporterIp: string, flows: ParsedFlow[]) {
      })
    }
  }
+  return true
 }

 function flushPending() {
@@ -129,7 +149,14 @@ function onMessage(msg: Buffer, rinfo: { address: string }) {
  try {
    const flows = parseFlowPacket(msg, rinfo.address)
    recordFlowPacket(rinfo.address)
-    if (flows.length) queueFlows(rinfo.address, flows)
+    if (!flows.length) return
+    if (!queueFlows(rinfo.address, flows)) {
+      recordFlowListenerError(
+        `IPFIX от ${rinfo.address}: нет jump-host с адресом wg-flow. Docker SNAT (172.x) при нескольких JH не различим.`,
+      )
+      return
+    }
+    recordFlowListenerError("")
  } catch (e) {
    recordFlowListenerError(e instanceof Error ? e.message : String(e))
  }
@@ -172,6 +199,7 @@ export function startTrafficFlowListener() {
 }

 export function listFlowTalkers(minutes = 5): FlowStatsDto {
+  const settings = getTrafficFlowSettingsRow()
  const rangeStart = new Date(Date.now() - minutes * 60_000).toISOString()
  const rows = db.select().from(flowBuckets).where(gte(flowBuckets.bucketAt, rangeStart)).all()
  const serverRows = db.select().from(servers).all()
@@ -217,7 +245,7 @@ export function listFlowTalkers(minutes = 5): FlowStatsDto {
  const talkers = [...agg.values()]
    .map((t) => ({ ...t, bps: (t.rawBytes * 8) / windowSec }))
    .sort((a, b) => b.bytes - a.bytes)
-    .slice(0, getTrafficFlowSettingsRow().topN)
+    .slice(0, settings.topN)
    .map(({ rawBytes: _raw, ...rest }) => rest)
  let topProto = "—"
  let topProtoBytes = 0
@@ -234,6 +262,9 @@ export function listFlowTalkers(minutes = 5): FlowStatsDto {
    uniqueDst: dsts.size,
    topProto,
    talkers,
+    lastExporterIp: settings.lastExporterIp ?? null,
+    lastError: settings.lastError || null,
+    packetsReceived: settings.packetsReceived,
  }
 }

@@ -0,0 +1,54 @@
+import assert from "node:assert/strict"
+import {
+  bareIpv4,
+  ipInCidr,
+  isNatMasqueradeExporter,
+  normalizeExporterIp,
+  pickServerIdForExporter,
+} from "./traffic-flow-map-exporter.js"
+
+assert.equal(normalizeExporterIp("::ffff:172.18.0.2"), "172.18.0.2")
+assert.equal(bareIpv4("10.255.254.3/32"), "10.255.254.3")
+assert.equal(ipInCidr("10.255.254.3", "10.255.254.0/24"), true)
+assert.equal(ipInCidr("172.18.0.2", "10.255.254.0/24"), false)
+assert.equal(isNatMasqueradeExporter("172.18.0.2", "10.255.254.0/24"), true)
+assert.equal(isNatMasqueradeExporter("10.255.254.3", "10.255.254.0/24"), false)
+assert.equal(isNatMasqueradeExporter("10.0.0.12", "10.255.254.0/24"), true)
+
+const byTunnel = new Map([["10.255.254.3", 7]])
+assert.equal(pickServerIdForExporter({
+  exporterIp: "10.255.254.3",
+  overlayPrefix: "10.255.254.0/24",
+  byTunnelIp: byTunnel,
+  peers: [],
+  hostIps: new Map(),
+}), 7)
+
+assert.equal(pickServerIdForExporter({
+  exporterIp: "172.18.0.2",
+  overlayPrefix: "10.255.254.0/24",
+  byTunnelIp: byTunnel,
+  peers: [{ serverId: 7, address: "10.255.254.3", allowedIps: ["10.255.254.3/32"] }],
+  hostIps: new Map(),
+}), 7)
+
+assert.equal(pickServerIdForExporter({
+  exporterIp: "172.18.0.2",
+  overlayPrefix: "10.255.254.0/24",
+  byTunnelIp: new Map([["10.255.254.3", 7], ["10.255.254.4", 8]]),
+  peers: [
+    { serverId: 7, address: "10.255.254.3", allowedIps: ["10.255.254.3/32"] },
+    { serverId: 8, address: "10.255.254.4", allowedIps: ["10.255.254.4/32"] },
+  ],
+  hostIps: new Map(),
+}), null)
+
+assert.equal(pickServerIdForExporter({
+  exporterIp: "94.142.140.141",
+  overlayPrefix: "10.255.254.0/24",
+  byTunnelIp: byTunnel,
+  peers: [],
+  hostIps: new Map([["94.142.140.141", 7]]),
+}), 7)
+
+console.log("traffic-flow-map-exporter.test.ts: ok")
@@ -0,0 +1,82 @@
+export interface OverlayPeerRef {
+  serverId: number
+  address: string
+  allowedIps: string[]
+}
+
+export function normalizeExporterIp(ip: string): string {
+  const trimmed = ip.trim()
+  if (trimmed.toLowerCase().startsWith("::ffff:")) return trimmed.slice(7)
+  return trimmed
+}
+
+export function bareIpv4(value: string): string {
+  const raw = normalizeExporterIp(value).split("/")[0]?.trim() ?? ""
+  return raw
+}
+
+function ipv4ToInt(ip: string): number | null {
+  const parts = ip.split(".")
+  if (parts.length !== 4) return null
+  const n = parts.map((x) => Number(x))
+  if (n.some((x) => !Number.isInteger(x) || x < 0 || x > 255)) return null
+  return ((n[0]! << 24) | (n[1]! << 16) | (n[2]! << 8) | n[3]!) >>> 0
+}
+
+export function ipInCidr(ip: string, cidr: string): boolean {
+  const host = bareIpv4(ip)
+  const [base, bitsRaw] = cidr.split("/")
+  const bits = Number(bitsRaw ?? 32)
+  const a = ipv4ToInt(host)
+  const b = ipv4ToInt(bareIpv4(base ?? ""))
+  if (a == null || b == null || !Number.isFinite(bits) || bits < 0 || bits > 32) return false
+  const mask = bits === 0 ? 0 : (0xffffffff << (32 - bits)) >>> 0
+  return (a & mask) === (b & mask)
+}
+
+/** Docker userland-proxy / bridge SNAT, не адрес из оверлея wg-flow. */
+export function isNatMasqueradeExporter(ip: string, overlayPrefix: string): boolean {
+  const host = bareIpv4(ip)
+  if (!host) return false
+  if (ipInCidr(host, overlayPrefix)) return false
+  return ipInCidr(host, "10.0.0.0/8")
+    || ipInCidr(host, "172.16.0.0/12")
+    || ipInCidr(host, "192.168.0.0/16")
+    || ipInCidr(host, "127.0.0.0/8")
+}
+
+export function pickServerIdForExporter(opts: {
+  exporterIp: string
+  overlayPrefix: string
+  byTunnelIp: Map<string, number>
+  peers: OverlayPeerRef[]
+  hostIps: Map<string, number>
+}): number | null {
+  const exporter = bareIpv4(opts.exporterIp)
+  if (!exporter) return null
+
+  const exact = opts.byTunnelIp.get(exporter)
+  if (exact != null) return exact
+
+  for (const [ip, id] of opts.byTunnelIp) {
+    if (bareIpv4(ip) === exporter) return id
+  }
+
+  for (const peer of opts.peers) {
+    if (bareIpv4(peer.address) === exporter) return peer.serverId
+    if (peer.allowedIps.some((cidr) => ipInCidr(exporter, cidr) || bareIpv4(cidr) === exporter)) {
+      return peer.serverId
+    }
+  }
+
+  const byHost = opts.hostIps.get(exporter)
+  if (byHost != null) return byHost
+
+  if (!isNatMasqueradeExporter(exporter, opts.overlayPrefix)) return null
+
+  const tunnelIds = [...new Set(opts.byTunnelIp.values())]
+  if (tunnelIds.length === 1) return tunnelIds[0] ?? null
+  const peerIds = [...new Set(opts.peers.map((p) => p.serverId))]
+  if (peerIds.length === 1) return peerIds[0] ?? null
+  return null
+}
@@ -92,7 +92,12 @@ async function ensureWgInputAccept(client: MikrotikClient, listenPort: number):
  return true
 }

-async function ensureTrafficFlow(client: MikrotikClient, collectorIp: string, port: number): Promise<void> {
+async function ensureTrafficFlow(
+  client: MikrotikClient,
+  collectorIp: string,
+  port: number,
+  srcAddress: string,
+): Promise<void> {
  const body = toRosBody({
    enabled: "yes",
    interfaces: "all",
@@ -111,6 +116,7 @@ async function ensureTrafficFlow(client: MikrotikClient, collectorIp: string, po
  const existing = targets.find((t) => String(t["dst-address"] ?? "") === collectorIp)
  const targetBody = toRosBody({
    "dst-address": collectorIp,
+    "src-address": srcAddress,
    port: String(port),
    version: "ipfix",
  })
@@ -232,8 +238,8 @@ export async function applyFlowOverlay(
      steps.push("Firewall input WG уже есть")
    }

-    await ensureTrafficFlow(client, settings.collectorIp, settings.flowListenPort)
-    steps.push(`Traffic Flow → ${settings.collectorIp}:${settings.flowListenPort} ipfix`)
+    await ensureTrafficFlow(client, settings.collectorIp, settings.flowListenPort, address)
+    steps.push(`Traffic Flow → ${settings.collectorIp}:${settings.flowListenPort} ipfix (src ${address})`)

    const listed = await listWireGuardInterfaces({ serverId: String(server.id), includePrivateKey: false })
    const created = listed.interfaces.find((i) => i.name === IFACE_NAME)
@@ -38,4 +38,32 @@ assert.equal(usablePublicHost("192.168.1.10"), "")
 assert.equal(usablePublicHost("mm.example.com:443"), "mm.example.com")
 assert.equal(usablePublicHost("203.0.113.10"), "203.0.113.10")

+resetFlowTemplatesForTests()
+{
+  const tpl = Buffer.alloc(16 + 16 + 20)
+  tpl.writeUInt16BE(10, 0)
+  tpl.writeUInt16BE(tpl.length, 2)
+  tpl.writeUInt16BE(2, 16)
+  tpl.writeUInt16BE(16, 18)
+  tpl.writeUInt16BE(256, 20)
+  tpl.writeUInt16BE(2, 22)
+  tpl.writeUInt16BE(8, 24)
+  tpl.writeUInt16BE(4, 26)
+  tpl.writeUInt16BE(12, 28)
+  tpl.writeUInt16BE(4, 30)
+  const data = Buffer.alloc(16 + 12)
+  data.writeUInt16BE(10, 0)
+  data.writeUInt16BE(data.length, 2)
+  data.writeUInt16BE(256, 16)
+  data.writeUInt16BE(12, 18)
+  data[20] = 10; data[21] = 1; data[22] = 1; data[23] = 8
+  data[24] = 8; data[25] = 8; data[26] = 8; data[27] = 8
+  const fromTpl = parseFlowPacket(tpl, "172.18.0.2")
+  assert.equal(fromTpl.length, 0)
+  const fromData = parseFlowPacket(data, "172.18.0.2")
+  assert.equal(fromData.length, 1)
+  assert.equal(fromData[0]?.src, "10.1.1.8")
+  assert.equal(fromData[0]?.dst, "8.8.8.8")
+}
+
 console.log("traffic-flow-parse.test.ts: ok")
@@ -24,6 +24,48 @@ function ipv4(buf: Buffer, offset: number): string {
  return `${buf[offset]}.${buf[offset + 1]}.${buf[offset + 2]}.${buf[offset + 3]}`
 }

+function ipv6(buf: Buffer, offset: number): string {
+  const parts: string[] = []
+  for (let i = 0; i < 8; i++) parts.push(buf.readUInt16BE(offset + i * 2).toString(16))
+  return parts.join(":")
+}
+
+const VAR_LEN = 0xffff
+
+function consumeField(
+  buf: Buffer,
+  off: number,
+  length: number,
+  limit: number,
+): { data: Buffer; next: number } | null {
+  if (length === VAR_LEN) {
+    if (off >= limit) return null
+    const first = buf[off]!
+    if (first < 255) {
+      const end = off + 1 + first
+      if (end > limit) return null
+      return { data: buf.subarray(off + 1, end), next: end }
+    }
+    if (off + 3 > limit) return null
+    const len = buf.readUInt16BE(off + 1)
+    const end = off + 3 + len
+    if (end > limit) return null
+    return { data: buf.subarray(off + 3, end), next: end }
+  }
+  const end = off + length
+  if (end > limit) return null
+  return { data: buf.subarray(off, end), next: end }
+}
+
+function fixedRecordSize(fields: FieldSpec[]): number | null {
+  let n = 0
+  for (const f of fields) {
+    if (f.length === VAR_LEN) return null
+    n += f.length
+  }
+  return n
+}
+
 function readUint(buf: Buffer, offset: number, length: number): number {
  if (length === 1) return buf.readUInt8(offset)
  if (length === 2) return buf.readUInt16BE(offset)
@@ -87,7 +129,12 @@ function parseIpfixTemplates(exporter: string, buf: Buffer, setStart: number, se
  templatesByExporter.set(exporter, map)
 }

-function recordFromFields(fields: FieldSpec[], buf: Buffer, offset: number): { flow: ParsedFlow; next: number } | null {
+function recordFromFields(
+  fields: FieldSpec[],
+  buf: Buffer,
+  offset: number,
+  limit: number,
+): { flow: ParsedFlow; next: number } | null {
  let off = offset
  let src = ""
  let dst = ""
@@ -98,41 +145,78 @@ function recordFromFields(fields: FieldSpec[], buf: Buffer, offset: number): { f
  let packets = 0
  let inIface = ""
  for (const f of fields) {
-    if (off + f.length > buf.length) return null
+    const field = consumeField(buf, off, f.length, limit)
+    if (!field) return null
+    const { data } = field
    switch (f.type) {
      case 8:
-        if (f.length === 4) src = ipv4(buf, off)
+        if (data.length === 4) src = ipv4(data, 0)
        break
      case 12:
-        if (f.length === 4) dst = ipv4(buf, off)
+        if (data.length === 4) dst = ipv4(data, 0)
+        break
+      case 27:
+        if (data.length === 16 && !src) src = ipv6(data, 0)
+        break
+      case 28:
+        if (data.length === 16 && !dst) dst = ipv6(data, 0)
+        break
+      case 225:
+        if (data.length === 4 && !src) src = ipv4(data, 0)
+        break
+      case 226:
+        if (data.length === 4 && !dst) dst = ipv4(data, 0)
        break
      case 4:
-        proto = readUint(buf, off, f.length)
+        proto = readUint(data, 0, data.length)
        break
      case 7:
-        srcPort = readUint(buf, off, f.length)
+        srcPort = readUint(data, 0, data.length)
        break
      case 11:
-        dstPort = readUint(buf, off, f.length)
+        dstPort = readUint(data, 0, data.length)
        break
      case 1:
-        bytes = readUint(buf, off, f.length)
+        bytes = readUint(data, 0, data.length)
        break
      case 2:
-        packets = readUint(buf, off, f.length)
+        packets = readUint(data, 0, data.length)
+        break
+      case 85:
+        if (!bytes) bytes = readUint(data, 0, data.length)
+        break
+      case 86:
+        if (!packets) packets = readUint(data, 0, data.length)
        break
      case 10:
-        inIface = String(readUint(buf, off, f.length))
+        inIface = String(readUint(data, 0, data.length))
        break
      default:
        break
    }
-    off += f.length
+    off = field.next
  }
-  if (!src && !dst) return { flow: { src, dst, proto, srcPort, dstPort, bytes, packets, inIface }, next: off }
  return { flow: { src, dst, proto, srcPort, dstPort, bytes, packets, inIface }, next: off }
 }

+function parseDataRecords(
+  tpl: Template,
+  buf: Buffer,
+  recOff: number,
+  setEnd: number,
+  out: ParsedFlow[],
+) {
+  const size = fixedRecordSize(tpl.fields)
+  while (recOff + 1 < setEnd) {
+    if (size != null && recOff + size > setEnd) break
+    const parsed = recordFromFields(tpl.fields, buf, recOff, setEnd)
+    if (!parsed) break
+    if (parsed.flow.src || parsed.flow.dst) out.push(parsed.flow)
+    if (parsed.next <= recOff) break
+    recOff = parsed.next
+  }
+}
+
 function parseIpfix(buf: Buffer, exporter: string): ParsedFlow[] {
  if (buf.length < 16) return []
  const total = buf.readUInt16BE(2)
@@ -148,16 +232,7 @@ function parseIpfix(buf: Buffer, exporter: string): ParsedFlow[] {
      parseIpfixTemplates(exporter, buf, off, setEnd, setId)
    } else if (setId >= 256) {
      const tpl = templatesByExporter.get(exporter)?.get(setId)
-      if (tpl) {
-        let recOff = off + 4
-        while (recOff + 1 < setEnd) {
-          const parsed = recordFromFields(tpl.fields, buf, recOff)
-          if (!parsed) break
-          if (parsed.flow.src || parsed.flow.dst) out.push(parsed.flow)
-          if (parsed.next <= recOff) break
-          recOff = parsed.next
-        }
-      }
+      if (tpl) parseDataRecords(tpl, buf, off + 4, setEnd, out)
    }
    off = setEnd
  }
@@ -191,16 +266,7 @@ function parseNetflowV9(buf: Buffer, exporter: string): ParsedFlow[] {
      templatesByExporter.set(exporter, map)
    } else if (setId >= 256) {
      const tpl = map.get(setId)
-      if (tpl) {
-        let recOff = off + 4
-        while (recOff + 1 < setEnd) {
-          const parsed = recordFromFields(tpl.fields, buf, recOff)
-          if (!parsed) break
-          if (parsed.flow.src || parsed.flow.dst) out.push(parsed.flow)
-          if (parsed.next <= recOff) break
-          recOff = parsed.next
-        }
-      }
+      if (tpl) parseDataRecords(tpl, buf, off + 4, setEnd, out)
    }
    off = setEnd
  }
@@ -110,7 +110,6 @@ export function recordFlowPacket(exporterIp: string) {
    lastDatagramAt: nowIso(),
    lastExporterIp: exporterIp,
    packetsReceived: row.packetsReceived + 1,
-    lastError: "",
    updatedAt: nowIso(),
  }).where(eq(trafficFlowSettings.id, 1)).run()
 }
@@ -19,7 +19,13 @@ function formatBytes(n: number): string {
  return `${n} Б`
 }

-function TrafficFlowsDataGrid({ rows }: { rows: FlowTalkerDto[] }) {
+function TrafficFlowsDataGrid({
+  rows,
+  emptyHint,
+}: {
+  rows: FlowTalkerDto[]
+  emptyHint?: string
+}) {
  const columns = useMemo<ColumnDef<FlowTalkerDto>[]>(
    () => [
      {
@@ -96,7 +102,10 @@ function TrafficFlowsDataGrid({ rows }: { rows: FlowTalkerDto[] }) {
    <DataGridShell
      table={table}
      recordCount={rows.length}
-      emptyMessage="Пока нет IPFIX. Поднимите wg-flow на хосте MM и подключите jump-host одним кликом."
+      emptyMessage={
+        emptyHint
+        || "Пока нет IPFIX. Поднимите wg-flow на хосте MM и подключите jump-host одним кликом."
+      }
    />
  )
 }
@@ -88,6 +88,9 @@ export const flowStatsDtoSchema = z.object({
  uniqueDst: z.number().int().nonnegative(),
  topProto: z.string(),
  talkers: z.array(flowTalkerDtoSchema),
+  lastExporterIp: z.string().nullable().optional(),
+  lastError: z.string().nullable().optional(),
+  packetsReceived: z.number().int().nonnegative().optional(),
 })

 export type FlowHostPeer = z.infer<typeof flowHostPeerSchema>