Compare commits

..
2 Commits
Author SHA1 Message Date
DenozordecandCursor 29d245cde3 feat(traffic): enhance flow analytics and brand classification
Docker images / prepare-release (push) Successful in 11s
Docker images / backend-image (push) Successful in 2m12s
Docker images / frontend-image (push) Successful in 4m3s
Docker images / notify-webhook (push) Skipped
Docker images / updater-image (push) Successful in 48s
Docker images / publish-release (push) Successful in 12s
- Introduced `pickInternetPeer` function to improve IP classification by selecting the appropriate public IP from source and destination.
- Updated `buildFlowAnalytics` and related functions to utilize the new peer selection logic, enhancing accuracy in flow analytics.
- Added tests for new classifications and ensured existing tests cover new scenarios for Google and Cloudflare.
- Refactored traffic flow brand mappings to include additional CIDR ranges for Google and Cloudflare.

Co-authored-by: Cursor <[email protected]>
2026-09-07 15:35:52 +07:00
DenozordecandCursor 7a491a325d fix(network-map): убрать тяжёлую классификацию из poll карты
Docker images / prepare-release (push) Successful in 9s
Docker images / backend-image (push) Successful in 1m53s
Docker images / frontend-image (push) Successful in 3m11s
Docker images / notify-webhook (push) Skipped
Docker images / updater-image (push) Successful in 49s
Docker images / publish-release (push) Successful in 12s
Не сканировать RIPE и каталог по каждой строке окна; сервисы по уникальным dst и кэшу ASN. Порог доли на карте настраиваемый и отключаемый.

