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 natSrcPort: number natDstPort: number } export type ParsedFlowInput = Partial & Pick 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: "", natSrcPort: 0, natDstPort: 0, } } 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 ?? "", natSrcPort: flow.natSrcPort ?? 0, natDstPort: flow.natDstPort ?? 0, 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>() function templatesForExporter(exporter: string): Map { const existing = templatesByExporter.get(exporter) if (existing) { templatesByExporter.delete(exporter) templatesByExporter.set(exporter, existing) return existing } const created = new Map() 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 usableIpfixIp(ip: string): boolean { const t = String(ip ?? "").trim() if (!t) return false return t !== "0.0.0.0" && t.toLowerCase() !== "::" && t.toLowerCase() !== "::0" } 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 = "" let natSrcPort = 0 let natDstPort = 0 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 (!usableIpfixIp(src) && usableIpfixIp(natSrc)) src = natSrc } break case 226: if (data.length === 4) { natDst = ipv4(data, 0) if (!usableIpfixIp(dst) && usableIpfixIp(natDst)) dst = natDst } break case 227: natSrcPort = readUint(data, 0, data.length) break case 228: natDstPort = readUint(data, 0, data.length) 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, natSrcPort, natDstPort, }), 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 }