From 2820683cbae7230de6d423e9b0f996e0974bd8ed Mon Sep 17 00:00:00 2001 From: Denozordec Date: Mon, 7 Sep 2026 02:13:21 +0700 Subject: [PATCH] =?UTF-8?q?feat(traffic):=20=D0=B4=D0=BE=D0=B1=D0=B0=D0=B2?= =?UTF-8?q?=D0=B8=D1=82=D1=8C=20=D0=BA=D0=B0=D1=80=D1=82=D1=83=20=D0=B8=20?= =?UTF-8?q?=D1=81=D1=80=D0=B5=D0=B7=D1=8B=20IPFIX=20=D0=BF=D0=BE=20=D1=81?= =?UTF-8?q?=D1=82=D1=80=D0=B0=D0=BD=D0=B0=D0=BC=20=D0=B8=20=D1=81=D0=B5?= =?UTF-8?q?=D1=81=D1=81=D0=B8=D1=8F=D0=BC?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Имена ifIndex пишутся по обоим ключам REST, дедуп 5-tuple по max байт, RIPEstat только из prefix-кэша, карта откуда-куда в Frame. Co-authored-by: Cursor --- app/(main)/traffic/page.tsx | 9 +- backend/package.json | 2 +- backend/src/db/index.ts | 16 + backend/src/db/schema.ts | 19 + backend/src/routes/traffic-flow.ts | 9 +- .../services/traffic-flow-analytics.test.ts | 100 +++++ .../src/services/traffic-flow-analytics.ts | 153 ++++++-- .../services/traffic-flow-classify.test.ts | 20 + backend/src/services/traffic-flow-classify.ts | 140 +++++++ .../src/services/traffic-flow-dedup.test.ts | 25 ++ backend/src/services/traffic-flow-dedup.ts | 44 +++ .../src/services/traffic-flow-ifaces.test.ts | 8 + backend/src/services/traffic-flow-ifaces.ts | 1 + backend/src/services/traffic-flow-ifindex.ts | 7 +- backend/src/services/traffic-flow-ingest.ts | 6 +- backend/src/services/traffic-flow-ip.ts | 54 +++ .../src/services/traffic-flow-parse.test.ts | 2 +- backend/src/services/traffic-flow-parse.ts | 2 +- .../src/services/traffic-flow-ripe.test.ts | 72 ++++ backend/src/services/traffic-flow-ripe.ts | 371 ++++++++++++++++++ .../data-grids/traffic-flows-data-grid.tsx | 9 +- components/traffic/flow-analytics-panel.tsx | 168 +++++--- components/traffic/flow-traffic-map.tsx | 184 +++++++++ hooks/use-flow-live.ts | 4 +- packages/contracts/src/traffic-flow.ts | 24 ++ shared/api/traffic-flow.ts | 5 +- 26 files changed, 1365 insertions(+), 89 deletions(-) create mode 100644 backend/src/services/traffic-flow-classify.test.ts create mode 100644 backend/src/services/traffic-flow-classify.ts create mode 100644 backend/src/services/traffic-flow-dedup.test.ts create mode 100644 backend/src/services/traffic-flow-dedup.ts create mode 100644 backend/src/services/traffic-flow-ip.ts create mode 100644 backend/src/services/traffic-flow-ripe.test.ts create mode 100644 backend/src/services/traffic-flow-ripe.ts create mode 100644 components/traffic/flow-traffic-map.tsx diff --git a/app/(main)/traffic/page.tsx b/app/(main)/traffic/page.tsx index c51e16e..0050b13 100644 --- a/app/(main)/traffic/page.tsx +++ b/app/(main)/traffic/page.tsx @@ -67,7 +67,7 @@ function flowEmptyHint(stats: FlowStatsDto | null): string | undefined { if (!stats) return undefined if (stats.lastError) return stats.lastError if (stats.packetsReceived) { - return `IPFIX приходит (${stats.lastExporterIp ?? "экспортёр"}), но разговоры ещё не записаны.` + return `IPFIX приходит (${stats.lastExporterIp ?? "экспортёр"}), но сессии ещё не записаны.` } if (stats.listenerBound === false) { return "Коллектор UDP не слушает. Подключите JH ещё раз — ingest включится автоматически." @@ -771,6 +771,7 @@ export default function TrafficPage() { const [flowClients, setFlowClients] = useState([]) const [flowAnalytics, setFlowAnalytics] = useState(null) const [flowIface, setFlowIface] = useState("__all__") + const [flowDedup, setFlowDedup] = useState(true) const [overlayOpen, setOverlayOpen] = useState(false) const [catalogServers, setCatalogServers] = useState([]) const effectiveMode: GroupMode = groupMode @@ -788,6 +789,7 @@ export default function TrafficPage() { serverId: flowScope === "servers" ? selectedId : undefined, userId: flowScope === "users" ? selectedId : undefined, iface: flowIface, + dedup: flowDedup, }) const toLiveServer = (s: LiveTrafficServer): ServerTraffic => { @@ -914,8 +916,9 @@ export default function TrafficPage() { serverId: flowScope === "servers" ? selectedId : undefined, userId: flowScope === "users" ? selectedId : undefined, iface: flowIface, + dedup: flowDedup, }).then(setFlowAnalytics).catch(() => setFlowAnalytics(null)) - }, [isLive, effectiveMode, selectedId, range, flowScope, flowIface, backendUrl]) + }, [isLive, effectiveMode, selectedId, range, flowScope, flowIface, flowDedup, backendUrl]) useEffect(() => { setFlowIface("__all__") @@ -1269,6 +1272,8 @@ export default function TrafficPage() { onRange={(r) => setRange(r as Range)} selectedIface={flowIface} onIface={setFlowIface} + dedup={flowDedup} + onDedup={setFlowDedup} liveHint={displayedFlow?.live ? "live" : undefined} emptyHint={flowEmptyHint(flowStats)} /> diff --git a/backend/package.json b/backend/package.json index 4f3eb3e..febecb1 100644 --- a/backend/package.json +++ b/backend/package.json @@ -14,7 +14,7 @@ "test:auth": "tsx src/lib/permissions.test.ts && tsx src/plugins/auth.smoke.test.ts", "test:wireguard": "npx tsx src/services/wireguard-config.test.ts", "test:traffic-rate": "tsx src/services/traffic-rate.test.ts", - "test:traffic-flow": "tsx src/services/traffic-flow-parse.test.ts && tsx src/services/traffic-flow-map-exporter.test.ts && tsx src/services/traffic-flow-ifaces.test.ts && tsx src/services/traffic-flow-analytics.test.ts", + "test:traffic-flow": "tsx src/services/traffic-flow-parse.test.ts && tsx src/services/traffic-flow-map-exporter.test.ts && tsx src/services/traffic-flow-ifaces.test.ts && tsx src/services/traffic-flow-dedup.test.ts && tsx src/services/traffic-flow-classify.test.ts && tsx src/services/traffic-flow-ripe.test.ts && tsx src/services/traffic-flow-analytics.test.ts", "test:users": "tsx src/modules/users/iface-type.test.ts && tsx src/modules/users/bindings.test.ts" }, "dependencies": { diff --git a/backend/src/db/index.ts b/backend/src/db/index.ts index 68dad3d..0758fe3 100644 --- a/backend/src/db/index.ts +++ b/backend/src/db/index.ts @@ -162,6 +162,22 @@ CREATE UNIQUE INDEX IF NOT EXISTS idx_flow_buckets_unique CREATE INDEX IF NOT EXISTS idx_flow_buckets_server_time ON flow_buckets(server_id, bucket_at); +CREATE TABLE IF NOT EXISTS flow_ip_meta ( + prefix TEXT PRIMARY KEY, + asn INTEGER NOT NULL DEFAULT 0, + country TEXT NOT NULL DEFAULT '', + lat REAL, + lng REAL, + holder TEXT NOT NULL DEFAULT '', + ok INTEGER NOT NULL DEFAULT 1, + fetched_at TEXT NOT NULL +); +CREATE TABLE IF NOT EXISTS flow_asn_meta ( + asn INTEGER PRIMARY KEY, + holder TEXT NOT NULL DEFAULT '', + fetched_at TEXT NOT NULL +); + CREATE TABLE IF NOT EXISTS uptime_settings ( id INTEGER PRIMARY KEY, enabled INTEGER NOT NULL DEFAULT 1, diff --git a/backend/src/db/schema.ts b/backend/src/db/schema.ts index e614bfd..25a4fa5 100644 --- a/backend/src/db/schema.ts +++ b/backend/src/db/schema.ts @@ -202,6 +202,23 @@ export const flowBuckets = sqliteTable("flow_buckets", { ), ]) +export const flowIpMeta = sqliteTable("flow_ip_meta", { + prefix: text("prefix").primaryKey(), + asn: integer("asn").notNull().default(0), + country: text("country").notNull().default(""), + lat: real("lat"), + lng: real("lng"), + holder: text("holder").notNull().default(""), + ok: integer("ok").notNull().default(1), + fetchedAt: text("fetched_at").notNull(), +}) + +export const flowAsnMeta = sqliteTable("flow_asn_meta", { + asn: integer("asn").primaryKey(), + holder: text("holder").notNull().default(""), + fetchedAt: text("fetched_at").notNull(), +}) + export const trafficSamples = sqliteTable("traffic_samples", { id: integer("id").primaryKey({ autoIncrement: true }), serverId: integer("server_id") @@ -641,6 +658,8 @@ export type RecursiveRouteRow = typeof recursiveRoutes.$inferSelect export type TrafficSettingsRow = typeof trafficSettings.$inferSelect export type TrafficFlowSettingsRow = typeof trafficFlowSettings.$inferSelect export type FlowBucketRow = typeof flowBuckets.$inferSelect +export type FlowIpMetaRow = typeof flowIpMeta.$inferSelect +export type FlowAsnMetaRow = typeof flowAsnMeta.$inferSelect export type ServersApiPingSettingsRow = typeof serversApiPingSettings.$inferSelect export type TrafficSampleRow = typeof trafficSamples.$inferSelect export type UptimeSettingsRow = typeof uptimeSettings.$inferSelect diff --git a/backend/src/routes/traffic-flow.ts b/backend/src/routes/traffic-flow.ts index 661f705..81a6251 100644 --- a/backend/src/routes/traffic-flow.ts +++ b/backend/src/routes/traffic-flow.ts @@ -43,13 +43,20 @@ function parseId(raw: unknown): number | undefined { return Number.isFinite(n) ? n : undefined } +function parseDedup(raw: unknown): boolean { + if (raw == null || raw === "") return true + const s = String(raw).toLowerCase() + return s !== "0" && s !== "false" && s !== "off" +} + function analyticsQuery(req: FastifyRequest) { - const q = req.query as { range?: string; serverId?: string; userId?: string; iface?: string } + const q = req.query as { range?: string; serverId?: string; userId?: string; iface?: string; dedup?: string } return { minutes: rangeToMinutes(q.range), serverId: parseId(q.serverId), userId: q.userId?.trim() || undefined, iface: q.iface?.trim() || undefined, + dedup: parseDedup(q.dedup), } } diff --git a/backend/src/services/traffic-flow-analytics.test.ts b/backend/src/services/traffic-flow-analytics.test.ts index c2e2804..634d420 100644 --- a/backend/src/services/traffic-flow-analytics.test.ts +++ b/backend/src/services/traffic-flow-analytics.test.ts @@ -5,7 +5,19 @@ import { resetFlowRingsForTests, } from "./traffic-flow-ingest.js" import { buildFlowAnalytics } from "./traffic-flow-analytics.js" +import { disableCatalogFetchForTests, resetFlowCatalogForTests, seedFlowCatalogForTests } from "./traffic-flow-classify.js" +import { + disableRipeEnqueueForTests, + disableRipePersistForTests, + resetRipeCacheForTests, + seedRipeCacheForTests, +} from "./traffic-flow-ripe.js" +disableCatalogFetchForTests() +resetFlowCatalogForTests() +disableRipePersistForTests() +resetRipeCacheForTests() +disableRipeEnqueueForTests() resetIfaceCacheForTests() resetFlowRingsForTests() rememberServerIfaces(7, [ @@ -65,4 +77,92 @@ try { resetIfaceCacheForTests() } +resetFlowRingsForTests() +resetIfaceCacheForTests() +rememberServerIfaces(7, [ + { ".id": "*2", name: "ether1" }, + { ".id": "*A", name: "wg-flow" }, +]) +ingestParsedFlowsForServerForTests(7, [ + { + src: "10.1.1.8", + dst: "8.8.8.8", + proto: 6, + srcPort: 51234, + dstPort: 443, + bytes: 12_000, + packets: 10, + inIface: "2", + outIface: "10", + }, + { + src: "10.1.1.8", + dst: "8.8.8.8", + proto: 6, + srcPort: 51234, + dstPort: 443, + bytes: 9_000, + packets: 9, + inIface: "10", + outIface: "", + }, +]) +try { + const summed = buildFlowAnalytics({ minutes: 5, serverId: 7, dedup: false }) + assert.equal(summed.bytes, 21_000) + assert.equal(summed.conversations, 2) + const deduped = buildFlowAnalytics({ minutes: 5, serverId: 7, dedup: true }) + assert.equal(deduped.bytes, 12_000) + assert.equal(deduped.conversations, 1) + assert.equal(deduped.dedupApplied, true) + assert.equal(deduped.interfaces.length, 2) +} finally { + resetFlowRingsForTests() + resetIfaceCacheForTests() +} + +resetFlowRingsForTests() +resetIfaceCacheForTests() +disableRipeEnqueueForTests() +seedRipeCacheForTests({ + prefix: "8.8.8.0/24", + asn: 15169, + country: "US", + lat: 37.4, + lng: -122.1, + holder: "GOOGLE", + ok: true, + fetchedAt: Date.now(), +}) +seedFlowCatalogForTests({ + cidrs: [{ cidr: "8.8.8.0/24", purpose: "steam-gaming" }], +}) +rememberServerIfaces(7, [{ ".id": "*2", name: "ether1" }]) +ingestParsedFlowsForServerForTests(7, [ + { + src: "10.1.1.8", + dst: "8.8.8.8", + proto: 6, + srcPort: 51234, + dstPort: 443, + bytes: 12_000, + packets: 10, + inIface: "2", + outIface: "", + }, +]) +try { + const geo = buildFlowAnalytics({ minutes: 5, serverId: 7 }) + assert.equal(geo.categories?.[0]?.label, "Игры") + assert.ok(geo.asns?.some((r) => r.label.includes("AS15169"))) + assert.equal(geo.countries?.[0]?.id, "US") + assert.equal(geo.mapEdges?.[0]?.toCountry, "US") + assert.equal(geo.conversationsList[0]?.dstCountry, "US") +} finally { + resetFlowRingsForTests() + resetIfaceCacheForTests() + resetRipeCacheForTests() + resetFlowCatalogForTests() +} + console.log("traffic-flow-analytics.test.ts: ok") diff --git a/backend/src/services/traffic-flow-analytics.ts b/backend/src/services/traffic-flow-analytics.ts index 1bc72d6..08b790c 100644 --- a/backend/src/services/traffic-flow-analytics.ts +++ b/backend/src/services/traffic-flow-analytics.ts @@ -7,6 +7,7 @@ import type { FlowClientsDto, FlowEntityCard, FlowExportersDto, + FlowMapEdge, FlowTalkerDto, } from "@mmapp/contracts/traffic-flow" import { protoName } from "./traffic-flow-parse.js" @@ -19,12 +20,17 @@ import { import { resolveIfaceName } from "./traffic-flow-ifaces.js" import { getTrafficFlowSettingsRow, listHostPeers } from "./traffic-flow-settings.js" import { applicationName, flowRowMatchesFilter } from "./traffic-flow-apps.js" +import { dedupFlowRowsMaxBytes, flowTupleKey } from "./traffic-flow-dedup.js" +import { enqueueRipeMisses, lookupRipeCached } from "./traffic-flow-ripe.js" +import { classifyFlowDst, refreshFlowCatalogInBackground } from "./traffic-flow-classify.js" export interface FlowAnalyticsQuery { minutes: number serverId?: number userId?: string iface?: string + /** Default true: один 5-tuple = max байт по ifaces. */ + dedup?: boolean } function bpsToMbps(bps: number): number { @@ -89,6 +95,18 @@ function snapshotStatus(serverId: number): FlowEntityCard["status"] { return "online" } +function topLabel(map: Map, fallback = "—"): string { + let best = fallback + let bestBytes = 0 + for (const [label, v] of map) { + if (v.bytes > bestBytes) { + bestBytes = v.bytes + best = label + } + } + return best +} + export function buildFlowAnalytics(q: FlowAnalyticsQuery): FlowAnalyticsDto { const settings = getTrafficFlowSettingsRow() const top = Math.min(50, Math.max(10, settings.topN)) @@ -98,39 +116,70 @@ export function buildFlowAnalytics(q: FlowAnalyticsQuery): FlowAnalyticsDto { const allow = q.userId ? userIfaceAllow(q.userId) : null const serverRows = db.select().from(servers).all() const nameById = new Map(serverRows.map((s) => [s.id, s.name || s.host])) + const countryById = new Map(serverRows.map((s) => [s.id, (s.country || "").toUpperCase() || "UN"])) + const ifaceFilter = q.iface && q.iface !== "__all__" ? q.iface : "" + const wantDedup = q.dedup !== false && !ifaceFilter + + refreshFlowCatalogInBackground() const applications = new Map() const protocols = new Map() const sources = new Map() const destinations = new Map() const ifacesMap = new Map() + const asns = new Map() + const countries = new Map() + const categories = new Map() + const services = new Map() const conv = new Map() + const edgeAcc = new Map }>() const srcs = new Set() const dsts = new Set() - let totalBytes = 0 - let totalPackets = 0 const matched: PendingFlowRow[] = [] for (const r of raw) { const resolved = resolveIfaceName(r.serverId, r.inIface) if (!flowRowMatchesFilter(r, resolved.name, q, allow)) continue matched.push(r) - totalBytes += r.bytes - totalPackets += r.packets - srcs.add(r.src) - dsts.add(r.dst) - const app = applicationName(r.proto, r.dstPort, r.srcPort) - bump(applications, app, r.bytes, r.packets) - bump(protocols, protoName(r.proto), r.bytes, r.packets) - bump(sources, r.src, r.bytes, r.packets) - bump(destinations, r.dst, r.bytes, r.packets) + const ifaceKey = resolved.name const prevIf = ifacesMap.get(ifaceKey) ?? { bytes: 0, packets: 0, index: resolved.index } prevIf.bytes += r.bytes prevIf.packets += r.packets ifacesMap.set(ifaceKey, prevIf) + } - const ckey = `${r.serverId}|${r.src}|${r.dst}|${r.proto}|${r.srcPort}|${r.dstPort}|${r.inIface}` + const working = wantDedup ? dedupFlowRowsMaxBytes(matched) : matched + const conversationsRaw = new Set(matched.map((r) => `${flowTupleKey(r)}|${r.inIface}`)).size + + let totalBytes = 0 + let totalPackets = 0 + for (const r of working) { + const resolved = resolveIfaceName(r.serverId, r.inIface) + totalBytes += r.bytes + totalPackets += r.packets + srcs.add(r.src) + dsts.add(r.dst) + const app = applicationName(r.proto, r.dstPort, r.srcPort) + const ripe = lookupRipeCached(r.dst) + const classified = classifyFlowDst(r.dst, r.proto, r.dstPort, r.srcPort, ripe) + bump(applications, app, r.bytes, r.packets) + bump(protocols, protoName(r.proto), r.bytes, r.packets) + bump(sources, r.src, r.bytes, r.packets) + bump(destinations, r.dst, r.bytes, r.packets) + bump(categories, classified.category, r.bytes, r.packets) + bump(services, classified.service, r.bytes, r.packets) + if (ripe?.ok && ripe.asn) { + const asnLabel = ripe.holder ? `AS${ripe.asn} ${ripe.holder}` : `AS${ripe.asn}` + bump(asns, asnLabel, r.bytes, r.packets) + } + if (ripe?.ok && ripe.country && ripe.country !== "—") { + bump(countries, ripe.country, r.bytes, r.packets) + } + + const ckey = wantDedup + ? flowTupleKey(r) + : `${flowTupleKey(r)}|${r.inIface}` const prev = conv.get(ckey) if (prev) { prev.rawBytes += r.bytes @@ -152,30 +201,53 @@ export function buildFlowAnalytics(q: FlowAnalyticsQuery): FlowAnalyticsDto { inIface: resolved.name, inIfaceIndex: resolved.index, application: app, + category: classified.category, + service: classified.service, + dstCountry: ripe?.country && ripe.country !== "—" ? ripe.country : undefined, + dstAsn: ripe?.asn || undefined, rawBytes: r.bytes, }) } + + const toCountry = ripe?.ok && ripe.country && ripe.country !== "—" ? ripe.country : "" + if (toCountry) { + const fromCountry = countryById.get(r.serverId) || "UN" + const ekey = `${r.serverId}|${toCountry}` + let edge = edgeAcc.get(ekey) + if (!edge) { + edge = { + fromId: String(r.serverId), + fromLabel: nameById.get(r.serverId) ?? String(r.serverId), + fromCountry, + toCountry, + toAsn: ripe?.asn ?? 0, + category: classified.category, + bytes: 0, + bps: 0, + catBytes: new Map(), + } + edgeAcc.set(ekey, edge) + } + edge.bytes += r.bytes + if (ripe?.asn) edge.toAsn = ripe.asn + edge.catBytes.set(classified.category, (edge.catBytes.get(classified.category) ?? 0) + r.bytes) + } } + enqueueRipeMisses(dsts) + const conversationsList = [...conv.values()] .map((t) => ({ ...t, bps: (t.rawBytes * 8) / windowSec })) .sort((a, b) => b.bytes - a.bytes) .slice(0, top) .map(({ rawBytes: _raw, ...rest }) => rest) - let topProto = "—" - let topProtoBytes = 0 - for (const [label, v] of protocols) { - if (v.bytes > topProtoBytes) { - topProtoBytes = v.bytes - topProto = label - } - } + const topProto = topLabel(protocols) + const topCategory = topLabel(categories) - const ifaceFilter = q.iface && q.iface !== "__all__" ? q.iface : "__all__" const ringServer = q.serverId ?? (matched[0]?.serverId ?? 0) const ring = ringServer - ? getRingMbps(ringServer, ifaceFilter === "__all__" ? "__all__" : (ifacesMap.get(ifaceFilter)?.index || ifaceFilter)) + ? getRingMbps(ringServer, ifaceFilter === "" ? "__all__" : (ifacesMap.get(ifaceFilter)?.index || ifaceFilter)) : { rx: Array(60).fill(0) as number[], tx: Array(60).fill(0) as number[], rxNow: 0, txNow: 0 } const fromBuckets = seriesFromRows(matched, q.minutes) @@ -190,21 +262,46 @@ export function buildFlowAnalytics(q: FlowAnalyticsQuery): FlowAnalyticsDto { bps: (v.bytes * 8) / windowSec, })) - const protoBreakdown = topN(protocols, windowSec, top) + const ifaceRawBytes = [...ifacesMap.values()].reduce((a, v) => a + v.bytes, 0) || 1 const listener = getFlowListenerState() + const mapEdges: FlowMapEdge[] = [...edgeAcc.values()] + .map((e) => { + let cat = e.category + let catBest = 0 + for (const [label, bytes] of e.catBytes) { + if (bytes > catBest) { + catBest = bytes + cat = label + } + } + return { + fromId: e.fromId, + fromLabel: e.fromLabel, + fromCountry: e.fromCountry, + toCountry: e.toCountry, + toAsn: e.toAsn, + category: cat, + bytes: e.bytes, + bps: (e.bytes * 8) / windowSec, + } + }) + .sort((a, b) => b.bytes - a.bytes) + .slice(0, top) return { bpsNow: (ring.rxNow + ring.txNow) * 1_000_000 || (totalBytes * 8) / windowSec, bytes: totalBytes, packets: totalPackets, conversations: conv.size, + conversationsRaw, uniqueSrc: srcs.size, uniqueDst: dsts.size, topProto, + topCategory, rxSeries, txSeries, applications: topN(applications, windowSec, top), - protocols: protoBreakdown, + protocols: topN(protocols, windowSec, top), sources: topN(sources, windowSec, top), destinations: topN(destinations, windowSec, top), interfaces: [...ifacesMap.entries()].map(([label, v]) => ({ @@ -213,11 +310,17 @@ export function buildFlowAnalytics(q: FlowAnalyticsQuery): FlowAnalyticsDto { bytes: v.bytes, packets: v.packets, bps: (v.bytes * 8) / windowSec, - percent: totalBytes > 0 ? (v.bytes / totalBytes) * 100 : 0, + percent: (v.bytes / ifaceRawBytes) * 100, })).sort((a, b) => b.bytes - a.bytes), + asns: topN(asns, windowSec, top), + countries: topN(countries, windowSec, top), + categories: topN(categories, windowSec, top), + services: topN(services, windowSec, top), + mapEdges, conversationsList, ifaces: ifaceRows, live: listener.bound, + dedupApplied: wantDedup, } } diff --git a/backend/src/services/traffic-flow-classify.test.ts b/backend/src/services/traffic-flow-classify.test.ts new file mode 100644 index 0000000..495c453 --- /dev/null +++ b/backend/src/services/traffic-flow-classify.test.ts @@ -0,0 +1,20 @@ +import assert from "node:assert/strict" +import { classifyFlowDst, disableCatalogFetchForTests, resetFlowCatalogForTests, seedFlowCatalogForTests } from "./traffic-flow-classify.js" + +disableCatalogFetchForTests() +resetFlowCatalogForTests() +seedFlowCatalogForTests({ + cidrs: [{ cidr: "192.0.2.0/24", purpose: "steam-gaming" }], +}) + +const hit = classifyFlowDst("192.0.2.10", 6, 443, 50000, null) +assert.equal(hit.category, "Игры") +assert.equal(hit.service, "steam-gaming") + +const miss = classifyFlowDst("203.0.113.9", 17, 53, 53000, null) +assert.equal(miss.category, "DNS") + +const cdn = classifyFlowDst("203.0.113.9", 6, 443, 1, { prefix: "203.0.113.0/24", asn: 13335, country: "US", lat: null, lng: null, holder: "CLOUDFLARENET", ok: true, fetchedAt: Date.now() }) +assert.equal(cdn.category, "CDN") + +console.log("traffic-flow-classify.test.ts: ok") diff --git a/backend/src/services/traffic-flow-classify.ts b/backend/src/services/traffic-flow-classify.ts new file mode 100644 index 0000000..30e7ac7 --- /dev/null +++ b/backend/src/services/traffic-flow-classify.ts @@ -0,0 +1,140 @@ +import { db } from "../db/index.js" +import { evobgpSettings } from "../db/schema.js" +import { ipInCidrV4, parseCidrV4 } from "./traffic-flow-ip.js" +import type { FlowIpMeta } from "./traffic-flow-ripe.js" +import { applicationName } from "./traffic-flow-apps.js" + +export interface FlowClassification { + service: string + category: string +} + +interface CatalogCidr { + cidr: string + purpose: string + prefixLen: number +} + +const CATALOG_TTL_MS = 10 * 60_000 +let cidrs: CatalogCidr[] = [] +let asnPurpose = new Map() +let fetchedAt = 0 +let catalogFetchEnabled = true +let inflight: Promise | null = null + +export function disableCatalogFetchForTests(): void { + catalogFetchEnabled = false +} + +export function resetFlowCatalogForTests(): void { + cidrs = [] + asnPurpose = new Map() + fetchedAt = 0 + inflight = null +} + +export function seedFlowCatalogForTests(input: { + cidrs?: Array<{ cidr: string; purpose: string }> + asns?: Array<{ asn: number; purpose: string }> +}): void { + cidrs = (input.cidrs ?? []) + .map((c) => ({ cidr: c.cidr, purpose: c.purpose, prefixLen: parseCidrV4(c.cidr)?.prefixLen ?? 0 })) + .sort((a, b) => b.prefixLen - a.prefixLen) + asnPurpose = new Map((input.asns ?? []).map((a) => [a.asn, a.purpose])) + fetchedAt = Date.now() +} + +export function categoryFromPurpose(purpose: string, proto: number, dstPort: number, srcPort: number): string { + const p = purpose.toLowerCase() + if (/gaming|steam|epic|riot/.test(p)) return "Игры" + if (/streaming|youtube|netflix|twitch|video/.test(p)) return "Видео / стриминг" + if (/cdn|cloudflare|akamai|fastly/.test(p)) return "CDN" + if (/voip|discord|zoom/.test(p)) return "Голос" + const app = applicationName(proto, dstPort, srcPort) + if (app === "DNS" || app === "SSH" || app === "BGP") return app + return "Проче" +} + +function matchCidr(ip: string): CatalogCidr | null { + for (const row of cidrs) { + if (ipInCidrV4(ip, row.cidr)) return row + } + return null +} + +export function classifyFlowDst( + dst: string, + proto: number, + dstPort: number, + srcPort: number, + ripe: FlowIpMeta | null, +): FlowClassification { + const hit = matchCidr(dst) + const asnName = ripe?.asn ? asnPurpose.get(ripe.asn) : undefined + const purpose = hit?.purpose || asnName || ripe?.holder || "" + const service = (hit?.purpose || asnName || ripe?.holder || "Проче").trim() || "Проче" + return { + service, + category: categoryFromPurpose(purpose, proto, dstPort, srcPort), + } +} + +async function fetchCatalog(): Promise { + if (!catalogFetchEnabled) return + if (Date.now() - fetchedAt < CATALOG_TTL_MS) return + if (inflight) return inflight + inflight = (async () => { + try { + const row = db.select().from(evobgpSettings).limit(1).all()[0] + if (!row?.enabled) return + const root = String(row.baseUrl ?? "").replace(/\/+$/, "") + const token = String(row.apiKey ?? "").replace(/^Bearer\s+/i, "").trim() + if (!root || !token) return + const ac = new AbortController() + const t = setTimeout(() => ac.abort(), 20_000) + try { + const res = await fetch(`${root}/v1/router-lists/catalog`, { + headers: { Authorization: `Bearer ${token}`, Accept: "application/json" }, + signal: ac.signal, + }) + if (!res.ok) return + const catalog = await res.json() as { + modules?: { items?: Array<{ id: string; name: string }> } + ip_ranges?: { items?: Array<{ module_id: string; entry: { prefix: string } }> } + asns?: { items?: Array<{ module_id: string; entry: { asn: number } }> } + } + const mods = new Map((catalog.modules?.items ?? []).map((m) => [m.id, m.name])) + const next: CatalogCidr[] = [] + for (const item of catalog.ip_ranges?.items ?? []) { + const prefix = String(item.entry?.prefix ?? "").trim() + const purpose = mods.get(item.module_id) ?? "" + const parsed = parseCidrV4(prefix) + if (!prefix || !parsed) continue + next.push({ cidr: prefix, purpose, prefixLen: parsed.prefixLen }) + } + next.sort((a, b) => b.prefixLen - a.prefixLen) + const nextAsn = new Map() + for (const item of catalog.asns?.items ?? []) { + const purpose = mods.get(item.module_id) + const asn = Number(item.entry?.asn) + if (purpose && Number.isFinite(asn) && asn > 0) nextAsn.set(asn, purpose) + } + cidrs = next + asnPurpose = nextAsn + fetchedAt = Date.now() + } finally { + clearTimeout(t) + } + } catch { + /* catalog optional */ + } finally { + inflight = null + } + })() + return inflight +} + +/** Background refresh — analytics never awaits the HTTP. */ +export function refreshFlowCatalogInBackground(): void { + void fetchCatalog() +} diff --git a/backend/src/services/traffic-flow-dedup.test.ts b/backend/src/services/traffic-flow-dedup.test.ts new file mode 100644 index 0000000..0131c7d --- /dev/null +++ b/backend/src/services/traffic-flow-dedup.test.ts @@ -0,0 +1,25 @@ +import assert from "node:assert/strict" +import { dedupFlowRowsMaxBytes, flowTupleKey } from "./traffic-flow-dedup.js" + +const a = { + serverId: 7, + src: "10.1.1.8", + dst: "8.8.8.8", + proto: 6, + srcPort: 1, + dstPort: 443, + inIface: "2", + bytes: 12_000, + packets: 10, +} +const b = { ...a, inIface: "10", bytes: 8_000, packets: 8 } +const out = dedupFlowRowsMaxBytes([a, b]) +assert.equal(out.length, 1) +assert.equal(out[0]?.bytes, 12_000) +assert.equal(out[0]?.inIface, "2") +assert.equal(flowTupleKey(a), flowTupleKey(b)) + +const sameIface = dedupFlowRowsMaxBytes([a, { ...a, bytes: 3_000, packets: 2 }]) +assert.equal(sameIface[0]?.bytes, 15_000) + +console.log("traffic-flow-dedup.test.ts: ok") diff --git a/backend/src/services/traffic-flow-dedup.ts b/backend/src/services/traffic-flow-dedup.ts new file mode 100644 index 0000000..e06ca61 --- /dev/null +++ b/backend/src/services/traffic-flow-dedup.ts @@ -0,0 +1,44 @@ +export interface FlowTupleRow { + serverId: number + src: string + dst: string + proto: number + srcPort: number + dstPort: number + inIface: string + bytes: number + packets: number +} + +export function flowTupleKey(r: Pick): string { + return `${r.serverId}|${r.src}|${r.dst}|${r.proto}|${r.srcPort}|${r.dstPort}` +} + +function ifaceKey(r: FlowTupleRow): string { + return `${flowTupleKey(r)}|${r.inIface}` +} + +/** + * Один 5-tuple на двух ifIndex — это один поток: сначала сумма по бакетам/iface, + * затем max байт между интерфейсами (не sum). + */ +export function dedupFlowRowsMaxBytes(rows: T[]): T[] { + const byIface = new Map() + for (const row of rows) { + const key = ifaceKey(row) + const prev = byIface.get(key) + if (!prev) { + byIface.set(key, { ...row }) + continue + } + prev.bytes += row.bytes + prev.packets += row.packets + } + const byTuple = new Map() + for (const row of byIface.values()) { + const key = flowTupleKey(row) + const prev = byTuple.get(key) + if (!prev || row.bytes > prev.bytes) byTuple.set(key, row) + } + return [...byTuple.values()] +} diff --git a/backend/src/services/traffic-flow-ifaces.test.ts b/backend/src/services/traffic-flow-ifaces.test.ts index 80d9cff..3d04e97 100644 --- a/backend/src/services/traffic-flow-ifaces.test.ts +++ b/backend/src/services/traffic-flow-ifaces.test.ts @@ -27,6 +27,14 @@ assert.equal(resolveIfaceName(7, "0").name, "—") assert.equal(resolveIfaceName(7, "ether1").name, "ether1") assert.equal(resolveIfaceName(7, "99").name, "#99") +resetIfaceCacheForTests() +rememberServerIfaces(8, [ + { ifindex: "10", ".id": "*12", name: "gre1" }, +]) +assert.equal(rosIdToIfIndex("*12"), 18) +assert.equal(resolveIfaceName(8, "10").name, "gre1") +assert.equal(resolveIfaceName(8, "18").name, "gre1") + assert.equal(applicationName(6, 443), "HTTPS") assert.equal(applicationName(17, 53), "DNS") assert.equal(applicationName(6, 22), "SSH") diff --git a/backend/src/services/traffic-flow-ifaces.ts b/backend/src/services/traffic-flow-ifaces.ts index 44c401b..f2544ba 100644 --- a/backend/src/services/traffic-flow-ifaces.ts +++ b/backend/src/services/traffic-flow-ifaces.ts @@ -9,6 +9,7 @@ import { } from "./traffic-flow-ifindex.js" export { + ifaceCacheFresh, ifaceCacheHas, rememberServerIfaces, resetIfaceCacheForTests, diff --git a/backend/src/services/traffic-flow-ifindex.ts b/backend/src/services/traffic-flow-ifindex.ts index 0c00bc4..65df4c3 100644 --- a/backend/src/services/traffic-flow-ifindex.ts +++ b/backend/src/services/traffic-flow-ifindex.ts @@ -25,10 +25,9 @@ export function rememberServerIfaces(serverId: number, rows: RosIfaceIndexRow[]) const name = String(row.name ?? "").trim() if (!name) continue const fromProp = Number.parseInt(String(row.ifindex ?? ""), 10) - const idx = Number.isFinite(fromProp) && fromProp > 0 - ? fromProp - : rosIdToIfIndex(row[".id"]) - if (idx != null && idx > 0) map.set(idx, name) + const fromId = rosIdToIfIndex(row[".id"]) + if (Number.isFinite(fromProp) && fromProp > 0) map.set(fromProp, name) + if (fromId != null && fromId > 0) map.set(fromId, name) } cache.set(serverId, map) fetchedAt.set(serverId, Date.now()) diff --git a/backend/src/services/traffic-flow-ingest.ts b/backend/src/services/traffic-flow-ingest.ts index 3648709..02a71aa 100644 --- a/backend/src/services/traffic-flow-ingest.ts +++ b/backend/src/services/traffic-flow-ingest.ts @@ -11,7 +11,7 @@ import { recordFlowListenerError, recordFlowPacket, } from "./traffic-flow-settings.js" -import { ifaceCacheHas, refreshServerIfaces, resolveIfaceName } from "./traffic-flow-ifaces.js" +import { ifaceCacheFresh, refreshServerIfaces, resolveIfaceName } from "./traffic-flow-ifaces.js" import { applicationName } from "./traffic-flow-apps.js" export interface FlowListenerState { @@ -145,7 +145,9 @@ function resolveServerId(exporterIp: string): number | null { function queueFlows(exporterIp: string, flows: ParsedFlow[]): boolean { const serverId = resolveServerId(exporterIp) if (serverId == null) return false - if (!ifaceCacheHas(serverId)) void refreshServerIfaces(serverId) + const needsRefresh = !ifaceCacheFresh(serverId) + || flows.some((f) => resolveIfaceName(serverId, f.inIface).name.startsWith("#")) + if (needsRefresh) void refreshServerIfaces(serverId, true) const bucketAt = minuteBucketIso() for (const flow of flows) { addToTick(serverId, flow.inIface, flow.outIface, flow.bytes) diff --git a/backend/src/services/traffic-flow-ip.ts b/backend/src/services/traffic-flow-ip.ts new file mode 100644 index 0000000..b51d61a --- /dev/null +++ b/backend/src/services/traffic-flow-ip.ts @@ -0,0 +1,54 @@ +/** IPv4 helpers for RIPEstat prefix cache and EvoBGP CIDR match. */ + +export function ipv4ToInt(ip: string): number | null { + const parts = String(ip ?? "").trim().split(".") + if (parts.length !== 4) return null + let n = 0 + for (const p of parts) { + if (!/^\d+$/.test(p)) return null + const o = Number(p) + if (o < 0 || o > 255) return null + n = ((n << 8) >>> 0) + o + } + return n >>> 0 +} + +export function parseCidrV4(cidr: string): { net: number; mask: number; prefixLen: number } | null { + const raw = String(cidr ?? "").trim() + const [ip, lenRaw] = raw.split("/") + const addr = ipv4ToInt(ip ?? "") + const prefixLen = Number.parseInt(lenRaw ?? "", 10) + if (addr == null || !Number.isFinite(prefixLen) || prefixLen < 0 || prefixLen > 32) return null + const mask = prefixLen === 0 ? 0 : (0xffffffff << (32 - prefixLen)) >>> 0 + return { net: (addr & mask) >>> 0, mask, prefixLen } +} + +export function ipInCidrV4(ip: string, cidr: string): boolean { + const addr = ipv4ToInt(ip) + const parsed = parseCidrV4(cidr) + if (addr == null || !parsed) return false + return ((addr & parsed.mask) >>> 0) === parsed.net +} + +export function isNonPublicIp(ip: string): boolean { + const trimmed = String(ip ?? "").trim() + if (!trimmed) return true + if (trimmed.includes(":")) { + const lower = trimmed.toLowerCase() + return lower === "::1" || lower.startsWith("fe80:") || lower.startsWith("fc") || lower.startsWith("fd") || lower === "::" + } + const n = ipv4ToInt(trimmed) + if (n == null) return true + const inRange = (cidr: string) => ipInCidrV4(trimmed, cidr) + return ( + inRange("0.0.0.0/8") + || inRange("10.0.0.0/8") + || inRange("127.0.0.0/8") + || inRange("169.254.0.0/16") + || inRange("172.16.0.0/12") + || inRange("192.168.0.0/16") + || inRange("100.64.0.0/10") + || inRange("224.0.0.0/4") + || inRange("255.255.255.255/32") + ) +} diff --git a/backend/src/services/traffic-flow-parse.test.ts b/backend/src/services/traffic-flow-parse.test.ts index 7578b14..7fc3084 100644 --- a/backend/src/services/traffic-flow-parse.test.ts +++ b/backend/src/services/traffic-flow-parse.test.ts @@ -96,7 +96,7 @@ resetFlowTemplatesForTests() parseFlowPacket(tpl, "10.255.254.3") const named = parseFlowPacket(data, "10.255.254.3") assert.equal(named.length, 1) - assert.equal(named[0]?.inIface, "13") + assert.equal(named[0]?.inIface, "ether1") assert.equal(named[0]?.src, "10.1.1.8") } diff --git a/backend/src/services/traffic-flow-parse.ts b/backend/src/services/traffic-flow-parse.ts index 98c3822..a99cd26 100644 --- a/backend/src/services/traffic-flow-parse.ts +++ b/backend/src/services/traffic-flow-parse.ts @@ -206,7 +206,7 @@ function recordFromFields( } off = field.next } - if (!inIface && ifaceName) inIface = ifaceName + if (ifaceName) inIface = ifaceName return { flow: { src, dst, proto, srcPort, dstPort, bytes, packets, inIface, outIface }, next: off } } diff --git a/backend/src/services/traffic-flow-ripe.test.ts b/backend/src/services/traffic-flow-ripe.test.ts new file mode 100644 index 0000000..b91f2eb --- /dev/null +++ b/backend/src/services/traffic-flow-ripe.test.ts @@ -0,0 +1,72 @@ +import assert from "node:assert/strict" +import { + disableRipeEnqueueForTests, + disableRipePersistForTests, + enqueueRipeMisses, + flushRipeQueueForTests, + lookupRipeCached, + resetRipeCacheForTests, + ripeFetchCountForTests, + seedRipeCacheForTests, + setRipeFetchForTests, +} from "./traffic-flow-ripe.js" + +disableRipePersistForTests() +resetRipeCacheForTests() +disableRipeEnqueueForTests() + +assert.equal(lookupRipeCached("10.1.1.8")?.ok, false) +assert.equal(lookupRipeCached("192.168.0.1")?.ok, false) +assert.equal(lookupRipeCached("100.64.1.2")?.ok, false) +assert.equal(ripeFetchCountForTests(), 0) + +seedRipeCacheForTests({ + prefix: "1.2.3.0/24", + asn: 64500, + country: "NL", + lat: 52.3, + lng: 4.9, + holder: "TEST", + ok: true, + fetchedAt: Date.now(), +}) +assert.equal(lookupRipeCached("1.2.3.10")?.country, "NL") +assert.equal(lookupRipeCached("1.2.3.10")?.asn, 64500) +assert.equal(ripeFetchCountForTests(), 0) + +resetRipeCacheForTests() +disableRipePersistForTests() +setRipeFetchForTests(async (input) => { + const url = String(input) + const body = url.includes("network-info") + ? { data: { prefix: "8.8.8.0/24", asns: ["15169"] } } + : url.includes("maxmind-geo-lite") + ? { data: { located_resources: [{ locations: [{ country: "US", latitude: 37.4, longitude: -122.1 }] }] } } + : { data: { holder: "GOOGLE" } } + return new Response(JSON.stringify(body), { status: 200, headers: { "Content-Type": "application/json" } }) +}) +enqueueRipeMisses(["8.8.8.8"]) +await flushRipeQueueForTests() +assert.equal(lookupRipeCached("8.8.8.8")?.country, "US") +assert.equal(lookupRipeCached("8.8.8.10")?.prefix, "8.8.8.0/24") +const afterFirst = ripeFetchCountForTests() +assert.ok(afterFirst >= 2) +enqueueRipeMisses(["8.8.8.10"]) +await flushRipeQueueForTests() +assert.equal(ripeFetchCountForTests(), afterFirst) + +resetRipeCacheForTests() +disableRipePersistForTests() +setRipeFetchForTests(async () => { + throw new Error("timeout") +}) +enqueueRipeMisses(["203.0.113.50"]) +await flushRipeQueueForTests() +const neg = lookupRipeCached("203.0.113.50") +assert.equal(neg?.ok, false) +const afterNeg = ripeFetchCountForTests() +enqueueRipeMisses(["203.0.113.50"]) +await flushRipeQueueForTests() +assert.equal(ripeFetchCountForTests(), afterNeg) + +console.log("traffic-flow-ripe.test.ts: ok") diff --git a/backend/src/services/traffic-flow-ripe.ts b/backend/src/services/traffic-flow-ripe.ts new file mode 100644 index 0000000..53ab431 --- /dev/null +++ b/backend/src/services/traffic-flow-ripe.ts @@ -0,0 +1,371 @@ +import { sqliteDatabase } from "../db/index.js" +import { ipInCidrV4, ipv4ToInt, isNonPublicIp, parseCidrV4 } from "./traffic-flow-ip.js" + +export interface FlowIpMeta { + prefix: string + asn: number + country: string + lat: number | null + lng: number | null + holder: string + ok: boolean + fetchedAt: number +} + +const HIT_TTL_MS = 24 * 60 * 60_000 +const NEG_TTL_MS = 6 * 60 * 60_000 +const MAX_NEW_PREFIX_PER_MIN = 30 +const CONCURRENCY = 3 +const RIPE_BASE = "https://stat.ripe.net/data" +const UA = "MikrotikManager-flow/1.0" + +const mem = new Map() +const asnHolder = new Map() +const inflight = new Map>() +const queue: string[] = [] +const queued = new Set() +const recentFetches: number[] = [] + +let persistEnabled = true +let enqueueEnabled = true +let loaded = false +let workerRunning = false +let fetchImpl: typeof fetch = globalThis.fetch.bind(globalThis) +let fetchCount = 0 + +export function disableRipePersistForTests(): void { + persistEnabled = false +} + +export function disableRipeEnqueueForTests(): void { + enqueueEnabled = false +} + +export function resetRipeCacheForTests(): void { + mem.clear() + asnHolder.clear() + inflight.clear() + queue.length = 0 + queued.clear() + recentFetches.length = 0 + loaded = persistEnabled ? false : true + workerRunning = false + fetchCount = 0 + enqueueEnabled = true + fetchImpl = globalThis.fetch.bind(globalThis) +} + +export function seedRipeCacheForTests(entry: FlowIpMeta): void { + mem.set(entry.prefix, { ...entry }) + loaded = true +} + +export function setRipeFetchForTests(fn: typeof fetch): void { + fetchImpl = fn + fetchCount = 0 +} + +export function ripeFetchCountForTests(): number { + return fetchCount +} + +export async function flushRipeQueueForTests(timeoutMs = 4000): Promise { + const start = Date.now() + while (Date.now() - start < timeoutMs) { + if (!queue.length && !inflight.size && !workerRunning) return + await new Promise((r) => setTimeout(r, 20)) + } +} + +function ttlMs(ok: boolean): number { + return ok ? HIT_TTL_MS : NEG_TTL_MS +} + +function isFresh(entry: FlowIpMeta): boolean { + return Date.now() - entry.fetchedAt < ttlMs(entry.ok) +} + +function loadSqlite(): void { + if (loaded || !persistEnabled) { + loaded = true + return + } + loaded = true + try { + const rows = sqliteDatabase.prepare(` + SELECT prefix, asn, country, lat, lng, holder, ok, fetched_at + FROM flow_ip_meta + `).all() as Array<{ + prefix: string + asn: number | null + country: string + lat: number | null + lng: number | null + holder: string + ok: number + fetched_at: string + }> + for (const r of rows) { + const fetchedAt = Date.parse(r.fetched_at) + mem.set(r.prefix, { + prefix: r.prefix, + asn: Number(r.asn ?? 0) || 0, + country: r.country || "—", + lat: r.lat == null ? null : Number(r.lat), + lng: r.lng == null ? null : Number(r.lng), + holder: r.holder || "", + ok: r.ok !== 0, + fetchedAt: Number.isFinite(fetchedAt) ? fetchedAt : 0, + }) + } + const asns = sqliteDatabase.prepare(`SELECT asn, holder, fetched_at FROM flow_asn_meta`).all() as Array<{ + asn: number + holder: string + fetched_at: string + }> + for (const a of asns) { + const fetchedAt = Date.parse(a.fetched_at) + asnHolder.set(a.asn, { holder: a.holder || "", fetchedAt: Number.isFinite(fetchedAt) ? fetchedAt : 0 }) + } + } catch { + /* table may not exist in isolated tests */ + } +} + +function persist(entry: FlowIpMeta): void { + if (!persistEnabled) return + try { + sqliteDatabase.prepare(` + INSERT INTO flow_ip_meta (prefix, asn, country, lat, lng, holder, ok, fetched_at) + VALUES (@prefix, @asn, @country, @lat, @lng, @holder, @ok, @fetchedAt) + ON CONFLICT(prefix) DO UPDATE SET + asn=excluded.asn, country=excluded.country, lat=excluded.lat, lng=excluded.lng, + holder=excluded.holder, ok=excluded.ok, fetched_at=excluded.fetched_at + `).run({ + prefix: entry.prefix, + asn: entry.asn, + country: entry.country, + lat: entry.lat, + lng: entry.lng, + holder: entry.holder, + ok: entry.ok ? 1 : 0, + fetchedAt: new Date(entry.fetchedAt).toISOString(), + }) + } catch { + /* ignore persist errors */ + } +} + +function persistAsn(asn: number, holder: string): void { + if (!persistEnabled || !asn) return + try { + sqliteDatabase.prepare(` + INSERT INTO flow_asn_meta (asn, holder, fetched_at) + VALUES (@asn, @holder, @fetchedAt) + ON CONFLICT(asn) DO UPDATE SET holder=excluded.holder, fetched_at=excluded.fetched_at + `).run({ + asn, + holder, + fetchedAt: new Date().toISOString(), + }) + } catch { + /* ignore */ + } +} + +function negative(prefix: string): FlowIpMeta { + return { + prefix, + asn: 0, + country: "—", + lat: null, + lng: null, + holder: "", + ok: false, + fetchedAt: Date.now(), + } +} + +export function lookupRipeCached(ip: string): FlowIpMeta | null { + loadSqlite() + const trimmed = String(ip ?? "").trim() + if (!trimmed) return null + if (isNonPublicIp(trimmed)) { + return negative(`${trimmed.includes(":") ? trimmed : trimmed}/32`) + } + let best: FlowIpMeta | null = null + let bestLen = -1 + for (const entry of mem.values()) { + if (!isFresh(entry)) continue + const parsed = parseCidrV4(entry.prefix) + if (!parsed) continue + if (!ipInCidrV4(trimmed, entry.prefix)) continue + if (parsed.prefixLen > bestLen) { + best = entry + bestLen = parsed.prefixLen + } + } + return best +} + +async function ripeJson(path: string, resource: string): Promise { + fetchCount += 1 + const url = `${RIPE_BASE}/${path}/data.json?resource=${encodeURIComponent(resource)}` + const ac = new AbortController() + const t = setTimeout(() => ac.abort(), 12_000) + try { + const res = await fetchImpl(url, { + headers: { Accept: "application/json", "User-Agent": UA }, + signal: ac.signal, + }) + if (!res.ok) throw new Error(`HTTP ${res.status}`) + return await res.json() + } finally { + clearTimeout(t) + } +} + +function pickPrefix(data: unknown): string { + const d = data as { data?: { prefix?: string } } + return String(d?.data?.prefix ?? "").trim() +} + +function pickAsns(data: unknown): number { + const d = data as { data?: { asns?: unknown } } + const raw = d?.data?.asns + const first = Array.isArray(raw) ? raw[0] : raw + const n = Number.parseInt(String(first ?? "").replace(/^AS/i, ""), 10) + return Number.isFinite(n) ? n : 0 +} + +function pickGeo(data: unknown): { country: string; lat: number | null; lng: number | null } { + const d = data as { + data?: { + located_resources?: Array<{ + locations?: Array<{ country?: string; latitude?: number; longitude?: number }> + }> + } + } + const loc = d?.data?.located_resources?.[0]?.locations?.[0] + const country = String(loc?.country ?? "").trim().toUpperCase() + const lat = loc?.latitude == null ? null : Number(loc.latitude) + const lng = loc?.longitude == null ? null : Number(loc.longitude) + return { + country: country || "—", + lat: Number.isFinite(lat) ? lat : null, + lng: Number.isFinite(lng) ? lng : null, + } +} + +function pickHolder(data: unknown): string { + const d = data as { data?: { holder?: string } } + return String(d?.data?.holder ?? "").trim() +} + +function allowNewPrefix(): boolean { + const now = Date.now() + while (recentFetches.length && now - recentFetches[0]! > 60_000) recentFetches.shift() + return recentFetches.length < MAX_NEW_PREFIX_PER_MIN +} + +async function resolveIp(ip: string): Promise { + const cached = lookupRipeCached(ip) + if (cached) return cached + const pending = inflight.get(ip) + if (pending) return pending + + const job = (async () => { + if (!allowNewPrefix()) return null + recentFetches.push(Date.now()) + try { + const net = await ripeJson("network-info", ip) + const prefix = pickPrefix(net) || `${ip}/32` + const existing = mem.get(prefix) + if (existing && isFresh(existing)) return existing + const asn = pickAsns(net) + let geo = { country: "—", lat: null as number | null, lng: null as number | null } + try { + geo = pickGeo(await ripeJson("maxmind-geo-lite", prefix)) + } catch { + /* best-effort */ + } + let holder = asnHolder.get(asn)?.holder ?? "" + if (asn && (!holder || Date.now() - (asnHolder.get(asn)?.fetchedAt ?? 0) > HIT_TTL_MS)) { + try { + holder = pickHolder(await ripeJson("as-overview", `AS${asn}`)) + asnHolder.set(asn, { holder, fetchedAt: Date.now() }) + persistAsn(asn, holder) + } catch { + /* best-effort */ + } + } + const entry: FlowIpMeta = { + prefix, + asn, + country: geo.country, + lat: geo.lat, + lng: geo.lng, + holder, + ok: Boolean(asn || (geo.country && geo.country !== "—")), + fetchedAt: Date.now(), + } + mem.set(prefix, entry) + persist(entry) + return entry + } catch { + const prefix = `${ip}/32` + const entry = negative(prefix) + mem.set(prefix, entry) + persist(entry) + return entry + } finally { + inflight.delete(ip) + } + })() + + inflight.set(ip, job) + return job +} + +async function runWorker(): Promise { + if (workerRunning) return + workerRunning = true + try { + while (queue.length) { + const batch: string[] = [] + while (batch.length < CONCURRENCY && queue.length) { + const ip = queue.shift() + if (!ip) break + queued.delete(ip) + if (lookupRipeCached(ip)) continue + if (ipv4ToInt(ip) == null && !ip.includes(":")) continue + batch.push(ip) + } + if (!batch.length) { + if (!allowNewPrefix()) { + await new Promise((r) => setTimeout(r, 1000)) + } + continue + } + await Promise.all(batch.map((ip) => resolveIp(ip))) + } + } finally { + workerRunning = false + if (queue.length) void runWorker() + } +} + +/** HTTP / SSE never await this — cache miss is filled on a later tick. */ +export function enqueueRipeMisses(ips: Iterable): void { + if (!enqueueEnabled) return + loadSqlite() + for (const raw of ips) { + const ip = String(raw ?? "").trim() + if (!ip || isNonPublicIp(ip)) continue + if (lookupRipeCached(ip)) continue + if (queued.has(ip) || inflight.has(ip)) continue + queued.add(ip) + queue.push(ip) + } + if (queue.length) void runWorker() +} diff --git a/components/data-grids/traffic-flows-data-grid.tsx b/components/data-grids/traffic-flows-data-grid.tsx index 60b3327..a1ac52a 100644 --- a/components/data-grids/traffic-flows-data-grid.tsx +++ b/components/data-grids/traffic-flows-data-grid.tsx @@ -64,7 +64,14 @@ function TrafficFlowsDataGrid({ accessorFn: (r) => r.application ?? r.protoName, header: () => App, cell: ({ row }) => ( - {row.original.application ?? row.original.protoName} + + {row.original.application ?? row.original.protoName} + {row.original.category || row.original.service ? ( + + {[row.original.category, row.original.service].filter(Boolean).join(" · ")} + + ) : null} + ), meta: { headerClassName: DATA_GRID_CELL_PAD, cellClassName: DATA_GRID_CELL_PAD }, }, diff --git a/components/traffic/flow-analytics-panel.tsx b/components/traffic/flow-analytics-panel.tsx index 7d459d0..0df90d0 100644 --- a/components/traffic/flow-analytics-panel.tsx +++ b/components/traffic/flow-analytics-panel.tsx @@ -3,11 +3,13 @@ import { useMemo, useState } from "react" import { type ColumnDef, getCoreRowModel, useReactTable } from "@tanstack/react-table" import type { FlowAnalyticsDto, FlowBreakdownRow, FlowEntityCard } from "@mmapp/contracts/traffic-flow" -import { ArrowDownIcon, ArrowUpIcon, GitBranchIcon } from "lucide-react" +import { ArrowDownIcon, ArrowUpIcon, GitBranchIcon, GlobeIcon, LayersIcon, UsersIcon } from "lucide-react" import { Badge } from "@/components/reui/badge" import { KpiStatGrid } from "@/components/reui-kit/kpi-stat-grid" import { TrafficRxTxChart } from "@/components/reui-kit/traffic-rx-tx-chart" import { Tabs, TabsContent, TabsList, TabsTrigger } from "@/components/ui/tabs" +import { Switch } from "@/components/ui/switch" +import { Label } from "@/components/ui/label" import { DataGridShell } from "@/components/data-grids/shared/data-grid-shell" import { DATA_GRID_CELL_PAD, @@ -15,7 +17,7 @@ import { DATA_GRID_CELL_PAD_LAST, } from "@/components/data-grids/shared/data-grid-layout" import { TrafficFlowsDataGrid } from "@/components/data-grids/traffic-flows-data-grid" -import { Sparkline } from "@/components/sparkline" +import { FlowTrafficMap } from "@/components/traffic/flow-traffic-map" import { StatusDot } from "@/components/status-dot" import { Flag } from "@/components/flag" import { fmtRate } from "@/lib/fmt-rate" @@ -106,7 +108,15 @@ export function FlowEntityCardView({ ) } -function FlowBreakdownGrid({ rows, empty }: { rows: FlowBreakdownRow[]; empty?: string }) { +function FlowBreakdownGrid({ + rows, + empty, + country, +}: { + rows: FlowBreakdownRow[] + empty?: string + country?: boolean +}) { const columns = useMemo[]>( () => [ { @@ -115,7 +125,10 @@ function FlowBreakdownGrid({ rows, empty }: { rows: FlowBreakdownRow[]; empty?: header: () => Имя, cell: ({ row }) => (
- {row.original.label} + + {country ? : null} + {row.original.label} +
void selectedIface: string onIface: (name: string) => void + dedup: boolean + onDedup: (value: boolean) => void liveHint?: string emptyHint?: string }) { const [slice, setSlice] = useState("applications") + const [mapCountry, setMapCountry] = useState(null) const rxNow = analytics ? analytics.bpsNow / 1_000_000 : (card?.rxNow ?? 0) const bytes = analytics?.bytes ?? card?.bytes ?? 0 + const sessionRows = (analytics?.conversationsList ?? []).filter((row) => + mapCountry ? row.dstCountry === mapCountry : true, + ) if (!card) { return ( @@ -211,22 +232,34 @@ export function FlowAnalyticsDetail({ {card.site}
-
- {RANGE_KEYS.map((r) => ( - - ))} +
+
+ + +
+
+ {RANGE_KEYS.map((r) => ( + + ))} +
@@ -265,6 +298,7 @@ export function FlowAnalyticsDetail({ ) })}
+

по iface, без дедупа

) : null} @@ -295,11 +329,28 @@ export function FlowAnalyticsDetail({ }, { id: "flows", - label: "Разговоры", + label: "Сессии", value: String(analytics?.conversations ?? card.sessions), + hint: analytics?.conversationsRaw != null && analytics.conversationsRaw !== analytics.conversations + ? `до дедупа ${analytics.conversationsRaw}` + : undefined, icon: , iconClassName: "text-warning", }, + { + id: "uniq", + label: "Уник. src / dst", + value: `${analytics?.uniqueSrc ?? 0} / ${analytics?.uniqueDst ?? 0}`, + icon: , + iconClassName: "text-muted-foreground", + }, + { + id: "category", + label: "Топ категория", + value: analytics?.topCategory ?? "—", + icon: , + iconClassName: "text-primary", + }, ]} /> @@ -312,15 +363,41 @@ export function FlowAnalyticsDetail({ setSlice(String(v))} className="gap-3"> Приложения + Категории + Сервисы + ASN + Страны + Карта Протоколы Источники Назначения - Разговоры + Сессии Интерфейсы + + + + + + + + + + + + + + { + setMapCountry(iso) + setSlice("sessions") + }} + /> + @@ -330,40 +407,33 @@ export function FlowAnalyticsDetail({ - + + {mapCountry ? ( +
+ + фильтр страны {mapCountry} + +
+ ) : null} - + - -
-
-

Объём за период

-

{formatBytes(bytes)}

- -
-
-

Уник. адреса

-

- {analytics?.uniqueSrc ?? 0} - src · - {analytics?.uniqueDst ?? 0} - dst -

-
-
) } diff --git a/components/traffic/flow-traffic-map.tsx b/components/traffic/flow-traffic-map.tsx new file mode 100644 index 0000000..f7e9193 --- /dev/null +++ b/components/traffic/flow-traffic-map.tsx @@ -0,0 +1,184 @@ +"use client" + +import { useMemo, useState } from "react" +import { type ColumnDef, getCoreRowModel, useReactTable } from "@tanstack/react-table" +import type { FlowMapEdge } from "@mmapp/contracts/traffic-flow" +import { Frame, FramePanel } from "@/components/reui/frame" +import { DataGridShell } from "@/components/data-grids/shared/data-grid-shell" +import { + DATA_GRID_CELL_PAD, + DATA_GRID_CELL_PAD_FIRST, + DATA_GRID_CELL_PAD_LAST, +} from "@/components/data-grids/shared/data-grid-layout" +import { Flag } from "@/components/flag" +import { fmtRate } from "@/lib/fmt-rate" + +const W = 640 +const H = 280 +const STROKES = [ + "var(--chart-1)", + "var(--chart-2)", + "var(--chart-3)", + "var(--color-success)", + "var(--chart-5)", +] + +function formatBytes(n: number): string { + if (n >= 1_000_000_000) return `${(n / 1_000_000_000).toFixed(2)} ГБ` + if (n >= 1_000_000) return `${(n / 1_000_000).toFixed(1)} МБ` + if (n >= 1000) return `${(n / 1000).toFixed(1)} КБ` + return `${n} Б` +} + +function FlowTrafficMap({ + edges, + onSelectCountry, +}: { + edges: FlowMapEdge[] + onSelectCountry?: (country: string) => void +}) { + const [hover, setHover] = useState(null) + + const layout = useMemo(() => { + const sources = [...new Map(edges.map((e) => [e.fromId, e])).values()] + const dests = [...new Map(edges.map((e) => [e.toCountry, e])).values()] + const srcY = (i: number) => sources.length <= 1 ? H / 2 : 36 + (i * (H - 72)) / Math.max(1, sources.length - 1) + const dstY = (i: number) => dests.length <= 1 ? H / 2 : 36 + (i * (H - 72)) / Math.max(1, dests.length - 1) + const srcPos = new Map(sources.map((s, i) => [s.fromId, { x: 88, y: srcY(i), label: s.fromLabel, country: s.fromCountry }])) + const dstPos = new Map(dests.map((d, i) => [d.toCountry, { x: 552, y: dstY(i), country: d.toCountry }])) + const maxBytes = Math.max(...edges.map((e) => e.bytes), 1) + return { srcPos, dstPos, maxBytes } + }, [edges]) + + const columns = useMemo[]>( + () => [ + { + id: "from", + accessorKey: "fromLabel", + header: () => Источник, + cell: ({ row }) => ( + + {row.original.fromCountry && row.original.fromCountry !== "UN" + ? + : null} + {row.original.fromLabel} + + ), + meta: { headerClassName: DATA_GRID_CELL_PAD_FIRST, cellClassName: DATA_GRID_CELL_PAD_FIRST }, + }, + { + id: "to", + accessorKey: "toCountry", + header: () => Назначение, + cell: ({ row }) => ( + + + {row.original.toCountry} + {row.original.toAsn ? AS{row.original.toAsn} : null} + + ), + meta: { headerClassName: DATA_GRID_CELL_PAD, cellClassName: DATA_GRID_CELL_PAD }, + }, + { + id: "cat", + accessorKey: "category", + header: () => Категория, + cell: ({ row }) => {row.original.category}, + meta: { headerClassName: DATA_GRID_CELL_PAD, cellClassName: DATA_GRID_CELL_PAD }, + }, + { + id: "rate", + accessorFn: (r) => r.bps, + header: () => Скорость, + cell: ({ row }) => {fmtRate(row.original.bps / 1_000_000)}, + meta: { headerClassName: DATA_GRID_CELL_PAD, cellClassName: DATA_GRID_CELL_PAD }, + }, + { + id: "bytes", + accessorKey: "bytes", + header: () => Байты, + cell: ({ row }) => {formatBytes(row.original.bytes)}, + meta: { headerClassName: DATA_GRID_CELL_PAD_LAST, cellClassName: DATA_GRID_CELL_PAD_LAST }, + }, + ], + [], + ) + + const table = useReactTable({ + data: edges, + columns, + getCoreRowModel: getCoreRowModel(), + getRowId: (row, i) => `${row.fromId}-${row.toCountry}-${i}`, + }) + + if (!edges.length) { + return ( +

+ Страны подтянутся из кэша RIPEstat +

+ ) + } + + return ( +
+ + + + {edges.map((e, i) => { + const from = layout.srcPos.get(e.fromId) + const to = layout.dstPos.get(e.toCountry) + if (!from || !to) return null + const id = `${e.fromId}-${e.toCountry}` + const midX = (from.x + to.x) / 2 + const d = `M ${from.x} ${from.y} C ${midX} ${from.y}, ${midX} ${to.y}, ${to.x} ${to.y}` + const sw = 1.25 + 7 * (e.bytes / layout.maxBytes) + const active = hover === id + return ( + setHover(id)} + onMouseLeave={() => setHover(null)} + onClick={() => onSelectCountry?.(e.toCountry)} + > + + {`${e.fromLabel} → ${e.toCountry} · ${e.category} · ${formatBytes(e.bytes)}`} + + + ) + })} + {[...layout.srcPos.values()].map((n) => ( + + + + {n.label} + + + ))} + {[...layout.dstPos.values()].map((n) => ( + + + + {n.country} + + + ))} + + {hover ? ( +

+ Нажмите дугу, чтобы отфильтровать сессии по стране назначения +

+ ) : null} + + + +
+ ) +} + +export { FlowTrafficMap } diff --git a/hooks/use-flow-live.ts b/hooks/use-flow-live.ts index 4a7bdae..c479885 100644 --- a/hooks/use-flow-live.ts +++ b/hooks/use-flow-live.ts @@ -22,6 +22,7 @@ export function useFlowLive(opts: { serverId?: string userId?: string iface?: string + dedup?: boolean }): { sample: FlowAnalyticsDto | null; error: string | null } { const [sample, setSample] = useState(null) const [error, setError] = useState(null) @@ -41,6 +42,7 @@ export function useFlowLive(opts: { serverId: opts.serverId, userId: opts.userId, iface: opts.iface, + dedup: opts.dedup, })}` const url = resolveApiUrl(opts.backendUrl, path) @@ -84,7 +86,7 @@ export function useFlowLive(opts: { })() return () => ac.abort() - }, [opts.enabled, opts.backendUrl, opts.range, opts.serverId, opts.userId, opts.iface]) + }, [opts.enabled, opts.backendUrl, opts.range, opts.serverId, opts.userId, opts.iface, opts.dedup]) return { sample, error } } diff --git a/packages/contracts/src/traffic-flow.ts b/packages/contracts/src/traffic-flow.ts index e956903..6f39e47 100644 --- a/packages/contracts/src/traffic-flow.ts +++ b/packages/contracts/src/traffic-flow.ts @@ -81,6 +81,10 @@ export const flowTalkerDtoSchema = z.object({ inIface: z.string(), inIfaceIndex: z.string().optional(), application: z.string().optional(), + category: z.string().optional(), + service: z.string().optional(), + dstCountry: z.string().optional(), + dstAsn: z.number().int().optional(), }) export const flowStatsDtoSchema = z.object({ @@ -133,14 +137,27 @@ export const flowEntityCardSchema = z.object({ bytes: z.number().nonnegative(), }) +export const flowMapEdgeSchema = z.object({ + fromId: z.string(), + fromLabel: z.string(), + fromCountry: z.string(), + toCountry: z.string(), + toAsn: z.number().int().nonnegative(), + category: z.string(), + bytes: z.number().nonnegative(), + bps: z.number().nonnegative(), +}) + export const flowAnalyticsDtoSchema = z.object({ bpsNow: z.number().nonnegative(), bytes: z.number().nonnegative(), packets: z.number().nonnegative(), conversations: z.number().int().nonnegative(), + conversationsRaw: z.number().int().nonnegative().optional(), uniqueSrc: z.number().int().nonnegative(), uniqueDst: z.number().int().nonnegative(), topProto: z.string(), + topCategory: z.string().optional(), rxSeries: z.array(z.number()), txSeries: z.array(z.number()), applications: z.array(flowBreakdownRowSchema), @@ -148,9 +165,15 @@ export const flowAnalyticsDtoSchema = z.object({ sources: z.array(flowBreakdownRowSchema), destinations: z.array(flowBreakdownRowSchema), interfaces: z.array(flowBreakdownRowSchema), + asns: z.array(flowBreakdownRowSchema).optional(), + countries: z.array(flowBreakdownRowSchema).optional(), + categories: z.array(flowBreakdownRowSchema).optional(), + services: z.array(flowBreakdownRowSchema).optional(), + mapEdges: z.array(flowMapEdgeSchema).optional(), conversationsList: z.array(flowTalkerDtoSchema), ifaces: z.array(flowIfaceChipSchema), live: z.boolean(), + dedupApplied: z.boolean().optional(), }) export const flowExportersDtoSchema = z.object({ @@ -172,6 +195,7 @@ export type FlowStatsDto = z.infer export type FlowBreakdownRow = z.infer export type FlowIfaceChip = z.infer export type FlowEntityCard = z.infer +export type FlowMapEdge = z.infer export type FlowAnalyticsDto = z.infer export type FlowExportersDto = z.infer export type FlowClientsDto = z.infer diff --git a/shared/api/traffic-flow.ts b/shared/api/traffic-flow.ts index c110ac1..521653d 100644 --- a/shared/api/traffic-flow.ts +++ b/shared/api/traffic-flow.ts @@ -59,12 +59,15 @@ function flowQuery(params: { serverId?: string userId?: string iface?: string + dedup?: boolean }): string { const q = new URLSearchParams() if (params.range) q.set("range", params.range) if (params.serverId) q.set("serverId", params.serverId) if (params.userId) q.set("userId", params.userId) if (params.iface && params.iface !== "__all__") q.set("iface", params.iface) + if (params.dedup === false) q.set("dedup", "0") + else if (params.dedup === true) q.set("dedup", "1") const s = q.toString() return s ? `?${s}` : "" } @@ -79,7 +82,7 @@ export async function getFlowClients(baseUrl: string, range = "5m"): Promise { return requestJson(baseUrl, `/api/traffic/flow/analytics${flowQuery(params)}`) }