fix(netflow): исключить специальные IPv4 из внешнего обогащения

This commit is contained in:
Denozordec committed 2026-10-11 12:59:57 +07:00
1 parent f3156b44cf
commit b619ef208e
14 files changed
+448 -81

No files matched your search

@@ -84,6 +84,7 @@ if (enId == null) throw new Error("no en server")
await ensurePartitionFor(pool, "flow_daily_facts", "month", new Date("2026-09-01T00:00:00Z"))
await ensurePartitionFor(pool, "flow_hour_facts", "day", new Date("2026-09-10T00:00:00Z"))
await ensurePartitionFor(pool, "server_snapshots", "week", new Date("2026-09-10T00:00:00Z"))
await dbQuery(`DELETE FROM flow_daily_facts WHERE server_id IN ($1, $2)`, [serverId, enId])
await dbQuery(`DELETE FROM flow_hour_facts WHERE server_id IN ($1, $2)`, [serverId, enId])
await dbQuery(`DELETE FROM user_interface_bindings WHERE server_id = $1`, [serverId])
@@ -44,6 +44,12 @@ if (!(await withPgOrSkip())) {
process.exit(0)
}
const fixture = await dbQuery<{ id: number }>(
"INSERT INTO servers (name, host) VALUES ($1, $2) RETURNING id",
["analytics-test-" + Date.now(), "127.0.0.1"],
)
const serverId = fixture.rows[0]!.id
try {
disableCatalogFetchForTests()
resetFlowCatalogForTests()
disableRipePersistForTests()
@@ -60,13 +66,13 @@ seedFlowTopologyForTests({
wanIfaces: new Map(),
plane: { clientIfaceNames: new Set(), enHosts: new Set(), jhHosts: new Set() },
})
rememberServerIfaces(7, [
rememberServerIfaces(serverId, [
{ ".id": "*2", name: "ether1" },
{ ".id": "*B", name: "ether2" },
{ ".id": "*A", name: "wg-flow" },
])
ingestParsedFlowsForServerForTests(7, [
ingestParsedFlowsForServerForTests(serverId, [
{
src: "10.1.1.8",
dst: "8.8.8.8",
@@ -92,7 +98,7 @@ ingestParsedFlowsForServerForTests(7, [
])
try {
const all = await buildFlowAnalytics({ minutes: 5, serverId: 7 })
const all = await buildFlowAnalytics({ minutes: 5, serverId })
assert.equal(all.applications[0]?.label, "HTTPS")
assert.ok(all.protocols.some((p) => p.label === "TCP"))
assert.equal(all.ifaces[0]?.name, "ether1")
@@ -104,11 +110,11 @@ try {
assert.equal(conv.application, "HTTPS")
assert.ok(!/^\d+$/.test(conv.inIface))
const filtered = await buildFlowAnalytics({ minutes: 5, serverId: 7, iface: "ether1" })
const filtered = await buildFlowAnalytics({ minutes: 5, serverId, iface: "ether1" })
assert.ok(filtered.bytes >= 12_000)
assert.equal(filtered.ifaces[0]?.name, "ether1")
const miss = await buildFlowAnalytics({ minutes: 5, serverId: 7, iface: "wg-flow" })
const miss = await buildFlowAnalytics({ minutes: 5, serverId, iface: "wg-flow" })
assert.equal(miss.conversations, 0)
const other = await buildFlowAnalytics({ minutes: 5, serverId: 99 })
@@ -120,12 +126,12 @@ try {
resetFlowRingsForTests()
resetIfaceCacheForTests()
rememberServerIfaces(7, [
rememberServerIfaces(serverId, [
{ ".id": "*2", name: "ether1" },
{ ".id": "*B", name: "ether2" },
{ ".id": "*A", name: "wg-flow" },
])
ingestParsedFlowsForServerForTests(7, [
ingestParsedFlowsForServerForTests(serverId, [
{
src: "10.1.1.8",
dst: "8.8.8.8",
@@ -150,10 +156,10 @@ ingestParsedFlowsForServerForTests(7, [
},
])
try {
const summed = await buildFlowAnalytics({ minutes: 5, serverId: 7, dedup: false })
const summed = await buildFlowAnalytics({ minutes: 5, serverId, dedup: false })
assert.equal(summed.bytes, 21_000)
assert.equal(summed.conversations, 2)
const deduped = await buildFlowAnalytics({ minutes: 5, serverId: 7, dedup: true })
const deduped = await buildFlowAnalytics({ minutes: 5, serverId, dedup: true })
assert.equal(deduped.bytes, 12_000)
assert.equal(deduped.conversations, 1)
assert.equal(deduped.dedupApplied, true)
@@ -179,8 +185,8 @@ seedRipeCacheForTests({
seedFlowCatalogForTests({
cidrs: [{ cidr: "8.8.8.0/24", purpose: "steam-gaming" }],
})
rememberServerIfaces(7, [{ ".id": "*2", name: "ether1" }])
ingestParsedFlowsForServerForTests(7, [
rememberServerIfaces(serverId, [{ ".id": "*2", name: "ether1" }])
ingestParsedFlowsForServerForTests(serverId, [
{
src: "10.1.1.8",
dst: "8.8.8.8",
@@ -194,7 +200,7 @@ ingestParsedFlowsForServerForTests(7, [
},
])
try {
const geo = await buildFlowAnalytics({ minutes: 5, serverId: 7 })
const geo = await buildFlowAnalytics({ minutes: 5, serverId })
assert.equal(geo.categories?.[0]?.label, "Игры")
assert.ok(geo.asns?.some((r) => r.label.includes("AS15169")))
assert.equal(geo.countries?.[0]?.id, "US")
@@ -224,8 +230,8 @@ seedRipeCacheForTests({
ok: true,
fetchedAt: Date.now(),
})
rememberServerIfaces(7, [{ ".id": "*2", name: "ether1" }])
ingestParsedFlowsForServerForTests(7, [
rememberServerIfaces(serverId, [{ ".id": "*2", name: "ether1" }])
ingestParsedFlowsForServerForTests(serverId, [
{
src: "10.1.1.8",
dst: "1.1.1.1",
@@ -239,7 +245,7 @@ ingestParsedFlowsForServerForTests(7, [
},
])
try {
const cf = await buildFlowAnalytics({ minutes: 5, serverId: 7 })
const cf = await buildFlowAnalytics({ minutes: 5, serverId })
assert.equal(cf.countries?.[0]?.id, "US")
assert.ok(cf.mapEdges?.every((e) => e.toCountry !== "?"))
assert.equal(cf.services?.[0]?.label, "Cloudflare")
@@ -254,7 +260,7 @@ try {
}
{
ingestParsedFlowsForServerForTests(7, [
ingestParsedFlowsForServerForTests(serverId, [
{
src: "10.1.1.8",
dst: "8.8.8.8",
@@ -267,7 +273,7 @@ try {
outIface: "",
},
])
const degraded = await buildFlowAnalytics({ minutes: 5, serverId: 7, skipHeavy: true })
const degraded = await buildFlowAnalytics({ minutes: 5, serverId, skipHeavy: true })
assert.equal(degraded.degraded, true)
assert.equal(degraded.conversationsList.length, 0)
assert.ok((degraded.bytes ?? 0) >= 12_000)
@@ -283,22 +289,22 @@ try {
const { pool } = await import("../db/index.js")
const { ensurePartitionFor } = await import("../db/partitions.js")
await ensurePartitionFor(pool, "flow_daily_dims", "month", new Date("2026-09-01T00:00:00Z"))
await dbQuery(`DELETE FROM flow_daily_dims WHERE server_id = 7 AND day >= '2026-09-01' AND day < '2026-10-01'`)
await dbQuery(`DELETE FROM flow_daily_dims WHERE server_id = $1 AND day >= '2026-09-01' AND day < '2026-10-01'`, [serverId])
await dbQuery(`
INSERT INTO flow_daily_dims (server_id, day, dim, key, bytes, packets)
VALUES
(7, '2026-09-01', 'country', 'US', 1000, 10),
(7, '2026-09-02', 'country', 'US', 500, 5),
(7, '2026-09-01', 'service', 'steam', 800, 8),
(7, '2026-09-01', 'asn', '15169', 900, 9),
(7, '2026-09-01', 'asn', 'other', 100, 1)
`)
const monthly = await getFlowMonthly("2026-09", 7)
($1, '2026-09-01', 'country', 'US', 1000, 10),
($1, '2026-09-02', 'country', 'US', 500, 5),
($1, '2026-09-01', 'service', 'steam', 800, 8),
($1, '2026-09-01', 'asn', '15169', 900, 9),
($1, '2026-09-01', 'asn', 'other', 100, 1)
`, [serverId])
const monthly = await getFlowMonthly("2026-09", serverId)
assert.equal(monthly.bytes, 1500)
assert.equal(monthly.countries[0]?.id, "US")
assert.equal(monthly.countries[0]?.bytes, 1500)
assert.ok(monthly.asns.some((row) => row.id === "other"))
await dbQuery(`DELETE FROM flow_daily_dims WHERE server_id = 7 AND day >= '2026-09-01' AND day < '2026-10-01'`)
await dbQuery(`DELETE FROM flow_daily_dims WHERE server_id = $1 AND day >= '2026-09-01' AND day < '2026-10-01'`, [serverId])
}
}
@@ -308,18 +314,18 @@ try {
resetRipeCacheForTests()
resetFlowCatalogForTests()
disableRipeEnqueueForTests()
rememberServerIfaces(7, [
rememberServerIfaces(serverId, [
{ ".id": "*2", name: "gre-client" },
{ ".id": "*3", name: "NSK-SERVHOST-RTK" },
{ ".id": "*4", name: "gre-en-nsk" },
])
const topo: FlowTopology = {
clientIfaces: new Map([[7, new Set(["gre-client"])]]),
clientByIface: new Map([["7|gre-client", {
clientIfaces: new Map([[serverId, new Set(["gre-client"])]]),
clientByIface: new Map([[serverId + "|gre-client", {
userId: "u1",
login: "alice",
name: "Alice",
serverId: 7,
serverId,
interfaceName: "gre-client",
}]]),
enNodes: [{ id: 9, name: "NSK-SERVHOST-RTK", hosts: ["198.51.100.1"] }],
@@ -343,7 +349,7 @@ try {
ok: true,
fetchedAt: Date.now(),
})
ingestParsedFlowsForServerForTests(7, [
ingestParsedFlowsForServerForTests(serverId, [
{
src: "10.100.1.17",
dst: "173.194.160.163",
@@ -380,7 +386,7 @@ try {
},
])
try {
const def = await buildFlowAnalytics({ minutes: 5, serverId: 7 })
const def = await buildFlowAnalytics({ minutes: 5, serverId })
assert.equal(def.bytes, 12_000)
assert.equal(def.bytesPayload, 12_000)
assert.equal(def.bytesOverlay, 5_000_000)
@@ -398,7 +404,7 @@ try {
assert.equal(path.clientName, "Alice")
assert.equal(path.enName, "NSK-SERVHOST-RTK")
assert.equal(path.dst, "173.194.160.163")
const withAll = await buildFlowAnalytics({ minutes: 5, serverId: 7, excludeOverlay: false, excludeMesh: false })
const withAll = await buildFlowAnalytics({ minutes: 5, serverId, excludeOverlay: false, excludeMesh: false })
assert.equal(withAll.bytes, 12_000 + 5_000_000 + 8000)
assert.ok(withAll.conversationsList.some((r) => r.plane === "overlay"))
assert.ok(withAll.conversationsList.some((r) => r.plane === "client_mesh"))
@@ -446,8 +452,8 @@ try {
ok: true,
fetchedAt: Date.now(),
})
rememberServerIfaces(7, [{ ".id": "*2", name: "ether1" }])
ingestParsedFlowsForServerForTests(7, [
rememberServerIfaces(serverId, [{ ".id": "*2", name: "ether1" }])
ingestParsedFlowsForServerForTests(serverId, [
{
src: "74.125.104.196/32",
dst: "10.200.100.53/32",
@@ -483,7 +489,7 @@ try {
},
])
try {
const rev = await buildFlowAnalytics({ minutes: 5, serverId: 7 })
const rev = await buildFlowAnalytics({ minutes: 5, serverId })
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")
@@ -515,7 +521,7 @@ try {
if (await withPgOrSkip()) {
resetFlowRingsForTests()
resetIfaceCacheForTests()
const sidRow = await dbQuery<{ id: number }>(`SELECT id FROM servers LIMIT 1`)
const sidRow = await dbQuery<{ id: number }>(`SELECT id FROM servers WHERE id = $1`, [serverId])
const sid = sidRow.rows[0]?.id
if (sid) {
const { pool } = await import("../db/index.js")
@@ -550,4 +556,10 @@ try {
}
}
} finally {
for (const table of ["flow_daily_dims", "flow_hour_facts", "flow_daily_facts", "flow_buckets", "traffic_samples", "user_interface_bindings", "server_snapshots"]) {
await dbQuery("DELETE FROM " + table + " WHERE server_id = $1", [serverId])
}
await dbQuery("DELETE FROM servers WHERE id = $1", [serverId])
}
console.log("traffic-flow-analytics.test.ts: ok")
@@ -18,6 +18,20 @@ 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 приватным")
for (const ip of [
"198.18.0.0", "198.19.255.255", "::ffff:198.18.139.1/32",
"192.0.0.0", "192.0.0.8", "192.0.0.11", "192.0.0.255",
"192.0.2.0", "192.0.2.255", "198.51.100.0", "198.51.100.255",
"203.0.113.0", "203.0.113.255", "192.88.99.2",
"240.0.0.0", "254.255.255.255", "255.255.255.255",
]) assert.equal(isNonPublicIp(ip), true, ip)
for (const ip of [
"198.17.255.255", "198.20.0.0", "192.0.0.9", "192.0.0.10/32",
"192.0.1.255", "192.0.3.0", "198.51.99.255", "198.51.101.0",
"203.0.112.255", "203.0.114.0", "192.31.196.1", "192.52.193.1",
"192.175.48.1", "223.255.255.255",
]) assert.equal(isNonPublicIp(ip), false, ip)
assert.equal(
pickInternetPeer("173.194.151.65", "10.200.100.53", 443, 57182),
"173.194.151.65",
+12 -12
View File
@@ -48,6 +48,16 @@ export function ipInCidrV4(ip: string, cidr: string): boolean {
return ((addr & parsed.mask) >>> 0) === parsed.net
}
// IPv4 ranges not globally reachable in the IANA special-purpose registry.
// Parse fixed ranges once: this predicate also runs for each flow endpoint.
const nonPublicV4Ranges = [
"0.0.0.0/8", "10.0.0.0/8", "100.64.0.0/10", "127.0.0.0/8",
"169.254.0.0/16", "172.16.0.0/12", "192.0.0.0/24", "192.0.2.0/24",
"192.88.99.2/32", "192.168.0.0/16", "198.18.0.0/15", "198.51.100.0/24",
"203.0.113.0/24", "224.0.0.0/4", "240.0.0.0/4",
].map(cidr => parseCidrV4(cidr)!)
const publicProtocolAnycastV4 = new Set([0xc0000009, 0xc000000a]) // 192.0.0.9/10
export function isNonPublicIp(ip: string): boolean {
const trimmed = canonicalIp(ip)
if (!trimmed) return true
@@ -57,18 +67,8 @@ export function isNonPublicIp(ip: string): boolean {
}
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")
)
if (publicProtocolAnycastV4.has(n)) return false
return nonPublicV4Ranges.some(({ net, mask }) => ((n & mask) >>> 0) === net)
}
const PEER_WELL_KNOWN_PORTS = new Set([80, 443, 53, 853])
@@ -290,7 +290,7 @@ rememberServerIfaces(7, [
googleRipe()
ingestParsedFlowsForServerForTests(7, [
payloadFlow("8.8.8.8", 600),
payloadFlow("203.0.113.50", 9400),
payloadFlow("11.0.113.50", 9400),
])
try {
resetFlowMapHopsCacheForTests()
@@ -335,7 +335,7 @@ rememberServerIfaces(7, [
googleRipe()
ingestParsedFlowsForServerForTests(7, [
payloadFlow("8.8.8.8", 400),
payloadFlow("203.0.113.50", 9600),
payloadFlow("11.0.113.50", 9600),
])
try {
resetFlowMapHopsCacheForTests()
@@ -370,7 +370,7 @@ rememberServerIfaces(7, [
ingestParsedFlowsForServerForTests(7, [
payloadFlow("8.8.8.8", 400),
payloadFlow("104.18.35.51", 400),
payloadFlow("203.0.113.50", 9200),
payloadFlow("11.0.113.50", 9200),
])
try {
resetFlowMapHopsCacheForTests()
@@ -424,7 +424,7 @@ rememberServerIfaces(7, [
])
ingestParsedFlowsForServerForTests(7, [
payloadFlow("64.233.161.1", 10_000),
payloadFlow("203.0.113.50", 20_000),
payloadFlow("11.0.113.50", 20_000),
])
try {
resetFlowMapHopsCacheForTests()
@@ -451,15 +451,15 @@ try {
}
const smallBrands: Array<{ ip: string; asn: number; holder: string; bytes: number; id: string }> = [
{ ip: "203.0.113.1", asn: 714, holder: "APPLE-ENGINEERING", bytes: 400, id: "svc:apple" },
{ ip: "203.0.113.2", asn: 36459, holder: "GITHUB", bytes: 390, id: "svc:github" },
{ ip: "203.0.113.3", asn: 54876, holder: "GITLAB", bytes: 380, id: "svc:gitlab" },
{ ip: "203.0.113.4", asn: 8403, holder: "SPOTIFY", bytes: 370, id: "svc:spotify" },
{ ip: "203.0.113.5", asn: 13414, holder: "TWITTER", bytes: 360, id: "svc:x" },
{ ip: "203.0.113.6", asn: 47541, holder: "VKONTAKTE", bytes: 350, id: "svc:vk" },
{ ip: "203.0.113.7", asn: 30103, holder: "ZOOM", bytes: 340, id: "svc:zoom" },
{ ip: "203.0.113.8", asn: 395701, holder: "EPIC-GAMES", bytes: 330, id: "svc:epic" },
{ ip: "203.0.113.9", asn: 6507, holder: "RIOT-GAMES", bytes: 320, id: "svc:riot" },
{ ip: "11.0.113.1", asn: 714, holder: "APPLE-ENGINEERING", bytes: 400, id: "svc:apple" },
{ ip: "11.0.113.2", asn: 36459, holder: "GITHUB", bytes: 390, id: "svc:github" },
{ ip: "11.0.113.3", asn: 54876, holder: "GITLAB", bytes: 380, id: "svc:gitlab" },
{ ip: "11.0.113.4", asn: 8403, holder: "SPOTIFY", bytes: 370, id: "svc:spotify" },
{ ip: "11.0.113.5", asn: 13414, holder: "TWITTER", bytes: 360, id: "svc:x" },
{ ip: "11.0.113.6", asn: 47541, holder: "VKONTAKTE", bytes: 350, id: "svc:vk" },
{ ip: "11.0.113.7", asn: 30103, holder: "ZOOM", bytes: 340, id: "svc:zoom" },
{ ip: "11.0.113.8", asn: 395701, holder: "EPIC-GAMES", bytes: 330, id: "svc:epic" },
{ ip: "11.0.113.9", asn: 6507, holder: "RIOT-GAMES", bytes: 320, id: "svc:riot" },
]
resetFlowRingsForTests()
@@ -513,7 +513,7 @@ ingestParsedFlowsForServerForTests(7, [
inIface: "3",
outIface: "3",
},
payloadFlow("203.0.113.50", 1000),
payloadFlow("11.0.113.50", 1000),
])
try {
resetFlowMapHopsCacheForTests()
@@ -880,7 +880,7 @@ rememberServerIfaces(7, [
googleRipe()
ingestParsedFlowsForServerForTests(7, [
payloadFlow("8.8.8.8", 600),
payloadFlow("203.0.113.50", 9400),
payloadFlow("11.0.113.50", 9400),
])
try {
resetFlowMapHopsCacheForTests()
+12 -3
View File
@@ -58,17 +58,26 @@ enqueueRipeMisses(["8.8.8.10"])
await flushRipeQueueForTests()
assert.equal(ripeFetchCountForTests(), afterFirst)
seedRipeCacheForTests({ prefix: "198.18.0.0/15", asn: 64500, country: "NL", lat: 52, lng: 4, holder: "STALE", ok: true, fetchedAt: Date.now() })
for (const ip of ["198.18.139.1", "198.19.255.255", "192.0.2.1", "198.51.100.1", "203.0.113.1", "240.0.0.1"]) {
assert.equal(lookupRipeCached(ip)?.ok, false, "Special ranges must bypass stale public cache")
}
enqueueRipeMisses(["198.18.139.1", "198.19.255.255", "192.0.2.1", "198.51.100.1", "203.0.113.1", "240.0.0.1"])
await flushRipeQueueForTests()
assert.equal(ripeFetchCountForTests(), afterFirst, "Special ranges must never query RIPE")
resetRipeCacheForTests()
disableRipePersistForTests()
setRipeFetchForTests(async () => {
throw new Error("timeout")
})
enqueueRipeMisses(["203.0.113.50"])
enqueueRipeMisses(["8.9.8.50"])
await flushRipeQueueForTests()
const neg = lookupRipeCached("203.0.113.50")
const neg = lookupRipeCached("8.9.8.50")
assert.equal(neg?.ok, false)
const afterNeg = ripeFetchCountForTests()
enqueueRipeMisses(["203.0.113.50"])
assert.ok(afterNeg > 0, "Upstream failure test must actually query a public address")
enqueueRipeMisses(["8.9.8.50"])
await flushRipeQueueForTests()
assert.equal(ripeFetchCountForTests(), afterNeg)