feat: NetFlow/IPFIX traffic analytics — per-client usage, app breakdown, built-in collector (v0.13.0-beta)
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 <noreply@anthropic.com>
This commit is contained in:
2GT-Rich
co-authored by
Claude Fable 5
Claude Fable 5
parent
367034c2d6
commit
b841eb6e7b
Generated
+12
-2
@@ -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",
|
||||
|
||||
@@ -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",
|
||||
|
||||
@@ -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<void> {
|
||||
|
||||
@@ -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<void> {
|
||||
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<void> {
|
||||
const shutdown = async () => {
|
||||
console.log('Shutting down...');
|
||||
await stopBulkAddWorker();
|
||||
await netflowCollector.stop();
|
||||
await pollerService.stop();
|
||||
await redis.quit().catch(() => {});
|
||||
await pool.end();
|
||||
|
||||
@@ -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++}
|
||||
`;
|
||||
|
||||
@@ -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<string, unknown> = {};
|
||||
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<string, string>[],
|
||||
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<DeviceRow>(`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<string, string> = {
|
||||
'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;
|
||||
|
||||
@@ -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' });
|
||||
});
|
||||
|
||||
|
||||
@@ -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<string, string> = {
|
||||
'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<string, { time: string; upload: number; download: number }> = {};
|
||||
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<string, { upload: number; download: number }>();
|
||||
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<string, { hostname: string | null; custom_name: string | null; vendor: string | null; ip_address: string | null }>();
|
||||
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<string, { time: string; upload: number; download: number }> = {};
|
||||
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;
|
||||
@@ -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<void> {
|
||||
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<string, unknown> = {};
|
||||
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<string, unknown>): Promise<void> {
|
||||
// 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
|
||||
|
||||
@@ -2706,4 +2706,31 @@ export class DeviceCollector {
|
||||
async removeWireGuardPeer(id: string): Promise<void> {
|
||||
await this.client.execute('/interface/wireguard/peers/remove', { '.id': id });
|
||||
}
|
||||
|
||||
// Traffic Flow (NetFlow/IPFIX export) ─────────────────────────────────────────
|
||||
|
||||
async getTrafficFlowSettings(): Promise<Record<string, string> | null> {
|
||||
const rows = await this.client.execute('/ip/traffic-flow/print', {}).catch(() => [] as Record<string, string>[]);
|
||||
return rows[0] || null;
|
||||
}
|
||||
|
||||
async setTrafficFlowSettings(params: Record<string, string>): Promise<void> {
|
||||
await this.client.execute('/ip/traffic-flow/set', params);
|
||||
}
|
||||
|
||||
async getTrafficFlowTargets(): Promise<Record<string, string>[]> {
|
||||
return this.client.execute('/ip/traffic-flow/target/print', { detail: '' }).catch(() => [] as Record<string, string>[]);
|
||||
}
|
||||
|
||||
async addTrafficFlowTarget(params: Record<string, string>): Promise<void> {
|
||||
await this.client.execute('/ip/traffic-flow/target/add', params);
|
||||
}
|
||||
|
||||
async updateTrafficFlowTarget(id: string, params: Record<string, string>): Promise<void> {
|
||||
await this.client.execute('/ip/traffic-flow/target/set', { '.id': id, ...params });
|
||||
}
|
||||
|
||||
async removeTrafficFlowTarget(id: string): Promise<void> {
|
||||
await this.client.execute('/ip/traffic-flow/target/remove', { '.id': id });
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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<string, Bucket>();
|
||||
|
||||
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<string, Map<number, number>>(); // 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<string, number>(); // 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<string, AggregatedRow>();
|
||||
const clientBytes = new Map<string, number>();
|
||||
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<string, AggregatedRow>();
|
||||
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());
|
||||
}
|
||||
}
|
||||
@@ -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<string, { deviceId: number; deviceName: string }>();
|
||||
private clientMacByIp = new Map<string, string>();
|
||||
|
||||
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<string, number>();
|
||||
|
||||
private flushTimer: ReturnType<typeof setInterval> | null = null;
|
||||
private mapTimer: ReturnType<typeof setInterval> | null = null;
|
||||
private reconcileTimer: ReturnType<typeof setInterval> | 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<number, ExporterStats>();
|
||||
|
||||
async start(): Promise<void> {
|
||||
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<void> {
|
||||
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<void> {
|
||||
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<NetflowSettings> {
|
||||
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<string, unknown> = {};
|
||||
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<void> {
|
||||
// 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<string, { deviceId: number; deviceName: string }>();
|
||||
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<string, string>();
|
||||
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<void> {
|
||||
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<string, { upload: number; download: number; apps: Map<string, number> }>();
|
||||
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();
|
||||
@@ -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');
|
||||
});
|
||||
});
|
||||
@@ -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:/);
|
||||
});
|
||||
});
|
||||
@@ -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<string, string> = {
|
||||
[`${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;
|
||||
}
|
||||
@@ -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<string, Template>;
|
||||
|
||||
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;
|
||||
}
|
||||
Reference in New Issue
Block a user