diff --git a/app/(main)/statistics/page.tsx b/app/(main)/statistics/page.tsx index c30d097..28accd8 100644 --- a/app/(main)/statistics/page.tsx +++ b/app/(main)/statistics/page.tsx @@ -393,6 +393,15 @@ export default function StatisticsPage() { ) : null} + {!slices.serverId && isLive && !emptyCube ? ( + + Интернет сети + + KPI — уникальный payload без overlay и транзита EN. Сумма WAN по серверам не равна интернету сети; для uplink откройте сервер. + + + ) : null} + , iconClassName: "text-muted-foreground", }, diff --git a/backend/package.json b/backend/package.json index 94ce2b5..8042602 100644 --- a/backend/package.json +++ b/backend/package.json @@ -15,7 +15,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-ifindex.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-purge.test.ts && tsx src/services/traffic-flow-geoip.test.ts && tsx src/services/traffic-flow-facts.test.ts && tsx src/services/statistics-aggregate.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-ifindex.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-purge.test.ts && tsx src/services/traffic-flow-geoip.test.ts && tsx src/services/traffic-flow-facts.test.ts && tsx src/services/traffic-flow-facts-filter.test.ts && tsx src/services/statistics-aggregate.test.ts", "test:users": "tsx src/modules/users/iface-type.test.ts && tsx src/modules/users/bindings.test.ts", "test:pg": "tsx src/db/sql-bind.test.ts && tsx src/db/sqlite-json.test.ts && tsx src/db/traffic-flags.test.ts && tsx src/db/pg-schema.test.ts", "test:backups": "tsx src/services/s3-backup-client.test.ts", diff --git a/backend/src/services/statistics-aggregate.test.ts b/backend/src/services/statistics-aggregate.test.ts index e7dbf72..047cec9 100644 --- a/backend/src/services/statistics-aggregate.test.ts +++ b/backend/src/services/statistics-aggregate.test.ts @@ -2,6 +2,7 @@ import assert from "node:assert/strict" import { getStatistics, getStatisticsPivot, parseStatisticsPeriod, pivotDimsConflict } from "./statistics-aggregate.js" import { rememberServerIfaces, resetIfaceCacheForTests } from "./traffic-flow-ifindex.js" import { setRefreshIfacesForTests } from "./traffic-flow-ifaces.js" +import { invalidateFlowCatalogCache } from "./traffic-flow-topology.js" import { withPgOrSkip } from "../test/pg.js" import { dbQuery } from "../db/index.js" import { ensurePartitionFor } from "../db/partitions.js" @@ -29,15 +30,25 @@ if (!(await withPgOrSkip())) { } const inserted = await dbQuery<{ id: number }>(` - INSERT INTO servers (name, host) VALUES ('stats-cube', '127.0.0.1') RETURNING id + INSERT INTO servers (name, host, type, wan_uplinks) + VALUES ('stats-cube', '127.0.0.1', 'jump-host', '[{"iface":"wan1"}]'::jsonb) + RETURNING id `) const serverId = inserted.rows[0]?.id if (serverId == null) throw new Error("no server") +const enInserted = await dbQuery<{ id: number }>(` + INSERT INTO servers (name, host, type) + VALUES ('stats-en', '198.51.100.1', 'exit-node') + RETURNING id +`) +const enId = enInserted.rows[0]?.id +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 dbQuery(`DELETE FROM flow_daily_facts WHERE server_id = $1`, [serverId]) -await dbQuery(`DELETE FROM flow_hour_facts WHERE server_id = $1`, [serverId]) +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]) await dbQuery(`DELETE FROM app_users WHERE id = 'u-stats-1'`) @@ -52,16 +63,29 @@ await dbQuery(` `, [serverId]) resetIfaceCacheForTests() -rememberServerIfaces(serverId, [{ name: "gre-client", ifindex: "2" }]) +rememberServerIfaces(serverId, [ + { name: "gre-client", ifindex: "2" }, + { name: "wan1", ifindex: "8" }, + { name: "gre-en", ifindex: "9" }, +]) +rememberServerIfaces(enId, [ + { name: "ether1", ifindex: "2" }, + { name: "gre-jh", ifindex: "5" }, +]) setRefreshIfacesForTests(async () => {}) +invalidateFlowCatalogCache() await dbQuery(` INSERT INTO flow_daily_facts (server_id, day, iface, country, service, asn, bytes, packets) VALUES ($1, '2026-09-10', '2', 'US', 'https', 15169, 800, 10), ($1, '2026-09-10', '2', 'DE', 'dns', 15133, 200, 4), - ($1, '2026-09-10', 'wan1', 'NL', 'other', 0, 70, 1) -`, [serverId]) + ($1, '2026-09-10', 'wan1', 'NL', 'other', 0, 70, 1), + ($1, '2026-09-10', '0', 'US', 'https', 0, 999, 3), + ($1, '2026-09-10', 'gre-en', 'US', 'https', 15169, 400, 2), + ($2, '2026-09-10', 'gre-jh', 'US', 'https', 15169, 500, 5), + ($2, '2026-09-10', 'ether1', 'US', 'https', 15169, 200, 2) +`, [serverId, enId]) try { const all = await getStatistics({ from: "2026-09-01", to: "2026-09-30" }) @@ -71,14 +95,41 @@ try { assert.ok(all.countries.some((r) => r.id === "US")) assert.ok(all.users.some((r) => r.id === "u-stats-1")) const unbound = all.users.find((r) => r.id === STATISTICS_UNBOUND_USER_ID) - assert.ok(unbound) - assert.equal(unbound.bytes, 70) + assert.equal(unbound, undefined, "WAN не в Прочие / Без привязки") assert.ok(all.servers.some((r) => r.id === String(serverId))) + assert.ok(!all.servers.some((r) => r.id === String(enId)), "EN-транзит не в сетевом KPI") const greIface = all.interfaces.find((r) => r.label.includes("gre-client")) assert.ok(greIface) assert.equal(greIface.bytes, 1000) assert.equal(greIface.id, `${serverId}:gre-client`) assert.ok(!all.interfaces.some((r) => /· (?:#)?\d+$/.test(r.label))) + assert.ok(!all.interfaces.some((r) => r.label.includes(" · —") || r.label.endsWith("· —"))) + assert.ok(!all.interfaces.some((r) => r.label.includes("gre-en"))) + const wanRow = all.interfaces.find((r) => r.id === `${serverId}:wan1`) + assert.ok(wanRow) + assert.ok(wanRow.label.includes("WAN · интернет")) + assert.equal(wanRow.bytes, 70) + + const nodeSlice = await getStatistics({ + from: "2026-09-01", + to: "2026-09-30", + serverId, + }) + assert.equal(nodeSlice.kpis.bytes, 1070) + const nodeWan = nodeSlice.interfaces.find((r) => r.id === `${serverId}:wan1`) + assert.ok(nodeWan) + assert.ok(nodeWan.label.includes("WAN · интернет")) + assert.ok(!nodeSlice.interfaces.some((r) => r.label.includes(" · —"))) + assert.ok(!nodeSlice.users.some((r) => r.id === STATISTICS_UNBOUND_USER_ID)) + + const enSlice = await getStatistics({ + from: "2026-09-01", + to: "2026-09-30", + serverId: enId, + }) + assert.equal(enSlice.kpis.bytes, 200) + assert.ok(enSlice.interfaces.some((r) => r.label.includes("WAN · интернет") && r.label.includes("ether1"))) + assert.ok(!enSlice.interfaces.some((r) => r.label.includes("gre-jh"))) const sliced = await getStatistics({ from: "2026-09-01", @@ -127,9 +178,10 @@ try { } finally { setRefreshIfacesForTests(null) resetIfaceCacheForTests() - await dbQuery(`DELETE FROM flow_daily_facts WHERE server_id = $1`, [serverId]) - await dbQuery(`DELETE FROM flow_hour_facts WHERE server_id = $1`, [serverId]) - await dbQuery(`DELETE FROM servers WHERE id = $1`, [serverId]) + invalidateFlowCatalogCache() + 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 servers WHERE id IN ($1, $2)`, [serverId, enId]) } console.log("statistics-aggregate.test.ts: ok") diff --git a/backend/src/services/statistics-aggregate.ts b/backend/src/services/statistics-aggregate.ts index 9b7b925..8e8f6ee 100644 --- a/backend/src/services/statistics-aggregate.ts +++ b/backend/src/services/statistics-aggregate.ts @@ -15,8 +15,17 @@ import { displayFactIface, expandBindingIfaces, factIfaceAliases, + listCachedIfaceNames, } from "./traffic-flow-ifindex.js" import { refreshServerIfaces } from "./traffic-flow-ifaces.js" +import { + isDashDisplayIface, + isJunkFactIface, + isOverlayGreIface, + isWanFactIface, + wanIfaceLabel, +} from "./traffic-flow-facts-filter.js" +import { getServerCatalog, loadFlowTopology, type FlowTopology } from "./traffic-flow-topology.js" const TOP_N = 200 const HOUR_WINDOW_MS = 48 * 3600_000 @@ -91,6 +100,10 @@ interface FilterCtx { userIfaces: Array<{ serverId: number; iface: string }> | null unboundOnly: boolean boundIfaces: Array<{ serverId: number; iface: string }> + overlayIfaces: Array<{ serverId: number; iface: string }> + wanIfaces: Array<{ serverId: number; iface: string }> + excludeServerIds: number[] + topo: FlowTopology | null } function ifaceFilterAliases(iface: string, serverId?: number): string[] { @@ -169,16 +182,31 @@ function factWhere(alias: string, grain: "hour" | "day", ctx: FilterCtx): { sql: } } if (ctx.unboundOnly) { - if (ctx.boundIfaces.length === 0) { - /* весь трафик без привязок */ - } else { - const tuples = ctx.boundIfaces.map(() => "(?, ?)").join(", ") + const skip = [...ctx.boundIfaces, ...ctx.wanIfaces] + if (skip.length) { + const tuples = skip.map(() => "(?, ?)").join(", ") parts.push(`(${alias}.server_id, ${alias}.iface) NOT IN (${tuples})`) - for (const u of ctx.boundIfaces) { - params.push(u.serverId, u.iface) - } + for (const u of skip) params.push(u.serverId, u.iface) } } + parts.push(`${alias}.iface NOT IN ('0', '—', '__unknown__', 'wg-flow', '')`) + const greLike = `(LOWER(${alias}.iface) LIKE 'gre%' OR LOWER(${alias}.iface) LIKE '%gre-tunnel%')` + if (ctx.boundIfaces.length) { + const tuples = ctx.boundIfaces.map(() => "(?, ?)").join(", ") + parts.push(`(NOT ${greLike} OR (${alias}.server_id, ${alias}.iface) IN (${tuples}))`) + for (const u of ctx.boundIfaces) params.push(u.serverId, u.iface) + } else { + parts.push(`NOT ${greLike}`) + } + if (ctx.overlayIfaces.length) { + const tuples = ctx.overlayIfaces.map(() => "(?, ?)").join(", ") + parts.push(`(${alias}.server_id, ${alias}.iface) NOT IN (${tuples})`) + for (const u of ctx.overlayIfaces) params.push(u.serverId, u.iface) + } + if (ctx.excludeServerIds.length) { + parts.push(`${alias}.server_id NOT IN (${ctx.excludeServerIds.map(() => "?").join(", ")})`) + params.push(...ctx.excludeServerIds) + } return { sql: parts.join(" AND "), params } } @@ -274,12 +302,61 @@ function userBindJoinSql(tuples: UserBindTuple[]): { sql: string; params: unknow } } +function expandIfaceTuples( + items: Array<{ serverId: number; iface: string }>, +): Array<{ serverId: number; iface: string }> { + const seen = new Set() + const out: Array<{ serverId: number; iface: string }> = [] + for (const t of items) { + for (const iface of factIfaceAliases(t.iface, t.serverId)) { + const k = `${t.serverId}\0${iface}` + if (seen.has(k)) continue + seen.add(k) + out.push({ serverId: t.serverId, iface }) + } + } + return out +} + +async function loadPayloadScope(serverId?: number): Promise<{ + overlayIfaces: Array<{ serverId: number; iface: string }> + wanIfaces: Array<{ serverId: number; iface: string }> + excludeServerIds: number[] + topo: FlowTopology +}> { + const topo = await loadFlowTopology() + const catalog = await getServerCatalog() + await warmIfaceCache(catalog.list.map((s) => s.id)) + const overlayRaw: Array<{ serverId: number; iface: string }> = [] + const wanRaw: Array<{ serverId: number; iface: string }> = [] + for (const s of catalog.list) { + if (serverId != null && s.id !== serverId) continue + const wanSet = topo.wanIfaces.get(s.id) + const wanNames = wanSet && wanSet.size > 0 + ? [...wanSet] + : s.type === "home-router" ? [] : ["ether1"] + for (const name of wanNames) wanRaw.push({ serverId: s.id, iface: name }) + for (const name of listCachedIfaceNames(s.id)) { + if (isOverlayGreIface(topo, s.id, name)) overlayRaw.push({ serverId: s.id, iface: name }) + } + } + return { + overlayIfaces: expandIfaceTuples(overlayRaw), + wanIfaces: expandIfaceTuples(wanRaw), + excludeServerIds: serverId != null + ? [] + : catalog.list.filter((s) => s.type === "exit-node").map((s) => s.id), + topo, + } +} + async function buildFilterCtx(query: StatisticsQuery, period: ParsedPeriod): Promise { const bindTuples = await loadBindUserTuples() const boundIfaces = uniqueBoundIfaces(bindTuples) const unboundOnly = query.userId === STATISTICS_UNBOUND_USER_ID const userIfaces = unboundOnly ? null : await resolveUserIfaces(query.userId) if (userIfaces && userIfaces.length === 0) return null + const scope = await loadPayloadScope(query.serverId) return { ...period, serverId: query.serverId, @@ -290,6 +367,10 @@ async function buildFilterCtx(query: StatisticsQuery, period: ParsedPeriod): Pro userIfaces, unboundOnly, boundIfaces, + overlayIfaces: scope.overlayIfaces, + wanIfaces: scope.wanIfaces, + excludeServerIds: scope.excludeServerIds, + topo: scope.topo, } } @@ -370,7 +451,11 @@ export async function getStatistics(query: StatisticsQuery): Promise looksLikeIfIndex(r.iface)).map((r) => r.serverId)) - const ifaceRows = collapseServerIfaceRows(ifaceRowsRaw) + const ifaceRows = collapseServerIfaceRows(ifaceRowsRaw).filter((r) => { + if (isJunkFactIface(r.iface) || isDashDisplayIface(r.iface)) return false + if (ctx.topo && isOverlayGreIface(ctx.topo, r.serverId, r.iface)) return false + return true + }) const ifaceCount = ifaceRows.length let userRows: Array<{ id: string; bytes: number; packets: number }> = [] @@ -442,12 +527,16 @@ export async function getStatistics(query: StatisticsQuery): Promise ({ - id: `${r.serverId}:${r.iface}`, - label: `${serverNames.get(r.serverId) || r.serverId} · ${r.iface}`, - bytes: Number(r.bytes) || 0, - packets: Number(r.packets) || 0, - })), + ifaceRows.map((r) => { + const serverName = serverNames.get(r.serverId) || String(r.serverId) + const wan = ctx.topo ? isWanFactIface(ctx.topo, r.serverId, r.iface) : false + return { + id: `${r.serverId}:${r.iface}`, + label: wan ? wanIfaceLabel(serverName, r.iface) : `${serverName} · ${r.iface}`, + bytes: Number(r.bytes) || 0, + packets: Number(r.packets) || 0, + } + }), bytes, period.windowSec, ) @@ -466,13 +555,14 @@ export async function getStatistics(query: StatisticsQuery): Promise "(?, ?)").join(", ") + const tuples = skipUnbound.map(() => "(?, ?)").join(", ") const unboundParams = [...where.params] - for (const u of ctx.boundIfaces) unboundParams.push(u.serverId, u.iface) + for (const u of skipUnbound) unboundParams.push(u.serverId, u.iface) const unboundRows = await dbAll<{ bytes: number; packets: number }>(` SELECT COALESCE(SUM(f.bytes), 0) AS bytes, COALESCE(SUM(f.packets), 0) AS packets FROM ${table} f @@ -486,7 +576,7 @@ export async function getStatistics(query: StatisticsQuery): Promise; col: Map }> { + const topo = await loadFlowTopology() const serverNames = new Map() const allServers = await db.select({ id: servers.id, name: servers.name, host: servers.host }).from(servers) for (const s of allServers) serverNames.set(String(s.id), s.name || s.host) @@ -733,7 +824,11 @@ async function loadPivotLabels( const iface = id.slice(colon + 1) const sidNum = Number(sid) const name = Number.isFinite(sidNum) ? displayFactIface(sidNum, iface) : iface - return `${serverNames.get(sid) || sid} · ${name}` + const serverName = serverNames.get(sid) || sid + if (Number.isFinite(sidNum) && isWanFactIface(topo, sidNum, name)) { + return wanIfaceLabel(serverName, name) + } + return `${serverName} · ${name}` } return id } diff --git a/backend/src/services/traffic-flow-engine.ts b/backend/src/services/traffic-flow-engine.ts index 7ab1f5d..11f054c 100644 --- a/backend/src/services/traffic-flow-engine.ts +++ b/backend/src/services/traffic-flow-engine.ts @@ -11,7 +11,14 @@ import { invalidateTrafficFlowSettingsCache } from "./traffic-flow-settings.js" import { isIsoCountry } from "./traffic-flow-brands.js" import { maybeRefreshIfaces } from "./traffic-flow-ifaces.js" import { canonicalFactIface } from "./traffic-flow-ifindex.js" +import { shouldWriteFlowFact } from "./traffic-flow-facts-filter.js" import { pickInternetPeer } from "./traffic-flow-ip.js" +import { + getServerCatalog, + loadFlowTopology, + peekFlowTopology, + peekServerCatalog, +} from "./traffic-flow-topology.js" import { bumpFlowFact, factsPendingSize, @@ -338,6 +345,11 @@ export function queueParsedFlows(serverId: number, flows: ParsedFlowInput[]): vo const bucketAt = minuteBucketIso() const hourAt = hourBucketIso() const ripeMisses: string[] = [] + const topo = peekFlowTopology() + const catalog = peekServerCatalog() + if (!topo) void loadFlowTopology().catch(() => {}) + if (!catalog) void getServerCatalog().catch(() => {}) + const serverType = catalog?.byId.get(serverId)?.type for (const raw of flows) { const flow = normalizeParsedFlow(raw) addToTick(serverId, flow, flow.bytes) @@ -358,16 +370,29 @@ export function queueParsedFlows(serverId: number, flows: ParsedFlowInput[]): vo bumpDim(serverId, bucketAt, "service", classified.service, flow.bytes, flow.packets) if (country) bumpDim(serverId, bucketAt, "country", country, flow.bytes, flow.packets) bumpDim(serverId, bucketAt, "asn", asnKey, flow.bytes, flow.packets) - bumpFlowFact({ + if (shouldWriteFlowFact({ serverId, - bucketAt: hourAt, - iface: canonicalFactIface(serverId, flow.inIface), - country: country || "XX", - service: classified.service, - asn: ripe?.ok && ripe.asn ? ripe.asn : 0, - bytes: flow.bytes, - packets: flow.packets, - }) + serverType, + inIface: flow.inIface, + outIface: flow.outIface, + proto: flow.proto, + srcPort: flow.srcPort, + dstPort: flow.dstPort, + src: flow.src, + dst: flow.dst, + topo, + })) { + bumpFlowFact({ + serverId, + bucketAt: hourAt, + iface: canonicalFactIface(serverId, flow.inIface), + country: country || "XX", + service: classified.service, + asn: ripe?.ok && ripe.asn ? ripe.asn : 0, + bytes: flow.bytes, + packets: flow.packets, + }) + } const key = pendingKey(serverId, bucketAt, flow) const prev = pending.get(key) diff --git a/backend/src/services/traffic-flow-facts-filter.test.ts b/backend/src/services/traffic-flow-facts-filter.test.ts new file mode 100644 index 0000000..c67b54b --- /dev/null +++ b/backend/src/services/traffic-flow-facts-filter.test.ts @@ -0,0 +1,132 @@ +import assert from "node:assert/strict" +import { + isJunkFactIface, + isOverlayGreIface, + isWanFactIface, + shouldWriteFlowFact, +} from "./traffic-flow-facts-filter.js" +import { seedFlowTopologyForTests, type FlowTopology } from "./traffic-flow-topology.js" +import { rememberServerIfaces, resetIfaceCacheForTests } from "./traffic-flow-ifindex.js" + +function topo(partial: Partial = {}): FlowTopology { + const wanIfaces = partial.wanIfaces ?? new Map([[1, new Set(["ether1"])]]) + const clientIfaces = partial.clientIfaces ?? new Map([[1, new Set(["gre-client"])]]) + const clientByIface = partial.clientByIface ?? new Map() + const enHosts = partial.enHosts ?? new Set(["198.51.100.1"]) + const jhHosts = partial.jhHosts ?? new Set(["203.0.113.10"]) + return { + clientIfaces, + clientByIface, + enNodes: partial.enNodes ?? [{ id: 2, name: "en", hosts: ["198.51.100.1"] }], + enHosts, + jhHosts, + wanIfaces, + plane: partial.plane ?? { + clientIfaceNames: new Set(["gre-client"]), + enHosts, + jhHosts, + }, + } +} + +resetIfaceCacheForTests() +rememberServerIfaces(1, [{ name: "ether1", ifindex: "2" }]) +seedFlowTopologyForTests(topo()) + +assert.equal(isJunkFactIface("0"), true) +assert.equal(isJunkFactIface(""), true) +assert.equal(isJunkFactIface("ether1"), false) +assert.equal(isWanFactIface(topo(), 1, "ether1"), true) +assert.equal(isOverlayGreIface(topo(), 1, "gre-en"), true) +assert.equal(isOverlayGreIface(topo(), 1, "gre-client"), false) +assert.equal(isOverlayGreIface(topo(), 1, "ether1"), false) + +const overlayOuter = shouldWriteFlowFact({ + serverId: 1, + serverType: "jump-host", + inIface: "ether1", + outIface: "gre-en", + proto: 47, + srcPort: 0, + dstPort: 0, + src: "203.0.113.10", + dst: "198.51.100.1", + topo: topo(), +}) +assert.equal(overlayOuter, false, "overlay proto 47 на ether1 не в facts") + +const payloadGre = shouldWriteFlowFact({ + serverId: 1, + serverType: "jump-host", + inIface: "gre-client", + outIface: "gre-en", + proto: 6, + srcPort: 51234, + dstPort: 443, + src: "10.100.1.17", + dst: "8.8.8.8", + topo: topo(), +}) +assert.equal(payloadGre, true, "payload на GRE — да") + +const payloadWan = shouldWriteFlowFact({ + serverId: 1, + serverType: "jump-host", + inIface: "ether1", + outIface: "gre-client", + proto: 6, + srcPort: 443, + dstPort: 51234, + src: "8.8.8.8", + dst: "10.100.1.17", + topo: topo(), +}) +assert.equal(payloadWan, true, "payload на ether1 WAN — да") + +const junkZero = shouldWriteFlowFact({ + serverId: 1, + serverType: "jump-host", + inIface: "0", + proto: 6, + srcPort: 443, + dstPort: 80, + src: "1.1.1.1", + dst: "8.8.8.8", + topo: topo(), +}) +assert.equal(junkZero, false) + +const enTopo = topo({ + clientIfaces: new Map([[2, new Set()]]), + wanIfaces: new Map([[2, new Set(["ether1"])]]), +}) +const enTransit = shouldWriteFlowFact({ + serverId: 2, + serverType: "exit-node", + inIface: "gre-jh", + outIface: "ether1", + proto: 6, + srcPort: 51234, + dstPort: 443, + src: "10.100.1.17", + dst: "8.8.8.8", + topo: enTopo, +}) +assert.equal(enTransit, false, "EN-транзит без клиента — нет") + +const enWan = shouldWriteFlowFact({ + serverId: 2, + serverType: "exit-node", + inIface: "ether1", + proto: 6, + srcPort: 443, + dstPort: 80, + src: "8.8.8.8", + dst: "198.51.100.1", + topo: enTopo, +}) +assert.equal(enWan, true, "WAN payload на EN — да") + +seedFlowTopologyForTests(null) +resetIfaceCacheForTests() +console.log("traffic-flow-facts-filter.test.ts: ok") diff --git a/backend/src/services/traffic-flow-facts-filter.ts b/backend/src/services/traffic-flow-facts-filter.ts new file mode 100644 index 0000000..b2b901c --- /dev/null +++ b/backend/src/services/traffic-flow-facts-filter.ts @@ -0,0 +1,85 @@ +import { mapRosInterfaceType } from "../modules/users/iface-type.js" +import { canonicalFactIface } from "./traffic-flow-ifindex.js" +import { classifyFlowPlane } from "./traffic-flow-planes.js" +import { resolveClient, type FlowTopology } from "./traffic-flow-topology.js" + +const JUNK_IFACE = new Set(["", "0", "—", "__unknown__", "wg-flow"]) + +export function isJunkFactIface(iface: string | null | undefined): boolean { + const n = String(iface ?? "").trim() + if (JUNK_IFACE.has(n)) return true + return /^#?0$/.test(n) +} + +export function isDashDisplayIface(iface: string): boolean { + return String(iface ?? "").trim() === "—" +} + +export function isGreIfaceName(name: string): boolean { + return mapRosInterfaceType("", name) === "gre" +} + +/** WAN uplink: `wanIfaces` топологии, иначе ether1 у JH/EN без wan_uplinks. */ +export function isWanFactIface( + topo: FlowTopology | null | undefined, + serverId: number, + iface: string, +): boolean { + const name = String(iface ?? "").trim() + if (!name || isJunkFactIface(name) || isDashDisplayIface(name)) return false + const wan = topo?.wanIfaces.get(serverId) + if (wan && wan.size > 0) return wan.has(name) + return /^ether1$/i.test(name) +} + +/** GRE между своими серверами / транзит EN: gre-имя, не клиент, не WAN. */ +export function isOverlayGreIface( + topo: FlowTopology | null | undefined, + serverId: number, + iface: string, +): boolean { + const name = String(iface ?? "").trim() + if (!name || isWanFactIface(topo, serverId, name)) return false + if (topo?.clientIfaces.get(serverId)?.has(name)) return false + if (name === "wg-flow") return false + return isGreIfaceName(name) +} + +export function shouldWriteFlowFact(opts: { + serverId: number + serverType?: string + inIface: string + outIface?: string + proto: number + srcPort: number + dstPort: number + src: string + dst: string + topo?: FlowTopology | null +}): boolean { + const inName = canonicalFactIface(opts.serverId, opts.inIface) || String(opts.inIface ?? "").trim() + if (isJunkFactIface(inName) || isJunkFactIface(opts.inIface)) return false + const outRaw = String(opts.outIface ?? "").trim() + const outName = outRaw ? (canonicalFactIface(opts.serverId, outRaw) || outRaw) : "" + const plane = classifyFlowPlane({ + src: opts.src, + dst: opts.dst, + proto: opts.proto, + srcPort: opts.srcPort, + dstPort: opts.dstPort, + inIface: inName, + outIface: outName || undefined, + }, opts.topo?.plane) + if (plane !== "payload") return false + if (opts.serverType === "exit-node" && opts.topo) { + const client = + resolveClient(opts.topo, opts.serverId, inName) + ?? (outName ? resolveClient(opts.topo, opts.serverId, outName) : null) + if (!client && isOverlayGreIface(opts.topo, opts.serverId, inName)) return false + } + return true +} + +export function wanIfaceLabel(serverName: string, iface: string): string { + return `${serverName} · ${iface} · WAN · интернет` +} diff --git a/backend/src/services/traffic-flow-ifindex.ts b/backend/src/services/traffic-flow-ifindex.ts index c1d3bf8..bb4829e 100644 --- a/backend/src/services/traffic-flow-ifindex.ts +++ b/backend/src/services/traffic-flow-ifindex.ts @@ -161,6 +161,12 @@ export function expandBindingIfaces( return out } +export function listCachedIfaceNames(serverId: number): string[] { + const map = cache.get(serverId) + if (!map) return [] + return [...new Set(map.values())] +} + export function ifaceCacheHas(serverId: number): boolean { return cache.has(serverId) } diff --git a/backend/src/services/traffic-flow-topology.ts b/backend/src/services/traffic-flow-topology.ts index dd84a81..93c22ba 100644 --- a/backend/src/services/traffic-flow-topology.ts +++ b/backend/src/services/traffic-flow-topology.ts @@ -48,6 +48,15 @@ export function invalidateFlowCatalogCache(): void { serverCatalogCache = null } +export function peekFlowTopology(): FlowTopology | null { + if (seeded) return seeded + return topologyCache?.topo ?? null +} + +export function peekServerCatalog(): { list: ServerCatalogEntry[]; byId: Map } | null { + return serverCatalogCache +} + export async function getServerCatalog(): Promise<{ list: ServerCatalogEntry[]; byId: Map }> { const now = Date.now() if (serverCatalogCache && now - serverCatalogCache.at < CATALOG_TTL_MS) {