feat(traffic): добавить приём Traffic Flow с jump-host

Чтобы видеть «кто с кем», а не только объём порта: IPFIX внутри WG на хосте Docker MM, REST-счётчики не трогаем.

Co-authored-by: Cursor <[email protected]>

backend/src/services/traffic-flow-parse.ts

@@ -0,0 +1,233 @@
+export interface ParsedFlow {
+  src: string
+  dst: string
+  proto: number
+  srcPort: number
+  dstPort: number
+  bytes: number
+  packets: number
+  inIface: string
+}
+
+interface FieldSpec {
+  type: number
+  length: number
+}
+
+interface Template {
+  fields: FieldSpec[]
+}
+
+const templatesByExporter = new Map<string, Map<number, Template>>()
+
+function ipv4(buf: Buffer, offset: number): string {
+  return `${buf[offset]}.${buf[offset + 1]}.${buf[offset + 2]}.${buf[offset + 3]}`
+}
+
+function readUint(buf: Buffer, offset: number, length: number): number {
+  if (length === 1) return buf.readUInt8(offset)
+  if (length === 2) return buf.readUInt16BE(offset)
+  if (length === 4) return buf.readUInt32BE(offset)
+  if (length === 8) {
+    const big = buf.readBigUInt64BE(offset)
+    const n = Number(big)
+    return Number.isFinite(n) ? n : 0
+  }
+  let v = 0
+  for (let i = 0; i < length; i++) v = (v << 8) + buf[offset + i]
+  return v >>> 0
+}
+
+function parseNetflowV5(buf: Buffer): ParsedFlow[] {
+  if (buf.length < 24) return []
+  const count = buf.readUInt16BE(2)
+  const out: ParsedFlow[] = []
+  let off = 24
+  for (let i = 0; i < count && off + 48 <= buf.length; i++) {
+    out.push({
+      src: ipv4(buf, off),
+      dst: ipv4(buf, off + 4),
+      packets: buf.readUInt32BE(off + 16),
+      bytes: buf.readUInt32BE(off + 20),
+      srcPort: buf.readUInt16BE(off + 32),
+      dstPort: buf.readUInt16BE(off + 34),
+      proto: buf.readUInt8(off + 38),
+      inIface: String(buf.readUInt16BE(off + 12)),
+    })
+    off += 48
+  }
+  return out
+}
+
+function parseIpfixTemplates(exporter: string, buf: Buffer, setStart: number, setEnd: number, setId: number) {
+  let off = setStart + 4
+  const map = templatesByExporter.get(exporter) ?? new Map<number, Template>()
+  while (off + 4 <= setEnd) {
+    const templateId = buf.readUInt16BE(off)
+    const fieldCount = buf.readUInt16BE(off + 2)
+    off += 4
+    if (setId === 3) {
+      // options template: skip scope count
+      if (off + 2 > setEnd) break
+      off += 2
+    }
+    const fields: FieldSpec[] = []
+    for (let i = 0; i < fieldCount && off + 4 <= setEnd; i++) {
+      const type = buf.readUInt16BE(off)
+      const length = buf.readUInt16BE(off + 2)
+      off += 4
+      if (type & 0x8000) {
+        if (off + 4 > setEnd) break
+        off += 4
+      }
+      fields.push({ type: type & 0x7fff, length })
+    }
+    if (templateId >= 256) map.set(templateId, { fields })
+  }
+  templatesByExporter.set(exporter, map)
+}
+
+function recordFromFields(fields: FieldSpec[], buf: Buffer, offset: number): { flow: ParsedFlow; next: number } | null {
+  let off = offset
+  let src = ""
+  let dst = ""
+  let proto = 0
+  let srcPort = 0
+  let dstPort = 0
+  let bytes = 0
+  let packets = 0
+  let inIface = ""
+  for (const f of fields) {
+    if (off + f.length > buf.length) return null
+    switch (f.type) {
+      case 8:
+        if (f.length === 4) src = ipv4(buf, off)
+        break
+      case 12:
+        if (f.length === 4) dst = ipv4(buf, off)
+        break
+      case 4:
+        proto = readUint(buf, off, f.length)
+        break
+      case 7:
+        srcPort = readUint(buf, off, f.length)
+        break
+      case 11:
+        dstPort = readUint(buf, off, f.length)
+        break
+      case 1:
+        bytes = readUint(buf, off, f.length)
+        break
+      case 2:
+        packets = readUint(buf, off, f.length)
+        break
+      case 10:
+        inIface = String(readUint(buf, off, f.length))
+        break
+      default:
+        break
+    }
+    off += f.length
+  }
+  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 parseIpfix(buf: Buffer, exporter: string): ParsedFlow[] {
+  if (buf.length < 16) return []
+  const total = buf.readUInt16BE(2)
+  const end = Math.min(buf.length, total)
+  let off = 16
+  const out: ParsedFlow[] = []
+  while (off + 4 <= end) {
+    const setId = buf.readUInt16BE(off)
+    const setLen = buf.readUInt16BE(off + 2)
+    if (setLen < 4 || off + setLen > end) break
+    const setEnd = off + setLen
+    if (setId === 2 || setId === 3) {
+      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
+        }
+      }
+    }
+    off = setEnd
+  }
+  return out
+}
+
+function parseNetflowV9(buf: Buffer, exporter: string): ParsedFlow[] {
+  if (buf.length < 20) return []
+  const count = buf.readUInt16BE(2)
+  let off = 20
+  const out: ParsedFlow[] = []
+  const map = templatesByExporter.get(exporter) ?? new Map<number, Template>()
+  for (let s = 0; s < count && off + 4 <= buf.length; s++) {
+    const setId = buf.readUInt16BE(off)
+    const setLen = buf.readUInt16BE(off + 2)
+    if (setLen < 4 || off + setLen > buf.length) break
+    const setEnd = off + setLen
+    if (setId === 0) {
+      let tOff = off + 4
+      while (tOff + 4 <= setEnd) {
+        const templateId = buf.readUInt16BE(tOff)
+        const fieldCount = buf.readUInt16BE(tOff + 2)
+        tOff += 4
+        const fields: FieldSpec[] = []
+        for (let i = 0; i < fieldCount && tOff + 4 <= setEnd; i++) {
+          fields.push({ type: buf.readUInt16BE(tOff), length: buf.readUInt16BE(tOff + 2) })
+          tOff += 4
+        }
+        if (templateId >= 256) map.set(templateId, { fields })
+      }
+      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
+        }
+      }
+    }
+    off = setEnd
+  }
+  return out
+}
+
+export function parseFlowPacket(buf: Buffer, exporterIp: string): ParsedFlow[] {
+  if (buf.length < 2) return []
+  const version = buf.readUInt16BE(0)
+  if (version === 5) return parseNetflowV5(buf)
+  if (version === 9) return parseNetflowV9(buf, exporterIp)
+  if (version === 10) return parseIpfix(buf, exporterIp)
+  return []
+}
+
+export function protoName(proto: number): string {
+  switch (proto) {
+    case 1: return "ICMP"
+    case 6: return "TCP"
+    case 17: return "UDP"
+    case 47: return "GRE"
+    case 50: return "ESP"
+    case 89: return "OSPF"
+    default: return String(proto)
+  }
+}
+
+export function resetFlowTemplatesForTests() {
+  templatesByExporter.clear()
+}