From 5895e905b7f328d31b6086f9167a54a4004dd731 Mon Sep 17 00:00:00 2001 From: 2GT-Rich <107219997+2GT-Rich@users.noreply.github.com> Date: Fri, 31 Jul 2026 21:48:50 -0700 Subject: [PATCH] fix(netflow): bound template cache, throttle UDP ingress, canonical IPv6 (v0.17.7-beta) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Follow-up to the decoder DoS fix (7e11115), closing the remaining hardening items from the same adversarial-review triage. Bounded template cache (was unbounded): - Templates now carry a lastUsed timestamp, refreshed whenever one decodes data. - New exported pruneTemplateCache() evicts templates idle past a TTL then enforces a hard cap, least-recently-used first. The collector runs it on its existing 60s map-refresh timer (30 min TTL, 10k entries) — no new timer. Bounded per-source bookkeeping: - unknownExporterIds is capped; sources beyond the cap share one overflow pseudo-exporter instead of allocating a new id each (client attribution doesn't depend on exporter identity, so only per-exporter stats coarsen). - Pseudo ids now come from a monotonic counter rather than map size, so they can't collide after eviction. UDP ingress guards (the socket is unauthenticated by protocol design): - Drop datagrams too small to carry a header before doing any parsing work. - New PacketRateLimiter: fixed-window per-source cap (2000 pkt/s), with the counter map cleared each window and a cap on distinct sources per window so spoofed source IPs can't grow it. - getStats() now reports packetsDropped and templatesCached. Canonical IPv6 rendering: - formatIPv6 applies RFC 5952 zero-compression (2001:db8::1, not the fully expanded form) and unwraps IPv4-mapped addresses (::ffff:192.0.2.1 -> 192.0.2.1) so those flows match client/exporter IPs instead of appearing as separate IPv6 endpoints. Adds 7 tests (rate limiter windows//caps/isolation, cache TTL + LRU eviction, lastUsed refresh, IPv6 canonical + mapped + no-run forms); 126 backend tests pass. Verified live: flooding the collector with 3000 packets from one source dropped 614 once over the cap while the backend stayed healthy and served the API, malicious templates were still refused (cache held only the 2 real ones), and real exporter flows kept decoding right through it (260 -> 336). Co-Authored-By: Claude Opus 4.8 --- README.md | 2 +- backend/package.json | 2 +- .../src/services/netflow/NetflowCollector.ts | 63 +++++++++++++-- .../netflow/__tests__/decoder.test.ts | 80 ++++++++++++++++++- .../netflow/__tests__/rateLimiter.test.ts | 44 ++++++++++ backend/src/services/netflow/decoder.ts | 67 ++++++++++++++-- backend/src/services/netflow/rateLimiter.ts | 65 +++++++++++++++ frontend/package.json | 2 +- frontend/src/version.ts | 2 +- 9 files changed, 311 insertions(+), 16 deletions(-) create mode 100644 backend/src/services/netflow/__tests__/rateLimiter.test.ts create mode 100644 backend/src/services/netflow/rateLimiter.ts diff --git a/README.md b/README.md index b5e5e4c..3a75175 100644 --- a/README.md +++ b/README.md @@ -2,7 +2,7 @@ A self-hosted, full-stack network management platform for MikroTik devices. Monitor, configure, and manage your entire MikroTik infrastructure — routers, switches, and wireless access points — from a single web interface. -![Version](https://img.shields.io/badge/version-0.17.6_Beta-blue) +![Version](https://img.shields.io/badge/version-0.17.7_Beta-blue) ![License](https://img.shields.io/badge/license-AGPLv3-blue) ![Docker](https://img.shields.io/badge/docker-compose-2496ED?logo=docker&logoColor=white) ![TypeScript](https://img.shields.io/badge/TypeScript-5.3-3178C6?logo=typescript&logoColor=white) diff --git a/backend/package.json b/backend/package.json index 8de973c..837ab66 100644 --- a/backend/package.json +++ b/backend/package.json @@ -1,6 +1,6 @@ { "name": "mikrotik-manager-backend", - "version": "0.17.6-beta", + "version": "0.17.7-beta", "description": "Mikrotik Network Management Platform - Backend API", "main": "dist/index.js", "scripts": { diff --git a/backend/src/services/netflow/NetflowCollector.ts b/backend/src/services/netflow/NetflowCollector.ts index c20c3c3..04412b3 100644 --- a/backend/src/services/netflow/NetflowCollector.ts +++ b/backend/src/services/netflow/NetflowCollector.ts @@ -8,14 +8,31 @@ import * as dgram from 'dgram'; import { Point } from '@influxdata/influxdb-client'; import { query } from '../../config/database'; import { getWriteApi } from '../../config/influxdb'; -import { decodePacket, TemplateCache } from './decoder'; +import { decodePacket, pruneTemplateCache, TemplateCache } from './decoder'; import { classifyApp, APP_LAN } from './appCategories'; import { FlowAggregator, CLIENT_UNKNOWN, Direction } from './FlowAggregator'; +import { PacketRateLimiter } from './rateLimiter'; const FLUSH_INTERVAL_MS = 60_000; const MAP_REFRESH_MS = 60_000; const SETTINGS_RECONCILE_MS = 60_000; +// Smallest packet that could carry a usable header (v9 needs 20, IPFIX 16). +const MIN_PACKET_BYTES = 16; + +// Templates unused for this long are dropped, and the cache is hard-capped, so a +// hostile or churning sender can't grow it without bound. +const TEMPLATE_TTL_MS = 30 * 60_000; +const MAX_TEMPLATES = 10_000; + +// Bounds on per-source bookkeeping for senders we can't match to a device. +const MAX_UNKNOWN_EXPORTERS = 500; +const OVERFLOW_EXPORTER_ID = -999_999; + +// Generous for real exporters (a busy RouterOS device sends far less), tight +// enough that a flood can't monopolise the event loop. +const RATE_LIMIT = { windowMs: 1_000, maxPerSource: 2_000, maxSources: 1_000 }; + export interface ExporterStats { deviceId: number; deviceName: string; @@ -32,6 +49,8 @@ export interface CollectorStats { flowsAttributed: number; packetsFromUnknownExporter: number; recordsWithoutTemplate: number; + packetsDropped: number; + templatesCached: number; exporters: ExporterStats[]; } @@ -76,8 +95,14 @@ export class NetflowCollector { // Pseudo-exporter ids for sources that don't match a managed device (e.g. // routers exporting from behind NAT). Negative so they can never collide - // with real device ids. + // with real device ids. Capped at MAX_UNKNOWN_EXPORTERS; sources beyond that + // share OVERFLOW_EXPORTER_ID (client attribution doesn't depend on exporter + // identity, so this only coarsens per-exporter stats). private unknownExporterIds = new Map(); + private nextPseudoId = -1; + + private rateLimiter = new PacketRateLimiter(RATE_LIMIT); + private packetsDropped = 0; private flushTimer: ReturnType | null = null; private mapTimer: ReturnType | null = null; @@ -99,6 +124,8 @@ export class NetflowCollector { }, FLUSH_INTERVAL_MS); this.mapTimer = setInterval(() => { this.refreshMaps().catch(() => {}); + // Expire idle templates and cap the cache on the same cadence. + pruneTemplateCache(this.templates, TEMPLATE_TTL_MS, MAX_TEMPLATES); }, MAP_REFRESH_MS); this.reconcileTimer = setInterval(() => { this.reconcile().catch(() => {}); @@ -138,6 +165,9 @@ export class NetflowCollector { flowsAttributed: this.flowsAttributed, packetsFromUnknownExporter: this.packetsFromUnknownExporter, recordsWithoutTemplate: this.recordsWithoutTemplate, + // packetsDropped already counts both undersized and rate-limited drops. + packetsDropped: this.packetsDropped, + templatesCached: this.templates.size, exporters: Array.from(this.exporterStats.values()), }; } @@ -194,6 +224,19 @@ export class NetflowCollector { private onMessage(msg: Buffer, rinfo: dgram.RemoteInfo): void { this.packetsReceived++; + + // Cheap guards before any parsing work: undersized datagrams can't carry a + // usable header, and a single source must not be able to monopolise the + // event loop (the socket is unauthenticated by protocol design). + if (msg.length < MIN_PACKET_BYTES) { + this.packetsDropped++; + return; + } + if (this.rateLimiter.shouldDrop(rinfo.address)) { + this.packetsDropped++; + return; + } + let exporter = this.exporterByIp.get(rinfo.address); if (!exporter) { // Source doesn't match any managed device — common when the exporting @@ -205,11 +248,19 @@ export class NetflowCollector { if (!this.settings.acceptUnknown) return; let pseudoId = this.unknownExporterIds.get(rinfo.address); if (pseudoId === undefined) { - pseudoId = -(this.unknownExporterIds.size + 1); - this.unknownExporterIds.set(rinfo.address, pseudoId); - console.log(`[NetFlow] Accepting flows from unidentified exporter ${rinfo.address} (NAT in path?)`); + if (this.unknownExporterIds.size >= MAX_UNKNOWN_EXPORTERS) { + // Too many distinct unidentified sources to track individually; fold + // them into one bucket so the maps stay bounded. + exporter = { deviceId: OVERFLOW_EXPORTER_ID, deviceName: 'Unidentified (many sources)' }; + } else { + pseudoId = this.nextPseudoId--; + this.unknownExporterIds.set(rinfo.address, pseudoId); + console.log(`[NetFlow] Accepting flows from unidentified exporter ${rinfo.address} (NAT in path?)`); + } + } + if (!exporter) { + exporter = { deviceId: pseudoId as number, deviceName: `Unidentified (${rinfo.address})` }; } - exporter = { deviceId: pseudoId, deviceName: `Unidentified (${rinfo.address})` }; } const result = decodePacket(msg, rinfo.address, this.templates); diff --git a/backend/src/services/netflow/__tests__/decoder.test.ts b/backend/src/services/netflow/__tests__/decoder.test.ts index bad03ac..3fda2ea 100644 --- a/backend/src/services/netflow/__tests__/decoder.test.ts +++ b/backend/src/services/netflow/__tests__/decoder.test.ts @@ -1,4 +1,4 @@ -import { decodePacket, TemplateCache } from '../decoder'; +import { decodePacket, pruneTemplateCache, TemplateCache } from '../decoder'; // ─── Fixture builders ───────────────────────────────────────────────────────── @@ -268,6 +268,47 @@ describe('IPFIX decoding', () => { expect(result.flows).toHaveLength(1); expect(result.flows[0].srcAddr).toMatch(/^fd00:/); }); + + it('renders IPv6 canonically and unwraps IPv4-mapped addresses', () => { + const V6_FIELDS: FieldSpec[] = [ + { id: 27, length: 16 }, + { id: 28, length: 16 }, + { id: 1, length: 4 }, + ]; + // Decode one flow whose src/dst are the given raw 16-byte addresses + const decodePair = (src: Buffer, dst: Buffer, templateId: number) => { + const cache: TemplateCache = new Map(); + const packet = ipfixPacket(7, [ + flowset(2, templateBody(templateId, V6_FIELDS)), + flowset(templateId, Buffer.concat([src, dst, u32(100)])), + ]); + return decodePacket(packet, '10.0.0.1', cache).flows[0]; + }; + + // 2001:db8::1 — interior zero run compressed, not fully expanded + const compressible = Buffer.alloc(16); + compressible.writeUInt16BE(0x2001, 0); + compressible.writeUInt16BE(0x0db8, 2); + compressible[15] = 1; + // ::ffff:192.0.2.1 — IPv4-mapped + const mapped = Buffer.alloc(16); + mapped.writeUInt16BE(0xffff, 10); + Buffer.from([192, 0, 2, 1]).copy(mapped, 12); + + const flow = decodePair(compressible, mapped, 310); + expect(flow.srcAddr).toBe('2001:db8::1'); + expect(flow.dstAddr).toBe('192.0.2.1'); + + // Loopback (all zeros but the last group) and a no-run address + const loopback = Buffer.alloc(16); + loopback[15] = 1; + const noRun = Buffer.alloc(16); + for (let i = 0; i < 8; i++) noRun.writeUInt16BE(0x2000 + i, i * 2); + + const flow2 = decodePair(loopback, noRun, 311); + expect(flow2.srcAddr).toBe('::1'); + expect(flow2.dstAddr).toBe('2000:2001:2002:2003:2004:2005:2006:2007'); + }); }); describe('malformed / hostile packets', () => { @@ -332,6 +373,43 @@ describe('malformed / hostile packets', () => { expect(result.flows).toHaveLength(0); }); + it('bounds the template cache by TTL and hard cap', () => { + const cache: TemplateCache = new Map(); + const mkTemplate = (lastUsed: number) => ({ + fields: [{ id: 1, length: 4, enterprise: false }], + minLength: 4, + lastUsed, + }); + + // Idle templates age out; recently-used ones survive + cache.set('a', mkTemplate(0)); + cache.set('b', mkTemplate(9_000)); + expect(pruneTemplateCache(cache, 5_000, 100, 10_000)).toBe(1); + expect([...cache.keys()]).toEqual(['b']); + + // Hard cap evicts least-recently-used first + cache.clear(); + for (let i = 0; i < 50; i++) cache.set(`t${i}`, mkTemplate(i)); + pruneTemplateCache(cache, 1_000_000, 10, 100); + expect(cache.size).toBe(10); + expect(cache.has('t49')).toBe(true); // newest kept + expect(cache.has('t0')).toBe(false); // oldest evicted + }); + + it('keeps a template alive while it is actively decoding data', () => { + const cache: TemplateCache = new Map(); + decodePacket(v9Packet(0, [flowset(0, templateBody(256, V9_FIELDS))]), '10.0.0.1', cache); + const key = [...cache.keys()][0]; + cache.get(key)!.lastUsed = 0; // pretend it went idle + + decodePacket( + v9Packet(0, [flowset(256, record('192.168.1.10', '1.1.1.1', 50000, 53, 17, 120, 2))]), + '10.0.0.1', + cache + ); + expect(cache.get(key)!.lastUsed).toBeGreaterThan(0); + }); + it('still decodes valid traffic after rejecting a malformed template', () => { const cache: TemplateCache = new Map(); decodePacket( diff --git a/backend/src/services/netflow/__tests__/rateLimiter.test.ts b/backend/src/services/netflow/__tests__/rateLimiter.test.ts new file mode 100644 index 0000000..115ad1b --- /dev/null +++ b/backend/src/services/netflow/__tests__/rateLimiter.test.ts @@ -0,0 +1,44 @@ +import { PacketRateLimiter } from '../rateLimiter'; + +const OPTS = { windowMs: 1000, maxPerSource: 3, maxSources: 2 }; + +describe('PacketRateLimiter', () => { + it('allows traffic up to the per-source limit and drops the excess', () => { + const rl = new PacketRateLimiter(OPTS, 0); + expect(rl.shouldDrop('10.0.0.1', 0)).toBe(false); + expect(rl.shouldDrop('10.0.0.1', 10)).toBe(false); + expect(rl.shouldDrop('10.0.0.1', 20)).toBe(false); + expect(rl.shouldDrop('10.0.0.1', 30)).toBe(true); // 4th in window + expect(rl.dropped).toBe(1); + }); + + it('resets each window so a well-behaved sender is never permanently blocked', () => { + const rl = new PacketRateLimiter(OPTS, 0); + for (let i = 0; i < 5; i++) rl.shouldDrop('10.0.0.1', i); + expect(rl.shouldDrop('10.0.0.1', 50)).toBe(true); + + // New window + expect(rl.shouldDrop('10.0.0.1', 1000)).toBe(false); + expect(rl.trackedSources).toBe(1); // counters cleared, map stays bounded + }); + + it('limits each source independently', () => { + const rl = new PacketRateLimiter(OPTS, 0); + for (let i = 0; i < 4; i++) rl.shouldDrop('10.0.0.1', i); + // A different source still gets its own budget + expect(rl.shouldDrop('10.0.0.2', 5)).toBe(false); + }); + + it('caps distinct sources per window so spoofed IPs cannot grow the map', () => { + const rl = new PacketRateLimiter(OPTS, 0); + expect(rl.shouldDrop('1.1.1.1', 0)).toBe(false); + expect(rl.shouldDrop('2.2.2.2', 1)).toBe(false); + // Third distinct source exceeds maxSources + expect(rl.shouldDrop('3.3.3.3', 2)).toBe(true); + expect(rl.trackedSources).toBe(2); + + // Thousands of spoofed sources must not grow it past the cap + for (let i = 0; i < 5000; i++) rl.shouldDrop(`10.1.${i % 255}.${i % 251}`, 3); + expect(rl.trackedSources).toBe(2); + }); +}); diff --git a/backend/src/services/netflow/decoder.ts b/backend/src/services/netflow/decoder.ts index 9380162..9718603 100644 --- a/backend/src/services/netflow/decoder.ts +++ b/backend/src/services/netflow/decoder.ts @@ -24,12 +24,44 @@ interface Template { fields: TemplateField[]; // Minimum bytes a record can occupy (var-length fields count as 1) minLength: number; + // Epoch ms of the last time this template was stored or used to decode data. + // Drives TTL/LRU eviction so a hostile or churning sender can't grow the cache + // without bound (see pruneTemplateCache). + lastUsed: number; } // Keyed by `${exporterKey}|${sourceId}|${templateId}` so two exporters (or two // observation domains on one exporter) can't clobber each other's templates. export type TemplateCache = Map; +/** + * Evict templates unused for longer than `ttlMs`, then enforce `max` entries by + * dropping the least-recently-used. Returns how many were removed. + */ +export function pruneTemplateCache( + cache: TemplateCache, + ttlMs: number, + max: number, + now: number = Date.now() +): number { + let removed = 0; + const cutoff = now - ttlMs; + for (const [key, tpl] of cache) { + if (tpl.lastUsed < cutoff) { + cache.delete(key); + removed++; + } + } + if (cache.size > max) { + const byAge = [...cache.entries()].sort((a, b) => a[1].lastUsed - b[1].lastUsed); + for (let i = 0; i < byAge.length - max; i++) { + cache.delete(byAge[i][0]); + removed++; + } + } + return removed; +} + export interface DecodeResult { flows: FlowRecord[]; templatesParsed: number; @@ -65,11 +97,35 @@ function formatIPv4(buf: Buffer, offset: number): string { } function formatIPv6(buf: Buffer, offset: number): string { - const parts: string[] = []; - for (let i = 0; i < 8; i++) { - parts.push(buf.readUInt16BE(offset + i * 2).toString(16)); + const groups: number[] = []; + for (let i = 0; i < 8; i++) groups.push(buf.readUInt16BE(offset + i * 2)); + + // IPv4-mapped (::ffff:a.b.c.d) — report the underlying IPv4 address so these + // flows match client/exporter IPs and don't look like separate IPv6 endpoints. + if (groups[0] === 0 && groups[1] === 0 && groups[2] === 0 && groups[3] === 0 && + groups[4] === 0 && groups[5] === 0xffff) { + return formatIPv4(buf, offset + 12); } - return parts.join(':'); + + // RFC 5952: compress the longest run of 2+ zero groups to '::' + let bestStart = -1; + let bestLen = 0; + let runStart = -1; + let runLen = 0; + for (let i = 0; i < 8; i++) { + if (groups[i] === 0) { + if (runStart === -1) runStart = i; + runLen++; + if (runLen > bestLen) { bestStart = runStart; bestLen = runLen; } + } else { + runStart = -1; + runLen = 0; + } + } + + const parts = groups.map((g) => g.toString(16)); + if (bestLen < 2) return parts.join(':'); + return `${parts.slice(0, bestStart).join(':')}::${parts.slice(bestStart + bestLen).join(':')}`; } function parseTemplates( @@ -113,7 +169,7 @@ function parseTemplates( minLength += length === 0xffff ? 1 : length; } if (!ok || minLength === 0) break; - cache.set(`${exporterKey}|${sourceId}|${templateId}`, { fields, minLength }); + cache.set(`${exporterKey}|${sourceId}|${templateId}`, { fields, minLength, lastUsed: Date.now() }); parsed++; } return parsed; @@ -223,6 +279,7 @@ export function decodePacket(buf: Buffer, exporterKey: string, cache: TemplateCa } else if (!isOptionsSet && setId >= 256) { const template = cache.get(`${exporterKey}|${sourceId}|${setId}`); if (template) { + template.lastUsed = Date.now(); // keep actively-used templates alive parseDataSet(buf, setStart, setEnd, template, result.flows); } else { result.recordsWithoutTemplate++; diff --git a/backend/src/services/netflow/rateLimiter.ts b/backend/src/services/netflow/rateLimiter.ts new file mode 100644 index 0000000..dbe173b --- /dev/null +++ b/backend/src/services/netflow/rateLimiter.ts @@ -0,0 +1,65 @@ +// Per-source packet rate limiting for the NetFlow UDP collector. +// +// The collector socket is unauthenticated by nature (NetFlow/IPFIX has no auth), +// so anything able to reach the port can push packets as fast as it likes. Each +// packet costs template lookups and record parsing on the single Node event +// loop, so an unthrottled sender can starve the rest of the backend. This caps +// how much work any one source can force per window and drops the excess. +// +// Fixed-window rather than token-bucket: it's cheaper on the hot path, and the +// counters map is cleared every window so it can't grow without bound. A cap on +// distinct sources per window bounds it further against spoofed source IPs. + +export interface RateLimiterOptions { + /** Length of the counting window in ms. */ + windowMs: number; + /** Packets allowed per source within a window. */ + maxPerSource: number; + /** Distinct sources tracked per window; beyond this, new sources are dropped. */ + maxSources: number; +} + +export class PacketRateLimiter { + private counts = new Map(); + private windowStart: number; + private droppedCount = 0; + + constructor(private readonly opts: RateLimiterOptions, now: number = Date.now()) { + this.windowStart = now; + } + + /** + * Record a packet from `source`. Returns true if it should be DROPPED. + */ + shouldDrop(source: string, now: number = Date.now()): boolean { + if (now - this.windowStart >= this.opts.windowMs) { + this.counts.clear(); // bounds the map every window + this.windowStart = now; + } + + const current = this.counts.get(source); + if (current === undefined && this.counts.size >= this.opts.maxSources) { + // Too many distinct sources this window (likely spoofed) — drop rather + // than track, so the map can't be grown arbitrarily. + this.droppedCount++; + return true; + } + + const next = (current ?? 0) + 1; + this.counts.set(source, next); + if (next > this.opts.maxPerSource) { + this.droppedCount++; + return true; + } + return false; + } + + get dropped(): number { + return this.droppedCount; + } + + /** Number of distinct sources seen in the current window (for tests/stats). */ + get trackedSources(): number { + return this.counts.size; + } +} diff --git a/frontend/package.json b/frontend/package.json index 5028086..f49f92d 100644 --- a/frontend/package.json +++ b/frontend/package.json @@ -1,6 +1,6 @@ { "name": "mikrotik-manager-frontend", - "version": "0.17.6-beta", + "version": "0.17.7-beta", "private": true, "type": "module", "scripts": { diff --git a/frontend/src/version.ts b/frontend/src/version.ts index 29c2567..87e032e 100644 --- a/frontend/src/version.ts +++ b/frontend/src/version.ts @@ -1 +1 @@ -export const APP_VERSION = 'v0.17.6 Beta'; +export const APP_VERSION = 'v0.17.7 Beta';