fix(netflow): исправить классификацию IPv6 и кеш префиксов RIPE

This commit is contained in:
Denozordec committed 2026-10-11 16:21:43 +07:00
1 parent 59938118bd
commit af66faaae4
12 files changed
+493 -23

No files matched your search

@@ -1,6 +1,7 @@
import assert from "node:assert/strict"
import {
canonicalIp,
parseCidrV6,
isNonPublicIp,
pickInternetDest,
pickInternetPeer,
@@ -18,6 +19,29 @@ 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 [
"::", "0:0:0:0:0:0:0:1", "::192.0.2.1", "::ffff:c000:201", "0:0:0:0:0:ffff:ac10:1",
"64:ff9b:1::1", "100::", "100::ffff:ffff:ffff:ffff", "100:0:0:1::1",
"2001:2::1", "2001:1::4", "2001:1ff:ffff:ffff:ffff:ffff:ffff:ffff",
"2001:db8::1", "2001:db8:ffff:ffff:ffff:ffff:ffff:ffff", "3fff::", "3fff:fff:ffff:ffff:ffff:ffff:ffff:ffff",
"5f00::1", "fc00::1", "fdff:ffff:ffff:ffff:ffff:ffff:ffff:ffff",
"fe80::1%ether1", "febf:ffff:ffff:ffff:ffff:ffff:ffff:ffff", "fec0::1", "ff02::1",
"2001:db8::1/128", "2001:DB8::1", "2001:::1", "abcd:1234", "::ffff:999.1.1.1",
]) assert.equal(isNonPublicIp(ip), true, ip)
for (const ip of [
"2001:4860:4860::8888", "2606:4700:4700::1111/128", "2001:4860:0:0:0:0:0:8888",
"64:ff9b::808:808", "100:0:0:2::1", "2001:1::1", "2001:1::2", "2001:1::3",
"2001:3::1", "2001:4:112::1", "2001:20::1", "2001:2f:ffff:ffff:ffff:ffff:ffff:ffff",
"2001:30::1", "2001:3f:ffff:ffff:ffff:ffff:ffff:ffff", "2001::1", "2002::1",
"2001:200::1", "3ffe:ffff::1", "3fff:1000::1", "2620:4f:8000::1",
"::ffff:808:808", "0:0:0:0:0:ffff:808:808", "::ffff:8.8.8.8/128",
]) assert.equal(isNonPublicIp(ip), false, ip)
assert.equal(canonicalIp("::ffff:808:808"), "8.8.8.8", "Hex mapped IPv6 uses IPv4 host identity")
assert.equal(canonicalIp("0:0:0:0:0:ffff:808:808/128"), "8.8.8.8")
assert.equal(parseCidrV6("2001:4860::/0")?.net, 0n)
assert.equal(parseCidrV6("2001:4860::1/128")?.shift, 0n)
for (const cidr of ["::/129", "::/-1", "::/32junk", "::/32/64", "::", "bad::ip/48"]) assert.equal(parseCidrV6(cidr), null)
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",
+60 -4
View File
@@ -1,4 +1,6 @@
/** IPv4 helpers for RIPEstat prefix cache and EvoBGP CIDR match. */
import { isIP } from "node:net"
/** IP helpers for RIPEstat prefix cache and EvoBGP CIDR match. */
/**
* Host-семантика PostgreSQL `host(inet)`: снимает `/32` `/128`, `::ffff:`.
@@ -9,12 +11,19 @@ export function canonicalIp(raw: string | undefined | null): string {
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)
}
if (t.toLowerCase().startsWith("::ffff:") && isIP(t.slice(7)) === 4) t = t.slice(7)
else if (/(?:^|:)ffff:/i.test(t)) {
const n = ipv6ToInt(t)
if (n !== null && n >> 32n === 0xffffn) {
const v4 = Number(n & 0xffffffffn)
t = [v4 >>> 24, (v4 >>> 16) & 255, (v4 >>> 8) & 255, v4 & 255].join(".")
}
}
return t.trim()
}
@@ -58,12 +67,59 @@ const nonPublicV4Ranges = [
].map(cidr => parseCidrV4(cidr)!)
const publicProtocolAnycastV4 = new Set([0xc0000009, 0xc000000a]) // 192.0.0.9/10
// Only parse validated IPv6. Embedded IPv4 occupies the final two words.
export function ipv6ToInt(ip: string): bigint | null {
if (isIP(ip) !== 6) return null
let value = ip
if (value.includes(".")) {
const last = value.lastIndexOf(":")
const v4 = ipv4ToInt(value.slice(last + 1))!
value = value.slice(0, last + 1) + (v4 >>> 16).toString(16) + ":" + (v4 & 0xffff).toString(16)
}
const [left, right] = value.split("::")
const head = left ? left.split(":") : []
const tail = right ? right.split(":") : []
const words = right === undefined ? head : [...head, ...Array<string>(8 - head.length - tail.length).fill("0"), ...tail]
return words.reduce((n, word) => (n << 16n) | BigInt(parseInt(word, 16)), 0n)
}
function v6Range(cidr: string) {
const [ip, length] = cidr.split("/")
const shift = BigInt(128 - Number(length))
return { prefix: ipv6ToInt(ip!)! >> shift, shift }
}
export function parseCidrV6(cidr: string): { net: bigint; shift: bigint; prefixLen: number } | null {
const parts = cidr.split("/")
if (parts.length !== 2 || !/^\d+$/.test(parts[1]!)) return null
const prefixLen = Number(parts[1])
const addr = ipv6ToInt(parts[0]!)
if (addr === null || prefixLen < 0 || prefixLen > 128) return null
const shift = BigInt(128 - prefixLen)
return { net: addr >> shift, shift, prefixLen }
}
// IANA non-global assignments plus deprecated site-local and multicast.
// More-specific globally reachable allocations inside 2001::/23 take priority.
const nonPublicV6Ranges = [
"::/96", "64:ff9b:1::/48", "100::/64", "100:0:0:1::/64",
"2001::/23", "2001:db8::/32", "3fff::/20", "5f00::/16",
"fc00::/7", "fe80::/10", "fec0::/10", "ff00::/8",
].map(v6Range)
const publicProtocolV6Ranges = [
"2001::/32", // Teredo: IANA reachability is N/A, not non-global.
"2001:1::1/128", "2001:1::2/128", "2001:1::3/128",
"2001:3::/32", "2001:4:112::/48", "2001:20::/28", "2001:30::/28",
].map(v6Range)
export function isNonPublicIp(ip: string): boolean {
const trimmed = canonicalIp(ip)
if (!trimmed) return true
if (trimmed.includes(":")) {
const lower = trimmed.toLowerCase()
return lower === "::1" || lower.startsWith("fe80:") || lower.startsWith("fc") || lower.startsWith("fd") || lower === "::"
const n = ipv6ToInt(trimmed)
if (n === null) return true
if (publicProtocolV6Ranges.some(({ prefix, shift }) => n >> shift === prefix)) return false
return nonPublicV6Ranges.some(({ prefix, shift }) => n >> shift === prefix)
}
const n = ipv4ToInt(trimmed)
if (n == null) return true
+64 -1
View File
@@ -57,6 +57,7 @@ assert.ok(afterFirst >= 2)
enqueueRipeMisses(["8.8.8.10"])
await flushRipeQueueForTests()
assert.equal(ripeFetchCountForTests(), afterFirst)
assert.equal(lookupRipeCached("::ffff:808:808")?.asn, 15169, "Hex mapped address shares IPv4 cache")
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"]) {
@@ -156,4 +157,66 @@ enqueueRipeMisses(["8.8.4.4"])
await flushRipeQueueForTests()
assert.ok(ripeFetchCountForTests() > 0)
console.log("traffic-flow-ripe.test.ts: ok (including database pause/resume)")
resetRipeCacheForTests()
disableRipePersistForTests()
const ipv6Requests: string[] = []
setRipeFetchForTests(async input => {
const url = new URL(String(input))
ipv6Requests.push(url.searchParams.get("resource")!)
const body = url.pathname.includes("network-info")
? { data: { prefix: "2001:4860:4860::/48", asns: [15169] } }
: url.pathname.includes("maxmind-geo-lite")
? { data: { located_resources: [{ locations: [{ country: "US" }] }] } }
: { data: { holder: "GOOGLE" } }
return new Response(JSON.stringify(body), { status: 200 })
})
enqueueRipeMisses(["2001:4860:4860::8888/128"])
await flushRipeQueueForTests()
assert.equal(ipv6Requests[0], "2001:4860:4860::8888", "Host CIDR must not be sent as a RIPE resource")
assert.equal(lookupRipeCached("2001:4860:4860::8888")?.asn, 15169)
assert.equal(lookupRipeCached("2001:4860:4860:0:0:0:0:8844")?.country, "US")
const ipv6Fetches = ripeFetchCountForTests()
enqueueRipeMisses(["2001:4860:4860::8844"])
await flushRipeQueueForTests()
assert.equal(ripeFetchCountForTests(), ipv6Fetches)
seedRipeCacheForTests({ prefix: "2001:db8::/32", asn: 64500, country: "US", lat: null, lng: null, holder: "STALE", ok: true, fetchedAt: Date.now() })
for (const ip of ["2001:db8::1", "3fff::1", "fe90::1", "ff02::1", "100:0:0:1::1", "abcd:1234"]) {
const entry = lookupRipeCached(ip)
assert.equal(entry?.ok, false, ip)
assert.equal(entry?.prefix, `${ip}/128`, ip)
}
enqueueRipeMisses(["2001:db8::1", "3fff::1", "fe90::1", "ff02::1", "100:0:0:1::1", "abcd:1234"])
await flushRipeQueueForTests()
assert.equal(ripeFetchCountForTests(), ipv6Fetches, "Special/invalid IPv6 must not query RIPE")
const v6Entry = (prefix: string, asn: number, fetchedAt = Date.now()) => ({ prefix, asn, country: "US", lat: null, lng: null, holder: "TEST", ok: true, fetchedAt })
seedRipeCacheForTests(v6Entry("2001:4860::/32", 1))
seedRipeCacheForTests(v6Entry("2001:4860:4860::8888/128", 2))
assert.equal(lookupRipeCached("2001:4860:4860::8888")?.asn, 2, "Longest matching prefix wins")
assert.equal(lookupRipeCached("2001:4860:4860::8844")?.asn, 15169)
assert.equal(lookupRipeCached("2001:4860:1234::1")?.asn, 1, "Wide prefix fallback")
seedRipeCacheForTests(v6Entry("2001:4860:4860::8888/128", 3))
assert.equal(lookupRipeCached("2001:4860:4860::8888")?.asn, 3, "Replacement removes stale indexed row")
seedRipeCacheForTests(v6Entry("2001:4860:4860::8888/128", 4, Date.now() - 25 * 60 * 60_000))
assert.equal(lookupRipeCached("2001:4860:4860::8888")?.asn, 15169, "Expired longest prefix falls back")
for (let i = 0; i < 3000; i++) seedRipeCacheForTests(v6Entry(`2606:4700:${i.toString(16)}::/48`, 13335))
assert.equal(lookupRipeCached("2001:4860:4860::8844")?.asn, 15169)
assert.ok(ripeLastCandidateCountForTests() < 8, "IPv6 lookup must not scan unrelated /48 buckets")
resetRipeCacheForTests()
disableRipePersistForTests()
setRipeFetchForTests(async () => { throw new Error("timeout") })
enqueueRipeMisses(["2606:4700:4700::1111"])
await flushRipeQueueForTests()
assert.equal(lookupRipeCached("2606:4700:4700::1111")?.prefix, "2606:4700:4700::1111/128")
assert.equal(lookupRipeCached("2606:4700:4700::1111")?.ok, false)
assert.equal(lookupRipeCached("2606:4700:4700::1001"), null, "Negative /128 must not poison sibling host")
const afterV6Failure = ripeFetchCountForTests()
assert.ok(afterV6Failure > 0)
enqueueRipeMisses(["2606:4700:4700:0:0:0:0:1111"])
await flushRipeQueueForTests()
assert.equal(ripeFetchCountForTests(), afterV6Failure, "Expanded spelling hits negative host cache")
resetRipeCacheForTests()
assert.equal(lookupRipeCached("2606:4700:4700::1111"), null, "Invalidation clears IPv6 indexes")
console.log("traffic-flow-ripe.test.ts: ok (including IPv6 indexes and database pause/resume)")
+57 -5
View File
@@ -1,6 +1,6 @@
import { databaseBackgroundWork } from "./database-background-work.js"
import { dbAll, dbQuery } from "../db/index.js"
import { canonicalIp, ipv4ToInt, isNonPublicIp, parseCidrV4 } from "./traffic-flow-ip.js"
import { canonicalIp, ipv4ToInt, ipv6ToInt, isNonPublicIp, parseCidrV4, parseCidrV6 } from "./traffic-flow-ip.js"
import { resolveRipeCountry } from "./traffic-flow-brands.js"
export interface FlowIpMeta {
@@ -40,6 +40,14 @@ interface RipeIndexed {
/** /24 → кандидаты с prefixLen ≥ 24. Более широкие префиксы — в `wideIndex`. */
const v24Index = new Map<number, RipeIndexed[]>()
const wideIndex: RipeIndexed[] = []
interface RipeIndexedV6 {
entry: FlowIpMeta
net: bigint
shift: bigint
prefixLen: number
}
const v48Index = new Map<bigint, RipeIndexedV6[]>()
const wideV6Index: RipeIndexedV6[] = []
let lastCandidateCount = 0
let persistEnabled = true
@@ -75,6 +83,8 @@ export function invalidateRipeCache(): void {
recentFetches.length = 0
v24Index.clear()
wideIndex.length = 0
v48Index.clear()
wideV6Index.length = 0
lastCandidateCount = 0
loaded = persistEnabled ? false : true
}
@@ -115,6 +125,20 @@ function isFresh(entry: FlowIpMeta): boolean {
}
function unindexPrefix(prefix: string): void {
if (prefix.includes(":")) {
const parsed = parseCidrV6(prefix)
if (!parsed) return
if (parsed.prefixLen >= 48) {
const key = parsed.net >> BigInt(parsed.prefixLen - 48)
const next = v48Index.get(key)?.filter(row => row.entry.prefix !== prefix)
if (next?.length) v48Index.set(key, next)
else v48Index.delete(key)
} else {
const idx = wideV6Index.findIndex(row => row.entry.prefix === prefix)
if (idx >= 0) wideV6Index.splice(idx, 1)
}
return
}
const parsed = parseCidrV4(prefix)
if (!parsed) return
if (parsed.prefixLen >= 24) {
@@ -131,6 +155,18 @@ function unindexPrefix(prefix: string): void {
}
function indexEntry(entry: FlowIpMeta): void {
if (entry.prefix.includes(":")) {
const parsed = parseCidrV6(entry.prefix)
if (!parsed) return
const row = { entry, ...parsed }
if (parsed.prefixLen >= 48) {
const key = parsed.net >> BigInt(parsed.prefixLen - 48)
const list = v48Index.get(key)
if (list) list.push(row)
else v48Index.set(key, [row])
} else wideV6Index.push(row)
return
}
const parsed = parseCidrV4(entry.prefix)
if (!parsed) return
const row: RipeIndexed = {
@@ -269,13 +305,29 @@ function negative(prefix: string): FlowIpMeta {
}
}
function hostPrefix(ip: string): string {
return `${ip}/${ip.includes(":") ? 128 : 32}`
}
export function lookupRipeCached(ip: string): FlowIpMeta | null {
loadSqlite()
const trimmed = canonicalIp(ip)
lastCandidateCount = 0
if (!trimmed) return null
if (isNonPublicIp(trimmed)) {
return negative(`${trimmed.includes(":") ? trimmed : trimmed}/32`)
return negative(hostPrefix(trimmed))
}
if (trimmed.includes(":")) {
const addr = ipv6ToInt(trimmed)
if (addr === null) return null
const bucket = v48Index.get(addr >> 80n)
const candidates = bucket ? bucket.concat(wideV6Index) : wideV6Index
lastCandidateCount = candidates.length
let best: RipeIndexedV6 | null = null
for (const row of candidates) {
if (isFresh(row.entry) && addr >> row.shift === row.net && (!best || row.prefixLen > best.prefixLen)) best = row
}
return best ? { ...best.entry, country: resolveRipeCountry(best.entry.country, best.entry.asn, best.entry.holder) || "—" } : null
}
const addr = ipv4ToInt(trimmed)
if (addr == null) return null
@@ -368,7 +420,7 @@ async function resolveIp(ip: string): Promise<FlowIpMeta | null> {
recentFetches.push(Date.now())
try {
const net = await ripeJson("network-info", ip)
const prefix = pickPrefix(net) || `${ip}/32`
const prefix = pickPrefix(net) || hostPrefix(ip)
const existing = mem.get(prefix)
if (existing && isFresh(existing)) return existing
const asn = pickAsns(net)
@@ -403,7 +455,7 @@ async function resolveIp(ip: string): Promise<FlowIpMeta | null> {
persist(entry)
return entry
} catch {
const prefix = `${ip}/32`
const prefix = hostPrefix(ip)
const entry = negative(prefix)
remember(entry)
persist(entry)
@@ -451,7 +503,7 @@ export function enqueueRipeMisses(ips: Iterable<string>): void {
loadSqlite()
for (const raw of ips) {
if (queue.length >= MAX_QUEUE) break
const ip = String(raw ?? "").trim()
const ip = canonicalIp(raw)
if (!ip || isNonPublicIp(ip)) continue
if (lookupRipeCached(ip)) continue
if (queued.has(ip) || inflight.has(ip)) continue