fix(netflow): bound template cache, throttle UDP ingress, canonical IPv6 (v0.17.7-beta)

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 <noreply@anthropic.com>

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": {

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<string, number>();
+  private nextPseudoId = -1;
+
+  private rateLimiter = new PacketRateLimiter(RATE_LIMIT);
+  private packetsDropped = 0;
 
   private flushTimer: ReturnType<typeof setInterval> | null = null;
   private mapTimer: ReturnType<typeof setInterval> | 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);

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(

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);
+  });
+});

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<string, Template>;
 
+/**
+ * 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++;

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<string, number>();
+  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;
+  }
+}