fix(traffic): записывать потоки при подмене UDP-источника в Docker
Docker images / prepare-release (push) Successful in 11s
Docker images / backend-image (push) Successful in 2m11s
Docker images / frontend-image (push) Successful in 2m57s
Docker images / notify-webhook (push) Skipped
Docker images / updater-image (push) Successful in 46s
Docker images / publish-release (push) Successful in 11s

Пакеты IPFIX доходили, но разговоры отбрасывались, если Docker подменял адрес jump-host на 172.x.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
DenozordecandCursor committed 2026-09-07 00:02:50 +07:00
1 parent 13889005f8
commit 5e512407e5
12 files changed
+333 -46

No files matched your search

+98 -32
View File
@@ -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
}