From b841eb6e7b72df21c71ec4b377f7048c6be96f7a Mon Sep 17 00:00:00 2001 From: 2GT-Rich <107219997+2GT-Rich@users.noreply.github.com> Date: Wed, 10 Jun 2026 10:58:41 -0700 Subject: [PATCH] =?UTF-8?q?feat:=20NetFlow/IPFIX=20traffic=20analytics=20?= =?UTF-8?q?=E2=80=94=20per-client=20usage,=20app=20breakdown,=20built-in?= =?UTF-8?q?=20collector=20(v0.13.0-beta)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds Meraki-style traffic analytics powered by MikroTik Traffic Flow: Backend - Built-in NetFlow v9 + IPFIX collector (UDP 2055) with hand-rolled decoder, template caching, enterprise-field and IPv6 support - Flow attribution: client IP → MAC via the clients table, upload/download direction, NAT-duplicate records dropped - App classification by protocol/port (~25 categories: HTTPS, QUIC, DNS, SSH, Email, WireGuard, ...) - Deduplication: per-client best-exporter-per-window so flows crossing two managed routers are never double-counted - NAT-tolerant ingest: flows from sources matching no managed device (NAT / Tailscale subnet routers in path) are accepted as per-source unidentified exporters instead of silently dropped (netflow_accept_unknown setting) - Storage: InfluxDB client_traffic time-series (bounded cardinality via top-N fold) + Postgres client_traffic_daily rollups; daily retention purge - One-toggle device config push via RouterOS API (/ip traffic-flow + target) - New /api/traffic/* analytics endpoints + /api/network-services/netflow - docker-compose: NETFLOW_PORT (default 2055/udp) published on the backend Frontend - New Traffic page: fleet up/down chart, top talkers with client click-through, app-category donut, 1h–30d ranges - New NetFlow config page: collector settings, live ingest stats, per-device export toggles with health badges, NAT-source banner - Client detail: App Traffic card (auto-hides without data) - Clients page: sortable "Data (today)" column - Settings: NetFlow retention controls Includes 17 unit tests (decoder fixtures, dedup, attribution, classifier); verified end-to-end against live RouterOS devices. Co-Authored-By: Claude Fable 5 --- README.md | 13 +- backend/package-lock.json | 14 +- backend/package.json | 3 +- backend/src/db/migrate.ts | 23 ++ backend/src/index.ts | 7 + backend/src/routes/clients.ts | 7 +- backend/src/routes/networkServices.ts | 163 ++++++++ backend/src/routes/settings.ts | 6 + backend/src/routes/trafficAnalytics.ts | 233 +++++++++++ backend/src/services/PollerService.ts | 53 +++ .../src/services/mikrotik/DeviceCollector.ts | 27 ++ .../src/services/netflow/FlowAggregator.ts | 139 +++++++ .../src/services/netflow/NetflowCollector.ts | 373 ++++++++++++++++++ .../netflow/__tests__/aggregator.test.ts | 107 +++++ .../netflow/__tests__/decoder.test.ts | 271 +++++++++++++ backend/src/services/netflow/appCategories.ts | 79 ++++ backend/src/services/netflow/decoder.ts | 227 +++++++++++ docker-compose.yml | 5 + frontend/package.json | 2 +- frontend/src/App.tsx | 4 + frontend/src/components/layout/Sidebar.tsx | 4 +- frontend/src/pages/ClientDetailPage.tsx | 138 ++++++- frontend/src/pages/ClientsPage.tsx | 12 + .../src/pages/NetworkServicesNetflowPage.tsx | 324 +++++++++++++++ frontend/src/pages/SettingsPage.tsx | 2 + frontend/src/pages/TrafficAnalyticsPage.tsx | 261 ++++++++++++ frontend/src/services/api.ts | 67 ++++ frontend/src/types/index.ts | 2 + frontend/src/version.ts | 2 +- 29 files changed, 2553 insertions(+), 15 deletions(-) create mode 100644 backend/src/routes/trafficAnalytics.ts create mode 100644 backend/src/services/netflow/FlowAggregator.ts create mode 100644 backend/src/services/netflow/NetflowCollector.ts create mode 100644 backend/src/services/netflow/__tests__/aggregator.test.ts create mode 100644 backend/src/services/netflow/__tests__/decoder.test.ts create mode 100644 backend/src/services/netflow/appCategories.ts create mode 100644 backend/src/services/netflow/decoder.ts create mode 100644 frontend/src/pages/NetworkServicesNetflowPage.tsx create mode 100644 frontend/src/pages/TrafficAnalyticsPage.tsx diff --git a/README.md b/README.md index a66f764..fe7f68a 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.12.3_Beta-blue) +![Version](https://img.shields.io/badge/version-0.13.0_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) @@ -144,6 +144,17 @@ Each service supports multi-device management with conflict detection: | **NTP** | Server (broadcast/manycast), client (unicast/multicast), sync status | | **WireGuard** | Interface management, peer configuration, public key display, RX/TX stats | | **Syslog** | Logging actions (remote/memory/disk/echo) and routing rules; single-device or push-to-all with per-entry coverage; enable/disable rules | +| **NetFlow** | One-toggle Traffic Flow (NetFlow v9/IPFIX) export per device, pointed at the built-in collector; live export status per device | + +### Traffic Analytics (NetFlow) +- **Built-in NetFlow/IPFIX collector** — receives Traffic Flow exports from your routers on UDP 2055 (no external tools needed); decoder supports both NetFlow v9 and IPFIX +- **Per-client usage accounting** — flows are attributed to known clients (IP → MAC) so every client shows real upload/download totals; "Data (today)" column on the Clients page and an App Traffic card on each client's detail page +- **Application breakdown** — flows are classified by protocol and port into readable categories (HTTPS, QUIC, DNS, SSH, Email, WireGuard, …) with fleet-wide and per-client views +- **Top talkers** — ranked client bandwidth usage over 1h/24h/7d/30d ranges on the dedicated Traffic page +- **Automatic deduplication** — a flow crossing two managed routers is exported twice; the collector keeps only the best exporter per client each window so totals are never double-counted +- **NAT-tolerant ingest** — when routers export from behind NAT or a VPN/Tailscale subnet router, packets arrive from an address that matches no managed device; the collector accepts these as per-source "unidentified" exporters (configurable) so flows are still attributed to clients instead of silently dropped, and the NetFlow page shows a banner naming the NAT'd source. Per-device export status/flow counters require un-NAT'd sources — exempt `udp/2055` from masquerade on the gateway to restore device identity (and exact cross-device dedup) +- **Fully UI-configurable** — enable the collector, set its address/port/version in the NetFlow page, then toggle export per device; the platform pushes `/ip traffic-flow` configuration to each router via the RouterOS API +- Configurable retention for detailed time-series (default 30 days) and daily rollups (default 365 days) ### Network Topology - **LLDP-authoritative** — LLDP links are treated as ground truth; CDP/MNDP links to the same neighbor are automatically suppressed, eliminating spurious "Shared Segment" nodes diff --git a/backend/package-lock.json b/backend/package-lock.json index 3ff1e89..d4d8876 100644 --- a/backend/package-lock.json +++ b/backend/package-lock.json @@ -1,14 +1,15 @@ { "name": "mikrotik-manager-backend", - "version": "0.11.7-beta", + "version": "0.12.3-beta", "lockfileVersion": 3, "requires": true, "packages": { "": { "name": "mikrotik-manager-backend", - "version": "0.11.7-beta", + "version": "0.12.3-beta", "dependencies": { "@influxdata/influxdb-client": "^1.33.2", + "@influxdata/influxdb-client-apis": "^1.35.0", "bcryptjs": "^2.4.3", "bullmq": "^5.1.1", "compression": "^1.7.4", @@ -992,6 +993,15 @@ "integrity": "sha512-woWMi8PDpPQpvTsRaUw4Ig+nOGS/CWwAwS66Fa1Vr/EkW+NEwxI8YfPBsdBMn33jK2Y86/qMiiuX/ROHIkJLTw==", "license": "MIT" }, + "node_modules/@influxdata/influxdb-client-apis": { + "version": "1.35.0", + "resolved": "https://registry.npmjs.org/@influxdata/influxdb-client-apis/-/influxdb-client-apis-1.35.0.tgz", + "integrity": "sha512-+7h6smVPHYBge2rNKgYh/5k+SriYvPMsoJDfbUiQt1vJtpWnElwgBDLDl7Cr6d9XPC+FCI9GP4GQEMv7y8WxdA==", + "license": "MIT", + "peerDependencies": { + "@influxdata/influxdb-client": "*" + } + }, "node_modules/@ioredis/commands": { "version": "1.5.1", "resolved": "https://registry.npmjs.org/@ioredis/commands/-/commands-1.5.1.tgz", diff --git a/backend/package.json b/backend/package.json index fb56ae1..d0bc0f0 100644 --- a/backend/package.json +++ b/backend/package.json @@ -1,6 +1,6 @@ { "name": "mikrotik-manager-backend", - "version": "0.12.3-beta", + "version": "0.13.0-beta", "description": "Mikrotik Network Management Platform - Backend API", "main": "dist/index.js", "scripts": { @@ -14,6 +14,7 @@ }, "dependencies": { "@influxdata/influxdb-client": "^1.33.2", + "@influxdata/influxdb-client-apis": "^1.35.0", "bcryptjs": "^2.4.3", "bullmq": "^5.1.1", "compression": "^1.7.4", diff --git a/backend/src/db/migrate.ts b/backend/src/db/migrate.ts index 9892f9b..31a22c8 100644 --- a/backend/src/db/migrate.ts +++ b/backend/src/db/migrate.ts @@ -388,6 +388,19 @@ CREATE TABLE IF NOT EXISTS config_templates ( ALTER TABLE users ADD COLUMN IF NOT EXISTS totp_secret VARCHAR(64); ALTER TABLE users ADD COLUMN IF NOT EXISTS totp_enabled BOOLEAN NOT NULL DEFAULT false; +-- Per-client daily traffic rollups from the NetFlow collector. mac_address +-- also holds the pseudo-clients 'unknown' (unmapped local IPs) and 'other' +-- (clients folded by the top-N cardinality cap). +CREATE TABLE IF NOT EXISTS client_traffic_daily ( + mac_address VARCHAR(17) NOT NULL, + day DATE NOT NULL, + upload_bytes BIGINT NOT NULL DEFAULT 0, + download_bytes BIGINT NOT NULL DEFAULT 0, + app_breakdown JSONB NOT NULL DEFAULT '{}', + PRIMARY KEY (mac_address, day) +); +CREATE INDEX IF NOT EXISTS idx_client_traffic_daily_day ON client_traffic_daily(day DESC); + -- Wireless security profiles (WPA/WPA2/WPA3 config) CREATE TABLE IF NOT EXISTS wireless_security_profiles ( id SERIAL PRIMARY KEY, @@ -424,6 +437,16 @@ const DEFAULT_SETTINGS = [ { key: 'config_snapshot_enabled', value: true }, { key: 'config_snapshot_interval_min', value: 60 }, { key: 'config_snapshot_retention', value: 30 }, + { key: 'netflow_enabled', value: false }, + { key: 'netflow_collector_address', value: '' }, + { key: 'netflow_collector_port', value: 2055 }, + { key: 'netflow_version', value: '9' }, + { key: 'netflow_active_timeout', value: '1m' }, + { key: 'netflow_inactive_timeout', value: '15s' }, + { key: 'netflow_topn_clients', value: 50 }, + { key: 'netflow_accept_unknown', value: true }, + { key: 'netflow_retention_days', value: 30 }, + { key: 'netflow_daily_retention_days', value: 365 }, ]; export async function runMigrations(): Promise { diff --git a/backend/src/index.ts b/backend/src/index.ts index 2027760..3871580 100644 --- a/backend/src/index.ts +++ b/backend/src/index.ts @@ -23,6 +23,7 @@ import { startBulkAddWorker, stopBulkAddWorker, } from './services/DeviceBulkAddWorker'; +import { netflowCollector } from './services/netflow/NetflowCollector'; import { verifyToken } from './middleware/auth'; import { decrypt } from './utils/crypto'; import { corsMiddlewareOptions, socketIoCorsOptions } from './utils/corsOrigins'; @@ -47,6 +48,7 @@ import configTemplatesRoutes from './routes/configTemplates'; import configHistoryRoutes from './routes/configHistory'; import wirelessRoutes from './routes/wireless'; import networkServicesRoutes from './routes/networkServices'; +import trafficAnalyticsRoutes from './routes/trafficAnalytics'; import credentialPresetsRoutes from './routes/credentialPresets'; import systemRoutes from './routes/system'; import { auditMiddleware } from './middleware/auditMiddleware'; @@ -214,6 +216,7 @@ app.use('/api/routers', routersRoutes); app.use('/api/alerts', alertsRoutes); app.use('/api/wireless', wirelessRoutes); app.use('/api/network-services', networkServicesRoutes); +app.use('/api/traffic', trafficAnalyticsRoutes); app.use('/api/credential-presets', credentialPresetsRoutes); app.use('/api/audit-log', auditLogRoutes); app.use('/api/tags', tagsRoutes); @@ -262,6 +265,9 @@ async function start(): Promise { await pollerService.start(); await startBulkAddWorker(); + // NetFlow/IPFIX collector (binds its UDP socket only when netflow_enabled) + await netflowCollector.start(); + // Start HTTP server httpServer.listen(PORT, '0.0.0.0', () => { console.log(`✓ Mikrotik Manager backend running on port ${PORT}`); @@ -271,6 +277,7 @@ async function start(): Promise { const shutdown = async () => { console.log('Shutting down...'); await stopBulkAddWorker(); + await netflowCollector.stop(); await pollerService.stop(); await redis.quit().catch(() => {}); await pool.end(); diff --git a/backend/src/routes/clients.ts b/backend/src/routes/clients.ts index ce4b095..4911135 100644 --- a/backend/src/routes/clients.ts +++ b/backend/src/routes/clients.ts @@ -42,7 +42,10 @@ router.get('/', async (req: Request, res: Response) => { // Deduplicate by MAC address across devices: prefer active rows, then most recently seen. // The inner DISTINCT ON picks the best row per MAC; the outer query applies final sort + pagination. const sql = ` - SELECT * FROM ( + SELECT deduped.*, + -- float8 so node-pg returns a number (bigint would arrive as a string) + (COALESCE(ctd.upload_bytes, 0) + COALESCE(ctd.download_bytes, 0))::float8 AS traffic_today_bytes + FROM ( SELECT DISTINCT ON (c.mac_address) c.*, d.name as device_name, wi.ssid as ssid @@ -53,6 +56,8 @@ router.get('/', async (req: Request, res: Response) => { ${where} ORDER BY c.mac_address, (c.client_type = 'wireless') DESC, c.active DESC, c.last_seen DESC NULLS LAST ) deduped + LEFT JOIN client_traffic_daily ctd + ON ctd.mac_address = LOWER(deduped.mac_address) AND ctd.day = CURRENT_DATE ORDER BY last_seen DESC NULLS LAST LIMIT $${idx++} OFFSET $${idx++} `; diff --git a/backend/src/routes/networkServices.ts b/backend/src/routes/networkServices.ts index ec93b84..82eb3b3 100644 --- a/backend/src/routes/networkServices.ts +++ b/backend/src/routes/networkServices.ts @@ -2,6 +2,7 @@ import { Router, Request, Response } from 'express'; import { query } from '../config/database'; import { requireAuth, requireWrite } from '../middleware/auth'; import { DeviceCollector, DeviceRow } from '../services/mikrotik/DeviceCollector'; +import { netflowCollector } from '../services/netflow/NetflowCollector'; const router = Router(); router.use(requireAuth); @@ -488,4 +489,166 @@ router.delete('/syslog/rule/:id', requireWrite, async (req: Request, res: Respon }); }); +// ─── NetFlow / Traffic Flow ─────────────────────────────────────────────────── + +async function getNetflowAppSettings(): Promise<{ + address: string; port: number; version: string; activeTimeout: string; inactiveTimeout: string; +}> { + const rows = await query<{ key: string; value: unknown }>( + `SELECT key, value FROM app_settings + WHERE key IN ('netflow_collector_address', 'netflow_collector_port', 'netflow_version', + 'netflow_active_timeout', 'netflow_inactive_timeout')` + ); + const map: Record = {}; + for (const row of rows) map[row.key] = row.value; + return { + address: String(map['netflow_collector_address'] || ''), + port: Number(map['netflow_collector_port']) || 2055, + version: String(map['netflow_version'] || '9'), + activeTimeout: String(map['netflow_active_timeout'] || '1m'), + inactiveTimeout: String(map['netflow_inactive_timeout'] || '15s'), + }; +} + +function targetMatchesCollector( + targets: Record[], + collector: { address: string; port: number } +): boolean { + return targets.some( + (t) => t['dst-address'] === collector.address && Number(t['port']) === collector.port && t['disabled'] !== 'true' + ); +} + +// GET /api/network-services/netflow/fleet — traffic-flow state for all online devices +router.get('/netflow/fleet', async (_req: Request, res: Response) => { + const settings = await getNetflowAppSettings(); + const stats = netflowCollector.getStats(); + const statsByDevice = new Map(stats.exporters.map((e) => [e.deviceId, e])); + const devices = await query(`SELECT * FROM devices WHERE status = 'online' ORDER BY name`); + + const results = await Promise.allSettled( + devices.map(async (device: DeviceRow) => { + const collector = new DeviceCollector(device); + try { + await collector.connect(); + const [tfSettings, targets] = await Promise.all([ + collector.getTrafficFlowSettings(), + collector.getTrafficFlowTargets(), + ]); + const exporterStats = statsByDevice.get(device.id); + return { + id: device.id, + name: device.name, + ip_address: device.ip_address, + enabled: tfSettings?.['enabled'] === 'true' || tfSettings?.['enabled'] === 'yes', + interfaces: tfSettings?.['interfaces'] || '', + targets: targets.map((t) => ({ + id: t['.id'], + dst_address: t['dst-address'], + port: Number(t['port']) || 0, + version: t['version'] || '', + disabled: t['disabled'] === 'true', + })), + target_matches_collector: targetMatchesCollector(targets, settings), + flows_received: exporterStats?.flows || 0, + last_flow_at: exporterStats?.lastSeen || null, + }; + } catch (err) { + return { + id: device.id, name: device.name, ip_address: device.ip_address, + enabled: null, interfaces: '', targets: [], target_matches_collector: false, + flows_received: 0, last_flow_at: null, error: (err as Error).message, + }; + } finally { + collector.disconnect(); + } + }) + ); + + res.json({ + collector: { ...settings, listening: stats.listening }, + devices: results.map((r) => (r.status === 'fulfilled' ? r.value : { error: (r.reason as Error).message })), + }); +}); + +// GET /api/network-services/netflow?deviceId=X — single device traffic-flow state +router.get('/netflow', async (req: Request, res: Response) => { + const deviceId = deviceIdParam(req, res); if (!deviceId) return; + const settings = await getNetflowAppSettings(); + await withDevice(deviceId, res, async (collector) => { + const [tfSettings, targets] = await Promise.all([ + collector.getTrafficFlowSettings(), + collector.getTrafficFlowTargets(), + ]); + res.json({ + settings: tfSettings, + targets, + target_matches_collector: targetMatchesCollector(targets, settings), + }); + }); +}); + +// PUT /api/network-services/netflow?deviceId=X — body { enabled: boolean } +// Enable: upsert our export target (matched by dst-address + port) and turn +// traffic-flow on. Disable: remove our target; only turn traffic-flow off if +// no other (user-managed) targets remain. +router.put('/netflow', requireWrite, async (req: Request, res: Response) => { + const deviceId = deviceIdParam(req, res); if (!deviceId) return; + const { enabled } = req.body; + if (typeof enabled !== 'boolean') return res.status(400).json({ error: 'enabled (boolean) is required' }); + + const settings = await getNetflowAppSettings(); + if (enabled && !settings.address) { + return res.status(400).json({ error: 'Set the collector address in NetFlow settings before enabling devices' }); + } + + await withDevice(deviceId, res, async (collector) => { + const targets = await collector.getTrafficFlowTargets(); + const ours = targets.find( + (t) => t['dst-address'] === settings.address && Number(t['port']) === settings.port + ); + + if (enabled) { + const targetParams: Record = { + 'dst-address': settings.address, + port: String(settings.port), + version: settings.version === 'ipfix' ? 'ipfix' : '9', + }; + if (settings.version !== 'ipfix') { + // Re-send templates frequently so the collector can decode data + // promptly after a backend restart clears its template cache. + targetParams['v9-template-refresh'] = '20'; + targetParams['v9-template-timeout'] = '1m'; + } + if (ours) { + await collector.updateTrafficFlowTarget(ours['.id'], targetParams); + } else { + await collector.addTrafficFlowTarget(targetParams); + } + await collector.setTrafficFlowSettings({ + enabled: 'yes', + interfaces: 'all', + 'active-flow-timeout': settings.activeTimeout, + 'inactive-flow-timeout': settings.inactiveTimeout, + }); + } else { + if (ours) await collector.removeTrafficFlowTarget(ours['.id']); + const remaining = (await collector.getTrafficFlowTargets()).filter((t) => t['.id'] !== ours?.['.id']); + if (remaining.length === 0) { + await collector.setTrafficFlowSettings({ enabled: 'no' }); + } + } + + const [tfSettings, updatedTargets] = await Promise.all([ + collector.getTrafficFlowSettings(), + collector.getTrafficFlowTargets(), + ]); + res.json({ + settings: tfSettings, + targets: updatedTargets, + target_matches_collector: targetMatchesCollector(updatedTargets, settings), + }); + }); +}); + export default router; diff --git a/backend/src/routes/settings.ts b/backend/src/routes/settings.ts index 7722f2a..f669b02 100644 --- a/backend/src/routes/settings.ts +++ b/backend/src/routes/settings.ts @@ -2,6 +2,7 @@ import { Router, Request, Response } from 'express'; import bcrypt from 'bcryptjs'; import { query, queryOne } from '../config/database'; import { requireAuth, requireAdmin } from '../middleware/auth'; +import { netflowCollector } from '../services/netflow/NetflowCollector'; const router = Router(); router.use(requireAuth); @@ -26,6 +27,11 @@ router.put('/', requireAdmin, async (req: Request, res: Response) => { [key, JSON.stringify(value)] ); } + // Apply NetFlow listener changes immediately instead of waiting for the + // collector's periodic settings reconcile. + if (Object.keys(updates).some((k) => k.startsWith('netflow_'))) { + netflowCollector.reconcile().catch(() => {}); + } res.json({ message: 'Settings updated' }); }); diff --git a/backend/src/routes/trafficAnalytics.ts b/backend/src/routes/trafficAnalytics.ts new file mode 100644 index 0000000..82c6cdf --- /dev/null +++ b/backend/src/routes/trafficAnalytics.ts @@ -0,0 +1,233 @@ +import { Router, Request, Response } from 'express'; +import { getQueryApi, bucket } from '../config/influxdb'; +import { query } from '../config/database'; +import { requireAuth } from '../middleware/auth'; +import { netflowCollector } from '../services/netflow/NetflowCollector'; + +const router = Router(); +router.use(requireAuth); + +function rangeToFlux(range: string): string { + const allowed = ['1h', '3h', '6h', '12h', '24h', '7d', '30d']; + return allowed.includes(range) ? range : '24h'; +} + +// Aggregation window sized to the range so charts stay readable (points are +// written every 60s by the collector flush). +function windowForRange(range: string): string { + const map: Record = { + '1h': '1m', '3h': '5m', '6h': '5m', '12h': '10m', + '24h': '15m', '7d': '1h', '30d': '6h', + }; + return map[range] || '15m'; +} + +// MACs are used inside Flux queries — restrict to safe charsets (real MACs +// plus the 'unknown'/'other' pseudo-clients). +function sanitizeMac(raw: string): string | null { + const mac = raw.toLowerCase(); + if (mac === 'unknown' || mac === 'other') return mac; + return /^[0-9a-f]{2}(:[0-9a-f]{2}){5}$/.test(mac) ? mac : null; +} + +// GET /api/traffic/status — collector state for the config page +router.get('/status', async (_req: Request, res: Response) => { + res.json(netflowCollector.getStats()); +}); + +// GET /api/traffic/timeseries?range=24h — fleet-wide upload/download over time +router.get('/timeseries', async (req: Request, res: Response) => { + const range = rangeToFlux(String(req.query.range || '24h')); + const every = windowForRange(range); + const queryApi = getQueryApi(); + + const fluxQuery = ` + from(bucket: "${bucket}") + |> range(start: -${range}) + |> filter(fn: (r) => r._measurement == "client_traffic") + |> filter(fn: (r) => r._field == "bytes") + |> group(columns: ["direction"]) + |> aggregateWindow(every: ${every}, fn: sum, createEmpty: false) + `; + + const pivoted: Record = {}; + try { + await queryApi.collectRows(fluxQuery, (row, tableMeta) => { + const time = tableMeta.get(row, '_time') as string; + const direction = tableMeta.get(row, 'direction') as string; + const value = Number(tableMeta.get(row, '_value')) || 0; + if (!pivoted[time]) pivoted[time] = { time, upload: 0, download: 0 }; + if (direction === 'upload') pivoted[time].upload += value; + else if (direction === 'download') pivoted[time].download += value; + }); + } catch { + // No data yet + } + + res.json(Object.values(pivoted).sort((a, b) => a.time.localeCompare(b.time))); +}); + +// GET /api/traffic/top-clients?range=24h&limit=10 — top talkers with client names +router.get('/top-clients', async (req: Request, res: Response) => { + const range = rangeToFlux(String(req.query.range || '24h')); + const limit = Math.min(parseInt(String(req.query.limit || '10'), 10) || 10, 50); + const queryApi = getQueryApi(); + + const fluxQuery = ` + from(bucket: "${bucket}") + |> range(start: -${range}) + |> filter(fn: (r) => r._measurement == "client_traffic") + |> filter(fn: (r) => r._field == "bytes") + |> group(columns: ["mac", "direction"]) + |> sum() + `; + + const perMac = new Map(); + try { + await queryApi.collectRows(fluxQuery, (row, tableMeta) => { + const mac = tableMeta.get(row, 'mac') as string; + const direction = tableMeta.get(row, 'direction') as string; + const value = Number(tableMeta.get(row, '_value')) || 0; + let entry = perMac.get(mac); + if (!entry) { + entry = { upload: 0, download: 0 }; + perMac.set(mac, entry); + } + if (direction === 'upload') entry.upload += value; + else entry.download += value; + }); + } catch { + // No data yet + } + + const ranked = Array.from(perMac.entries()) + .map(([mac, e]) => ({ mac, upload_bytes: e.upload, download_bytes: e.download, total_bytes: e.upload + e.download })) + .sort((a, b) => b.total_bytes - a.total_bytes) + .slice(0, limit); + + // Enrich real MACs with client identity from Postgres + const realMacs = ranked.map((r) => r.mac).filter((m) => m !== 'unknown' && m !== 'other'); + const names = new Map(); + if (realMacs.length > 0) { + const rows = await query<{ mac_address: string; hostname: string | null; custom_name: string | null; vendor: string | null; ip_address: string | null }>( + `SELECT DISTINCT ON (mac_address) mac_address, hostname, custom_name, vendor, ip_address + FROM clients WHERE LOWER(mac_address) = ANY($1) + ORDER BY mac_address, last_seen DESC NULLS LAST`, + [realMacs] + ); + for (const row of rows) names.set(row.mac_address.toLowerCase(), row); + } + + res.json( + ranked.map((r) => { + const info = names.get(r.mac); + return { + ...r, + hostname: info?.hostname || null, + custom_name: info?.custom_name || null, + vendor: info?.vendor || null, + ip_address: info?.ip_address || null, + }; + }) + ); +}); + +// GET /api/traffic/apps?range=24h&mac=xx:xx:.. — app-category byte totals +// (fleet-wide, or for a single client when mac is given) +router.get('/apps', async (req: Request, res: Response) => { + const range = rangeToFlux(String(req.query.range || '24h')); + const queryApi = getQueryApi(); + + let macFilter = ''; + if (req.query.mac) { + const mac = sanitizeMac(String(req.query.mac)); + if (!mac) return res.status(400).json({ error: 'Invalid mac' }); + macFilter = `|> filter(fn: (r) => r.mac == "${mac}")`; + } + + const fluxQuery = ` + from(bucket: "${bucket}") + |> range(start: -${range}) + |> filter(fn: (r) => r._measurement == "client_traffic") + |> filter(fn: (r) => r._field == "bytes") + ${macFilter} + |> group(columns: ["app"]) + |> sum() + `; + + const apps: { app: string; bytes: number }[] = []; + try { + await queryApi.collectRows(fluxQuery, (row, tableMeta) => { + apps.push({ + app: tableMeta.get(row, 'app') as string, + bytes: Number(tableMeta.get(row, '_value')) || 0, + }); + }); + } catch { + // No data yet + } + + res.json(apps.sort((a, b) => b.bytes - a.bytes)); +}); + +// GET /api/traffic/client/:mac?range=24h — per-client time series + app breakdown +router.get('/client/:mac', async (req: Request, res: Response) => { + const mac = sanitizeMac(req.params.mac); + if (!mac) return res.status(400).json({ error: 'Invalid mac' }); + const range = rangeToFlux(String(req.query.range || '24h')); + const every = windowForRange(range); + const queryApi = getQueryApi(); + + const seriesFlux = ` + from(bucket: "${bucket}") + |> range(start: -${range}) + |> filter(fn: (r) => r._measurement == "client_traffic") + |> filter(fn: (r) => r._field == "bytes") + |> filter(fn: (r) => r.mac == "${mac}") + |> group(columns: ["direction"]) + |> aggregateWindow(every: ${every}, fn: sum, createEmpty: false) + `; + + const pivoted: Record = {}; + try { + await queryApi.collectRows(seriesFlux, (row, tableMeta) => { + const time = tableMeta.get(row, '_time') as string; + const direction = tableMeta.get(row, 'direction') as string; + const value = Number(tableMeta.get(row, '_value')) || 0; + if (!pivoted[time]) pivoted[time] = { time, upload: 0, download: 0 }; + if (direction === 'upload') pivoted[time].upload += value; + else if (direction === 'download') pivoted[time].download += value; + }); + } catch { + // No data yet + } + + const appsFlux = ` + from(bucket: "${bucket}") + |> range(start: -${range}) + |> filter(fn: (r) => r._measurement == "client_traffic") + |> filter(fn: (r) => r._field == "bytes") + |> filter(fn: (r) => r.mac == "${mac}") + |> group(columns: ["app"]) + |> sum() + `; + + const apps: { app: string; bytes: number }[] = []; + try { + await queryApi.collectRows(appsFlux, (row, tableMeta) => { + apps.push({ + app: tableMeta.get(row, 'app') as string, + bytes: Number(tableMeta.get(row, '_value')) || 0, + }); + }); + } catch { + // No data yet + } + + res.json({ + series: Object.values(pivoted).sort((a, b) => a.time.localeCompare(b.time)), + apps: apps.sort((a, b) => b.bytes - a.bytes), + }); +}); + +export default router; diff --git a/backend/src/services/PollerService.ts b/backend/src/services/PollerService.ts index 89fc2fb..8ea54b0 100644 --- a/backend/src/services/PollerService.ts +++ b/backend/src/services/PollerService.ts @@ -207,6 +207,17 @@ export class PollerService { ); } + // NetFlow data retention — runs once per day. Purges old client_traffic + // points from InfluxDB and old daily rollups from Postgres. + const netflowPruneKey = 'task:netflow_retention'; + const lastNetflowPrune = await this.getTimestamp(netflowPruneKey); + if (now - lastNetflowPrune > 86_400_000) { + await this.setTimestamp(netflowPruneKey, now, 86_400 + 3_600); + this.purgeNetflowData().catch((e) => + console.error('[Poller] NetFlow retention error:', e) + ); + } + // Stale client pruning — runs once per hour // Deletes inactive client records not seen for longer than retention_clients_days. const pruneKey = 'task:prune_clients'; @@ -353,6 +364,48 @@ export class PollerService { } } + private async purgeNetflowData(): Promise { + const rows = await query<{ key: string; value: unknown }>( + `SELECT key, value FROM app_settings + WHERE key IN ('netflow_retention_days', 'netflow_daily_retention_days')` + ); + const map: Record = {}; + for (const row of rows) map[row.key] = row.value; + const detailDays = Number(map['netflow_retention_days']) || 30; + const dailyDays = Number(map['netflow_daily_retention_days']) || 365; + + // InfluxDB: delete client_traffic points older than the detail retention + const { DeleteAPI } = await import('@influxdata/influxdb-client-apis'); + const { getInfluxClient, org, bucket } = await import('../config/influxdb'); + const stop = new Date(Date.now() - detailDays * 86_400_000).toISOString(); + await new DeleteAPI(getInfluxClient()) + .postDelete({ + org, + bucket, + body: { + start: '1970-01-01T00:00:00Z', + stop, + predicate: '_measurement="client_traffic"', + }, + }) + .catch((e) => console.error('[Poller] NetFlow Influx purge error:', (e as Error).message)); + + // Postgres: delete daily rollups older than the rollup retention + const result = await query<{ count: string }>( + `WITH deleted AS ( + DELETE FROM client_traffic_daily + WHERE day < CURRENT_DATE - ($1 || ' days')::interval + RETURNING 1 + ) + SELECT COUNT(*) AS count FROM deleted`, + [dailyDays] + ); + const count = parseInt(result[0]?.count || '0', 10); + if (count > 0) { + console.log(`[Poller] NetFlow retention pruned ${count} daily rollup row(s) (> ${dailyDays} days)`); + } + } + private async pruneStaleClients(settings: Record): Promise { // Delete inactive clients not seen for longer than the configured retention period. // Default: 7 days. Preserves any client that was active within the window so diff --git a/backend/src/services/mikrotik/DeviceCollector.ts b/backend/src/services/mikrotik/DeviceCollector.ts index 3f55bf8..f9516fa 100644 --- a/backend/src/services/mikrotik/DeviceCollector.ts +++ b/backend/src/services/mikrotik/DeviceCollector.ts @@ -2706,4 +2706,31 @@ export class DeviceCollector { async removeWireGuardPeer(id: string): Promise { await this.client.execute('/interface/wireguard/peers/remove', { '.id': id }); } + + // Traffic Flow (NetFlow/IPFIX export) ───────────────────────────────────────── + + async getTrafficFlowSettings(): Promise | null> { + const rows = await this.client.execute('/ip/traffic-flow/print', {}).catch(() => [] as Record[]); + return rows[0] || null; + } + + async setTrafficFlowSettings(params: Record): Promise { + await this.client.execute('/ip/traffic-flow/set', params); + } + + async getTrafficFlowTargets(): Promise[]> { + return this.client.execute('/ip/traffic-flow/target/print', { detail: '' }).catch(() => [] as Record[]); + } + + async addTrafficFlowTarget(params: Record): Promise { + await this.client.execute('/ip/traffic-flow/target/add', params); + } + + async updateTrafficFlowTarget(id: string, params: Record): Promise { + await this.client.execute('/ip/traffic-flow/target/set', { '.id': id, ...params }); + } + + async removeTrafficFlowTarget(id: string): Promise { + await this.client.execute('/ip/traffic-flow/target/remove', { '.id': id }); + } } diff --git a/backend/src/services/netflow/FlowAggregator.ts b/backend/src/services/netflow/FlowAggregator.ts new file mode 100644 index 0000000..1556d35 --- /dev/null +++ b/backend/src/services/netflow/FlowAggregator.ts @@ -0,0 +1,139 @@ +// In-memory flow aggregation between flushes (60s windows). +// +// Counters are kept PER EXPORTER per (client, direction, app) so that a flow +// crossing two managed routers — and therefore exported twice — can be +// deduplicated at drain time: for each client we keep only the exporter that +// saw the most bytes in the window (it converges on the gateway, which sees +// all of a client's traffic) and discard the rest. Tie-break: lowest exporter +// id, for determinism. + +export type Direction = 'upload' | 'download'; + +// Pseudo-client keys (not real MACs, but stored in the same mac slot) +export const CLIENT_UNKNOWN = 'unknown'; +export const CLIENT_OTHER = 'other'; + +export interface FlowSample { + exporterId: number; // managed device id that exported the flow + clientKey: string; // client MAC address, or CLIENT_UNKNOWN + direction: Direction; + app: string; + bytes: number; + packets: number; +} + +export interface AggregatedRow { + clientKey: string; + direction: Direction; + app: string; + bytes: number; + packets: number; +} + +interface Bucket { + exporterId: number; + clientKey: string; + direction: Direction; + app: string; + bytes: number; + packets: number; +} + +export class FlowAggregator { + private buckets = new Map(); + + add(sample: FlowSample): void { + if (sample.bytes <= 0 && sample.packets <= 0) return; + const key = `${sample.exporterId}|${sample.clientKey}|${sample.direction}|${sample.app}`; + const existing = this.buckets.get(key); + if (existing) { + existing.bytes += sample.bytes; + existing.packets += sample.packets; + } else { + this.buckets.set(key, { ...sample }); + } + } + + get size(): number { + return this.buckets.size; + } + + // Drain the window: dedup across exporters, then fold clients beyond the + // top-N (by total bytes) into CLIENT_OTHER to bound series cardinality. + // Pseudo-clients (unknown) are never folded — they're already aggregates. + drain(topN: number): AggregatedRow[] { + const all = Array.from(this.buckets.values()); + this.buckets.clear(); + if (all.length === 0) return []; + + // 1. Pick the winning exporter per client (max bytes, tie → lowest id) + const totals = new Map>(); // clientKey → exporterId → bytes + for (const b of all) { + let perExporter = totals.get(b.clientKey); + if (!perExporter) { + perExporter = new Map(); + totals.set(b.clientKey, perExporter); + } + perExporter.set(b.exporterId, (perExporter.get(b.exporterId) || 0) + b.bytes); + } + const winner = new Map(); // clientKey → exporterId + for (const [clientKey, perExporter] of totals) { + let bestId = -1; + let bestBytes = -1; + for (const [exporterId, bytes] of perExporter) { + if (bytes > bestBytes || (bytes === bestBytes && exporterId < bestId)) { + bestId = exporterId; + bestBytes = bytes; + } + } + winner.set(clientKey, bestId); + } + + // 2. Keep only the winning exporter's rows, merged by (client, dir, app) + const merged = new Map(); + const clientBytes = new Map(); + for (const b of all) { + if (winner.get(b.clientKey) !== b.exporterId) continue; + const key = `${b.clientKey}|${b.direction}|${b.app}`; + const row = merged.get(key); + if (row) { + row.bytes += b.bytes; + row.packets += b.packets; + } else { + merged.set(key, { + clientKey: b.clientKey, + direction: b.direction, + app: b.app, + bytes: b.bytes, + packets: b.packets, + }); + } + clientBytes.set(b.clientKey, (clientBytes.get(b.clientKey) || 0) + b.bytes); + } + + // 3. Top-N fold + const realClients = Array.from(clientBytes.keys()).filter((c) => c !== CLIENT_UNKNOWN); + const keep = new Set( + realClients + .sort((a, c) => (clientBytes.get(c) || 0) - (clientBytes.get(a) || 0)) + .slice(0, topN) + ); + + const result = new Map(); + for (const row of merged.values()) { + const clientKey = + row.clientKey === CLIENT_UNKNOWN || keep.has(row.clientKey) + ? row.clientKey + : CLIENT_OTHER; + const key = `${clientKey}|${row.direction}|${row.app}`; + const out = result.get(key); + if (out) { + out.bytes += row.bytes; + out.packets += row.packets; + } else { + result.set(key, { ...row, clientKey }); + } + } + return Array.from(result.values()); + } +} diff --git a/backend/src/services/netflow/NetflowCollector.ts b/backend/src/services/netflow/NetflowCollector.ts new file mode 100644 index 0000000..c20c3c3 --- /dev/null +++ b/backend/src/services/netflow/NetflowCollector.ts @@ -0,0 +1,373 @@ +// NetFlow/IPFIX collector — receives Traffic Flow exports from managed +// RouterOS devices on a UDP socket, attributes flows to known clients +// (IP → MAC via the clients table), classifies them by protocol/port, and +// aggregates into 60-second windows written to InfluxDB (time series) and +// Postgres (daily per-client rollups). + +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 { classifyApp, APP_LAN } from './appCategories'; +import { FlowAggregator, CLIENT_UNKNOWN, Direction } from './FlowAggregator'; + +const FLUSH_INTERVAL_MS = 60_000; +const MAP_REFRESH_MS = 60_000; +const SETTINGS_RECONCILE_MS = 60_000; + +export interface ExporterStats { + deviceId: number; + deviceName: string; + packets: number; + flows: number; + lastSeen: string | null; +} + +export interface CollectorStats { + listening: boolean; + port: number; + packetsReceived: number; + flowsDecoded: number; + flowsAttributed: number; + packetsFromUnknownExporter: number; + recordsWithoutTemplate: number; + exporters: ExporterStats[]; +} + +interface NetflowSettings { + enabled: boolean; + port: number; + topN: number; + acceptUnknown: boolean; +} + +function isPrivateIPv4(ip: string): boolean { + const parts = ip.split('.'); + if (parts.length !== 4) return false; + const a = Number(parts[0]); + const b = Number(parts[1]); + if (a === 10) return true; + if (a === 172 && b >= 16 && b <= 31) return true; + if (a === 192 && b === 168) return true; + if (a === 169 && b === 254) return true; // link-local + if (a === 100 && b >= 64 && b <= 127) return true; // CGNAT + return false; +} + +function isLocalAddress(ip: string): boolean { + if (ip.includes(':')) { + const lower = ip.toLowerCase(); + return lower.startsWith('fc') || lower.startsWith('fd') || lower.startsWith('fe8'); + } + return isPrivateIPv4(ip); +} + +export class NetflowCollector { + private socket: dgram.Socket | null = null; + private templates: TemplateCache = new Map(); + private aggregator = new FlowAggregator(); + + private exporterByIp = new Map(); + private clientMacByIp = new Map(); + + private settings: NetflowSettings = { enabled: false, port: 2055, topN: 50, acceptUnknown: true }; + private listening = false; + + // 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. + private unknownExporterIds = new Map(); + + private flushTimer: ReturnType | null = null; + private mapTimer: ReturnType | null = null; + private reconcileTimer: ReturnType | null = null; + + // Stats (since process start) + private packetsReceived = 0; + private flowsDecoded = 0; + private flowsAttributed = 0; + private packetsFromUnknownExporter = 0; + private recordsWithoutTemplate = 0; + private exporterStats = new Map(); + + async start(): Promise { + await this.refreshMaps().catch(() => {}); + await this.reconcile(); + this.flushTimer = setInterval(() => { + this.flush().catch((e) => console.error('[NetFlow] Flush error:', e)); + }, FLUSH_INTERVAL_MS); + this.mapTimer = setInterval(() => { + this.refreshMaps().catch(() => {}); + }, MAP_REFRESH_MS); + this.reconcileTimer = setInterval(() => { + this.reconcile().catch(() => {}); + }, SETTINGS_RECONCILE_MS); + } + + async stop(): Promise { + if (this.flushTimer) clearInterval(this.flushTimer); + if (this.mapTimer) clearInterval(this.mapTimer); + if (this.reconcileTimer) clearInterval(this.reconcileTimer); + this.closeSocket(); + await this.flush().catch(() => {}); + } + + // Re-read settings and (re)bind or close the socket as needed. Called on an + // interval and directly by the settings route when netflow_* keys change. + async reconcile(): Promise { + const next = await this.readSettings(); + const needsRebind = + next.enabled !== this.settings.enabled || + (next.enabled && next.port !== this.settings.port); + this.settings = next; + if (!needsRebind) return; + + this.closeSocket(); + if (next.enabled) { + this.bindSocket(next.port); + } + } + + getStats(): CollectorStats { + return { + listening: this.listening, + port: this.settings.port, + packetsReceived: this.packetsReceived, + flowsDecoded: this.flowsDecoded, + flowsAttributed: this.flowsAttributed, + packetsFromUnknownExporter: this.packetsFromUnknownExporter, + recordsWithoutTemplate: this.recordsWithoutTemplate, + exporters: Array.from(this.exporterStats.values()), + }; + } + + private async readSettings(): Promise { + try { + const rows = await query<{ key: string; value: unknown }>( + `SELECT key, value FROM app_settings + WHERE key IN ('netflow_enabled', 'netflow_collector_port', 'netflow_topn_clients', + 'netflow_accept_unknown')` + ); + const map: Record = {}; + for (const row of rows) map[row.key] = row.value; + return { + enabled: map['netflow_enabled'] === true, + // In Docker the container always binds 2055 (NETFLOW_BIND_PORT) and the + // host port is remapped via compose; netflow_collector_port is the + // externally reachable port pushed to devices as the export target. + port: Number(process.env.NETFLOW_BIND_PORT) || Number(map['netflow_collector_port']) || 2055, + topN: Number(map['netflow_topn_clients']) || 50, + acceptUnknown: map['netflow_accept_unknown'] !== false, + }; + } catch { + return this.settings; + } + } + + private bindSocket(port: number): void { + const socket = dgram.createSocket('udp4'); + socket.on('message', (msg, rinfo) => this.onMessage(msg, rinfo)); + socket.on('error', (err) => { + console.error(`[NetFlow] Socket error: ${err.message}`); + this.closeSocket(); + }); + socket.bind(port, () => { + this.listening = true; + console.log(`[NetFlow] Collector listening on udp/${port}`); + }); + this.socket = socket; + } + + private closeSocket(): void { + if (this.socket) { + try { + this.socket.close(); + } catch { + /* already closed */ + } + this.socket = null; + } + if (this.listening) console.log('[NetFlow] Collector stopped'); + this.listening = false; + } + + private onMessage(msg: Buffer, rinfo: dgram.RemoteInfo): void { + this.packetsReceived++; + let exporter = this.exporterByIp.get(rinfo.address); + if (!exporter) { + // Source doesn't match any managed device — common when the exporting + // routers sit behind NAT relative to the collector, so every packet + // arrives from the NAT gateway's address. Client attribution doesn't + // depend on exporter identity, so (unless disabled) process these as a + // per-source pseudo-exporter instead of dropping real traffic data. + this.packetsFromUnknownExporter++; + 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?)`); + } + exporter = { deviceId: pseudoId, deviceName: `Unidentified (${rinfo.address})` }; + } + + const result = decodePacket(msg, rinfo.address, this.templates); + this.flowsDecoded += result.flows.length; + this.recordsWithoutTemplate += result.recordsWithoutTemplate; + + let stats = this.exporterStats.get(exporter.deviceId); + if (!stats) { + stats = { deviceId: exporter.deviceId, deviceName: exporter.deviceName, packets: 0, flows: 0, lastSeen: null }; + this.exporterStats.set(exporter.deviceId, stats); + } + stats.packets++; + stats.flows += result.flows.length; + stats.deviceName = exporter.deviceName; + if (result.flows.length > 0) stats.lastSeen = new Date().toISOString(); + + for (const flow of result.flows) { + this.attribute(exporter.deviceId, flow.srcAddr, flow.dstAddr, flow.srcPort, flow.dstPort, flow.protocol, flow.bytes, flow.packets); + } + } + + // Attribution rules: + // - exactly one endpoint is a known client → that client's traffic + // (client = source → upload, client = destination → download) + // - both endpoints known clients → LAN traffic, attributed to both + // - a local/private endpoint we can't map to a client → "unknown" bucket + // - neither endpoint local (e.g. post-NAT duplicate record where the + // source is the router's WAN address) → dropped + private attribute( + exporterId: number, + srcAddr: string, + dstAddr: string, + srcPort: number, + dstPort: number, + protocol: number, + bytes: number, + packets: number + ): void { + const srcMac = this.clientMacByIp.get(srcAddr); + const dstMac = this.clientMacByIp.get(dstAddr); + const srcLocal = srcMac !== undefined || isLocalAddress(srcAddr); + const dstLocal = dstMac !== undefined || isLocalAddress(dstAddr); + + const add = (clientKey: string, direction: Direction, app: string) => { + this.aggregator.add({ exporterId, clientKey, direction, app, bytes, packets }); + this.flowsAttributed++; + }; + + if (srcMac && dstMac) { + // Client-to-client on the local network + add(srcMac, 'upload', APP_LAN); + add(dstMac, 'download', APP_LAN); + return; + } + if (srcLocal && dstLocal) { + // Local flow we can't fully attribute (e.g. printer ↔ unknown host) + add(srcMac || dstMac || CLIENT_UNKNOWN, srcMac ? 'upload' : 'download', APP_LAN); + return; + } + if (srcLocal && !dstLocal) { + add(srcMac || CLIENT_UNKNOWN, 'upload', classifyApp(protocol, srcPort, dstPort)); + return; + } + if (!srcLocal && dstLocal) { + add(dstMac || CLIENT_UNKNOWN, 'download', classifyApp(protocol, dstPort, srcPort)); + return; + } + // Neither endpoint local — transit or post-NAT duplicate; drop. + } + + private async refreshMaps(): Promise { + // Exporters: management IP + every cached /ip/address (CIDR stripped) + const devices = await query<{ id: number; name: string; ip_address: string; ip_addresses_jsonb: { address: string }[] | null }>( + `SELECT id, name, ip_address, ip_addresses_jsonb FROM devices` + ); + const exporters = new Map(); + for (const d of devices) { + exporters.set(d.ip_address, { deviceId: d.id, deviceName: d.name }); + for (const entry of d.ip_addresses_jsonb || []) { + const ip = (entry.address || '').split('/')[0]; + if (ip && !exporters.has(ip)) exporters.set(ip, { deviceId: d.id, deviceName: d.name }); + } + } + this.exporterByIp = exporters; + + // Clients: newest MAC per IP wins (DHCP reuse) + const clients = await query<{ mac_address: string; ip_address: string }>( + `SELECT DISTINCT ON (ip_address) mac_address, ip_address + FROM clients + WHERE ip_address IS NOT NULL AND ip_address != '' + ORDER BY ip_address, last_seen DESC NULLS LAST` + ); + const clientMap = new Map(); + for (const c of clients) { + clientMap.set(c.ip_address, c.mac_address.toLowerCase()); + } + // A device's own addresses are not client traffic endpoints we can bill + // to a MAC — exporter match takes priority over a stale client row. + for (const ip of exporters.keys()) clientMap.delete(ip); + this.clientMacByIp = clientMap; + } + + private async flush(): Promise { + const rows = this.aggregator.drain(this.settings.topN); + if (rows.length === 0) return; + + const writeApi = getWriteApi(); + const now = new Date(); + for (const row of rows) { + writeApi.writePoint( + new Point('client_traffic') + .tag('mac', row.clientKey) + .tag('direction', row.direction) + .tag('app', row.app) + .intField('bytes', row.bytes) + .intField('packets', row.packets) + .timestamp(now) + ); + } + await writeApi.flush().catch((e) => console.error('[NetFlow] Influx write error:', e)); + + // Daily rollups: totals per mac + per-app byte breakdown + const perClient = new Map }>(); + for (const row of rows) { + let entry = perClient.get(row.clientKey); + if (!entry) { + entry = { upload: 0, download: 0, apps: new Map() }; + perClient.set(row.clientKey, entry); + } + if (row.direction === 'upload') entry.upload += row.bytes; + else entry.download += row.bytes; + entry.apps.set(row.app, (entry.apps.get(row.app) || 0) + row.bytes); + } + + for (const [mac, entry] of perClient) { + try { + await query( + `INSERT INTO client_traffic_daily (mac_address, day, upload_bytes, download_bytes, app_breakdown) + VALUES ($1, CURRENT_DATE, $2, $3, $4::jsonb) + ON CONFLICT (mac_address, day) DO UPDATE SET + upload_bytes = client_traffic_daily.upload_bytes + EXCLUDED.upload_bytes, + download_bytes = client_traffic_daily.download_bytes + EXCLUDED.download_bytes, + app_breakdown = ( + SELECT COALESCE(jsonb_object_agg(k, v), '{}'::jsonb) + FROM ( + SELECT COALESCE(a.key, b.key) AS k, + to_jsonb(COALESCE((a.value)::bigint, 0) + COALESCE((b.value)::bigint, 0)) AS v + FROM jsonb_each_text(client_traffic_daily.app_breakdown) a + FULL OUTER JOIN jsonb_each_text(EXCLUDED.app_breakdown) b ON a.key = b.key + ) merged + )`, + [mac, entry.upload, entry.download, JSON.stringify(Object.fromEntries(entry.apps))] + ); + } catch (e) { + console.error('[NetFlow] Daily rollup error:', e); + } + } + } +} + +export const netflowCollector = new NetflowCollector(); diff --git a/backend/src/services/netflow/__tests__/aggregator.test.ts b/backend/src/services/netflow/__tests__/aggregator.test.ts new file mode 100644 index 0000000..a080261 --- /dev/null +++ b/backend/src/services/netflow/__tests__/aggregator.test.ts @@ -0,0 +1,107 @@ +import { FlowAggregator, CLIENT_UNKNOWN, CLIENT_OTHER } from '../FlowAggregator'; +import { classifyApp } from '../appCategories'; + +const MAC_A = 'aa:aa:aa:aa:aa:01'; +const MAC_B = 'aa:aa:aa:aa:aa:02'; + +describe('FlowAggregator dedup', () => { + it('keeps only the exporter that saw the most bytes for each client', () => { + const agg = new FlowAggregator(); + // Gateway (device 1) sees all of client A's traffic; an intermediate + // switch/router (device 2) sees a subset of the same flows. + agg.add({ exporterId: 1, clientKey: MAC_A, direction: 'download', app: 'HTTPS', bytes: 10_000, packets: 20 }); + agg.add({ exporterId: 2, clientKey: MAC_A, direction: 'download', app: 'HTTPS', bytes: 7_000, packets: 14 }); + + const rows = agg.drain(50); + expect(rows).toHaveLength(1); + expect(rows[0].bytes).toBe(10_000); + }); + + it('breaks byte ties deterministically by lowest exporter id', () => { + const agg = new FlowAggregator(); + agg.add({ exporterId: 5, clientKey: MAC_A, direction: 'upload', app: 'DNS', bytes: 100, packets: 1 }); + agg.add({ exporterId: 3, clientKey: MAC_A, direction: 'upload', app: 'QUIC', bytes: 100, packets: 1 }); + + const rows = agg.drain(50); + expect(rows).toHaveLength(1); + expect(rows[0].app).toBe('QUIC'); // exporter 3 wins the tie + }); + + it('picks the winning exporter per client independently', () => { + const agg = new FlowAggregator(); + agg.add({ exporterId: 1, clientKey: MAC_A, direction: 'download', app: 'HTTPS', bytes: 500, packets: 1 }); + agg.add({ exporterId: 2, clientKey: MAC_A, direction: 'download', app: 'HTTPS', bytes: 100, packets: 1 }); + agg.add({ exporterId: 1, clientKey: MAC_B, direction: 'download', app: 'HTTPS', bytes: 100, packets: 1 }); + agg.add({ exporterId: 2, clientKey: MAC_B, direction: 'download', app: 'HTTPS', bytes: 900, packets: 1 }); + + const rows = agg.drain(50); + const a = rows.find((r) => r.clientKey === MAC_A); + const b = rows.find((r) => r.clientKey === MAC_B); + expect(a?.bytes).toBe(500); // device 1 wins for A + expect(b?.bytes).toBe(900); // device 2 wins for B + }); + + it('accumulates repeated samples within the window and clears on drain', () => { + const agg = new FlowAggregator(); + agg.add({ exporterId: 1, clientKey: MAC_A, direction: 'upload', app: 'HTTPS', bytes: 100, packets: 1 }); + agg.add({ exporterId: 1, clientKey: MAC_A, direction: 'upload', app: 'HTTPS', bytes: 250, packets: 2 }); + + const rows = agg.drain(50); + expect(rows).toHaveLength(1); + expect(rows[0].bytes).toBe(350); + expect(rows[0].packets).toBe(3); + expect(agg.drain(50)).toHaveLength(0); + }); +}); + +describe('FlowAggregator top-N fold', () => { + it('folds clients beyond the top-N into the "other" bucket', () => { + const agg = new FlowAggregator(); + for (let i = 0; i < 5; i++) { + const mac = `aa:aa:aa:aa:bb:0${i}`; + agg.add({ exporterId: 1, clientKey: mac, direction: 'download', app: 'HTTPS', bytes: (i + 1) * 1000, packets: 1 }); + } + + const rows = agg.drain(2); + const macs = new Set(rows.map((r) => r.clientKey)); + expect(macs.has('aa:aa:aa:aa:bb:04')).toBe(true); // biggest + expect(macs.has('aa:aa:aa:aa:bb:03')).toBe(true); // second + expect(macs.has(CLIENT_OTHER)).toBe(true); + const other = rows.find((r) => r.clientKey === CLIENT_OTHER); + expect(other?.bytes).toBe(1000 + 2000 + 3000); // the three folded clients + }); + + it('never folds the unknown pseudo-client and does not let it consume a top-N slot', () => { + const agg = new FlowAggregator(); + agg.add({ exporterId: 1, clientKey: CLIENT_UNKNOWN, direction: 'download', app: 'HTTPS', bytes: 999_999, packets: 1 }); + agg.add({ exporterId: 1, clientKey: MAC_A, direction: 'download', app: 'HTTPS', bytes: 10, packets: 1 }); + + const rows = agg.drain(1); + const macs = new Set(rows.map((r) => r.clientKey)); + expect(macs.has(CLIENT_UNKNOWN)).toBe(true); + expect(macs.has(MAC_A)).toBe(true); // still fits — unknown didn't take its slot + }); +}); + +describe('classifyApp', () => { + it('classifies by the remote port first', () => { + expect(classifyApp(6, 54321, 443)).toBe('HTTPS'); + expect(classifyApp(17, 54321, 443)).toBe('QUIC'); + expect(classifyApp(17, 50000, 53)).toBe('DNS'); + expect(classifyApp(6, 50000, 22)).toBe('SSH'); + }); + + it('falls back to the local port for inbound services', () => { + // Remote ephemeral port, local well-known port (client hosts SSH) + expect(classifyApp(6, 22, 51515)).toBe('SSH'); + }); + + it('handles non-port protocols and fallbacks', () => { + expect(classifyApp(1, 0, 0)).toBe('ICMP'); + expect(classifyApp(58, 0, 0)).toBe('ICMP'); + expect(classifyApp(47, 0, 0)).toBe('GRE'); + expect(classifyApp(6, 49152, 49153)).toBe('Other TCP'); + expect(classifyApp(17, 49152, 49153)).toBe('Other UDP'); + expect(classifyApp(132, 0, 0)).toBe('Other'); + }); +}); diff --git a/backend/src/services/netflow/__tests__/decoder.test.ts b/backend/src/services/netflow/__tests__/decoder.test.ts new file mode 100644 index 0000000..fb3c72c --- /dev/null +++ b/backend/src/services/netflow/__tests__/decoder.test.ts @@ -0,0 +1,271 @@ +import { decodePacket, TemplateCache } from '../decoder'; + +// ─── Fixture builders ───────────────────────────────────────────────────────── + +interface FieldSpec { id: number; length: number } + +function u16(n: number): Buffer { + const b = Buffer.alloc(2); + b.writeUInt16BE(n); + return b; +} + +function u32(n: number): Buffer { + const b = Buffer.alloc(4); + b.writeUInt32BE(n); + return b; +} + +function ipv4(addr: string): Buffer { + return Buffer.from(addr.split('.').map(Number)); +} + +function templateBody(templateId: number, fields: FieldSpec[]): Buffer { + return Buffer.concat([ + u16(templateId), + u16(fields.length), + ...fields.flatMap((f) => [u16(f.id), u16(f.length)]), + ]); +} + +function flowset(setId: number, body: Buffer, padTo4 = true): Buffer { + let length = 4 + body.length; + let padding = Buffer.alloc(0); + if (padTo4 && length % 4 !== 0) { + padding = Buffer.alloc(4 - (length % 4)); + length += padding.length; + } + return Buffer.concat([u16(setId), u16(length), body, padding]); +} + +function v9Packet(sourceId: number, flowsets: Buffer[]): Buffer { + const body = Buffer.concat(flowsets); + return Buffer.concat([ + u16(9), // version + u16(flowsets.length), // record count (not used by decoder) + u32(12345), // sysUptime + u32(1700000000), // unix secs + u32(1), // sequence + u32(sourceId), + body, + ]); +} + +function ipfixPacket(domainId: number, sets: Buffer[]): Buffer { + const body = Buffer.concat(sets); + return Buffer.concat([ + u16(10), // version + u16(16 + body.length), // total length + u32(1700000000), // export time + u32(1), // sequence + u32(domainId), + body, + ]); +} + +// Standard 7-field record layout used in most tests: +// srcIPv4(8), dstIPv4(12), srcPort(7), dstPort(11), protocol(4), bytes(1, 4B), packets(2, 4B) +const V9_FIELDS: FieldSpec[] = [ + { id: 8, length: 4 }, + { id: 12, length: 4 }, + { id: 7, length: 2 }, + { id: 11, length: 2 }, + { id: 4, length: 1 }, + { id: 1, length: 4 }, + { id: 2, length: 4 }, +]; + +function record(src: string, dst: string, srcPort: number, dstPort: number, proto: number, bytes: number, packets: number): Buffer { + return Buffer.concat([ + ipv4(src), + ipv4(dst), + u16(srcPort), + u16(dstPort), + Buffer.from([proto]), + u32(bytes), + u32(packets), + ]); +} + +// ─── Tests ──────────────────────────────────────────────────────────────────── + +describe('NetFlow v9 decoding', () => { + it('decodes a template followed by data records in the same packet', () => { + const cache: TemplateCache = new Map(); + const packet = v9Packet(0, [ + flowset(0, templateBody(256, V9_FIELDS)), + flowset(256, Buffer.concat([ + record('192.168.1.10', '93.184.216.34', 54321, 443, 6, 1500, 10), + record('93.184.216.34', '192.168.1.10', 443, 54321, 6, 90000, 60), + ])), + ]); + + const result = decodePacket(packet, '10.0.0.1', cache); + expect(result.templatesParsed).toBe(1); + expect(result.flows).toHaveLength(2); + expect(result.flows[0]).toEqual({ + srcAddr: '192.168.1.10', + dstAddr: '93.184.216.34', + srcPort: 54321, + dstPort: 443, + protocol: 6, + bytes: 1500, + packets: 10, + }); + expect(result.flows[1].bytes).toBe(90000); + }); + + it('drops data that arrives before its template, then decodes once the template is known', () => { + const cache: TemplateCache = new Map(); + const dataOnly = v9Packet(0, [ + flowset(256, record('192.168.1.10', '1.1.1.1', 50000, 53, 17, 120, 2)), + ]); + + const first = decodePacket(dataOnly, '10.0.0.1', cache); + expect(first.flows).toHaveLength(0); + expect(first.recordsWithoutTemplate).toBe(1); + + const templateOnly = v9Packet(0, [flowset(0, templateBody(256, V9_FIELDS))]); + decodePacket(templateOnly, '10.0.0.1', cache); + + const second = decodePacket(dataOnly, '10.0.0.1', cache); + expect(second.flows).toHaveLength(1); + expect(second.flows[0].dstPort).toBe(53); + }); + + it('keeps templates from different exporters separate', () => { + const cache: TemplateCache = new Map(); + const templateOnly = v9Packet(0, [flowset(0, templateBody(256, V9_FIELDS))]); + decodePacket(templateOnly, '10.0.0.1', cache); + + const dataOnly = v9Packet(0, [ + flowset(256, record('192.168.1.10', '1.1.1.1', 50000, 53, 17, 120, 2)), + ]); + // Different exporter never sent its template + const other = decodePacket(dataOnly, '10.0.0.2', cache); + expect(other.flows).toHaveLength(0); + expect(other.recordsWithoutTemplate).toBe(1); + }); + + it('ignores trailing padding in data flowsets', () => { + const cache: TemplateCache = new Map(); + decodePacket(v9Packet(0, [flowset(0, templateBody(256, V9_FIELDS))]), '10.0.0.1', cache); + + // One 21-byte record + flowset padding to a 4-byte boundary + const packet = v9Packet(0, [ + flowset(256, record('192.168.1.20', '8.8.8.8', 40000, 443, 17, 555, 5)), + ]); + const result = decodePacket(packet, '10.0.0.1', cache); + expect(result.flows).toHaveLength(1); + expect(result.flows[0].bytes).toBe(555); + }); + + it('ignores options templates and unknown packet versions', () => { + const cache: TemplateCache = new Map(); + const optionsSet = v9Packet(0, [flowset(1, Buffer.alloc(8))]); + expect(decodePacket(optionsSet, '10.0.0.1', cache).flows).toHaveLength(0); + + const v5ish = Buffer.alloc(24); + v5ish.writeUInt16BE(5, 0); + expect(decodePacket(v5ish, '10.0.0.1', cache).flows).toHaveLength(0); + }); +}); + +describe('IPFIX decoding', () => { + it('decodes IPFIX templates (set 2) and 64-bit counters', () => { + const cache: TemplateCache = new Map(); + const fields: FieldSpec[] = [ + { id: 8, length: 4 }, + { id: 12, length: 4 }, + { id: 7, length: 2 }, + { id: 11, length: 2 }, + { id: 4, length: 1 }, + { id: 1, length: 8 }, // octetDeltaCount as 64-bit + { id: 2, length: 8 }, + ]; + const bigBytes = Buffer.alloc(8); + bigBytes.writeBigUInt64BE(5_000_000_000n); + const bigPackets = Buffer.alloc(8); + bigPackets.writeBigUInt64BE(4_000_000n); + const dataRecord = Buffer.concat([ + ipv4('192.168.1.50'), + ipv4('142.250.80.78'), + u16(51000), + u16(443), + Buffer.from([17]), + bigBytes, + bigPackets, + ]); + + const packet = ipfixPacket(7, [ + flowset(2, templateBody(300, fields)), + flowset(300, dataRecord), + ]); + + const result = decodePacket(packet, '10.0.0.1', cache); + expect(result.templatesParsed).toBe(1); + expect(result.flows).toHaveLength(1); + expect(result.flows[0].bytes).toBe(5_000_000_000); + expect(result.flows[0].protocol).toBe(17); + }); + + it('skips enterprise-specific fields by length without corrupting the record walk', () => { + const cache: TemplateCache = new Map(); + // Template: srcIPv4, dstIPv4, enterprise field (id 0x8000|99, 4 bytes + PEN), bytes + const templateBuf = Buffer.concat([ + u16(301), + u16(6), + u16(8), u16(4), + u16(12), u16(4), + u16(0x8000 | 99), u16(4), u32(14988), // enterprise number (MikroTik PEN) + u16(1), u16(4), + u16(2), u16(4), + u16(4), u16(1), + ]); + const dataRecord = Buffer.concat([ + ipv4('192.168.1.60'), + ipv4('9.9.9.9'), + u32(0xdeadbeef), // enterprise value — must be skipped + u32(777), + u32(7), + Buffer.from([6]), + ]); + const packet = ipfixPacket(7, [ + flowset(2, templateBuf), + flowset(301, dataRecord), + ]); + + const result = decodePacket(packet, '10.0.0.1', cache); + expect(result.flows).toHaveLength(1); + expect(result.flows[0].bytes).toBe(777); + expect(result.flows[0].protocol).toBe(6); + }); + + it('decodes IPv6 flow records', () => { + const cache: TemplateCache = new Map(); + const fields: FieldSpec[] = [ + { id: 27, length: 16 }, + { id: 28, length: 16 }, + { id: 7, length: 2 }, + { id: 11, length: 2 }, + { id: 4, length: 1 }, + { id: 1, length: 4 }, + { id: 2, length: 4 }, + ]; + const src = Buffer.alloc(16); + src[0] = 0xfd; + src[15] = 0x01; + const dst = Buffer.alloc(16); + dst[0] = 0x26; + dst[1] = 0x07; + const dataRecord = Buffer.concat([src, dst, u16(50000), u16(443), Buffer.from([6]), u32(2048), u32(4)]); + + const packet = ipfixPacket(7, [ + flowset(2, templateBody(302, fields)), + flowset(302, dataRecord), + ]); + const result = decodePacket(packet, '10.0.0.1', cache); + expect(result.flows).toHaveLength(1); + expect(result.flows[0].srcAddr).toMatch(/^fd00:/); + }); +}); diff --git a/backend/src/services/netflow/appCategories.ts b/backend/src/services/netflow/appCategories.ts new file mode 100644 index 0000000..7ee6520 --- /dev/null +++ b/backend/src/services/netflow/appCategories.ts @@ -0,0 +1,79 @@ +// Flow → application category classification by IP protocol + well-known port. +// Port-based heuristics (no DPI): good enough to distinguish web, DNS, VPN, +// mail, etc. The remote endpoint's port is matched first — a client browsing +// uses an ephemeral local port and the service's well-known remote port. + +const TCP = 6; +const UDP = 17; +const ICMP = 1; +const ICMPV6 = 58; + +export const APP_LAN = 'LAN'; +export const APP_OTHER = 'Other'; + +// `${protocol}/${port}` → category +const PORT_MAP: Record = { + [`${TCP}/443`]: 'HTTPS', + [`${UDP}/443`]: 'QUIC', + [`${TCP}/80`]: 'HTTP', + [`${UDP}/80`]: 'HTTP', + [`${TCP}/53`]: 'DNS', + [`${UDP}/53`]: 'DNS', + [`${TCP}/853`]: 'DNS over TLS', + [`${UDP}/853`]: 'DNS over TLS', + [`${TCP}/22`]: 'SSH', + [`${TCP}/23`]: 'Telnet', + [`${TCP}/25`]: 'Email', + [`${TCP}/465`]: 'Email', + [`${TCP}/587`]: 'Email', + [`${TCP}/110`]: 'Email', + [`${TCP}/995`]: 'Email', + [`${TCP}/143`]: 'Email', + [`${TCP}/993`]: 'Email', + [`${UDP}/123`]: 'NTP', + [`${UDP}/51820`]: 'WireGuard', + [`${UDP}/500`]: 'IPsec', + [`${UDP}/4500`]: 'IPsec', + [`${TCP}/1723`]: 'VPN (PPTP)', + [`${TCP}/1194`]: 'OpenVPN', + [`${UDP}/1194`]: 'OpenVPN', + [`${TCP}/3389`]: 'RDP', + [`${UDP}/3389`]: 'RDP', + [`${TCP}/445`]: 'SMB', + [`${TCP}/139`]: 'SMB', + [`${TCP}/21`]: 'FTP', + [`${UDP}/67`]: 'DHCP', + [`${UDP}/68`]: 'DHCP', + [`${UDP}/547`]: 'DHCP', + [`${UDP}/5353`]: 'mDNS', + [`${UDP}/1900`]: 'SSDP', + [`${TCP}/8291`]: 'Management', + [`${TCP}/8728`]: 'Management', + [`${TCP}/8729`]: 'Management', + [`${UDP}/161`]: 'Management', + [`${UDP}/3478`]: 'Video/Voice Calls', + [`${TCP}/5060`]: 'Video/Voice Calls', + [`${UDP}/5060`]: 'Video/Voice Calls', + [`${TCP}/5061`]: 'Video/Voice Calls', +}; + +export function classifyApp( + protocol: number, + localPort: number, + remotePort: number +): string { + if (protocol === ICMP || protocol === ICMPV6) return 'ICMP'; + if (protocol === 47) return 'GRE'; + if (protocol === 50) return 'IPsec'; + + // Remote port first (the service the client is talking to), then local port + // (the client is hosting the service, e.g. inbound SSH). + const byRemote = PORT_MAP[`${protocol}/${remotePort}`]; + if (byRemote) return byRemote; + const byLocal = PORT_MAP[`${protocol}/${localPort}`]; + if (byLocal) return byLocal; + + if (protocol === TCP) return 'Other TCP'; + if (protocol === UDP) return 'Other UDP'; + return APP_OTHER; +} diff --git a/backend/src/services/netflow/decoder.ts b/backend/src/services/netflow/decoder.ts new file mode 100644 index 0000000..008282a --- /dev/null +++ b/backend/src/services/netflow/decoder.ts @@ -0,0 +1,227 @@ +// NetFlow v9 / IPFIX (v10) decoder — purpose-built for the fields RouterOS +// Traffic Flow exports. Both protocols share the same structure (header + +// sets; template sets describe field layouts, data sets reference them), and +// the field IDs we care about are identical in v9 field types and IPFIX +// information elements, so one decoder covers both. + +export interface FlowRecord { + srcAddr: string; + dstAddr: string; + srcPort: number; + dstPort: number; + protocol: number; + bytes: number; + packets: number; +} + +interface TemplateField { + id: number; + length: number; // 65535 = IPFIX variable-length + enterprise: boolean; +} + +interface Template { + fields: TemplateField[]; + // Minimum bytes a record can occupy (var-length fields count as 1) + minLength: 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; + +export interface DecodeResult { + flows: FlowRecord[]; + templatesParsed: number; + // Data records that arrived before their template (dropped) + recordsWithoutTemplate: number; +} + +const FIELD = { + BYTES: 1, // IN_BYTES / octetDeltaCount + PACKETS: 2, // IN_PKTS / packetDeltaCount + PROTOCOL: 4, // PROTOCOL / protocolIdentifier + SRC_PORT: 7, // L4_SRC_PORT / sourceTransportPort + SRC_IPV4: 8, // IPV4_SRC_ADDR / sourceIPv4Address + DST_PORT: 11, // L4_DST_PORT / destinationTransportPort + DST_IPV4: 12, // IPV4_DST_ADDR / destinationIPv4Address + SRC_IPV6: 27, // IPV6_SRC_ADDR / sourceIPv6Address + DST_IPV6: 28, // IPV6_DST_ADDR / destinationIPv6Address +} as const; + +function readUInt(buf: Buffer, offset: number, length: number): number { + if (length === 8) { + // 64-bit counters (IPFIX octetDeltaCount is often 8 bytes) + return Number(buf.readBigUInt64BE(offset)); + } + if (length >= 1 && length <= 6) { + return buf.readUIntBE(offset, length); + } + return 0; +} + +function formatIPv4(buf: Buffer, offset: number): string { + return `${buf[offset]}.${buf[offset + 1]}.${buf[offset + 2]}.${buf[offset + 3]}`; +} + +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)); + } + return parts.join(':'); +} + +function parseTemplates( + buf: Buffer, + start: number, + setEnd: number, + ipfix: boolean, + exporterKey: string, + sourceId: number, + cache: TemplateCache +): number { + let off = start; + let parsed = 0; + // A template set can contain multiple template records back to back + while (off + 4 <= setEnd) { + const templateId = buf.readUInt16BE(off); + const fieldCount = buf.readUInt16BE(off + 2); + off += 4; + if (templateId < 256 || fieldCount === 0 || fieldCount > 128) break; + + const fields: TemplateField[] = []; + let minLength = 0; + let ok = true; + for (let i = 0; i < fieldCount; i++) { + if (off + 4 > setEnd) { ok = false; break; } + let id = buf.readUInt16BE(off); + const length = buf.readUInt16BE(off + 2); + off += 4; + let enterprise = false; + if (ipfix && (id & 0x8000) !== 0) { + // Enterprise-specific element: 4-byte enterprise number follows + if (off + 4 > setEnd) { ok = false; break; } + off += 4; + id = id & 0x7fff; + enterprise = true; + } + fields.push({ id, length, enterprise }); + minLength += length === 0xffff ? 1 : length; + } + if (!ok) break; + cache.set(`${exporterKey}|${sourceId}|${templateId}`, { fields, minLength }); + parsed++; + } + return parsed; +} + +function parseDataSet( + buf: Buffer, + start: number, + setEnd: number, + template: Template, + flows: FlowRecord[] +): void { + let off = start; + // Records are packed back to back; trailing bytes < minLength are padding + while (off + template.minLength <= setEnd) { + let srcAddr = ''; + let dstAddr = ''; + let srcPort = 0; + let dstPort = 0; + let protocol = 0; + let bytes = -1; + let packets = 0; + let valid = true; + + for (const field of template.fields) { + let len = field.length; + if (len === 0xffff) { + // IPFIX variable-length: first byte is the length (255 → next 2 bytes) + if (off + 1 > setEnd) { valid = false; break; } + len = buf[off]; + off += 1; + if (len === 255) { + if (off + 2 > setEnd) { valid = false; break; } + len = buf.readUInt16BE(off); + off += 2; + } + } + if (off + len > setEnd) { valid = false; break; } + + if (!field.enterprise) { + switch (field.id) { + case FIELD.BYTES: bytes = readUInt(buf, off, len); break; + case FIELD.PACKETS: packets = readUInt(buf, off, len); break; + case FIELD.PROTOCOL: protocol = readUInt(buf, off, len); break; + case FIELD.SRC_PORT: srcPort = readUInt(buf, off, len); break; + case FIELD.DST_PORT: dstPort = readUInt(buf, off, len); break; + case FIELD.SRC_IPV4: if (len === 4) srcAddr = formatIPv4(buf, off); break; + case FIELD.DST_IPV4: if (len === 4) dstAddr = formatIPv4(buf, off); break; + case FIELD.SRC_IPV6: if (len === 16 && !srcAddr) srcAddr = formatIPv6(buf, off); break; + case FIELD.DST_IPV6: if (len === 16 && !dstAddr) dstAddr = formatIPv6(buf, off); break; + } + } + off += len; + } + + if (!valid) break; + // Only emit records that look like flows (options/stats records lack these) + if (srcAddr && dstAddr && bytes >= 0) { + flows.push({ srcAddr, dstAddr, srcPort, dstPort, protocol, bytes, packets }); + } + } +} + +export function decodePacket(buf: Buffer, exporterKey: string, cache: TemplateCache): DecodeResult { + const result: DecodeResult = { flows: [], templatesParsed: 0, recordsWithoutTemplate: 0 }; + if (buf.length < 16) return result; + + const version = buf.readUInt16BE(0); + let offset: number; + let end: number; + let sourceId: number; + let ipfix: boolean; + + if (version === 9) { + if (buf.length < 20) return result; + ipfix = false; + sourceId = buf.readUInt32BE(16); + offset = 20; + end = buf.length; + } else if (version === 10) { + ipfix = true; + const declaredLength = buf.readUInt16BE(2); + sourceId = buf.readUInt32BE(12); // observation domain ID + offset = 16; + end = Math.min(declaredLength, buf.length); + } else { + return result; // unsupported version (v5 etc.) + } + + while (offset + 4 <= end) { + const setId = buf.readUInt16BE(offset); + const setLength = buf.readUInt16BE(offset + 2); + if (setLength < 4 || offset + setLength > end) break; + const setStart = offset + 4; + const setEnd = offset + setLength; + + const isTemplateSet = ipfix ? setId === 2 : setId === 0; + const isOptionsSet = ipfix ? setId === 3 : setId === 1; + + if (isTemplateSet) { + result.templatesParsed += parseTemplates(buf, setStart, setEnd, ipfix, exporterKey, sourceId, cache); + } else if (!isOptionsSet && setId >= 256) { + const template = cache.get(`${exporterKey}|${sourceId}|${setId}`); + if (template) { + parseDataSet(buf, setStart, setEnd, template, result.flows); + } else { + result.recordsWithoutTemplate++; + } + } + offset = setEnd; + } + + return result; +} diff --git a/docker-compose.yml b/docker-compose.yml index fc1d21c..95053bc 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -34,6 +34,11 @@ services: ENCRYPTION_KEY: ${ENCRYPTION_KEY:-changeme32byteslongencryptionkey} CORS_ORIGIN: ${CORS_ORIGIN:-https://localhost,http://localhost,https://127.0.0.1,http://127.0.0.1} PORT: 3001 + NETFLOW_BIND_PORT: 2055 + ports: + # NetFlow/IPFIX collector — UDP goes straight to the backend (nginx not involved). + # Host port must match the netflow_collector_port app setting (default 2055). + - "${NETFLOW_PORT:-2055}:2055/udp" depends_on: postgres: condition: service_healthy diff --git a/frontend/package.json b/frontend/package.json index ad42f98..f152b4c 100644 --- a/frontend/package.json +++ b/frontend/package.json @@ -1,6 +1,6 @@ { "name": "mikrotik-manager-frontend", - "version": "0.12.3-beta", + "version": "0.13.0-beta", "private": true, "type": "module", "scripts": { diff --git a/frontend/src/App.tsx b/frontend/src/App.tsx index b488027..6bf6aa2 100644 --- a/frontend/src/App.tsx +++ b/frontend/src/App.tsx @@ -24,6 +24,8 @@ import NetworkServicesDNSPage from './pages/NetworkServicesDNSPage'; import NetworkServicesNTPPage from './pages/NetworkServicesNTPPage'; import NetworkServicesWireGuardPage from './pages/NetworkServicesWireGuardPage'; import NetworkServicesSyslogPage from './pages/NetworkServicesSyslogPage'; +import NetworkServicesNetflowPage from './pages/NetworkServicesNetflowPage'; +import TrafficAnalyticsPage from './pages/TrafficAnalyticsPage'; function ProtectedRoute({ children }: { children: React.ReactNode }) { const token = useAuthStore((s) => s.token); @@ -67,6 +69,8 @@ export default function App() { } /> } /> } /> + } /> + } /> } /> diff --git a/frontend/src/components/layout/Sidebar.tsx b/frontend/src/components/layout/Sidebar.tsx index f20cd7e..ffcc060 100644 --- a/frontend/src/components/layout/Sidebar.tsx +++ b/frontend/src/components/layout/Sidebar.tsx @@ -4,7 +4,7 @@ import { useQuery } from '@tanstack/react-query'; import { LayoutDashboard, Router, Users, Bell, GitBranch, HardDrive, Settings, Network, ChevronLeft, ChevronRight, Layers, ChevronDown, SlidersHorizontal, X, Wifi, - Server, Globe, Clock, Shield, FileText, + Server, Globe, Clock, Shield, FileText, Activity, BarChart3, } from 'lucide-react'; import clsx from 'clsx'; import { APP_VERSION } from '../../version'; @@ -14,6 +14,7 @@ const navItems = [ { to: '/dashboard', icon: LayoutDashboard, label: 'Dashboard' }, { to: '/devices', icon: Router, label: 'Devices' }, { to: '/clients', icon: Users, label: 'Clients' }, + { to: '/traffic', icon: Activity, label: 'Traffic' }, { to: '/events', icon: Bell, label: 'Events' }, { to: '/topology', icon: GitBranch, label: 'Topology' }, { to: '/backups', icon: HardDrive, label: 'Backups' }, @@ -42,6 +43,7 @@ const networkServicesSubItems = [ { to: '/network-services/ntp', icon: Clock, label: 'NTP' }, { to: '/network-services/wireguard', icon: Shield, label: 'WireGuard' }, { to: '/network-services/syslog', icon: FileText, label: 'Syslog' }, + { to: '/network-services/netflow', icon: BarChart3, label: 'NetFlow' }, ]; interface SidebarProps { diff --git a/frontend/src/pages/ClientDetailPage.tsx b/frontend/src/pages/ClientDetailPage.tsx index 42507e9..3be74f2 100644 --- a/frontend/src/pages/ClientDetailPage.tsx +++ b/frontend/src/pages/ClientDetailPage.tsx @@ -11,7 +11,7 @@ import { BarChart, Bar, Cell, ReferenceLine, } from 'recharts'; import { format, formatDistanceToNow, parseISO } from 'date-fns'; -import { clientsApi } from '../services/api'; +import { clientsApi, trafficApi } from '../services/api'; import type { ClientDetail } from '../services/api'; import type { SignalPoint } from '../services/api'; import clsx from 'clsx'; @@ -581,6 +581,127 @@ function TrafficCard({ mac }: { mac: string }) { ); } +// ─── App Traffic (NetFlow) ──────────────────────────────────────────────────── + +const APP_COLORS = [ + '#3b82f6', '#8b5cf6', '#10b981', '#f59e0b', '#ef4444', '#06b6d4', + '#ec4899', '#84cc16', '#f97316', '#6366f1', '#14b8a6', '#a855f7', +]; + +function AppTrafficCard({ mac }: { mac: string }) { + const [range, setRange] = useState<'1h' | '24h' | '7d' | '30d'>('24h'); + + const { data, isLoading } = useQuery({ + queryKey: ['client-app-traffic', mac, range], + queryFn: () => trafficApi.client(mac, range).then(r => r.data), + refetchInterval: 60_000, + }); + + const series = data?.series ?? []; + const apps = data?.apps ?? []; + + // NetFlow data only exists for clients on networks with export enabled — + // hide the card entirely rather than showing a permanently empty chart. + if (!isLoading && series.length === 0 && apps.length === 0) return null; + + const ranges = [ + { value: '1h', label: '1h' }, + { value: '24h', label: '24h' }, + { value: '7d', label: '7d' }, + { value: '30d', label: '30d' }, + ] as const; + + const chartData = series.map(p => ({ + time: p.time, + Upload: p.upload, + Download: p.download, + label: format(parseISO(p.time), range === '7d' || range === '30d' ? 'MMM d HH:mm' : 'HH:mm'), + })); + + const totalAppBytes = apps.reduce((s, a) => s + a.bytes, 0); + + return ( +
+
+

+ + App Traffic + (NetFlow) +

+
+ {ranges.map(r => ( + + ))} +
+
+ + {isLoading ? ( +
+ ) : ( + <> + {chartData.length > 0 && ( + + + + + formatBytes(v)} + tick={{ fontSize: 10, fill: 'currentColor' }} + className="text-gray-400 dark:text-slate-500" + width={60} + /> + [formatBytes(value), '']} + contentStyle={{ + fontSize: '11px', + backgroundColor: 'var(--tooltip-bg, #1e293b)', + borderColor: 'var(--tooltip-border, #334155)', + color: 'var(--tooltip-color, #e2e8f0)', + }} + /> + + + + + + )} + + {apps.length > 0 && ( +
+ {apps.slice(0, 8).map((a, i) => ( +
+ + {a.app} + + {totalAppBytes > 0 ? `${((a.bytes / totalAppBytes) * 100).toFixed(1)}%` : ''} + + {formatBytes(a.bytes)} +
+ ))} +
+ )} + + )} +
+ ); +} + // ─── Signal Strength Graph ──────────────────────────────────────────────────── function qualityColor(dbm: number): string { @@ -800,12 +921,15 @@ export default function ClientDetailPage() { {/* Details on left; Traffic + Signal stacked on right */}
- {client.client_type === 'wireless' && ( -
- - -
- )} +
+ + {client.client_type === 'wireless' && ( + <> + + + + )} +
); diff --git a/frontend/src/pages/ClientsPage.tsx b/frontend/src/pages/ClientsPage.tsx index c9f0880..3495d52 100644 --- a/frontend/src/pages/ClientsPage.tsx +++ b/frontend/src/pages/ClientsPage.tsx @@ -19,6 +19,13 @@ const REFRESH_OPTIONS = [ const STORAGE_KEY = 'clients-refresh-interval'; +function formatDataBytes(bytes: number): string { + if (bytes < 1024) return `${bytes} B`; + if (bytes < 1024 * 1024) return `${(bytes / 1024).toFixed(1)} KB`; + if (bytes < 1024 * 1024 * 1024) return `${(bytes / 1024 / 1024).toFixed(1)} MB`; + return `${(bytes / 1024 / 1024 / 1024).toFixed(2)} GB`; +} + function readStoredInterval(): number | null { const raw = localStorage.getItem(STORAGE_KEY); if (raw === 'null') return null; @@ -189,6 +196,7 @@ export default function ClientsPage() { case 'interface_name': av = a.interface_name || ''; bv = b.interface_name || ''; break; case 'vlan_id': av = a.vlan_id ?? -1; bv = b.vlan_id ?? -1; break; case 'device_name': av = a.device_name || ''; bv = b.device_name || ''; break; + case 'traffic_today_bytes': av = a.traffic_today_bytes ?? 0; bv = b.traffic_today_bytes ?? 0; break; case 'last_seen': av = a.last_seen || ''; bv = b.last_seen || ''; break; default: av = ''; bv = ''; } @@ -298,6 +306,7 @@ export default function ClientsPage() { { col: 'interface_name', label: 'Port', align: 'left' }, { col: 'vlan_id', label: 'VLAN', align: 'left' }, { col: 'device_name', label: 'Device', align: 'left' }, + { col: 'traffic_today_bytes', label: 'Data (today)', align: 'left' }, { col: 'last_seen', label: 'Last Seen', align: 'left' }, ] as { col: string; label: string; align: 'left' | 'right' }[]).map(({ col, label, align }) => ( {client.device_name || '—'} + + {client.traffic_today_bytes ? formatDataBytes(client.traffic_today_bytes) : '—'} + {client.last_seen ? formatDistanceToNow(new Date(client.last_seen), { addSuffix: true }) diff --git a/frontend/src/pages/NetworkServicesNetflowPage.tsx b/frontend/src/pages/NetworkServicesNetflowPage.tsx new file mode 100644 index 0000000..6689c58 --- /dev/null +++ b/frontend/src/pages/NetworkServicesNetflowPage.tsx @@ -0,0 +1,324 @@ +import { useEffect, useState } from 'react'; +import { Link } from 'react-router-dom'; +import { useQuery, useQueryClient } from '@tanstack/react-query'; +import { + BarChart3, RefreshCw, AlertTriangle, Save, CheckCircle, XCircle, Radio, +} from 'lucide-react'; +import clsx from 'clsx'; +import { formatDistanceToNow, parseISO } from 'date-fns'; +import { networkServicesApi, settingsApi, trafficApi } from '../services/api'; +import { useCanWrite } from '../hooks/useCanWrite'; + +function Toggle({ checked, onChange, disabled }: { checked: boolean; onChange: (v: boolean) => void; disabled?: boolean }) { + return ( + + ); +} + +export default function NetworkServicesNetflowPage() { + const canWrite = useCanWrite(); + const qc = useQueryClient(); + + // ── Collector settings ──────────────────────────────────────────────────── + const { data: settings } = useQuery({ + queryKey: ['app-settings'], + queryFn: () => settingsApi.get().then(r => r.data), + }); + + const [enabled, setEnabled] = useState(false); + const [address, setAddress] = useState(''); + const [port, setPort] = useState('2055'); + const [version, setVersion] = useState('9'); + const [loaded, setLoaded] = useState(false); + const [saving, setSaving] = useState(false); + const [saveMsg, setSaveMsg] = useState(''); + + useEffect(() => { + if (settings && !loaded) { + setEnabled(settings['netflow_enabled'] === true); + setAddress(String(settings['netflow_collector_address'] ?? '')); + setPort(String(settings['netflow_collector_port'] ?? '2055')); + setVersion(String(settings['netflow_version'] ?? '9')); + setLoaded(true); + } + }, [settings, loaded]); + + async function saveCollectorSettings(overrides: Record = {}) { + setSaving(true); setSaveMsg(''); + try { + await settingsApi.update({ + netflow_enabled: enabled, + netflow_collector_address: address.trim(), + netflow_collector_port: parseInt(port, 10) || 2055, + netflow_version: version, + ...overrides, + }); + qc.invalidateQueries({ queryKey: ['app-settings'] }); + qc.invalidateQueries({ queryKey: ['traffic-status'] }); + setSaveMsg('Saved'); + setTimeout(() => setSaveMsg(''), 3000); + } catch (e) { + setSaveMsg(`Failed: ${(e as Error).message}`); + } finally { + setSaving(false); + } + } + + async function handleListenerToggle(value: boolean) { + setEnabled(value); + await saveCollectorSettings({ netflow_enabled: value }); + } + + // ── Collector status (live) ─────────────────────────────────────────────── + const { data: status } = useQuery({ + queryKey: ['traffic-status'], + queryFn: () => trafficApi.status().then(r => r.data), + refetchInterval: 10_000, + }); + + // ── Fleet device state ──────────────────────────────────────────────────── + const { data: fleet, isLoading: fleetLoading, refetch, isFetching, error: fleetError } = useQuery({ + queryKey: ['ns-netflow-fleet'], + queryFn: () => networkServicesApi.netflowFleet().then(r => r.data), + }); + + const [togglePending, setTogglePending] = useState(null); + const [toggleError, setToggleError] = useState(''); + + async function handleDeviceToggle(deviceId: number, value: boolean) { + setTogglePending(deviceId); setToggleError(''); + try { + await networkServicesApi.setNetflow(deviceId, value); + qc.invalidateQueries({ queryKey: ['ns-netflow-fleet'] }); + } catch (e) { + const err = e as { response?: { data?: { error?: string } }; message: string }; + setToggleError(err.response?.data?.error || err.message); + } finally { + setTogglePending(null); + } + } + + const addressMissing = !address.trim(); + const devices = fleet?.devices ?? []; + const unidentified = (status?.exporters ?? []).filter(e => e.deviceId < 0); + + function deviceBadge(d: (typeof devices)[number]) { + if (d.error) { + return Unreachable; + } + if (!d.enabled) { + return Disabled; + } + if (!d.target_matches_collector) { + return Wrong target; + } + if (!d.last_flow_at) { + return No flows yet; + } + return Exporting; + } + + return ( +
+ {/* Header */} +
+
+

NetFlow

+

+ Per-client traffic analytics via NetFlow/IPFIX export — view results on the{' '} + Traffic page +

+
+ +
+ + {/* Collector card */} +
+
+ +

Collector

+ {status && ( + + + {status.listening ? `Listening on udp/${status.port}` : 'Stopped'} + + )} +
+
+
+ +
+
Enable NetFlow collector
+

+ Receives NetFlow v9 / IPFIX exports from your MikroTik devices and aggregates per-client traffic. +

+
+
+ +
+
+ + setAddress(e.target.value)} + placeholder="192.168.1.100" disabled={!canWrite} /> +

+ The IP of this server as reachable by your devices (the Docker host IP, not the container). +

+
+
+ + setPort(e.target.value)} + min="1" max="65535" disabled={!canWrite} /> +

+ Must match the host port mapping (NETFLOW_PORT, default 2055). +

+
+
+ + +
+
+ + {addressMissing && ( +
+ Set the collector address before enabling devices below. +
+ )} + +
+ {canWrite && ( + + )} + {saveMsg && ( + {saveMsg} + )} +
+ + {unidentified.length > 0 && ( +
+ +
+ Receiving flows from unidentified source{unidentified.length > 1 ? 's' : ''}{' '} + {unidentified.map(e => e.deviceName.replace(/^Unidentified \((.+)\)$/, '$1')).join(', ')}. + This usually means NAT sits between your routers and the collector. Flows are still + processed and attributed to clients, but can't be tied to a specific exporting device. +
+
+ )} + + {status && status.packetsReceived > 0 && ( +
+
+
Packets received
+
{status.packetsReceived.toLocaleString()}
+
+
+
Flows decoded
+
{status.flowsDecoded.toLocaleString()}
+
+
+
Flows attributed
+
{status.flowsAttributed.toLocaleString()}
+
+
+
Unknown exporters
+
{status.packetsFromUnknownExporter.toLocaleString()}
+
+
+ )} +
+
+ + {/* Devices */} +
+
+ +

+ Devices({devices.length} online) +

+
+ + {toggleError && ( +
+ {toggleError} +
+ )} + + {fleetLoading &&
Loading device state…
} + {!!fleetError && ( +
+ Failed: {(fleetError as Error).message} +
+ )} + {!fleetLoading && !fleetError && devices.length === 0 && ( +
No online devices found.
+ )} + + {devices.length > 0 && ( +
+ + + + + + + + + + + + {devices.map((d, i) => ( + + + + + + + + ))} + +
DeviceStatusFlowsLast FlowExport
+
{d.name}
+
{d.ip_address}
+ 		{deviceBadge(d)}{d.flows_received.toLocaleString()} + {d.last_flow_at ? formatDistanceToNow(parseISO(d.last_flow_at), { addSuffix: true }) : '—'} + +
+ handleDeviceToggle(d.id, v)} + disabled={!canWrite || !!d.error || togglePending === d.id || (addressMissing && !d.enabled)} + /> +
+
+
+ )} +
+
+ ); +} diff --git a/frontend/src/pages/SettingsPage.tsx b/frontend/src/pages/SettingsPage.tsx index 522c2fb..f5ea6b8 100644 --- a/frontend/src/pages/SettingsPage.tsx +++ b/frontend/src/pages/SettingsPage.tsx @@ -667,6 +667,8 @@ export default function SettingsPage() { {[ { key: 'retention_events_days', label: 'Events retention', desc: 'Auto-delete event log entries older than this many days' }, { key: 'retention_clients_days', label: 'Client retention', desc: 'Auto-delete inactive client records not seen within this many days' }, + { key: 'netflow_retention_days', label: 'Traffic detail retention', desc: 'Auto-delete per-client NetFlow time-series data older than this many days' }, + { key: 'netflow_daily_retention_days', label: 'Traffic daily rollup retention', desc: 'Auto-delete per-client daily traffic totals older than this many days' }, ].map(({ key, label, desc }) => (
diff --git a/frontend/src/pages/TrafficAnalyticsPage.tsx b/frontend/src/pages/TrafficAnalyticsPage.tsx new file mode 100644 index 0000000..98fb605 --- /dev/null +++ b/frontend/src/pages/TrafficAnalyticsPage.tsx @@ -0,0 +1,261 @@ +import { useState } from 'react'; +import { Link } from 'react-router-dom'; +import { useQuery } from '@tanstack/react-query'; +import { useNavigate } from 'react-router-dom'; +import { + AreaChart, Area, XAxis, YAxis, CartesianGrid, Tooltip, ResponsiveContainer, + PieChart, Pie, Cell, Legend, +} from 'recharts'; +import { Activity, ArrowDown, ArrowUp, BarChart3, Settings2 } from 'lucide-react'; +import { format, parseISO } from 'date-fns'; +import clsx from 'clsx'; +import { trafficApi } from '../services/api'; + +type Range = '1h' | '24h' | '7d' | '30d'; +const RANGES: Range[] = ['1h', '24h', '7d', '30d']; + +const APP_COLORS = [ + '#3b82f6', '#8b5cf6', '#10b981', '#f59e0b', '#ef4444', '#06b6d4', + '#ec4899', '#84cc16', '#f97316', '#6366f1', '#14b8a6', '#a855f7', + '#64748b', '#0ea5e9', '#d946ef', '#22c55e', +]; + +function formatBytes(bytes: number): string { + if (bytes < 1024) return `${bytes} B`; + if (bytes < 1024 * 1024) return `${(bytes / 1024).toFixed(1)} KB`; + if (bytes < 1024 * 1024 * 1024) return `${(bytes / 1024 / 1024).toFixed(1)} MB`; + if (bytes < 1024 ** 4) return `${(bytes / 1024 ** 3).toFixed(2)} GB`; + return `${(bytes / 1024 ** 4).toFixed(2)} TB`; +} + +function timeFormatter(range: Range) { + return (t: string) => { + try { + const d = parseISO(t); + return range === '7d' || range === '30d' ? format(d, 'MMM d') : format(d, 'HH:mm'); + } catch { + return t; + } + }; +} + +function clientLabel(c: { mac: string; custom_name: string | null; hostname: string | null; vendor: string | null }): string { + if (c.mac === 'unknown') return 'Unattributed (local)'; + if (c.mac === 'other') return 'Other clients'; + return c.custom_name || c.hostname || c.vendor || c.mac; +} + +export default function TrafficAnalyticsPage() { + const [range, setRange] = useState('24h'); + const navigate = useNavigate(); + + const { data: series = [], isLoading: seriesLoading } = useQuery({ + queryKey: ['traffic-timeseries', range], + queryFn: () => trafficApi.timeseries(range).then(r => r.data), + refetchInterval: 60_000, + }); + + const { data: topClients = [], isLoading: clientsLoading } = useQuery({ + queryKey: ['traffic-top-clients', range], + queryFn: () => trafficApi.topClients(range, 10).then(r => r.data), + refetchInterval: 60_000, + }); + + const { data: apps = [], isLoading: appsLoading } = useQuery({ + queryKey: ['traffic-apps', range], + queryFn: () => trafficApi.apps(range).then(r => r.data), + refetchInterval: 60_000, + }); + + const isLoading = seriesLoading || clientsLoading || appsLoading; + const hasData = series.length > 0 || topClients.length > 0 || apps.length > 0; + + const totalUpload = series.reduce((s, p) => s + p.upload, 0); + const totalDownload = series.reduce((s, p) => s + p.download, 0); + const totalAppBytes = apps.reduce((s, a) => s + a.bytes, 0); + const maxClientBytes = topClients[0]?.total_bytes || 1; + + // Donut: top 8 categories, rest folded into "Other" + const donutData = (() => { + const top = apps.slice(0, 8); + const rest = apps.slice(8).reduce((s, a) => s + a.bytes, 0); + return rest > 0 ? [...top, { app: 'Other', bytes: rest }] : top; + })(); + + return ( +
+ {/* Header */} +
+
+

Traffic

+

Per-client traffic analytics from NetFlow/IPFIX

+
+
+
+ {RANGES.map(r => ( + + ))} +
+ + Configure + +
+
+ + {/* Empty state */} + {!isLoading && !hasData && ( +
+ +

No traffic data yet

+

+ Enable the NetFlow collector and turn on traffic export for your routers to start + collecting per-client usage and protocol analytics. +

+ + Set up NetFlow + +
+ )} + + {hasData && ( + <> + {/* Totals */} +
+
+
+ +
+
+
Download ({range})
+
{formatBytes(totalDownload)}
+
+
+
+
+ +
+
+
Upload ({range})
+
{formatBytes(totalUpload)}
+
+
+
+ + {/* Throughput chart */} +
+

Network Traffic

+ + + + + + + + + + + + + + + formatBytes(v)} tick={{ fontSize: 11 }} width={70} stroke="currentColor" /> + { try { return format(parseISO(String(t)), 'MMM d, HH:mm'); } catch { return String(t); } }} + formatter={(value: number, name: string) => [formatBytes(value), name === 'download' ? 'Download' : 'Upload']} + /> + (v === 'download' ? 'Download' : 'Upload')} /> + + + + +
+ +
+ {/* Top talkers */} +
+
+ +

Top Talkers

+
+ {topClients.length === 0 + ?
No client traffic recorded.
+ : ( +
+ {topClients.map(c => { + const isReal = c.mac !== 'unknown' && c.mac !== 'other'; + return ( +
isReal && navigate(`/clients/${encodeURIComponent(c.mac)}`)} + className={clsx('px-5 py-3', isReal && 'cursor-pointer hover:bg-blue-50 dark:hover:bg-slate-700/40 transition-colors')}> +
+
+ {clientLabel(c)} + {isReal && ( + {c.ip_address || c.mac} + )} +
+ {formatBytes(c.total_bytes)} +
+
+
+
+
+ {formatBytes(c.download_bytes)} + {formatBytes(c.upload_bytes)} +
+
+ ); + })} +
+ )} +
+ + {/* App categories */} +
+
+ +

Traffic by Application

+
+ {apps.length === 0 + ?
No application data recorded.
+ : ( +
+ + + + {donutData.map((entry, i) => ( + + ))} + + [formatBytes(value), name]} /> + + +
+ {apps.slice(0, 10).map((a, i) => ( +
+ + {a.app} + + {totalAppBytes > 0 ? `${((a.bytes / totalAppBytes) * 100).toFixed(1)}%` : ''} + + {formatBytes(a.bytes)} +
+ ))} +
+
+ )} +
+
+ + )} +
+ ); +} diff --git a/frontend/src/services/api.ts b/frontend/src/services/api.ts index b2e1f51..08a265d 100644 --- a/frontend/src/services/api.ts +++ b/frontend/src/services/api.ts @@ -775,6 +775,73 @@ export const networkServicesApi = { api.put(`/network-services/wireguard/peer/${encodeURIComponent(id)}`, body, { params: { deviceId } }), deleteWireGuardPeer: (deviceId: number, id: string) => api.delete(`/network-services/wireguard/peer/${encodeURIComponent(id)}`, { params: { deviceId } }), + + // ── NetFlow ─────────────────────────────────────────────────────────────── + netflowFleet: () => + api.get<{ + collector: { address: string; port: number; version: string; listening: boolean }; + devices: NetflowDeviceState[]; + }>('/network-services/netflow/fleet', { timeout: 60_000 }), + getNetflow: (deviceId: number) => + api.get<{ settings: NS | null; targets: NS[]; target_matches_collector: boolean }>( + '/network-services/netflow', { params: { deviceId } } + ), + setNetflow: (deviceId: number, enabled: boolean) => + api.put<{ settings: NS | null; targets: NS[]; target_matches_collector: boolean }>( + '/network-services/netflow', { enabled }, { params: { deviceId } } + ), +}; + +export interface NetflowDeviceState { + id: number; + name: string; + ip_address: string; + enabled: boolean | null; + interfaces: string; + targets: { id: string; dst_address: string; port: number; version: string; disabled: boolean }[]; + target_matches_collector: boolean; + flows_received: number; + last_flow_at: string | null; + error?: string; +} + +// ─── Traffic Analytics (NetFlow) ────────────────────────────────────────────── + +export interface ClientTrafficPoint { time: string; upload: number; download: number } +export interface TrafficTopClient { + mac: string; + upload_bytes: number; + download_bytes: number; + total_bytes: number; + hostname: string | null; + custom_name: string | null; + vendor: string | null; + ip_address: string | null; +} +export interface TrafficApp { app: string; bytes: number } +export interface TrafficCollectorStats { + listening: boolean; + port: number; + packetsReceived: number; + flowsDecoded: number; + flowsAttributed: number; + packetsFromUnknownExporter: number; + recordsWithoutTemplate: number; + exporters: { deviceId: number; deviceName: string; packets: number; flows: number; lastSeen: string | null }[]; +} + +export const trafficApi = { + status: () => api.get('/traffic/status'), + timeseries: (range = '24h') => + api.get('/traffic/timeseries', { params: { range } }), + topClients: (range = '24h', limit = 10) => + api.get('/traffic/top-clients', { params: { range, limit } }), + apps: (range = '24h', mac?: string) => + api.get('/traffic/apps', { params: { range, mac } }), + client: (mac: string, range = '24h') => + api.get<{ series: ClientTrafficPoint[]; apps: TrafficApp[] }>( + `/traffic/client/${encodeURIComponent(mac)}`, { params: { range } } + ), }; // ─── Maintenance Windows ────────────────────────────────────────────────────── diff --git a/frontend/src/types/index.ts b/frontend/src/types/index.ts index 50c8fe3..c6ad370 100644 --- a/frontend/src/types/index.ts +++ b/frontend/src/types/index.ts @@ -179,6 +179,8 @@ export interface Client { client_type: 'wired' | 'wireless'; active: boolean; last_seen?: string; + /** Today's NetFlow upload+download total (0 when NetFlow is not collecting). */ + traffic_today_bytes?: number; } export interface Backup { diff --git a/frontend/src/version.ts b/frontend/src/version.ts index 5e65fd9..7ee87b3 100644 --- a/frontend/src/version.ts +++ b/frontend/src/version.ts @@ -1 +1 @@ -export const APP_VERSION = 'v0.12.3 Beta'; +export const APP_VERSION = 'v0.13.0 Beta';