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';