fix(netflow): сохранить IPv6 в смешанных шаблонах IPFIX

2026-10-11 16:44:00 +07:00
1 parent c718654bc7
commit 11813179a7
@@ -3,6 +3,7 @@ import {
  canonicalIp,
  parseCidrV6,
  isNonPublicIp,
+  isUnspecifiedIp,
  pickInternetDest,
  pickInternetPeer,
  pickMapInternetDest,
@@ -38,6 +39,8 @@ for (const ip of [
 ]) assert.equal(isNonPublicIp(ip), false, ip)
 assert.equal(canonicalIp("::ffff:808:808"), "8.8.8.8", "Hex mapped IPv6 uses IPv4 host identity")
 assert.equal(canonicalIp("0:0:0:0:0:ffff:808:808/128"), "8.8.8.8")
+for (const ip of ["::", "::0", "0::0", "0:0:0:0:0:0:0:0", "::ffff:0.0.0.0"]) assert.equal(isUnspecifiedIp(ip), true)
+for (const ip of ["::1", "0:0:0:0:0:0:0:1", "0:::0"]) assert.equal(isUnspecifiedIp(ip), false)
 assert.equal(parseCidrV6("2001:4860::/0")?.net, 0n)
 assert.equal(parseCidrV6("2001:4860::1/128")?.shift, 0n)
 for (const cidr of ["::/129", "::/-1", "::/32junk", "::/32/64", "::", "bad::ip/48"]) assert.equal(parseCidrV6(cidr), null)
@@ -134,6 +134,7 @@ export function isUnspecifiedIp(ip: string): boolean {
  if (!t) return true
  const lower = t.toLowerCase()
  return t === "0.0.0.0" || lower === "::" || lower === "::0"
+    || (lower.includes(":") && !/[1-9a-f]/i.test(lower) && ipv6ToInt(lower) === 0n)
 }

 function usableIp(ip: string | undefined): string {
@@ -0,0 +1,142 @@
+import assert from "node:assert/strict"
+import { createSocket } from "node:dgram"
+import { Worker } from "node:worker_threads"
+import { parseFlowPacket, resetFlowTemplatesForTests } from "./traffic-flow-parse.js"
+import { ipv6ToInt } from "./traffic-flow-ip.js"
+import { pool, closePool } from "../db/index.js"
+import { initDatabase } from "../db/bootstrap.js"
+import type { CollectorHeartbeat, WorkerToMain } from "./traffic-flow-collector-ipc.js"
+
+// A dual-stack template can contain zero IPv4 placeholders in either field order.
+const fields = [[27, 16], [8, 4], [12, 4], [28, 16], [4, 1], [7, 2], [11, 2], [1, 4], [2, 4], [10, 4]] as const
+const endpoints = [
+  ["fd00:144::10", "2001:4860:4860::8888"],
+  ["2001:4860:4860::8888", "fd00:144::10"],
+  ["fd00:144::10", "2606:4700:4700::1111"],
+  ["fd00:144::10", "2001:db8::1"],
+  ["fd00:144::10", "2001:2::1"],
+  ["fd00:144::10", "ff02::1"],
+]
+function header(buffer: Buffer, sequence = 0) {
+  buffer.writeUInt16BE(10, 0)
+  buffer.writeUInt16BE(buffer.length, 2)
+  buffer.writeUInt32BE(Math.floor(Date.now() / 1000), 4)
+  buffer.writeUInt32BE(sequence, 8)
+  buffer.writeUInt32BE(144, 12)
+}
+function address(buffer: Buffer, offset: number, ip: string) {
+  let n = ipv6ToInt(ip)!
+  for (let i = 7; i >= 0; i--) { buffer.writeUInt16BE(Number(n & 0xffffn), offset + i * 2); n >>= 16n }
+}
+function packets() {
+  const template = Buffer.alloc(24 + fields.length * 4)
+  header(template)
+  template.writeUInt16BE(2, 16)
+  template.writeUInt16BE(template.length - 16, 18)
+  template.writeUInt16BE(256, 20)
+  template.writeUInt16BE(fields.length, 22)
+  fields.forEach(([type, length], i) => { template.writeUInt16BE(type, 24 + i * 4); template.writeUInt16BE(length, 26 + i * 4) })
+  const recordLength = fields.reduce((n, [, length]) => n + length, 0)
+  const data = Buffer.alloc(20 + recordLength * endpoints.length)
+  header(data, endpoints.length)
+  data.writeUInt16BE(256, 16)
+  data.writeUInt16BE(data.length - 16, 18)
+  endpoints.forEach(([src, dst], i) => {
+    let offset = 20 + i * recordLength
+    address(data, offset, src!); offset += 16
+    offset += 8 // Zero IPv4 source/destination placeholders.
+    address(data, offset, dst!); offset += 16
+    data.writeUInt8(6, offset++)
+    data.writeUInt16BE(i === 1 ? 443 : 50000 + i, offset); offset += 2
+    data.writeUInt16BE(i === 1 ? 50000 : 443, offset); offset += 2
+    data.writeUInt32BE((i + 1) * 1200, offset); offset += 4
+    data.writeUInt32BE((i + 1) * 2, offset); offset += 4
+    data.writeUInt32BE(1, offset)
+  })
+  return { template, data }
+}
+const fixture = packets()
+resetFlowTemplatesForTests()
+parseFlowPacket(fixture.template, "127.0.0.1")
+const parsed = parseFlowPacket(fixture.data, "127.0.0.1")
+assert.equal(parsed.length, endpoints.length)
+parsed.forEach((flow, i) => {
+  assert.equal(ipv6ToInt(flow.src), ipv6ToInt(endpoints[i]![0]!), "Zero IPv4 must not overwrite preceding IPv6")
+  assert.equal(ipv6ToInt(flow.dst), ipv6ToInt(endpoints[i]![1]!), "Zero IPv4 must not block following IPv6")
+  assert.equal(flow.bytes, (i + 1) * 1200)
+})
+console.log("IPv6 dual-stack IPFIX parsing: passed")
+const ipv4Control = Buffer.from(fixture.data)
+ipv4Control.writeUInt32BE(0x08080808, 36)
+ipv4Control.writeUInt32BE(0x01010101, 40)
+const preferred = parseFlowPacket(ipv4Control, "127.0.0.1")[0]!
+assert.equal(preferred.src, "8.8.8.8", "Present IPv4 retains precedence")
+assert.equal(preferred.dst, "1.1.1.1", "Following IPv6 must not replace usable IPv4")
+
+const ownDb = process.env.MMAPP_OWN_TEST_DB
+if (!ownDb) {
+  console.log("IPv6 UDP/worker/SQL: skip (requires an owned isolated database)")
+  await closePool()
+} else {
+  let worker: Worker | undefined
+  let sender: ReturnType<typeof createSocket> | undefined
+  let stopped = false
+  let heartbeat: CollectorHeartbeat | undefined
+  let workerError: Error | undefined
+  const waitFor = async (predicate: () => boolean) => {
+    const until = Date.now() + 20_000
+    while (!predicate()) {
+      if (workerError) throw workerError
+      assert(Date.now() < until, "Collector event deadline exceeded")
+      await new Promise(resolve => setTimeout(resolve, 50))
+    }
+  }
+  try {
+    assert(/^mm_accept_144_\d+$/.test(ownDb))
+    assert.equal((await pool.query("SELECT current_database() AS name")).rows[0].name, ownDb)
+    await initDatabase()
+    const serverId = Number((await pool.query("INSERT INTO servers (name, host) VALUES ('ipv6-udp-fixture', '127.0.0.1') RETURNING id")).rows[0].id)
+    const reservation = createSocket("udp4")
+    await new Promise<void>(resolve => reservation.bind(0, "127.0.0.1", resolve))
+    const port = reservation.address().port
+    await new Promise<void>(resolve => reservation.close(resolve))
+    const workerFile = new URL(import.meta.url.endsWith(".ts") ? "./traffic-flow-collector-worker.ts" : "./traffic-flow-collector-worker.js", import.meta.url)
+    worker = new Worker(workerFile)
+    worker.on("error", error => { workerError = error })
+    worker.on("message", (message: WorkerToMain) => {
+      if (message.type === "heartbeat") heartbeat = message.payload
+      else if (message.type === "shutdown-complete") stopped = true
+      else if (message.type === "error") workerError = new Error(message.payload.message)
+    })
+    worker.postMessage({ type: "start", payload: { listenHost: "127.0.0.1", listenPort: port, topN: 100, retentionHours: 24, exporterMap: { overlayPrefix: "10.255.254.0/24", byTunnelIp: [["127.0.0.1", serverId]], peers: [], hostIps: [] } } })
+    await waitFor(() => Boolean(heartbeat?.bound))
+    sender = createSocket("udp4")
+    for (const packet of [fixture.template, fixture.data]) await new Promise<void>((resolve, reject) => sender!.send(packet, port, "127.0.0.1", error => error ? reject(error) : resolve()))
+    await waitFor(() => (heartbeat?.packetsReceived ?? 0) >= 2)
+    assert.equal(heartbeat!.dropped, 0)
+    assert.equal(heartbeat!.lastError, "")
+    worker.postMessage({ type: "shutdown" })
+    await waitFor(() => stopped)
+    const rows = (await pool.query("SELECT host(src) AS src, host(dst) AS dst, bytes, packets FROM flow_buckets WHERE server_id=$1 ORDER BY bytes", [serverId])).rows
+    assert.equal(rows.length, endpoints.length)
+    rows.forEach((row, i) => {
+      assert.equal(ipv6ToInt(row.src), ipv6ToInt(endpoints[i]![0]!))
+      assert.equal(ipv6ToInt(row.dst), ipv6ToInt(endpoints[i]![1]!))
+      assert.equal(Number(row.bytes), (i + 1) * 1200)
+      assert.equal(Number(row.packets), (i + 1) * 2)
+    })
+    for (const table of ["flow_minute_stats", "flow_hour_facts", "flow_daily_facts"]) {
+      const sum = (await pool.query(`SELECT SUM(bytes) AS bytes, SUM(packets) AS packets FROM ${table} WHERE server_id=$1`, [serverId])).rows[0]
+      // Raw/minute traffic includes all six flows. Internet facts include only
+      // the first three public peers; documentation/benchmark/multicast are excluded.
+      assert.equal(Number(sum.bytes), table === "flow_minute_stats" ? 25200 : 7200, table)
+      assert.equal(Number(sum.packets), table === "flow_minute_stats" ? 42 : 12, table)
+    }
+    console.log(JSON.stringify({ udpDatagramsReceived: heartbeat?.packetsReceived, rawRows: rows.length, allTrafficBytes: 25200, allTrafficPackets: 42, internetFactsBytes: 7200, internetFactsPackets: 12, workerShutdownAcknowledged: stopped, minuteHourDayTotalsPassed: true }))
+    console.log("IPV6_UDP_PIPELINE_ACCEPTED")
+  } finally {
+    if (sender) await new Promise<void>(resolve => sender!.close(resolve))
+    if (worker) await worker.terminate()
+    await closePool()
+  }
+}
@@ -1,4 +1,4 @@
-import { canonicalIp } from "./traffic-flow-ip.js"
+import { canonicalIp, isUnspecifiedIp } from "./traffic-flow-ip.js"

 export interface ParsedFlow {
  src: string
@@ -99,7 +99,7 @@ function ipv4(buf: Buffer, offset: number): string {
 function usableIpfixIp(ip: string): boolean {
  const t = String(ip ?? "").trim()
  if (!t) return false
-  return t !== "0.0.0.0" && t.toLowerCase() !== "::" && t.toLowerCase() !== "::0"
+  return !isUnspecifiedIp(t)
 }

 function ipv6(buf: Buffer, offset: number): string {
@@ -238,16 +238,22 @@ function recordFromFields(
    const { data } = field
    switch (f.type) {
      case 8:
-        if (data.length === 4) src = ipv4(data, 0)
+        if (data.length === 4) {
+          const value = ipv4(data, 0)
+          if (usableIpfixIp(value) || !usableIpfixIp(src)) src = value
+        }
        break
      case 12:
-        if (data.length === 4) dst = ipv4(data, 0)
+        if (data.length === 4) {
+          const value = ipv4(data, 0)
+          if (usableIpfixIp(value) || !usableIpfixIp(dst)) dst = value
+        }
        break
      case 27:
-        if (data.length === 16 && !src) src = ipv6(data, 0)
+        if (data.length === 16 && !usableIpfixIp(src)) src = ipv6(data, 0)
        break
      case 28:
-        if (data.length === 16 && !dst) dst = ipv6(data, 0)
+        if (data.length === 16 && !usableIpfixIp(dst)) dst = ipv6(data, 0)
        break
      case 15:
        if (data.length === 4 && !nextHop) nextHop = ipv4(data, 0)