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

export interface ParsedFlow {
  src: string
  dst: string
  proto: number
  srcPort: number
  dstPort: number
  bytes: number
  packets: number
  inIface: string
  outIface: string
  nextHop: string
  flowStartMs: number
  flowEndMs: number
  natSrc: string
  natDst: string
}

export type ParsedFlowInput = Partial<ParsedFlow> & Pick<ParsedFlow, "src" | "dst" | "proto" | "bytes">

export function emptyParsedFlow(): ParsedFlow {
  return {
    src: "",
    dst: "",
    proto: 0,
    srcPort: 0,
    dstPort: 0,
    bytes: 0,
    packets: 0,
    inIface: "",
    outIface: "",
    nextHop: "",
    flowStartMs: 0,
    flowEndMs: 0,
    natSrc: "",
    natDst: "",
  }
}

export function normalizeParsedFlow(flow: ParsedFlowInput): ParsedFlow {
  return {
    ...emptyParsedFlow(),
    ...flow,
    nextHop: flow.nextHop ?? "",
    flowStartMs: flow.flowStartMs ?? 0,
    flowEndMs: flow.flowEndMs ?? 0,
    natSrc: flow.natSrc ?? "",
    natDst: flow.natDst ?? "",
    inIface: flow.inIface ?? "",
    outIface: flow.outIface ?? "",
    srcPort: flow.srcPort ?? 0,
    dstPort: flow.dstPort ?? 0,
    packets: flow.packets ?? 0,
  }
}

interface FieldSpec {
  type: number
  length: number
}

interface Template {
  fields: FieldSpec[]
}

const MAX_TEMPLATE_EXPORTERS = 256
const templatesByExporter = new Map<string, Map<number, Template>>()

function templatesForExporter(exporter: string): Map<number, Template> {
  const existing = templatesByExporter.get(exporter)
  if (existing) {
    templatesByExporter.delete(exporter)
    templatesByExporter.set(exporter, existing)
    return existing
  }
  const created = new Map<number, Template>()
  templatesByExporter.set(exporter, created)
  while (templatesByExporter.size > MAX_TEMPLATE_EXPORTERS) {
    const oldest = templatesByExporter.keys().next().value
    if (oldest == null || oldest === exporter) break
    templatesByExporter.delete(oldest)
  }
  return created
}

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)
  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(normalizeParsedFlow({
      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)),
      outIface: String(buf.readUInt16BE(off + 14)),
    }))
    off += 48
  }
  return out
}

function parseIpfixTemplates(exporter: string, buf: Buffer, setStart: number, setEnd: number, setId: number) {
  let off = setStart + 4
  const map = templatesForExporter(exporter)
  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,
  limit: 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 = ""
  let outIface = ""
  let ifaceName = ""
  let nextHop = ""
  let flowStartMs = 0
  let flowEndMs = 0
  let natSrc = ""
  let natDst = ""
  for (const f of fields) {
    const field = consumeField(buf, off, f.length, limit)
    if (!field) return null
    const { data } = field
    switch (f.type) {
      case 8:
        if (data.length === 4) src = ipv4(data, 0)
        break
      case 12:
        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 15:
        if (data.length === 4 && !nextHop) nextHop = ipv4(data, 0)
        break
      case 18:
        if (data.length === 4 && !nextHop) nextHop = ipv4(data, 0)
        break
      case 62:
        if (data.length === 16 && !nextHop) nextHop = ipv6(data, 0)
        break
      case 225:
        if (data.length === 4) {
          natSrc = ipv4(data, 0)
          if (!src) src = natSrc
        }
        break
      case 226:
        if (data.length === 4) {
          natDst = ipv4(data, 0)
          if (!dst) dst = natDst
        }
        break
      case 4:
        proto = readUint(data, 0, data.length)
        break
      case 7:
        srcPort = readUint(data, 0, data.length)
        break
      case 11:
        dstPort = readUint(data, 0, data.length)
        break
      case 1:
        bytes = readUint(data, 0, data.length)
        break
      case 2:
        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(data, 0, data.length))
        break
      case 14:
        outIface = String(readUint(data, 0, data.length))
        break
      case 21:
        if (!flowEndMs) flowEndMs = readUint(data, 0, data.length)
        break
      case 22:
        if (!flowStartMs) flowStartMs = readUint(data, 0, data.length)
        break
      case 150:
        if (!flowStartMs) flowStartMs = readUint(data, 0, data.length) * 1000
        break
      case 151:
        if (!flowEndMs) flowEndMs = readUint(data, 0, data.length) * 1000
        break
      case 152:
        flowStartMs = readUint(data, 0, data.length)
        break
      case 153:
        flowEndMs = readUint(data, 0, data.length)
        break
      case 82:
        ifaceName = data.toString("utf8").replace(/\0/g, "").trim()
        break
      default:
        break
    }
    off = field.next
  }
  if (ifaceName && !inIface) inIface = ifaceName
  return {
    flow: normalizeParsedFlow({
      src, dst, proto, srcPort, dstPort, bytes, packets, inIface, outIface, nextHop, flowStartMs, flowEndMs, natSrc, natDst,
    }),
    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)
  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) parseDataRecords(tpl, buf, off + 4, setEnd, out)
    }
    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 = templatesForExporter(exporter)
  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) parseDataRecords(tpl, buf, off + 4, setEnd, out)
    }
    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()
}

export function templateExporterCountForTests(): number {
  return templatesByExporter.size
}