feat(traffic-flow): enhance flow analytics and IP handling

- Introduced `canonicalIp` function to standardize IP address formats across the application, improving consistency in flow processing.
- Updated traffic flow analytics to utilize new endpoint resolution logic, enhancing accuracy in traffic classification.
- Enhanced tests for traffic flow analytics and IP handling, ensuring comprehensive coverage for new functionalities.
- Improved traffic flow destination resolution with additional test cases for various internet services, including Google and Fastly.

Co-authored-by: Cursor <[email protected]>

backend/src/services/traffic-flow-analytics.test.ts

@@ -416,14 +416,44 @@ try {
   resetIfaceCacheForTests()
   resetRipeCacheForTests()
   disableRipeEnqueueForTests()
+  seedRipeCacheForTests({
+    prefix: "74.125.0.0/16",
+    asn: 15169,
+    country: "US",
+    lat: null,
+    lng: null,
+    holder: "GOOGLE",
+    ok: true,
+    fetchedAt: Date.now(),
+  })
+  seedRipeCacheForTests({
+    prefix: "104.18.0.0/16",
+    asn: 13335,
+    country: "US",
+    lat: null,
+    lng: null,
+    holder: "CLOUDFLARENET",
+    ok: true,
+    fetchedAt: Date.now(),
+  })
+  seedRipeCacheForTests({
+    prefix: "146.75.0.0/16",
+    asn: 54113,
+    country: "US",
+    lat: null,
+    lng: null,
+    holder: "FASTLY",
+    ok: true,
+    fetchedAt: Date.now(),
+  })
   rememberServerIfaces(7, [{ ".id": "*2", name: "ether1" }])
   ingestParsedFlowsForServerForTests(7, [
     {
-      src: "173.194.151.65",
-      dst: "10.200.100.53",
-      proto: 6,
+      src: "74.125.104.196/32",
+      dst: "10.200.100.53/32",
+      proto: 17,
       srcPort: 443,
-      dstPort: 57182,
+      dstPort: 62598,
       bytes: 12_000,
       packets: 10,
       inIface: "2",
@@ -440,15 +470,40 @@ try {
       inIface: "2",
       outIface: "2",
     },
+    {
+      src: "146.75.118.132/32",
+      dst: "10.200.100.53/32",
+      proto: 6,
+      srcPort: 80,
+      dstPort: 35026,
+      bytes: 4_000,
+      packets: 5,
+      inIface: "2",
+      outIface: "2",
+    },
   ])
   try {
     const rev = await buildFlowAnalytics({ minutes: 5, serverId: 7 })
-    const google = rev.conversationsList.find((r) => r.src === "173.194.151.65")
+    const google = rev.conversationsList.find((r) => r.src === "74.125.104.196")
     const cf = rev.conversationsList.find((r) => r.src === "104.18.35.51")
+    const fastly = rev.conversationsList.find((r) => r.src === "146.75.118.132")
     assert.equal(google?.service, "YouTube")
     assert.equal(google?.category, "Видео / стриминг")
+    assert.equal(google?.internetPeer, "74.125.104.196")
+    assert.equal(google?.internetPeerPort, 443)
+    assert.equal(google?.clientIp, "10.200.100.53")
+    assert.equal(google?.direction, "to_client")
+    assert.equal(google?.dstAsn, 15169)
+    assert.equal(google?.dstCountry, "US")
+    assert.ok(!String(google?.src).includes("/"), "DTO src без /32")
     assert.equal(cf?.service, "Cloudflare")
     assert.equal(cf?.category, "CDN")
+    assert.equal(fastly?.service, "Fastly")
+    assert.equal(fastly?.dstAsn, 54113)
+    assert.equal(fastly?.dstCountry, "US")
+    assert.ok(rev.asns?.some((r) => r.id === "54113"))
+    assert.ok(rev.countries?.some((r) => r.id === "US"))
+    assert.ok(!rev.services?.every((s) => s.label === "Прочее"), "сервисы не схлопнуты в Прочее")
   } finally {
     resetFlowRingsForTests()
     resetIfaceCacheForTests()

backend/src/services/traffic-flow-analytics.ts

@@ -265,8 +265,9 @@ async function buildFlowAnalyticsUncached(q: FlowAnalyticsQuery): Promise<FlowAn
       natSrcPort: r.natSrcPort,
       natDstPort: r.natDstPort,
     })
+    const ep = destMeta.endpoints
     if (destMeta.dest) peers.add(destMeta.dest)
-    const app = applicationName(r.proto, r.dstPort, r.srcPort)
+    const app = applicationName(r.proto, ep.peerPort || r.dstPort, ep.otherPort || r.srcPort)
     const ripe = destMeta.ripe
     const classified = destMeta.classified
     bump(applications, app, r.bytes, r.packets)
@@ -312,8 +313,8 @@ async function buildFlowAnalyticsUncached(q: FlowAnalyticsQuery): Promise<FlowAn
         conv.set(ckey, {
           serverId: String(r.serverId),
           serverName: nameById.get(r.serverId) ?? String(r.serverId),
-          src: r.src,
-          dst: r.dst,
+          src: ep.packetSrc || r.src,
+          dst: ep.packetDst || r.dst,
           proto: r.proto,
           protoName: protoName(r.proto),
           srcPort: r.srcPort,
@@ -332,6 +333,10 @@ async function buildFlowAnalyticsUncached(q: FlowAnalyticsQuery): Promise<FlowAn
           dstAsn: ripe?.asn || undefined,
           clientId: client?.userId,
           clientName: client?.name,
+          clientIp: ep.clientIp || undefined,
+          internetPeer: ep.internetPeer || undefined,
+          internetPeerPort: ep.peerPort || undefined,
+          direction: ep.direction,
           enId: en ? String(en.id) : undefined,
           enName: en?.name,
           plane,

backend/src/services/traffic-flow-dest.test.ts

@@ -4,7 +4,7 @@ import {
   resetEngineForTests,
 } from "./traffic-flow-engine.js"
 import { factsSnapshotForTests } from "./traffic-flow-facts.js"
-import { classifyInternetBrand, mapInternetBrand } from "./traffic-flow-dest.js"
+import { classifyInternetBrand, mapInternetBrand, resolveInternetDest } from "./traffic-flow-dest.js"
 import { disableCatalogFetchForTests, resetFlowCatalogForTests } from "./traffic-flow-classify.js"
 import {
   disableRipeEnqueueForTests,
@@ -182,6 +182,73 @@ assert.equal(mapInternetBrand("2001:4860:4860::8888", 17, 443, 50000, {
 assert.equal(mapInternetBrand("64.233.161.1", 17, 443, 50000, null).service, "YouTube")
 assert.equal(mapInternetBrand("142.250.1.10", 6, 443, 1, null).service, "YouTube")
 
+{
+  const meta = resolveInternetDest({
+    src: "74.125.104.196/32",
+    dst: "10.200.100.53/32",
+    proto: 17,
+    srcPort: 443,
+    dstPort: 62598,
+    serverId: 1,
+    inIface: "gre-client",
+    topo,
+  })
+  assert.equal(meta.dest, "74.125.104.196")
+  assert.equal(meta.classified.service, "YouTube")
+  assert.notEqual(meta.classified.service, "Прочее")
+  assert.equal(meta.endpoints.direction, "to_client")
+  assert.equal(meta.endpoints.clientIp, "10.200.100.53")
+  assert.equal(meta.asn, 0)
+}
+
+{
+  seedRipeCacheForTests({
+    prefix: "146.75.0.0/16",
+    asn: 54113,
+    country: "US",
+    lat: null,
+    lng: null,
+    holder: "FASTLY",
+    ok: true,
+    fetchedAt: Date.now(),
+  })
+  seedRipeCacheForTests({
+    prefix: "3.174.0.0/16",
+    asn: 16509,
+    country: "US",
+    lat: null,
+    lng: null,
+    holder: "AMAZON-AES",
+    ok: true,
+    fetchedAt: Date.now(),
+  })
+  const fastly = resolveInternetDest({
+    src: "146.75.118.132/32",
+    dst: "10.200.100.53/32",
+    proto: 6,
+    srcPort: 80,
+    dstPort: 35026,
+    serverId: 1,
+    inIface: "gre-client",
+    topo,
+  })
+  assert.equal(fastly.classified.service, "Fastly")
+  assert.equal(fastly.asn, 54113)
+  assert.equal(fastly.country, "US")
+  const aws = resolveInternetDest({
+    src: "3.174.2.35/32",
+    dst: "10.200.100.53/32",
+    proto: 6,
+    srcPort: 443,
+    dstPort: 43726,
+    serverId: 1,
+    inIface: "gre-client",
+    topo,
+  })
+  assert.equal(aws.classified.service, "AWS")
+  assert.equal(aws.asn, 16509)
+}
+
 resetEngineForTests()
 seedFlowTopologyForTests(null)
 resetRipeCacheForTests()

backend/src/services/traffic-flow-dest.ts

@@ -3,7 +3,11 @@ import { isIsoCountry, isNamedInternetService, OTHER_SERVICE, resolveFlowBrand }
 import { classifyFlowDst, type FlowClassification } from "./traffic-flow-classify.js"
 import { resolveFlowIp } from "./traffic-flow-geoip.js"
 import { canonicalFactIface } from "./traffic-flow-ifindex.js"
-import { pickInternetDest, type InternetDestCtx } from "./traffic-flow-ip.js"
+import {
+  resolveFlowEndpoints,
+  type FlowEndpoints,
+  type InternetDestCtx,
+} from "./traffic-flow-ip.js"
 import type { FlowIpMeta } from "./traffic-flow-ripe.js"
 import {
   flowOursHosts,
@@ -17,6 +21,7 @@ export interface InternetDestMeta {
   classified: FlowClassification
   country: string
   asn: number
+  endpoints: FlowEndpoints
 }
 
 export function destCtxForIface(
@@ -92,28 +97,35 @@ export function resolveInternetDest(opts: {
   natSrcPort?: number
   natDstPort?: number
 }): InternetDestMeta {
-  const dest = pickInternetDest(
-    opts.src,
-    opts.dst,
-    opts.srcPort,
-    opts.dstPort,
-    destCtxForIface(opts.topo, opts.serverId, opts.inIface, {
-      natSrc: opts.natSrc,
-      natDst: opts.natDst,
-      natSrcPort: opts.natSrcPort,
-      natDstPort: opts.natDstPort,
-    }),
-  )
-  const ripe = dest ? resolveFlowIp(dest) : null
-  const classified = dest
-    ? classifyFlowDst(dest, opts.proto, opts.dstPort, opts.srcPort, ripe, { ignoreTunnelProto: true })
-    : classifyFlowDst(opts.dst, opts.proto, opts.dstPort, opts.srcPort, ripe)
+  const ctx = destCtxForIface(opts.topo, opts.serverId, opts.inIface, {
+    natSrc: opts.natSrc,
+    natDst: opts.natDst,
+    natSrcPort: opts.natSrcPort,
+    natDstPort: opts.natDstPort,
+  })
+  const endpoints = resolveFlowEndpoints({
+    src: opts.src,
+    dst: opts.dst,
+    srcPort: opts.srcPort,
+    dstPort: opts.dstPort,
+    ctx,
+  })
+  const dest = endpoints.internetPeer
   if (!dest) {
-    return { dest: "", ripe: null, classified, country: "", asn: 0 }
+    return { dest: "", ripe: null, classified: OTHER_BRAND, country: "", asn: 0, endpoints }
   }
+  const ripe = resolveFlowIp(dest)
+  const classified = classifyFlowDst(
+    dest,
+    opts.proto,
+    endpoints.peerPort,
+    endpoints.otherPort,
+    ripe,
+    { ignoreTunnelProto: true },
+  )
   const country = ripe?.ok && isIsoCountry(ripe.country)
     ? ripe.country
     : (ripe?.ok ? "" : "unknown")
   const asn = ripe?.ok && ripe.asn ? ripe.asn : 0
-  return { dest, ripe, classified, country, asn }
+  return { dest, ripe, classified, country, asn, endpoints }
 }

backend/src/services/traffic-flow-engine.ts

@@ -9,6 +9,7 @@ import { invalidateTrafficFlowSettingsCache } from "./traffic-flow-settings.js"
 import { maybeRefreshIfaces } from "./traffic-flow-ifaces.js"
 import { canonicalFactIface } from "./traffic-flow-ifindex.js"
 import { shouldWriteFlowFact } from "./traffic-flow-facts-filter.js"
+import { canonicalIp } from "./traffic-flow-ip.js"
 import { resolveInternetDest } from "./traffic-flow-dest.js"
 import {
   getServerCatalog,
@@ -68,12 +69,12 @@ export interface PendingFlowRow {
 }
 
 function inetOrNull(value: string | null | undefined): string | null {
-  const s = String(value ?? "").trim()
+  const s = canonicalIp(value)
   return s.length > 0 ? s : null
 }
 
 export function isValidFlowInet(value: string): boolean {
-  const s = value.trim()
+  const s = canonicalIp(value)
   if (!s) return false
   const v4 = /^(\d{1,3})\.(\d{1,3})\.(\d{1,3})\.(\d{1,3})$/.exec(s)
   if (v4) {
@@ -100,16 +101,16 @@ function clampPort(n: number): number {
 }
 
 function sanitizeNatIp(value: string | null | undefined): string {
-  const s = String(value ?? "").trim()
+  const s = canonicalIp(value)
   if (!s || s === "0.0.0.0") return ""
   return isValidFlowInet(s) ? s : ""
 }
 
 function sanitizeFlowRow(r: PendingFlowRow): PendingFlowRow | null {
-  const src = (r.src || "").trim() || "0.0.0.0"
-  const dst = (r.dst || "").trim() || "0.0.0.0"
+  const src = canonicalIp(r.src) || "0.0.0.0"
+  const dst = canonicalIp(r.dst) || "0.0.0.0"
   if (!isValidFlowInet(src) || !isValidFlowInet(dst)) return null
-  const next = inetOrNull(r.nextHop)
+  const next = inetOrNull(canonicalIp(r.nextHop))
   return {
     ...r,
     src,

backend/src/services/traffic-flow-geoip.test.ts

@@ -90,6 +90,7 @@ setGeoipReadersForTests({
 const hit = resolveFlowIp("8.8.8.8")
 assert.equal(hit?.country, "US")
 assert.equal(hit?.asn, 15169)
+assert.equal(resolveFlowIp("8.8.8.8/32")?.asn, 15169, "GeoIP по inet::text /32")
 assert.equal(hit?.holder, "GOOGLE")
 assert.equal(hit?.ok, true)
 

backend/src/services/traffic-flow-geoip.ts

@@ -1,7 +1,7 @@
 import { existsSync } from "node:fs"
 import path from "node:path"
 import { open, type AsnResponse, type CountryResponse, type Reader } from "maxmind"
-import { isNonPublicIp } from "./traffic-flow-ip.js"
+import { canonicalIp, isNonPublicIp } from "./traffic-flow-ip.js"
 import { isIsoCountry, resolveRipeCountry } from "./traffic-flow-brands.js"
 import { lookupRipeCached, type FlowIpMeta } from "./traffic-flow-ripe.js"
 
@@ -117,7 +117,7 @@ function safeAsn(reader: Reader<AsnResponse>, ip: string): { asn: number; holder
  * (ok=true когда есть страна или ASN; null — данных нет, пусть пробует RIPE).
  */
 export function lookupGeoip(ip: string): FlowIpMeta | null {
-  const trimmed = String(ip ?? "").trim()
+  const trimmed = canonicalIp(ip)
   if (!trimmed) return null
   if (isNonPublicIp(trimmed)) return negativeMeta(trimmed)
   const { country: countryReader, asn: asnReader } = readers

backend/src/services/traffic-flow-ingest.test.ts

@@ -72,6 +72,7 @@ setPendingCapForTests(null)
 
 assert.equal(isValidFlowInet("10.0.0.1"), true)
 assert.equal(isValidFlowInet("8.8.8.8"), true)
+assert.equal(isValidFlowInet("8.8.8.8/32"), true)
 assert.equal(isValidFlowInet("0:0:0:0:0:0:0:1"), true)
 assert.equal(isValidFlowInet("not-an-ip"), false)
 assert.equal(isValidFlowInet("999.1.1.1"), false)

backend/src/services/traffic-flow-ingest.ts

@@ -29,6 +29,7 @@ import {
 } from "./traffic-flow-settings.js"
 import { applicationName } from "./traffic-flow-apps.js"
 import { resolveIfaceName } from "./traffic-flow-ifaces.js"
+import { canonicalIp } from "./traffic-flow-ip.js"
 import { getServerCatalog } from "./traffic-flow-topology.js"
 
 export type { PendingFlowRow }
@@ -357,10 +358,10 @@ export async function listStoredFlowRows(sinceIso: string): Promise<PendingFlowR
   const keep = Math.max(20, settings.topN)
   const stored = await dbAll<StoredBucketRow>(`
     SELECT
-      server_id, bucket_at, src::text AS src, dst::text AS dst, proto,
+      server_id, bucket_at, COALESCE(host(src), '') AS src, COALESCE(host(dst), '') AS dst, proto,
       src_port, dst_port, bytes, packets, in_iface, out_iface,
-      next_hop::text AS next_hop, flow_start_ms, flow_end_ms,
-      nat_src::text AS nat_src, nat_dst::text AS nat_dst, nat_src_port, nat_dst_port
+      COALESCE(host(next_hop), '') AS next_hop, flow_start_ms, flow_end_ms,
+      COALESCE(host(nat_src), '') AS nat_src, COALESCE(host(nat_dst), '') AS nat_dst, nat_src_port, nat_dst_port
     FROM (
       SELECT fb.*,
              ROW_NUMBER() OVER (
@@ -376,8 +377,8 @@ export async function listStoredFlowRows(sinceIso: string): Promise<PendingFlowR
     mergeInto(merged, {
       serverId: Number(r.server_id),
       bucketAt: isoBucketAt(r.bucket_at),
-      src: r.src,
-      dst: r.dst,
+      src: canonicalIp(r.src),
+      dst: canonicalIp(r.dst),
       proto: Number(r.proto) || 0,
       srcPort: Number(r.src_port) || 0,
       dstPort: Number(r.dst_port) || 0,
@@ -385,11 +386,11 @@ export async function listStoredFlowRows(sinceIso: string): Promise<PendingFlowR
       packets: Number(r.packets) || 0,
       inIface: r.in_iface ?? "",
       outIface: r.out_iface ?? "",
-      nextHop: r.next_hop ?? "",
+      nextHop: canonicalIp(r.next_hop ?? ""),
       flowStartMs: Number(r.flow_start_ms) || 0,
       flowEndMs: Number(r.flow_end_ms) || 0,
-      natSrc: r.nat_src ?? "",
-      natDst: r.nat_dst ?? "",
+      natSrc: canonicalIp(r.nat_src ?? ""),
+      natDst: canonicalIp(r.nat_dst ?? ""),
       natSrcPort: Number(r.nat_src_port) || 0,
       natDstPort: Number(r.nat_dst_port) || 0,
     })

backend/src/services/traffic-flow-ip.test.ts

@@ -1,14 +1,33 @@
 import assert from "node:assert/strict"
-import { isNonPublicIp, pickInternetDest, pickInternetPeer, pickMapInternetDest } from "./traffic-flow-ip.js"
+import {
+  canonicalIp,
+  isNonPublicIp,
+  pickInternetDest,
+  pickInternetPeer,
+  pickMapInternetDest,
+  resolveFlowEndpoints,
+} from "./traffic-flow-ip.js"
+
+assert.equal(canonicalIp("74.125.104.196/32"), "74.125.104.196")
+assert.equal(canonicalIp("10.200.100.53/32"), "10.200.100.53")
+assert.equal(canonicalIp("::ffff:8.8.8.8"), "8.8.8.8")
+assert.equal(canonicalIp("2001:4860:4860::8888/128"), "2001:4860:4860::8888")
 
 assert.equal(isNonPublicIp("10.200.100.53"), true)
+assert.equal(isNonPublicIp("10.200.100.53/32"), true)
 assert.equal(isNonPublicIp("173.194.151.65"), false)
+assert.equal(isNonPublicIp("74.125.104.196/32"), false, "PG inet::text не делает Google приватным")
 
 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("74.125.104.196/32", "10.200.100.53/32", 443, 62598),
+  "74.125.104.196",
+  "inet::text /32 reverse Google",
+)
 assert.equal(
   pickInternetPeer("10.200.100.53", "104.18.35.51", 53880, 443),
   "104.18.35.51",
@@ -92,6 +111,11 @@ assert.equal(
   "",
   "карта: RFC1918 без NAT → Прочее",
 )
+assert.equal(
+  pickInternetDest("173.194.151.65", "10.200.100.53", 12345, 57182, client),
+  "173.194.151.65",
+  "реверс googlevideo не :443 — публичный src",
+)
 assert.equal(
   pickMapInternetDest("173.194.151.65", "10.200.100.53", 12345, 57182, client),
   "173.194.151.65",
@@ -103,4 +127,32 @@ assert.equal(
   "карта: JH ours → EN ours всё ещё не dest",
 )
 
+{
+  const ep = resolveFlowEndpoints({
+    src: "74.125.104.196/32",
+    dst: "10.200.100.53/32",
+    srcPort: 443,
+    dstPort: 62598,
+  })
+  assert.equal(ep.internetPeer, "74.125.104.196")
+  assert.equal(ep.peerPort, 443)
+  assert.equal(ep.clientIp, "10.200.100.53")
+  assert.equal(ep.direction, "to_client")
+  assert.equal(ep.packetSrc, "74.125.104.196")
+  assert.equal(ep.packetDst, "10.200.100.53")
+}
+
+{
+  const ep = resolveFlowEndpoints({
+    src: "10.200.100.53",
+    dst: "104.18.35.51",
+    srcPort: 53880,
+    dstPort: 443,
+  })
+  assert.equal(ep.internetPeer, "104.18.35.51")
+  assert.equal(ep.peerPort, 443)
+  assert.equal(ep.clientIp, "10.200.100.53")
+  assert.equal(ep.direction, "from_client")
+}
+
 console.log("traffic-flow-ip.test.ts: ok")

backend/src/services/traffic-flow-ip.ts

@@ -1,7 +1,25 @@
 /** IPv4 helpers for RIPEstat prefix cache and EvoBGP CIDR match. */
 
+/**
+ * Host-семантика PostgreSQL `host(inet)`: снимает `/32` `/128`, `::ffff:`.
+ * IPFIX 5-tuple не меняем — только канонический вид адреса.
+ */
+export function canonicalIp(raw: string | undefined | null): string {
+  let t = String(raw ?? "").trim()
+  if (!t) return ""
+  const zone = t.indexOf("%")
+  if (zone >= 0) t = t.slice(0, zone)
+  if (t.toLowerCase().startsWith("::ffff:")) t = t.slice(7)
+  const slash = t.lastIndexOf("/")
+  if (slash >= 0) {
+    const plen = t.slice(slash + 1)
+    if (/^\d+$/.test(plen)) t = t.slice(0, slash)
+  }
+  return t.trim()
+}
+
 export function ipv4ToInt(ip: string): number | null {
-  const parts = String(ip ?? "").trim().split(".")
+  const parts = canonicalIp(ip).split(".")
   if (parts.length !== 4) return null
   let n = 0
   for (const p of parts) {
@@ -26,12 +44,12 @@ export function parseCidrV4(cidr: string): { net: number; mask: number; prefixLe
 export function ipInCidrV4(ip: string, cidr: string): boolean {
   const addr = ipv4ToInt(ip)
   const parsed = parseCidrV4(cidr)
-  if (addr == null || !parsed) return false
+  if (addr == null || parsed == null) return false
   return ((addr & parsed.mask) >>> 0) === parsed.net
 }
 
 export function isNonPublicIp(ip: string): boolean {
-  const trimmed = String(ip ?? "").trim()
+  const trimmed = canonicalIp(ip)
   if (!trimmed) return true
   if (trimmed.includes(":")) {
     const lower = trimmed.toLowerCase()
@@ -56,14 +74,14 @@ export function isNonPublicIp(ip: string): boolean {
 const PEER_WELL_KNOWN_PORTS = new Set([80, 443, 53, 853])
 
 export function isUnspecifiedIp(ip: string): boolean {
-  const t = String(ip ?? "").trim()
+  const t = canonicalIp(ip)
   if (!t) return true
   const lower = t.toLowerCase()
   return t === "0.0.0.0" || lower === "::" || lower === "::0"
 }
 
 function usableIp(ip: string | undefined): string {
-  const t = String(ip ?? "").trim()
+  const t = canonicalIp(ip)
   return isUnspecifiedIp(t) ? "" : t
 }
 
@@ -81,13 +99,22 @@ export interface InternetDestCtx {
 }
 
 export function isLocalIp(ip: string, ours?: ReadonlySet<string>): boolean {
-  if (isUnspecifiedIp(ip) || isNonPublicIp(ip)) return true
-  return Boolean(ours?.has(String(ip ?? "").trim()))
+  const host = canonicalIp(ip)
+  if (isUnspecifiedIp(host) || isNonPublicIp(host)) return true
+  if (!ours || ours.size === 0) return false
+  if (ours.has(host) || ours.has(String(ip ?? "").trim())) return true
+  for (const o of ours) {
+    if (canonicalIp(o) === host) return true
+  }
+  return false
 }
 
 /**
  * Интернет-назначение потока для ASN/страны/сервиса.
  * Пустая строка — dest нет (не GeoIP IP клиента / GRE-пира).
+ *
+ * boundClient: download CDN→overlay берём публичный src даже без :80/:443
+ * (googlevideo). Client-ISP → overlay:well-known — не dest (ASN клиента).
  */
 export function pickInternetDest(
   srcRaw: string,
@@ -112,8 +139,8 @@ export function pickInternetDest(
     if (srcIp) {
       const srcWk = PEER_WELL_KNOWN_PORTS.has(srcPortEff)
       const dstWk = PEER_WELL_KNOWN_PORTS.has(dstPortEff)
-      if (srcWk && !dstWk) return srcIp
-      return ""
+      if (!srcWk && dstWk) return ""
+      return srcIp
     }
     return ""
   }
@@ -133,8 +160,8 @@ export function pickInternetDest(
 }
 
 /**
- * Dest для карты: сначала pickInternetDest, иначе любой публичный src/dst/NAT
- * (без well-known ловушки boundClient — реверс googlevideo не только :80/:443/:53).
+ * Dest для карты = тот же internet peer, что аналитика/факты.
+ * (исторически отдельный fallback без well-known trap — теперь в pickInternetDest.)
  */
 export function pickMapInternetDest(
   srcRaw: string,
@@ -143,20 +170,82 @@ export function pickMapInternetDest(
   dstPort: number,
   ctx?: InternetDestCtx,
 ): string {
-  const dest = pickInternetDest(srcRaw, dstRaw, srcPort, dstPort, ctx)
-  if (dest) return dest
-  const ours = ctx?.ours
-  const internet = (ip: string) => Boolean(ip) && !isLocalIp(ip, ours)
-  for (const ip of [usableIp(ctx?.natDst), usableIp(ctx?.natSrc), usableIp(dstRaw), usableIp(srcRaw)]) {
-    if (internet(ip)) return ip
-  }
-  return ""
+  return pickInternetDest(srcRaw, dstRaw, srcPort, dstPort, ctx)
 }
 
 /**
  * Интернет-сторона потока без топологии: у IPFIX сервис часто в src (Google:443 → RFC1918).
- * Для куба статистики используйте pickInternetDest.
  */
 export function pickInternetPeer(src: string, dst: string, srcPort: number, dstPort: number): string {
-  return pickInternetDest(src, dst, srcPort, dstPort) || dst
+  return pickInternetDest(src, dst, srcPort, dstPort) || ""
+}
+
+export type FlowDirection = "to_client" | "from_client" | "transit"
+
+export interface FlowEndpoints {
+  packetSrc: string
+  packetDst: string
+  internetPeer: string
+  peerPort: number
+  otherPort: number
+  clientIp: string
+  direction: FlowDirection
+}
+
+function sameHost(a: string, b: string | undefined): boolean {
+  const x = canonicalIp(a)
+  const y = canonicalIp(b)
+  return Boolean(x) && x === y
+}
+
+/** Роли концов IPFIX-пакета. 5-tuple не переворачивается. */
+export function resolveFlowEndpoints(opts: {
+  src: string
+  dst: string
+  srcPort: number
+  dstPort: number
+  ctx?: InternetDestCtx
+}): FlowEndpoints {
+  const packetSrc = usableIp(opts.src)
+  const packetDst = usableIp(opts.dst)
+  const internetPeer = pickInternetDest(opts.src, opts.dst, opts.srcPort, opts.dstPort, opts.ctx)
+  const ours = opts.ctx?.ours
+  const srcLocal = Boolean(packetSrc) && isLocalIp(packetSrc, ours)
+  const dstLocal = Boolean(packetDst) && isLocalIp(packetDst, ours)
+
+  let peerPort = 0
+  let otherPort = 0
+  if (internetPeer) {
+    if (sameHost(internetPeer, packetSrc) || sameHost(internetPeer, opts.ctx?.natSrc)) {
+      peerPort = opts.srcPort
+      otherPort = opts.dstPort
+    } else if (sameHost(internetPeer, packetDst) || sameHost(internetPeer, opts.ctx?.natDst)) {
+      peerPort = opts.dstPort
+      otherPort = opts.srcPort
+    } else {
+      peerPort = opts.ctx?.natDstPort || opts.dstPort
+      otherPort = opts.srcPort
+    }
+  }
+
+  let clientIp = ""
+  if (srcLocal && !dstLocal) clientIp = packetSrc
+  else if (dstLocal && !srcLocal) clientIp = packetDst
+  else if (srcLocal) clientIp = packetSrc
+  else if (dstLocal) clientIp = packetDst
+
+  let direction: FlowDirection = "transit"
+  if (internetPeer && clientIp) {
+    direction = sameHost(internetPeer, packetSrc) ? "to_client" : "from_client"
+  }
+
+  return {
+    packetSrc,
+    packetDst,
+    internetPeer,
+    peerPort,
+    otherPort,
+    clientIp,
+    direction,
+  }
 }

backend/src/services/traffic-flow-map-hops.ts

@@ -10,7 +10,7 @@ import { getFlowListenerState, listFlowRowsForWindow } from "./traffic-flow-inge
 import { resolveIfaceName } from "./traffic-flow-ifaces.js"
 import { classifyFlowPlane, shouldKeepPlane } from "./traffic-flow-planes.js"
 import { destCtxForIface, mapInternetBrand } from "./traffic-flow-dest.js"
-import { pickMapInternetDest } from "./traffic-flow-ip.js"
+import { resolveFlowEndpoints } from "./traffic-flow-ip.js"
 import { geoipReadersStatus, resolveFlowIp } from "./traffic-flow-geoip.js"
 import { getTrafficFlowSettingsRow } from "./traffic-flow-settings.js"
 import { loadFlowTopology, resolveClient, resolveEn, getServerCatalog, type FlowTopology } from "./traffic-flow-topology.js"
@@ -336,18 +336,19 @@ async function buildFlowMapHopsUncached(q: FlowMapHopsQuery, minSharePct: number
     const inName = resolveIfaceName(r.serverId, r.inIface).name
     const outName = resolveIfaceName(r.serverId, r.outIface).name
     totalBytes += r.bytes
-    const dest = pickMapInternetDest(
-      r.src,
-      r.dst,
-      r.srcPort,
-      r.dstPort,
-      destCtxForIface(topo, r.serverId, inName, {
+    const ep = resolveFlowEndpoints({
+      src: r.src,
+      dst: r.dst,
+      srcPort: r.srcPort,
+      dstPort: r.dstPort,
+      ctx: destCtxForIface(topo, r.serverId, inName, {
         natSrc: r.natSrc,
         natDst: r.natDst,
         natSrcPort: r.natSrcPort,
         natDstPort: r.natDstPort,
       }),
-    )
+    })
+    const dest = ep.internetPeer
     const destKey = dest || "__other__"
     const client = resolveMapClient(topo, r.serverId, inName, outName)
     const prevDst = dstAcc.get(destKey)
@@ -358,8 +359,8 @@ async function buildFlowMapHopsUncached(q: FlowMapHopsQuery, minSharePct: number
       const acc: DstAcc = {
         bytes: r.bytes,
         proto: r.proto,
-        dstPort: r.dstPort,
-        srcPort: r.srcPort,
+        dstPort: ep.peerPort || r.dstPort,
+        srcPort: ep.otherPort || r.srcPort,
         fromBytes: new Map(),
       }
       bumpFrom(acc, String(r.serverId), r.bytes, client)

backend/src/services/traffic-flow-parse.ts

@@ -1,3 +1,5 @@
+import { canonicalIp } from "./traffic-flow-ip.js"
+
 export interface ParsedFlow {
   src: string
   dst: string
@@ -44,11 +46,13 @@ export function normalizeParsedFlow(flow: ParsedFlowInput): ParsedFlow {
   return {
     ...emptyParsedFlow(),
     ...flow,
-    nextHop: flow.nextHop ?? "",
+    src: canonicalIp(flow.src) || flow.src || "",
+    dst: canonicalIp(flow.dst) || flow.dst || "",
+    nextHop: canonicalIp(flow.nextHop ?? ""),
     flowStartMs: flow.flowStartMs ?? 0,
     flowEndMs: flow.flowEndMs ?? 0,
-    natSrc: flow.natSrc ?? "",
-    natDst: flow.natDst ?? "",
+    natSrc: canonicalIp(flow.natSrc ?? ""),
+    natDst: canonicalIp(flow.natDst ?? ""),
     natSrcPort: flow.natSrcPort ?? 0,
     natDstPort: flow.natDstPort ?? 0,
     inIface: flow.inIface ?? "",

backend/src/services/traffic-flow-ripe.ts

@@ -1,5 +1,5 @@
 import { dbAll, dbQuery } from "../db/index.js"
-import { ipv4ToInt, isNonPublicIp, parseCidrV4 } from "./traffic-flow-ip.js"
+import { canonicalIp, ipv4ToInt, isNonPublicIp, parseCidrV4 } from "./traffic-flow-ip.js"
 import { resolveRipeCountry } from "./traffic-flow-brands.js"
 
 export interface FlowIpMeta {
@@ -264,7 +264,7 @@ function negative(prefix: string): FlowIpMeta {
 
 export function lookupRipeCached(ip: string): FlowIpMeta | null {
   loadSqlite()
-  const trimmed = String(ip ?? "").trim()
+  const trimmed = canonicalIp(ip)
   lastCandidateCount = 0
   if (!trimmed) return null
   if (isNonPublicIp(trimmed)) {

backend/src/services/traffic-flow-topology.ts

@@ -2,6 +2,7 @@ import { db, dbAll } from "../db/index.js"
 import { parseJsonArray } from "../db/json.js"
 import { appUsers, servers, userInterfaceBindings } from "../db/schema.js"
 import { mapRosInterfaceType, parseRawInterfaces } from "../modules/users/iface-type.js"
+import { canonicalIp } from "./traffic-flow-ip.js"
 import type { PlaneTopology } from "./traffic-flow-planes.js"
 
 export interface FlowClientBinding {
@@ -176,10 +177,12 @@ export function flowOursHosts(topo: FlowTopology | null | undefined): Set<string
   const ours = new Set<string>()
   if (!topo) return ours
   for (const h of topo.enHosts) {
-    if (h) ours.add(h)
+    const ip = canonicalIp(h)
+    if (ip) ours.add(ip)
   }
   for (const h of topo.jhHosts) {
-    if (h) ours.add(h)
+    const ip = canonicalIp(h)
+    if (ip) ours.add(ip)
   }
   return ours
 }

components/data-grids/traffic-flows-data-grid.tsx

@@ -3,6 +3,7 @@
 import { useMemo } from "react"
 import { type ColumnDef, getCoreRowModel, useReactTable } from "@tanstack/react-table"
 import type { FlowTalkerDto } from "@mmapp/contracts/traffic-flow"
+import { ArrowDownIcon, ArrowUpIcon, MinusIcon } from "lucide-react"
 import { DataGridShell } from "@/components/data-grids/shared/data-grid-shell"
 import {
   DATA_GRID_CELL_PAD,
@@ -19,6 +20,40 @@ function formatBytes(n: number): string {
   return `${n} Б`
 }
 
+function formatEndpoint(ip: string | undefined, port: number | undefined): string {
+  const host = String(ip ?? "").trim()
+  if (!host) return "—"
+  return port ? `${host}:${port}` : host
+}
+
+function packetTuple(row: FlowTalkerDto): string {
+  const src = formatEndpoint(row.src, row.srcPort)
+  const dst = formatEndpoint(row.dst, row.dstPort)
+  return `${src} → ${dst}`
+}
+
+function FlowDirectionMark({ direction }: { direction: FlowTalkerDto["direction"] }) {
+  if (direction === "to_client") {
+    return (
+      <span className="inline-flex text-muted-foreground" title="Download: интернет → клиент">
+        <ArrowDownIcon className="size-3" />
+      </span>
+    )
+  }
+  if (direction === "from_client") {
+    return (
+      <span className="inline-flex text-muted-foreground" title="Upload: клиент → интернет">
+        <ArrowUpIcon className="size-3" />
+      </span>
+    )
+  }
+  return (
+    <span className="inline-flex text-muted-foreground" title="Транзит">
+      <MinusIcon className="size-3" />
+    </span>
+  )
+}
+
 function TrafficFlowsDataGrid({
   rows,
   emptyHint,
@@ -30,9 +65,16 @@ function TrafficFlowsDataGrid({
     () => [
       {
         id: "client",
-        accessorFn: (r) => r.clientName ?? "",
+        accessorFn: (r) => r.clientName || r.clientIp || "",
         header: () => <span className="text-xs font-medium text-muted-foreground">Клиент</span>,
-        cell: ({ row }) => <span className="text-xs">{row.original.clientName || "—"}</span>,
+        cell: ({ row }) => (
+          <span className="flex min-w-0 flex-col gap-0.5">
+            <span className="text-xs truncate">{row.original.clientName || "—"}</span>
+            {row.original.clientIp ? (
+              <span className="font-mono text-[10px] text-muted-foreground truncate">{row.original.clientIp}</span>
+            ) : null}
+          </span>
+        ),
         meta: { headerClassName: DATA_GRID_CELL_PAD_FIRST, cellClassName: DATA_GRID_CELL_PAD_FIRST },
       },
       {
@@ -43,27 +85,26 @@ function TrafficFlowsDataGrid({
         meta: { headerClassName: DATA_GRID_CELL_PAD, cellClassName: DATA_GRID_CELL_PAD },
       },
       {
-        id: "src",
-        accessorKey: "src",
-        header: () => <span className="text-xs font-medium text-muted-foreground">Src</span>,
-        cell: ({ row }) => (
-          <span className="font-mono text-xs">
-            {row.original.src}
-            {row.original.srcPort ? `:${row.original.srcPort}` : ""}
-          </span>
-        ),
+        id: "direction",
+        accessorFn: (r) => r.direction ?? "",
+        header: () => <span className="text-xs font-medium text-muted-foreground">Напр.</span>,
+        cell: ({ row }) => <FlowDirectionMark direction={row.original.direction} />,
         meta: { headerClassName: DATA_GRID_CELL_PAD, cellClassName: DATA_GRID_CELL_PAD },
       },
       {
-        id: "dst",
-        accessorKey: "dst",
-        header: () => <span className="text-xs font-medium text-muted-foreground">Dst</span>,
-        cell: ({ row }) => (
-          <span className="font-mono text-xs">
-            {row.original.dst}
-            {row.original.dstPort ? `:${row.original.dstPort}` : ""}
-          </span>
-        ),
+        id: "internet",
+        accessorFn: (r) => r.internetPeer ?? r.dst,
+        header: () => <span className="text-xs font-medium text-muted-foreground">Интернет</span>,
+        cell: ({ row }) => {
+          const r = row.original
+          const label = formatEndpoint(r.internetPeer, r.internetPeerPort)
+          const tuple = packetTuple(r)
+          return (
+            <span className="font-mono text-xs truncate max-w-[220px]" title={tuple}>
+              {label}
+            </span>
+          )
+        },
         meta: { headerClassName: DATA_GRID_CELL_PAD, cellClassName: DATA_GRID_CELL_PAD },
       },
       {

packages/contracts/src/traffic-flow.ts

@@ -93,6 +93,10 @@ export const flowTalkerDtoSchema = z.object({
   dstAsn: z.number().int().optional(),
   clientId: z.string().optional(),
   clientName: z.string().optional(),
+  clientIp: z.string().optional(),
+  internetPeer: z.string().optional(),
+  internetPeerPort: z.number().int().optional(),
+  direction: z.enum(["to_client", "from_client", "transit"]).optional(),
   enId: z.string().optional(),
   enName: z.string().optional(),
   plane: z.string().optional(),