Co-authored-by: Cursor <[email protected]>
2026-09-07 14:17:04 +07:00
21 changed files with 508 additions and 68 deletions
+17 -5
View File
@@ -938,6 +938,7 @@ export default function NetworkMapPage() {
const [mapHops, setMapHops] = useState<FlowMapHop[]>([])
const [mapServices, setMapServices] = useState<FlowMapService[]>([])
const [mapServiceEdges, setMapServiceEdges] = useState<FlowMapServiceEdge[]>([])
const [mapSharePct, setMapSharePct] = useState(5)
/** FQDN из GRE outer → IPv4 (ответ POST /api/network/resolve-hosts), для матчинга с WAN. */
const [greResolvedIpv4ByHost, setGreResolvedIpv4ByHost] = useState<Record<string, string>>({})
const [dataError, setDataError] = useState<string | null>(null)
@@ -1032,6 +1033,7 @@ export default function NetworkMapPage() {
setMapHops([])
setMapServices(MOCK_MAP_SERVICES)
setMapServiceEdges(MOCK_MAP_SERVICE_EDGES)
setMapSharePct(5)
setDataError(null)
})
return
@@ -1106,6 +1108,7 @@ export default function NetworkMapPage() {
setMapHops([])
setMapServices(MOCK_MAP_SERVICES)
setMapServiceEdges(MOCK_MAP_SERVICE_EDGES)
setMapSharePct(5)
})
return
}
@@ -1118,25 +1121,29 @@ export default function NetworkMapPage() {
return
}
let cancelled = false
let ac: AbortController | null = null
const tick = () => {
apiFetch<FlowMapHopsDto>("/api/traffic/flow/map-hops?range=5m")
ac?.abort()
ac = new AbortController()
apiFetch<FlowMapHopsDto>("/api/traffic/flow/map-hops?range=5m", { signal: ac.signal })
.then((res) => {
if (cancelled) return
setMapHops(res.hops ?? [])
setMapServices(res.services ?? [])
setMapServiceEdges(res.serviceEdges ?? [])
if (res.mapServiceMinSharePct != null) setMapSharePct(res.mapServiceMinSharePct)
})
.catch(() => {
.catch((err: unknown) => {
if (cancelled) return
setMapHops([])
setMapServices([])
setMapServiceEdges([])
const name = err instanceof Error ? err.name : ""
if (name === "AbortError") return
})
}
tick()
const id = window.setInterval(tick, 4000)
return () => {
cancelled = true
ac?.abort()
window.clearInterval(id)
}
}, [useLiveData, showNetflow, showServices, apiFetch])
@@ -1736,6 +1743,11 @@ export default function NetworkMapPage() {
</span>
</button>
))}
<p className="px-3 pt-1.5 pb-1 text-[10px] text-muted-foreground leading-snug">
{mapSharePct > 0
? `Порог доли сервиса ≥ ${mapSharePct}% · Настройки → NetFlow`
: "Порог доли выключен (все бренды, макс. 20) · Настройки → NetFlow"}
</p>
{(Object.keys(nodePositions).length > 0 || Object.keys(satPositions).length > 0) && (
<div className="border-t border-border/50 mt-1 pt-1">
<button
+1 -1
View File
@@ -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-dedup.test.ts && tsx src/services/traffic-flow-planes.test.ts && tsx src/services/traffic-flow-classify.test.ts && tsx src/services/traffic-flow-ripe.test.ts && tsx src/services/traffic-flow-brands.test.ts && tsx src/services/traffic-flow-ingest.test.ts && tsx src/services/traffic-flow-analytics.test.ts && tsx src/services/traffic-flow-map-hops.test.ts && tsx src/services/traffic-flow-hardening.test.ts && tsx src/services/traffic-flow-purge.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-planes.test.ts && tsx src/services/traffic-flow-ip.test.ts && tsx src/services/traffic-flow-classify.test.ts && tsx src/services/traffic-flow-ripe.test.ts && tsx src/services/traffic-flow-brands.test.ts && tsx src/services/traffic-flow-ingest.test.ts && tsx src/services/traffic-flow-analytics.test.ts && tsx src/services/traffic-flow-map-hops.test.ts && tsx src/services/traffic-flow-hardening.test.ts && tsx src/services/traffic-flow-purge.test.ts",
"test:users": "tsx src/modules/users/iface-type.test.ts && tsx src/modules/users/bindings.test.ts"
},
"dependencies": {
+8
View File
@@ -146,6 +146,7 @@ CREATE TABLE IF NOT EXISTS traffic_flow_settings (
hub_server_id INTEGER,
retention_hours INTEGER NOT NULL DEFAULT 24,
top_n INTEGER NOT NULL DEFAULT 200,
map_service_min_share_pct REAL NOT NULL DEFAULT 5,
last_datagram_at TEXT,
last_exporter_ip TEXT,
last_error TEXT,
@@ -834,6 +835,13 @@ SELECT 1, 0, '10.255.254.1', 4739, 51821, '10.255.254.0/24'
WHERE NOT EXISTS (SELECT 1 FROM traffic_flow_settings WHERE id = 1);
`)
{
const flowSettingsCols = sqlite.prepare(`PRAGMA table_info('traffic_flow_settings')`).all() as Array<{ name?: string }>
if (!flowSettingsCols.some((c) => c.name === "map_service_min_share_pct")) {
sqlite.exec(`ALTER TABLE traffic_flow_settings ADD COLUMN map_service_min_share_pct REAL NOT NULL DEFAULT 5`)
}
}
sqlite.exec(`
INSERT INTO uptime_settings (id, enabled, interval_sec, retention_days)
SELECT 1, 1, 15, 14
+1
View File
@@ -173,6 +173,7 @@ export const trafficFlowSettings = sqliteTable("traffic_flow_settings", {
hubServerId: integer("hub_server_id"),
retentionHours: integer("retention_hours").notNull().default(24),
topN: integer("top_n").notNull().default(200),
mapServiceMinSharePct: real("map_service_min_share_pct").notNull().default(5),
lastDatagramAt: text("last_datagram_at"),
lastExporterIp: text("last_exporter_ip"),
lastError: text("last_error"),
@@ -384,6 +384,51 @@ try {
}
}
{
resetFlowRingsForTests()
resetIfaceCacheForTests()
resetRipeCacheForTests()
disableRipeEnqueueForTests()
rememberServerIfaces(7, [{ ".id": "*2", name: "ether1" }])
ingestParsedFlowsForServerForTests(7, [
{
src: "173.194.151.65",
dst: "10.200.100.53",
proto: 6,
srcPort: 443,
dstPort: 57182,
bytes: 12_000,
packets: 10,
inIface: "2",
outIface: "2",
},
{
src: "104.18.35.51",
dst: "10.200.100.53",
proto: 6,
srcPort: 443,
dstPort: 53880,
bytes: 3_000,
packets: 4,
inIface: "2",
outIface: "2",
},
])
try {
const rev = buildFlowAnalytics({ minutes: 5, serverId: 7 })
const google = rev.conversationsList.find((r) => r.src === "173.194.151.65")
const cf = rev.conversationsList.find((r) => r.src === "104.18.35.51")
assert.equal(google?.service, "Google")
assert.equal(google?.category, "Веб")
assert.equal(cf?.service, "Cloudflare")
assert.equal(cf?.category, "CDN")
} finally {
resetFlowRingsForTests()
resetIfaceCacheForTests()
resetRipeCacheForTests()
}
}
{
resetFlowRingsForTests()
resetIfaceCacheForTests()
@@ -30,6 +30,7 @@ import { enqueueRipeMisses, lookupRipeCached } from "./traffic-flow-ripe.js"
import { classifyFlowDst, refreshFlowCatalogInBackground } from "./traffic-flow-classify.js"
import { isIsoCountry } from "./traffic-flow-brands.js"
import { classifyFlowPlane, flowBps, shouldKeepPlane } from "./traffic-flow-planes.js"
import { pickInternetPeer } from "./traffic-flow-ip.js"
import {
enGreIfaceNames,
latestWireBps,
@@ -171,6 +172,7 @@ export function buildFlowAnalytics(q: FlowAnalyticsQuery): FlowAnalyticsDto {
const pathAcc = new Map<string, FlowPathRow>()
const srcs = new Set<string>()
const dsts = new Set<string>()
const peers = new Set<string>()
const matched: PendingFlowRow[] = []
const skipHeavy = Boolean(q.skipHeavy)
let bytesPayload = 0
@@ -217,9 +219,11 @@ export function buildFlowAnalytics(q: FlowAnalyticsQuery): FlowAnalyticsDto {
totalPackets += r.packets
srcs.add(r.src)
dsts.add(r.dst)
const peer = pickInternetPeer(r.src, r.dst, r.srcPort, r.dstPort)
peers.add(peer)
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)
const ripe = lookupRipeCached(peer)
const classified = classifyFlowDst(peer, 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)
@@ -345,7 +349,7 @@ export function buildFlowAnalytics(q: FlowAnalyticsQuery): FlowAnalyticsDto {
}
}
enqueueRipeMisses(dsts)
enqueueRipeMisses(peers)
const conversationsList = [...conv.values()]
.map((t) => {
@@ -27,6 +27,9 @@ assert.equal(brandByAsn(16509)?.service, "AWS")
assert.equal(brandByAsn(57976)?.service, "Blizzard")
assert.equal(brandByAsn(401115)?.service, "ChatGPT")
assert.equal(lookupBrand("1.1.1.1", 13335)?.service, "Cloudflare")
assert.equal(lookupBrand("104.18.35.51", 0)?.service, "Cloudflare")
assert.equal(lookupBrand("173.194.151.65", 0)?.service, "Google")
assert.equal(lookupBrand("8.8.8.8", 0)?.service, "Google")
assert.equal(lookupBrand("203.0.113.9", 64500), null)
assert.equal(OTHER_SERVICE, "Прочее")
assert.equal(isNamedInternetService("Google", "Веб"), true)
+16 -6
View File
@@ -56,13 +56,23 @@ const ASN_HQ_COUNTRY = new Map<number, string>([
[211157, "NL"],
])
const GOOGLE: BrandHit = { service: "Google", category: "Веб" }
const CLOUDFLARE: BrandHit = { service: "Cloudflare", category: "CDN" }
const YOUTUBE: BrandHit = { service: "YouTube", category: "Видео / стриминг" }
const CIDR_BRANDS: Array<{ cidr: string; prefixLen: number; hit: BrandHit }> = [
{ cidr: "104.16.0.0/13", prefixLen: 13, hit: { service: "Cloudflare", category: "CDN" } },
{ cidr: "104.24.0.0/14", prefixLen: 14, hit: { service: "Cloudflare", category: "CDN" } },
{ cidr: "172.64.0.0/13", prefixLen: 13, hit: { service: "Cloudflare", category: "CDN" } },
{ cidr: "162.158.0.0/15", prefixLen: 15, hit: { service: "Cloudflare", category: "CDN" } },
{ cidr: "208.65.152.0/22", prefixLen: 22, hit: { service: "YouTube", category: "Видео / стриминг" } },
{ cidr: "208.117.224.0/19", prefixLen: 19, hit: { service: "YouTube", category: "Видео / стриминг" } },
{ cidr: "104.16.0.0/13", prefixLen: 13, hit: CLOUDFLARE },
{ cidr: "104.24.0.0/14", prefixLen: 14, hit: CLOUDFLARE },
{ cidr: "172.64.0.0/13", prefixLen: 13, hit: CLOUDFLARE },
{ cidr: "162.158.0.0/15", prefixLen: 15, hit: CLOUDFLARE },
{ cidr: "8.8.8.0/24", prefixLen: 24, hit: GOOGLE },
{ cidr: "8.8.4.0/24", prefixLen: 24, hit: GOOGLE },
{ cidr: "173.194.0.0/16", prefixLen: 16, hit: GOOGLE },
{ cidr: "172.217.0.0/16", prefixLen: 16, hit: GOOGLE },
{ cidr: "74.125.0.0/16", prefixLen: 16, hit: GOOGLE },
{ cidr: "142.250.0.0/15", prefixLen: 15, hit: GOOGLE },
{ cidr: "208.65.152.0/22", prefixLen: 22, hit: YOUTUBE },
{ cidr: "208.117.224.0/19", prefixLen: 19, hit: YOUTUBE },
].sort((a, b) => b.prefixLen - a.prefixLen)
const NON_ISO = new Set(["EU", "AP", "ZZ", "XX", "A1", "A2", "O1"])
@@ -36,6 +36,10 @@ const google = classifyFlowDst("173.194.160.163", 6, 443, 1, {
assert.equal(google.service, "Google")
assert.equal(google.category, "Веб")
const googleCidr = classifyFlowDst("173.194.151.65", 6, 57182, 443, null)
assert.equal(googleCidr.service, "Google")
assert.equal(googleCidr.category, "Веб")
const youtube = classifyFlowDst("173.194.160.163", 6, 443, 1, {
prefix: "173.194.0.0/16",
asn: 15169,
+5 -3
View File
@@ -7,6 +7,7 @@ import { classifyFlowDst } from "./traffic-flow-classify.js"
import { enqueueRipeMisses, lookupRipeCached } from "./traffic-flow-ripe.js"
import { isIsoCountry } from "./traffic-flow-brands.js"
import { maybeRefreshIfaces } from "./traffic-flow-ifaces.js"
import { pickInternetPeer } from "./traffic-flow-ip.js"
type SqliteHandle = InstanceType<typeof Database>
@@ -239,9 +240,10 @@ export function queueParsedFlows(serverId: number, flows: ParsedFlowInput[]): vo
const flow = normalizeParsedFlow(raw)
addToTick(serverId, flow, flow.bytes)
bumpRollup(serverId, bucketAt, flow, flow.bytes, flow.packets)
const ripe = lookupRipeCached(flow.dst)
if (flow.dst && !ripe) ripeMisses.push(flow.dst)
const classified = classifyFlowDst(flow.dst, flow.proto, flow.dstPort, flow.srcPort, ripe)
const peer = pickInternetPeer(flow.src, flow.dst, flow.srcPort, flow.dstPort)
const ripe = lookupRipeCached(peer)
if (peer && !ripe) ripeMisses.push(peer)
const classified = classifyFlowDst(peer, flow.proto, flow.dstPort, flow.srcPort, ripe)
const app = applicationName(flow.proto, flow.dstPort, flow.srcPort)
const country = ripe?.ok && isIsoCountry(ripe.country)
? ripe.country
@@ -0,0 +1,25 @@
import assert from "node:assert/strict"
import { isNonPublicIp, pickInternetPeer } from "./traffic-flow-ip.js"
assert.equal(isNonPublicIp("10.200.100.53"), true)
assert.equal(isNonPublicIp("173.194.151.65"), false)
assert.equal(
pickInternetPeer("173.194.151.65", "10.200.100.53", 443, 57182),
"173.194.151.65",
"reverse IPFIX: Google:443 → RFC1918",
)
assert.equal(
pickInternetPeer("10.200.100.53", "104.18.35.51", 53880, 443),
"104.18.35.51",
"client → Cloudflare:443",
)
assert.equal(pickInternetPeer("10.100.1.17", "8.8.8.8", 51234, 443), "8.8.8.8")
assert.equal(
pickInternetPeer("1.1.1.1", "8.8.8.8", 443, 51234),
"1.1.1.1",
"оба публичные — сторона с well-known портом",
)
assert.equal(pickInternetPeer("10.1.1.1", "10.2.2.2", 443, 80), "10.2.2.2")
console.log("traffic-flow-ip.test.ts: ok")
+20
View File
@@ -52,3 +52,23 @@ export function isNonPublicIp(ip: string): boolean {
|| inRange("255.255.255.255/32")
)
}
const PEER_WELL_KNOWN_PORTS = new Set([80, 443, 53, 853])
/**
* Интернет-сторона потока: у IPFIX сервис часто в src (Google:443 RFC1918:ephemeral).
* Классифицировать этот IP, не слепой dst.
*/
export function pickInternetPeer(src: string, dst: string, srcPort: number, dstPort: number): string {
const srcPub = !isNonPublicIp(src)
const dstPub = !isNonPublicIp(dst)
if (srcPub && !dstPub) return src
if (dstPub && !srcPub) return dst
if (srcPub && dstPub) {
const srcWk = PEER_WELL_KNOWN_PORTS.has(srcPort)
const dstWk = PEER_WELL_KNOWN_PORTS.has(dstPort)
if (srcWk && !dstWk) return src
if (dstWk && !srcWk) return dst
}
return dst
}
@@ -4,7 +4,7 @@ import {
ingestParsedFlowsForServerForTests,
resetFlowRingsForTests,
} from "./traffic-flow-ingest.js"
import { buildFlowMapHops } from "./traffic-flow-map-hops.js"
import { buildFlowMapHops, resetFlowMapHopsCacheForTests } from "./traffic-flow-map-hops.js"
import { seedFlowTopologyForTests, type FlowTopology } from "./traffic-flow-topology.js"
import { disableCatalogFetchForTests, resetFlowCatalogForTests } from "./traffic-flow-classify.js"
import {
@@ -114,6 +114,7 @@ ingestParsedFlowsForServerForTests(3, [
])
try {
resetFlowMapHopsCacheForTests()
const def = buildFlowMapHops({ minutes: 5 })
assert.equal(def.excludeOverlayApplied, true)
assert.equal(def.excludeMeshApplied, true)
@@ -142,6 +143,7 @@ try {
assert.ok(wan, "WAN hop from home-router")
assert.equal(wan.bytes, 3000)
resetFlowMapHopsCacheForTests()
const withAll = buildFlowMapHops({ minutes: 5, excludeOverlay: false, excludeMesh: false })
const overlayIface = withAll.hops.find((h) => h.iface === "gre-jh-en" && h.fromId === "7")
assert.ok(overlayIface && overlayIface.bytes >= 5_000_000)
@@ -200,7 +202,8 @@ ingestParsedFlowsForServerForTests(7, [
payloadFlow("203.0.113.50", 9400),
])
try {
const six = buildFlowMapHops({ minutes: 5 })
resetFlowMapHopsCacheForTests()
const six = buildFlowMapHops({ minutes: 5, minSharePct: 5 })
assert.equal(six.totalBytes, 10_000)
const google = six.services?.find((s) => s.id === "svc:google")
assert.ok(google, "Google ≥ 5%")
@@ -227,9 +230,13 @@ ingestParsedFlowsForServerForTests(7, [
payloadFlow("203.0.113.50", 9600),
])
try {
const four = buildFlowMapHops({ minutes: 5 })
resetFlowMapHopsCacheForTests()
const four = buildFlowMapHops({ minutes: 5, minSharePct: 5 })
assert.equal(four.totalBytes, 10_000)
assert.ok(!(four.services ?? []).some((s) => s.id === "svc:google"), "Google < 5% hidden")
resetFlowMapHopsCacheForTests()
const off = buildFlowMapHops({ minutes: 5, minSharePct: 0 })
assert.ok(off.services?.some((s) => s.id === "svc:google"), "порог 0 показывает Google 4%")
} finally {
resetFlowRingsForTests()
resetIfaceCacheForTests()
@@ -260,7 +267,8 @@ ingestParsedFlowsForServerForTests(7, [
payloadFlow("203.0.113.50", 1000),
])
try {
const greOnly = buildFlowMapHops({ minutes: 5, excludeOverlay: false })
resetFlowMapHopsCacheForTests()
const greOnly = buildFlowMapHops({ minutes: 5, excludeOverlay: false, minSharePct: 0 })
assert.ok(!(greOnly.services ?? []).some((s) => s.label === "GRE"), "GRE is not a destination service")
} finally {
seedFlowTopologyForTests(null)
@@ -270,4 +278,53 @@ try {
resetFlowCatalogForTests()
}
resetFlowRingsForTests()
resetIfaceCacheForTests()
resetRipeCacheForTests()
disableRipeEnqueueForTests()
seedFlowTopologyForTests(topo)
rememberServerIfaces(7, [
{ ".id": "*2", name: "gre-client" },
{ ".id": "*3", name: "gre-jh-en" },
])
ingestParsedFlowsForServerForTests(7, [
{
src: "173.194.151.65",
dst: "10.200.100.53",
proto: 6,
srcPort: 443,
dstPort: 57182,
bytes: 9_000,
packets: 90,
inIface: "2",
outIface: "3",
nextHop: "198.51.100.1",
},
{
src: "104.18.35.51",
dst: "10.200.100.53",
proto: 6,
srcPort: 443,
dstPort: 53880,
bytes: 1_000,
packets: 10,
inIface: "2",
outIface: "3",
nextHop: "198.51.100.1",
},
])
try {
resetFlowMapHopsCacheForTests()
const rev = buildFlowMapHops({ minutes: 5, minSharePct: 0 })
assert.ok(rev.services?.some((s) => s.id === "svc:google"), "реверс Google:443 → 10.x")
assert.ok(rev.services?.some((s) => s.id === "svc:cloudflare"), "реверс Cloudflare:443 → 10.x")
assert.ok(rev.serviceEdges?.some((e) => e.toId === "svc:google" && e.fromId === "9"))
} finally {
seedFlowTopologyForTests(null)
resetFlowRingsForTests()
resetIfaceCacheForTests()
resetRipeCacheForTests()
resetFlowCatalogForTests()
}
console.log("traffic-flow-map-hops.test.ts: ok")
+129 -32
View File
@@ -2,20 +2,24 @@ import { eq } from "drizzle-orm"
import type { FlowMapHop, FlowMapHopsDto, FlowMapService, FlowMapServiceEdge } from "@mmapp/contracts/traffic-flow"
import { db } from "../db/index.js"
import { servers, userInterfaceBindings } from "../db/schema.js"
import { flowRowMatchesFilter } from "./traffic-flow-apps.js"
import { applicationName, flowRowMatchesFilter } from "./traffic-flow-apps.js"
import {
isNamedInternetService,
lookupBrand,
mapServiceNodeId,
} from "./traffic-flow-brands.js"
import { classifyFlowDst, refreshFlowCatalogInBackground } from "./traffic-flow-classify.js"
import { dedupFlowRowsMaxBytes } from "./traffic-flow-dedup.js"
import { getFlowListenerState, listFlowRowsForWindow } from "./traffic-flow-ingest.js"
import { resolveIfaceName } from "./traffic-flow-ifaces.js"
import { classifyFlowPlane, shouldKeepPlane } from "./traffic-flow-planes.js"
import { enqueueRipeMisses, lookupRipeCached } from "./traffic-flow-ripe.js"
import { pickInternetPeer } from "./traffic-flow-ip.js"
import { lookupRipeCached, type FlowIpMeta } from "./traffic-flow-ripe.js"
import { getTrafficFlowSettingsRow } from "./traffic-flow-settings.js"
import { loadFlowTopology, resolveEn } from "./traffic-flow-topology.js"
export const MAP_SERVICE_SHARE_THRESHOLD = 0.05
export const DEFAULT_MAP_SERVICE_MIN_SHARE_PCT = 5
export const MAP_SERVICE_NODE_CAP = 20
const HOPS_CACHE_TTL_MS = 2000
export interface FlowMapHopsQuery {
minutes: number
@@ -25,6 +29,8 @@ export interface FlowMapHopsQuery {
dedup?: boolean
excludeMesh?: boolean
excludeOverlay?: boolean
/** Переопределение порога (тесты). Иначе из настроек NetFlow. */
minSharePct?: number
}
interface HopAcc {
@@ -39,6 +45,39 @@ interface HopAcc {
bytesRev: number
}
interface DstAcc {
bytes: number
proto: number
dstPort: number
srcPort: number
fromBytes: Map<string, number>
}
let hopsCache: { key: string; at: number; dto: FlowMapHopsDto } | null = null
export function resetFlowMapHopsCacheForTests(): void {
hopsCache = null
}
export function clampMapServiceMinSharePct(n: unknown): number {
const v = typeof n === "number" ? n : Number(n)
if (!Number.isFinite(v)) return DEFAULT_MAP_SERVICE_MIN_SHARE_PCT
return Math.min(100, Math.max(0, v))
}
function hopsQueryKey(q: FlowMapHopsQuery, minSharePct: number): string {
return JSON.stringify({
minutes: q.minutes,
serverId: q.serverId ?? null,
userId: q.userId ?? null,
iface: q.iface ?? null,
dedup: q.dedup !== false,
excludeMesh: q.excludeMesh !== false,
excludeOverlay: q.excludeOverlay !== false,
minSharePct,
})
}
function userIfaceAllow(userId: string): Map<number, Set<string>> | null {
if (!userId) return null
const binds = db.select().from(userInterfaceBindings).where(eq(userInterfaceBindings.userId, userId)).all()
@@ -89,8 +128,36 @@ function toHop(a: HopAcc, windowSec: number): FlowMapHop {
}
}
/** Hop-rates для карты сети: те же фильтры, что у общего NetFlow (dedup / mesh / overlay). */
export function buildFlowMapHops(q: FlowMapHopsQuery): FlowMapHopsDto {
/** Имя бренда без каталога EvoBGP — только ASN/CIDR кэш + proto. */
function classifyMapDstLite(
dst: string,
proto: number,
dstPort: number,
srcPort: number,
ripe: FlowIpMeta | null,
): { service: string; category: string } | null {
if (proto === 47 || proto === 50) return null
const app = applicationName(proto, dstPort, srcPort)
if (app === "WireGuard" || app === "DNS" || app === "SSH" || app === "BGP") return null
if (/youtube/i.test(ripe?.holder ?? "")) {
return { service: "YouTube", category: "Видео / стриминг" }
}
const brand = lookupBrand(dst, ripe?.asn ?? 0)
if (!brand || !isNamedInternetService(brand.service, brand.category)) return null
return brand
}
function resolveMinSharePct(q: FlowMapHopsQuery): number {
if (q.minSharePct != null) return clampMapServiceMinSharePct(q.minSharePct)
try {
const row = getTrafficFlowSettingsRow() as { mapServiceMinSharePct?: number }
return clampMapServiceMinSharePct(row.mapServiceMinSharePct ?? DEFAULT_MAP_SERVICE_MIN_SHARE_PCT)
} catch {
return DEFAULT_MAP_SERVICE_MIN_SHARE_PCT
}
}
function buildFlowMapHopsUncached(q: FlowMapHopsQuery, minSharePct: number): FlowMapHopsDto {
const windowSec = Math.max(60, q.minutes * 60)
const raw = listFlowRowsForWindow(q.minutes)
const allow = q.userId ? userIfaceAllow(q.userId) : null
@@ -122,13 +189,9 @@ export function buildFlowMapHops(q: FlowMapHopsQuery): FlowMapHopsDto {
const working = wantDedup ? dedupFlowRowsMaxBytes(matched) : matched
const hops = new Map<string, HopAcc>()
const svcTotals = new Map<string, { label: string; category: string; bytes: number }>()
const svcEdges = new Map<string, { fromId: string; toId: string; bytes: number; bytesFwd: number; bytesRev: number }>()
const dsts = new Set<string>()
const dstAcc = new Map<string, DstAcc>()
let totalBytes = 0
refreshFlowCatalogInBackground()
for (const r of working) {
const inRes = resolveIfaceName(r.serverId, r.inIface)
const outRes = resolveIfaceName(r.serverId, r.outIface)
@@ -215,37 +278,54 @@ export function buildFlowMapHops(q: FlowMapHopsQuery): FlowMapHopsDto {
}
totalBytes += r.bytes
dsts.add(r.dst)
const ripe = lookupRipeCached(r.dst)
const classified = classifyFlowDst(r.dst, r.proto, r.dstPort, r.srcPort, ripe)
if (isNamedInternetService(classified.service, classified.category)) {
const toId = mapServiceNodeId(classified.service)
const prevSvc = svcTotals.get(toId)
if (prevSvc) prevSvc.bytes += r.bytes
else svcTotals.set(toId, { label: classified.service, category: classified.category, bytes: r.bytes })
const svcFromId = String((enOut ?? enIn)?.id ?? r.serverId)
const peer = pickInternetPeer(r.src, r.dst, r.srcPort, r.dstPort)
const prevDst = dstAcc.get(peer)
if (prevDst) {
prevDst.bytes += r.bytes
prevDst.fromBytes.set(svcFromId, (prevDst.fromBytes.get(svcFromId) ?? 0) + r.bytes)
} else {
dstAcc.set(peer, {
bytes: r.bytes,
proto: r.proto,
dstPort: r.dstPort,
srcPort: r.srcPort,
fromBytes: new Map([[svcFromId, r.bytes]]),
})
}
}
const svcEn = enOut ?? enIn
const svcFromId = svcEn ? String(svcEn.id) : fromId
const edgeKey = `${svcFromId}|${toId}`
const svcTotals = new Map<string, { label: string; category: string; bytes: number }>()
const svcEdges = new Map<string, { fromId: string; toId: string; bytes: number; bytesFwd: number; bytesRev: number }>()
for (const [dst, acc] of dstAcc) {
const ripe = lookupRipeCached(dst)
const classified = classifyMapDstLite(dst, acc.proto, acc.dstPort, acc.srcPort, ripe)
if (!classified) continue
const toId = mapServiceNodeId(classified.service)
const prevSvc = svcTotals.get(toId)
if (prevSvc) prevSvc.bytes += acc.bytes
else svcTotals.set(toId, { label: classified.service, category: classified.category, bytes: acc.bytes })
for (const [fromId, bytes] of acc.fromBytes) {
const edgeKey = `${fromId}|${toId}`
const prevEdge = svcEdges.get(edgeKey)
if (prevEdge) {
prevEdge.bytes += r.bytes
prevEdge.bytesFwd += r.bytes
prevEdge.bytes += bytes
prevEdge.bytesFwd += bytes
} else {
svcEdges.set(edgeKey, {
fromId: svcFromId,
fromId,
toId,
bytes: r.bytes,
bytesFwd: r.bytes,
bytes,
bytesFwd: bytes,
bytesRev: 0,
})
}
}
}
enqueueRipeMisses(dsts)
const services: FlowMapService[] = [...svcTotals.entries()]
const minShare = minSharePct / 100
let services: FlowMapService[] = [...svcTotals.entries()]
.map(([id, s]) => ({
id,
label: s.label,
@@ -254,8 +334,11 @@ export function buildFlowMapHops(q: FlowMapHopsQuery): FlowMapHopsDto {
bps: (s.bytes * 8) / windowSec,
share: totalBytes > 0 ? s.bytes / totalBytes : 0,
}))
.filter((s) => s.share >= MAP_SERVICE_SHARE_THRESHOLD)
.sort((a, b) => b.bytes - a.bytes)
if (minSharePct > 0) {
services = services.filter((s) => s.share >= minShare)
}
services = services.slice(0, MAP_SERVICE_NODE_CAP)
const keepSvc = new Set(services.map((s) => s.id))
const serviceEdges: FlowMapServiceEdge[] = [...svcEdges.values()]
.filter((e) => keepSvc.has(e.toId))
@@ -273,15 +356,29 @@ export function buildFlowMapHops(q: FlowMapHopsQuery): FlowMapHopsDto {
return {
hops: [...hops.values()]
.map((a) => toHop(a, windowSec))
.sort((a, b) => b.bytes - a.bytes),
.sort((a, b) => a.bytes === b.bytes ? 0 : b.bytes - a.bytes),
live: listener.bound,
rangeMinutes: q.minutes,
windowSec,
totalBytes,
services,
serviceEdges,
mapServiceMinSharePct: minSharePct,
dedupApplied: wantDedup,
excludeMeshApplied: excludeMesh,
excludeOverlayApplied: excludeOverlay,
}
}
/** Hop-rates для карты сети: те же фильтры, что у общего NetFlow (dedup / mesh / overlay). */
export function buildFlowMapHops(q: FlowMapHopsQuery): FlowMapHopsDto {
const minSharePct = resolveMinSharePct(q)
const key = hopsQueryKey(q, minSharePct)
const now = Date.now()
if (hopsCache && hopsCache.key === key && now - hopsCache.at < HOPS_CACHE_TTL_MS) {
return hopsCache.dto
}
const dto = buildFlowMapHopsUncached(q, minSharePct)
hopsCache = { key, at: now, dto }
return dto
}
@@ -7,6 +7,7 @@ import {
lookupRipeCached,
resetRipeCacheForTests,
ripeFetchCountForTests,
ripeLastCandidateCountForTests,
seedRipeCacheForTests,
setRipeFetchForTests,
} from "./traffic-flow-ripe.js"
@@ -100,4 +101,36 @@ await flushRipeQueueForTests()
assert.equal(lookupRipeCached("1.0.0.1")?.country, "US")
assert.equal(lookupRipeCached("1.0.0.1")?.asn, 13335)
resetRipeCacheForTests()
disableRipePersistForTests()
for (let i = 0; i < 3000; i++) {
const o2 = Math.floor(i / 256)
const o3 = i % 256
seedRipeCacheForTests({
prefix: `203.${o2}.${o3}.0/24`,
asn: 64500,
country: "NL",
lat: null,
lng: null,
holder: "NOISE",
ok: true,
fetchedAt: Date.now(),
})
}
seedRipeCacheForTests({
prefix: "8.8.8.0/24",
asn: 15169,
country: "US",
lat: null,
lng: null,
holder: "GOOGLE",
ok: true,
fetchedAt: Date.now(),
})
assert.equal(lookupRipeCached("8.8.8.8")?.asn, 15169)
assert.ok(
ripeLastCandidateCountForTests() < 8,
`index should not scan all prefixes, got ${ripeLastCandidateCountForTests()}`,
)
console.log("traffic-flow-ripe.test.ts: ok")
+79 -13
View File
@@ -1,5 +1,5 @@
import { sqliteDatabase } from "../db/index.js"
import { ipInCidrV4, ipv4ToInt, isNonPublicIp, parseCidrV4 } from "./traffic-flow-ip.js"
import { ipv4ToInt, isNonPublicIp, parseCidrV4 } from "./traffic-flow-ip.js"
import { resolveRipeCountry } from "./traffic-flow-brands.js"
export interface FlowIpMeta {
@@ -28,6 +28,18 @@ const queue: string[] = []
const queued = new Set<string>()
const recentFetches: number[] = []
interface RipeIndexed {
entry: FlowIpMeta
net: number
mask: number
prefixLen: number
}
/** /24 → кандидаты с prefixLen ≥ 24. Более широкие префиксы — в `wideIndex`. */
const v24Index = new Map<number, RipeIndexed[]>()
const wideIndex: RipeIndexed[] = []
let lastCandidateCount = 0
let persistEnabled = true
let enqueueEnabled = true
let loaded = false
@@ -50,6 +62,9 @@ export function resetRipeCacheForTests(): void {
queue.length = 0
queued.clear()
recentFetches.length = 0
v24Index.clear()
wideIndex.length = 0
lastCandidateCount = 0
loaded = persistEnabled ? false : true
workerRunning = false
fetchCount = 0
@@ -58,10 +73,15 @@ export function resetRipeCacheForTests(): void {
}
export function seedRipeCacheForTests(entry: FlowIpMeta): void {
mem.set(entry.prefix, { ...entry })
remember(entry)
loaded = true
}
/** Сколько CIDR смотрели в последнем lookup (для теста индекса /24). */
export function ripeLastCandidateCountForTests(): number {
return lastCandidateCount
}
export function setRipeFetchForTests(fn: typeof fetch): void {
fetchImpl = fn
fetchCount = 0
@@ -87,6 +107,48 @@ function isFresh(entry: FlowIpMeta): boolean {
return Date.now() - entry.fetchedAt < ttlMs(entry.ok)
}
function unindexPrefix(prefix: string): void {
const parsed = parseCidrV4(prefix)
if (!parsed) return
if (parsed.prefixLen >= 24) {
const key = parsed.net >>> 8
const list = v24Index.get(key)
if (!list) return
const next = list.filter((row) => row.entry.prefix !== prefix)
if (next.length) v24Index.set(key, next)
else v24Index.delete(key)
return
}
const idx = wideIndex.findIndex((row) => row.entry.prefix === prefix)
if (idx >= 0) wideIndex.splice(idx, 1)
}
function indexEntry(entry: FlowIpMeta): void {
const parsed = parseCidrV4(entry.prefix)
if (!parsed) return
const row: RipeIndexed = {
entry,
net: parsed.net,
mask: parsed.mask,
prefixLen: parsed.prefixLen,
}
if (parsed.prefixLen >= 24) {
const key = parsed.net >>> 8
const list = v24Index.get(key)
if (list) list.push(row)
else v24Index.set(key, [row])
return
}
wideIndex.push(row)
}
function remember(entry: FlowIpMeta): void {
const prev = mem.get(entry.prefix)
if (prev) unindexPrefix(prev.prefix)
mem.set(entry.prefix, entry)
indexEntry(entry)
}
function loadSqlite(): void {
if (loaded || !persistEnabled) {
loaded = true
@@ -111,7 +173,7 @@ function loadSqlite(): void {
const fetchedAt = Date.parse(r.fetched_at)
const asn = Number(r.asn ?? 0) || 0
const holder = r.holder || ""
mem.set(r.prefix, {
remember({
prefix: r.prefix,
asn,
country: resolveRipeCountry(r.country || "", asn, holder) || "—",
@@ -193,20 +255,24 @@ function negative(prefix: string): FlowIpMeta {
export function lookupRipeCached(ip: string): FlowIpMeta | null {
loadSqlite()
const trimmed = String(ip ?? "").trim()
lastCandidateCount = 0
if (!trimmed) return null
if (isNonPublicIp(trimmed)) {
return negative(`${trimmed.includes(":") ? trimmed : trimmed}/32`)
}
const addr = ipv4ToInt(trimmed)
if (addr == null) return null
const bucket = v24Index.get(addr >>> 8)
const candidates = bucket ? bucket.concat(wideIndex) : wideIndex
lastCandidateCount = candidates.length
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
for (const row of candidates) {
if (!isFresh(row.entry)) continue
if (((addr & row.mask) >>> 0) !== row.net) continue
if (row.prefixLen > bestLen) {
best = row.entry
bestLen = row.prefixLen
}
}
return best
@@ -316,13 +382,13 @@ async function resolveIp(ip: string): Promise<FlowIpMeta | null> {
ok: Boolean(asn || country),
fetchedAt: Date.now(),
}
mem.set(prefix, entry)
remember(entry)
persist(entry)
return entry
} catch {
const prefix = `${ip}/32`
const entry = negative(prefix)
mem.set(prefix, entry)
remember(entry)
persist(entry)
return entry
} finally {
@@ -53,6 +53,7 @@ export function toTrafficFlowSettingsDto(
hubServerId: row.hubServerId ?? null,
retentionHours: row.retentionHours,
topN: row.topN,
mapServiceMinSharePct: Number(row.mapServiceMinSharePct ?? 5),
lastDatagramAt: row.lastDatagramAt ?? null,
lastExporterIp: row.lastExporterIp ?? null,
lastError: row.lastError || null,
@@ -75,6 +76,9 @@ export function updateTrafficFlowSettings(patch: TrafficFlowSettingsPatch) {
hubServerId: patch.hubServerId === undefined ? row.hubServerId : patch.hubServerId,
retentionHours: patch.retentionHours ?? row.retentionHours,
topN: patch.topN ?? row.topN,
mapServiceMinSharePct: patch.mapServiceMinSharePct == null
? row.mapServiceMinSharePct
: Math.min(100, Math.max(0, patch.mapServiceMinSharePct)),
updatedAt: nowIso(),
}).where(eq(trafficFlowSettings.id, 1)).run()
return getTrafficFlowSettingsRow()
@@ -47,6 +47,8 @@ function NetflowSettingsPanel({
const [endpoint, setEndpoint] = useState("")
const [retention, setRetention] = useState("24")
const [topN, setTopN] = useState("200")
const [shareOn, setShareOn] = useState(true)
const [sharePct, setSharePct] = useState("5")
const [ingestOn, setIngestOn] = useState(false)
const [purgeOpen, setPurgeOpen] = useState(false)
const [purgeBusy, setPurgeBusy] = useState(false)
@@ -62,6 +64,9 @@ function NetflowSettingsPanel({
setEndpoint(s.publicEndpoint)
setRetention(String(s.retentionHours))
setTopN(String(s.topN))
const pct = Number(s.mapServiceMinSharePct ?? 5)
setShareOn(pct > 0)
setSharePct(String(pct > 0 ? pct : 5))
setIngestOn(s.enabled)
}, [backendUrl, enabled])
@@ -83,6 +88,9 @@ function NetflowSettingsPanel({
publicEndpoint: endpoint,
retentionHours: Number.parseInt(retention, 10) || 24,
topN: Number.parseInt(topN, 10) || 200,
mapServiceMinSharePct: shareOn
? Math.min(100, Math.max(1, Number.parseFloat(sharePct) || 5))
: 0,
})
setSettings(res.settings)
toast.success("Настройки NetFlow сохранены")
@@ -193,6 +201,29 @@ function NetflowSettingsPanel({
<FormField label="Top-N разговоров">
<Input value={topN} onChange={(e) => setTopN(e.target.value)} inputMode="numeric" />
</FormField>
<div className="sm:col-span-2 flex flex-col gap-2">
<div className="flex items-center gap-3">
<FormToggle
checked={shareOn}
onChange={(on) => {
setShareOn(on)
if (on && (!sharePct || sharePct === "0")) setSharePct("5")
}}
/>
<span className="text-sm">Порог доли на карте</span>
</div>
<FormField
label="Минимум % окна"
hint="Узел сервиса, если доля байт окна ≥ N%. Выключить — показать все распознанные бренды (макс. 20)"
>
<Input
value={sharePct}
onChange={(e) => setSharePct(e.target.value)}
inputMode="decimal"
disabled={!shareOn}
/>
</FormField>
</div>
</div>
<p className="text-xs text-muted-foreground">
+15
View File
@@ -14259,6 +14259,21 @@
"dependencies": {
"zod": "^4.4.1"
}
},
"node_modules/@next/swc-win32-x64-msvc": {
"version": "16.2.4",
"resolved": "https://registry.npmjs.org/@next/swc-win32-x64-msvc/-/swc-win32-x64-msvc-16.2.4.tgz",
"integrity": "sha512-kMVGgsqhO5YTYODD9IPGGhA6iprWidQckK3LmPeW08PIFENRmgfb4MjXHO+p//d+ts2rpjvK5gXWzXSMrPl9cw==",
"cpu": [
"x64"
],
"optional": true,
"os": [
"win32"
],
"engines": {
"node": ">= 10"
}
}
}
}
+3
View File
@@ -21,6 +21,7 @@ export const trafficFlowSettingsDtoSchema = z.object({
hubServerId: z.number().int().positive().nullable(),
retentionHours: z.number().int().positive(),
topN: z.number().int().positive(),
mapServiceMinSharePct: z.number().min(0).max(100),
lastDatagramAt: z.string().nullable(),
lastExporterIp: z.string().nullable(),
lastError: z.string().nullable(),
@@ -40,6 +41,7 @@ export const trafficFlowSettingsPatchSchema = z.object({
hubServerId: z.number().int().positive().nullable().optional(),
retentionHours: z.number().int().positive().optional(),
topN: z.number().int().positive().max(1000).optional(),
mapServiceMinSharePct: z.number().min(0).max(100).optional(),
})
export const trafficFlowOverlayRequestSchema = z.object({
@@ -289,6 +291,7 @@ export const flowMapHopsDtoSchema = z.object({
totalBytes: z.number().nonnegative().optional(),
services: z.array(flowMapServiceDtoSchema).optional(),
serviceEdges: z.array(flowMapServiceEdgeDtoSchema).optional(),
mapServiceMinSharePct: z.number().min(0).max(100).optional(),
dedupApplied: z.boolean(),
excludeMeshApplied: z.boolean(),
excludeOverlayApplied: z.boolean(),
+1 -1
View File
File diff suppressed because one or more lines are too long