feat(traffic): enhance flow analytics and brand classification

- 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
13 changed files with 195 additions and 16 deletions
@@ -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": {
@@ -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)
@@ -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,
@@ -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")
@@ -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
+}
@@ -278,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")
@@ -12,6 +12,7 @@ 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 { 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"
@@ -278,12 +279,13 @@ function buildFlowMapHopsUncached(q: FlowMapHopsQuery, minSharePct: number): Flo

    totalBytes += r.bytes
    const svcFromId = String((enOut ?? enIn)?.id ?? r.serverId)
-    const prevDst = dstAcc.get(r.dst)
+    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(r.dst, {
+      dstAcc.set(peer, {
        bytes: r.bytes,
        proto: r.proto,
        dstPort: r.dstPort,
@@ -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"
+      }
    }
  }
 }