fix(traffic): записывать потоки при подмене UDP-источника в Docker

Пакеты IPFIX доходили, но разговоры отбрасывались, если Docker подменял адрес jump-host на 172.x.

Co-authored-by: Cursor <cursoragent@cursor.com>

app/(main)/data-collection/page.tsx

@@ -54,6 +54,7 @@ import {
   type SchedulerJobGridRow,
 } from "@/components/data-grids/data-collection-scheduler-data-grid"
 import { DataPageCard } from "@/components/data-page-card"
+import { NetflowSettingsPanel } from "@/components/traffic/netflow-settings-panel"
 import { cn } from "@/lib/utils"
 import {
   AlertCircleIcon,
@@ -1210,6 +1211,8 @@ export default function DataCollectionPage() {
                   </div>
               </DataPageCard>
 
+              {isLive ? <NetflowSettingsPanel backendUrl={backendUrl} enabled={isLive} /> : null}
+
               <OpsPanel
                 title="Журнал прогонов"
                 description="SQLite `scheduler_runs` — до 80 записей; раскройте строку для полей и текста ошибки."

app/(main)/traffic/page.tsx

@@ -11,13 +11,20 @@ import { StatusDot } from "@/components/status-dot"
 import { Flag } from "@/components/flag"
 import {
   RefreshCwIcon, DownloadIcon, TrendingUpIcon, TrendingDownIcon,
-  ArrowUpIcon, ArrowDownIcon, ActivityIcon, UsersIcon, CableIcon, ServerIcon, SearchIcon,
+  ArrowUpIcon, ArrowDownIcon, ActivityIcon, UsersIcon, CableIcon, ServerIcon, SearchIcon, GitBranchIcon, PlusIcon,
 } from "lucide-react"
 import { cn } from "@/lib/utils"
 import { fmtGB, fmtRate } from "@/lib/fmt-rate"
 import { useDataSource } from "@/lib/data-source"
 import { useTrafficLive } from "@/hooks/use-traffic-live"
 import { requestJson } from "@/shared/api/http-client"
+import { getTrafficFlows } from "@/shared/api/traffic-flow"
+import { listServers } from "@/shared/api/servers"
+import { TrafficFlowsDataGrid } from "@/components/data-grids/traffic-flows-data-grid"
+import { FlowOverlaySheet } from "@/components/traffic/flow-overlay-sheet"
+import { DataPageCard } from "@/components/data-page-card"
+import type { FlowStatsDto } from "@mmapp/contracts/traffic-flow"
+import type { ServerRead } from "@mmapp/contracts/servers"
 import { Badge } from "@/components/reui/badge"
 import {
   IFACE_TYPE_LABEL,
@@ -276,7 +283,7 @@ const TRAFFIC_RANGE_LABELS: Record<Range, string> = {
   "24h": "24ч",
 }
 
-type GroupMode = "servers" | "users" | "ifaces"
+type GroupMode = "servers" | "users" | "ifaces" | "flows"
 type SortField = "rx" | "tx" | "name" | "sessions"
 type SortDir   = "asc" | "desc"
 
@@ -701,6 +708,7 @@ const GROUP_MODES: Array<{ mode: GroupMode; icon: ReactNode; label: string }> =
   { mode: "servers", icon: <ServerIcon className="size-3" />, label: "Серверы" },
   { mode: "users",   icon: <UsersIcon  className="size-3" />, label: "Клиенты" },
   { mode: "ifaces",  icon: <CableIcon  className="size-3" />, label: "Интерфейсы" },
+  { mode: "flows",   icon: <GitBranchIcon className="size-3" />, label: "Потоки" },
 ]
 
 const SORT_FIELDS: Array<{ field: SortField; label: string; modesOnly?: GroupMode[] }> = [
@@ -731,6 +739,9 @@ export default function TrafficPage() {
   const [liveDetailServer, setLiveDetailServer] = useState<ServerTraffic | null>(null)
   const [liveUsers, setLiveUsers] = useState<UserTraffic[]>([])
   const [liveBoundIfaces, setLiveBoundIfaces] = useState<BoundIfaceTraffic[]>([])
+  const [flowStats, setFlowStats] = useState<FlowStatsDto | null>(null)
+  const [overlayOpen, setOverlayOpen] = useState(false)
+  const [catalogServers, setCatalogServers] = useState<ServerRead[]>([])
   const effectiveMode: GroupMode = groupMode
   const { sample: liveSample, error: liveStreamError } = useTrafficLive({
     enabled: isLive && effectiveMode === "servers" && liveServers.some((s) => s.id === selectedId),
@@ -821,6 +832,30 @@ export default function TrafficPage() {
     void loadLiveTraffic(range)
   }, [isLive, range, loadLiveTraffic])
 
+  const loadFlows = useCallback(async () => {
+    if (!isLive) return
+    setLiveBusy(true)
+    setLiveError(null)
+    try {
+      const stats = await getTrafficFlows(backendUrl, range)
+      setFlowStats(stats)
+    } catch (e) {
+      setLiveError(e instanceof Error ? e.message : "Не удалось загрузить потоки")
+    } finally {
+      setLiveBusy(false)
+    }
+  }, [isLive, backendUrl, range])
+
+  useEffect(() => {
+    if (!isLive || effectiveMode !== "flows") return
+    void loadFlows()
+  }, [isLive, effectiveMode, loadFlows])
+
+  useEffect(() => {
+    if (!isLive) return
+    void listServers(backendUrl).then(setCatalogServers).catch(() => setCatalogServers([]))
+  }, [isLive, backendUrl])
+
   const activeServerTraffic = isLive ? liveServers : serverTraffic
   const activeUserTraffic = isLive ? liveUsers : userTraffic
   const activeBoundIfaces = isLive ? liveBoundIfaces : boundIfaces
@@ -863,7 +898,7 @@ export default function TrafficPage() {
     setGroupMode(next)
     if (next === "servers") setSelectedId(activeServerTraffic[0]?.id ?? "srv1")
     else if (next === "users") setSelectedId((isLive ? liveUsers : userTraffic)[0]?.id ?? "u1")
-    else setSelectedId((isLive ? liveBoundIfaces : boundIfaces)[0]?.id ?? "")
+    else if (next === "ifaces") setSelectedId((isLive ? liveBoundIfaces : boundIfaces)[0]?.id ?? "")
     setSortField("rx")
     setSortDir("desc")
     setSearch("")
@@ -935,6 +970,68 @@ export default function TrafficPage() {
 
   const visibleSortFields = SORT_FIELDS.filter(s => !s.modesOnly || s.modesOnly.includes(effectiveMode))
 
+  const flowKpiItems = [
+    {
+      id: "exporters",
+      label: "Экспортёры",
+      value: String(flowStats?.exportersOnline ?? 0),
+      icon: <ServerIcon className="size-4" />,
+      iconClassName: "text-info",
+    },
+    {
+      id: "bytes",
+      label: "Байт/мин",
+      value: flowStats ? fmtRate((flowStats.bytesPerMin * 8) / 1_000_000) : "—",
+      icon: <ActivityIcon className="size-4" />,
+      iconClassName: "text-success",
+    },
+    {
+      id: "src",
+      label: "Уник. src",
+      value: String(flowStats?.uniqueSrc ?? 0),
+      icon: <ArrowUpIcon className="size-4" />,
+      iconClassName: "text-muted-foreground",
+    },
+    {
+      id: "proto",
+      label: "Топ протокол",
+      value: flowStats?.topProto ?? "—",
+      icon: <GitBranchIcon className="size-4" />,
+      iconClassName: "text-warning",
+    },
+  ]
+
+  const counterKpiItems = [
+              {
+                id: "rx",
+                label: "RX сейчас",
+                value: fmtRate(totalRx),
+                icon: <ArrowDownIcon className="size-4" />,
+                iconClassName: "text-success",
+              },
+              {
+                id: "tx",
+                label: "TX сейчас",
+                value: fmtRate(totalTx),
+                icon: <ArrowUpIcon className="size-4" />,
+                iconClassName: "text-info",
+              },
+              {
+                id: "peak-rx",
+                label: "Пик RX",
+                value: fmtRate(peakRx),
+                icon: <TrendingUpIcon className="size-4" />,
+                iconClassName: "text-warning",
+              },
+              {
+                id: "peak-tx",
+                label: "Пик TX",
+                value: fmtRate(peakTx),
+                icon: <TrendingUpIcon className="size-4" />,
+                iconClassName: "text-warning",
+              },
+            ]
+
   return (
     <div className="flex flex-col h-full">
       <PageHeader
@@ -944,7 +1041,10 @@ export default function TrafficPage() {
             <Button
               variant="outline"
               size="sm"
-              onClick={() => { void loadLiveTraffic(range) }}
+              onClick={() => {
+                if (effectiveMode === "flows") void loadFlows()
+                else void loadLiveTraffic(range)
+              }}
               disabled={isLive && liveBusy}
             >
               <RefreshCwIcon className={cn("size-4", isLive && liveBusy && "animate-spin")} />Обновить
@@ -975,40 +1075,65 @@ export default function TrafficPage() {
         <div className="flex flex-col gap-5">
           <KpiStatGrid
             aria-label="Сводка трафика"
-            items={[
-              {
-                id: "rx",
-                label: "RX сейчас",
-                value: fmtRate(totalRx),
-                icon: <ArrowDownIcon className="size-4" />,
-                iconClassName: "text-success",
-              },
-              {
-                id: "tx",
-                label: "TX сейчас",
-                value: fmtRate(totalTx),
-                icon: <ArrowUpIcon className="size-4" />,
-                iconClassName: "text-info",
-              },
-              {
-                id: "peak-rx",
-                label: "Пик RX",
-                value: fmtRate(peakRx),
-                icon: <TrendingUpIcon className="size-4" />,
-                iconClassName: "text-warning",
-              },
-              {
-                id: "peak-tx",
-                label: "Пик TX",
-                value: fmtRate(peakTx),
-                icon: <TrendingUpIcon className="size-4" />,
-                iconClassName: "text-warning",
-              },
-            ]}
+            items={effectiveMode === "flows" ? flowKpiItems : counterKpiItems}
           />
 
-          <div className="grid grid-cols-[300px_1fr] gap-5 items-start">
-
+          {effectiveMode === "flows" ? (
+            <div className="flex flex-col gap-3">
+              <div className="flex items-center justify-between gap-2 flex-wrap">
+                <p className="text-sm text-muted-foreground">
+                  IPFIX top-разговоры. Счётчики интерфейсов — в режимах Серверы / Клиенты / Интерфейсы.
+                </p>
+                <div className="flex items-center gap-2 flex-wrap">
+                  <div className="flex gap-1">
+                    {TRAFFIC_RANGE_KEYS.map((key) => (
+                      <button
+                        key={key}
+                        type="button"
+                        onClick={() => setRange(key)}
+                        className={cn(
+                          "text-[10px] px-2 py-0.5 rounded border transition-colors",
+                          range === key
+                            ? "border-primary bg-primary/10 text-primary font-medium"
+                            : "border-border text-muted-foreground hover:text-foreground",
+                        )}
+                      >
+                        {TRAFFIC_RANGE_LABELS[key]}
+                      </button>
+                    ))}
+                  </div>
+                  <Button size="sm" onClick={() => setOverlayOpen(true)} disabled={!isLive}>
+                    <PlusIcon className="size-4" />
+                    Подключить JH
+                  </Button>
+                </div>
+              </div>
+              {liveError && (
+                <div className="text-xs text-destructive bg-destructive/10 border border-destructive/20 rounded-md px-3 py-2">
+                  {liveError}
+                </div>
+              )}
+              <DataPageCard>
+                <TrafficFlowsDataGrid
+                  rows={flowStats?.talkers ?? []}
+                  emptyHint={
+                    flowStats?.packetsReceived
+                      ? (flowStats.lastError
+                        || `IPFIX приходит (${flowStats.lastExporterIp ?? "экспортёр"}), но разговоры ещё не записаны.`)
+                      : undefined
+                  }
+                />
+              </DataPageCard>
+              <FlowOverlaySheet
+                open={overlayOpen}
+                onOpenChange={setOverlayOpen}
+                servers={catalogServers}
+                backendUrl={backendUrl}
+                onDone={() => { void loadFlows() }}
+              />
+            </div>
+          ) : (
+            <div className="grid grid-cols-[300px_1fr] gap-5 items-start">
             {/* ── left panel ── */}
             <div className="flex flex-col gap-3">
 
@@ -1127,6 +1252,7 @@ export default function TrafficPage() {
             </Frame>
 
           </div>
+          )}
         </div>
       </div>
     </div>

backend/package.json

@@ -14,6 +14,7 @@
     "test:auth": "tsx src/lib/permissions.test.ts && tsx src/plugins/auth.smoke.test.ts",
     "test:wireguard": "npx tsx src/services/wireguard-config.test.ts",
     "test:traffic-rate": "tsx src/services/traffic-rate.test.ts",
+    "test:traffic-flow": "tsx src/services/traffic-flow-parse.test.ts && tsx src/services/traffic-flow-map-exporter.test.ts",
     "test:users": "tsx src/modules/users/iface-type.test.ts && tsx src/modules/users/bindings.test.ts"
   },
   "dependencies": {

backend/src/db/index.ts

@@ -121,6 +121,47 @@ CREATE INDEX IF NOT EXISTS idx_traffic_samples_server_time
 CREATE INDEX IF NOT EXISTS idx_traffic_samples_server_iface_time
   ON traffic_samples(server_id, interface_name, sampled_at);
 
+CREATE TABLE IF NOT EXISTS traffic_flow_settings (
+  id INTEGER PRIMARY KEY,
+  enabled INTEGER NOT NULL DEFAULT 0,
+  collector_ip TEXT NOT NULL DEFAULT '10.255.254.1',
+  flow_listen_port INTEGER NOT NULL DEFAULT 4739,
+  wg_listen_port INTEGER NOT NULL DEFAULT 51821,
+  prefix TEXT NOT NULL DEFAULT '10.255.254.0/24',
+  public_endpoint TEXT NOT NULL DEFAULT '',
+  host_public_key TEXT NOT NULL DEFAULT '',
+  host_private_key TEXT NOT NULL DEFAULT '',
+  hub_server_id INTEGER,
+  retention_hours INTEGER NOT NULL DEFAULT 24,
+  top_n INTEGER NOT NULL DEFAULT 200,
+  last_datagram_at TEXT,
+  last_exporter_ip TEXT,
+  last_error TEXT,
+  packets_received INTEGER NOT NULL DEFAULT 0,
+  peers_json TEXT NOT NULL DEFAULT '[]',
+  created_at TEXT NOT NULL DEFAULT (datetime('now')),
+  updated_at TEXT NOT NULL DEFAULT (datetime('now'))
+);
+
+CREATE TABLE IF NOT EXISTS flow_buckets (
+  id INTEGER PRIMARY KEY AUTOINCREMENT,
+  server_id INTEGER NOT NULL,
+  bucket_at TEXT NOT NULL,
+  src TEXT NOT NULL,
+  dst TEXT NOT NULL,
+  proto INTEGER NOT NULL DEFAULT 0,
+  src_port INTEGER NOT NULL DEFAULT 0,
+  dst_port INTEGER NOT NULL DEFAULT 0,
+  bytes INTEGER NOT NULL DEFAULT 0,
+  packets INTEGER NOT NULL DEFAULT 0,
+  in_iface TEXT NOT NULL DEFAULT '',
+  FOREIGN KEY (server_id) REFERENCES servers(id) ON DELETE CASCADE
+);
+CREATE UNIQUE INDEX IF NOT EXISTS idx_flow_buckets_unique
+  ON flow_buckets(server_id, bucket_at, src, dst, proto, src_port, dst_port);
+CREATE INDEX IF NOT EXISTS idx_flow_buckets_server_time
+  ON flow_buckets(server_id, bucket_at);
+
 CREATE TABLE IF NOT EXISTS uptime_settings (
   id INTEGER PRIMARY KEY,
   enabled INTEGER NOT NULL DEFAULT 1,
@@ -674,6 +715,9 @@ if (!serverCols.some((c) => c.name === "lan_subnet")) {
 if (!serverCols.some((c) => c.name === "wan_uplinks")) {
   sqlite.exec(`ALTER TABLE servers ADD COLUMN wan_uplinks TEXT NOT NULL DEFAULT '[]'`)
 }
+if (!serverCols.some((c) => c.name === "mgmt_tunnel_ip")) {
+  sqlite.exec(`ALTER TABLE servers ADD COLUMN mgmt_tunnel_ip TEXT NOT NULL DEFAULT ''`)
+}
 
 const alertTgCols = sqlite.prepare(`PRAGMA table_info('alert_telegram_settings')`).all() as Array<{ name?: string }>
 if (!alertTgCols.some((c) => c.name === "message_thread_id")) {
@@ -719,6 +763,12 @@ SELECT 1, 1, 30, 14
 WHERE NOT EXISTS (SELECT 1 FROM traffic_settings WHERE id = 1);
 `)
 
+sqlite.exec(`
+INSERT INTO traffic_flow_settings (id, enabled, collector_ip, flow_listen_port, wg_listen_port, prefix)
+SELECT 1, 0, '10.255.254.1', 4739, 51821, '10.255.254.0/24'
+WHERE NOT EXISTS (SELECT 1 FROM traffic_flow_settings WHERE id = 1);
+`)
+
 sqlite.exec(`
 INSERT INTO uptime_settings (id, enabled, interval_sec, retention_days)
 SELECT 1, 1, 15, 14

backend/src/db/schema.ts

@@ -32,6 +32,8 @@ export const servers = sqliteTable("servers", {
   lanSubnet: text("lan_subnet").notNull().default(""),
   /** JSON-массив WAN-аплинков [{ id, name, isp, iface, ip, maxDl, maxUl }, …] */
   wanUplinks: text("wan_uplinks").notNull().default("[]"),
+  /** Адрес в оверлее wg-flow (экспортёр IPFIX), например 10.255.254.5 */
+  mgmtTunnelIp: text("mgmt_tunnel_ip").notNull().default(""),
 
   createdAt: text("created_at").notNull().default(sql`(datetime('now'))`),
   updatedAt: text("updated_at").notNull().default(sql`(datetime('now'))`),
@@ -158,6 +160,48 @@ export const alertBgpPeerSamples = sqliteTable("alert_bgp_peer_samples", {
 
 // ── raw traffic samples (per server/interface/timepoint) ──────────────────────
 
+export const trafficFlowSettings = sqliteTable("traffic_flow_settings", {
+  id:              integer("id").primaryKey(),
+  enabled:         integer("enabled", { mode: "boolean" }).notNull().default(false),
+  collectorIp:     text("collector_ip").notNull().default("10.255.254.1"),
+  flowListenPort:  integer("flow_listen_port").notNull().default(4739),
+  wgListenPort:    integer("wg_listen_port").notNull().default(51821),
+  prefix:          text("prefix").notNull().default("10.255.254.0/24"),
+  publicEndpoint:  text("public_endpoint").notNull().default(""),
+  hostPublicKey:   text("host_public_key").notNull().default(""),
+  hostPrivateKey:  text("host_private_key").notNull().default(""),
+  hubServerId:     integer("hub_server_id"),
+  retentionHours:  integer("retention_hours").notNull().default(24),
+  topN:            integer("top_n").notNull().default(200),
+  lastDatagramAt:  text("last_datagram_at"),
+  lastExporterIp:  text("last_exporter_ip"),
+  lastError:       text("last_error"),
+  packetsReceived: integer("packets_received").notNull().default(0),
+  peersJson:       text("peers_json").notNull().default("[]"),
+  createdAt:       text("created_at").notNull().default(sql`(datetime('now'))`),
+  updatedAt:       text("updated_at").notNull().default(sql`(datetime('now'))`),
+})
+
+export const flowBuckets = sqliteTable("flow_buckets", {
+  id:        integer("id").primaryKey({ autoIncrement: true }),
+  serverId:  integer("server_id")
+               .notNull()
+               .references(() => servers.id, { onDelete: "cascade" }),
+  bucketAt:  text("bucket_at").notNull(),
+  src:       text("src").notNull(),
+  dst:       text("dst").notNull(),
+  proto:     integer("proto").notNull().default(0),
+  srcPort:   integer("src_port").notNull().default(0),
+  dstPort:   integer("dst_port").notNull().default(0),
+  bytes:     integer("bytes").notNull().default(0),
+  packets:   integer("packets").notNull().default(0),
+  inIface:   text("in_iface").notNull().default(""),
+}, (t) => [
+  uniqueIndex("idx_flow_buckets_unique").on(
+    t.serverId, t.bucketAt, t.src, t.dst, t.proto, t.srcPort, t.dstPort,
+  ),
+])
+
 export const trafficSamples = sqliteTable("traffic_samples", {
   id:            integer("id").primaryKey({ autoIncrement: true }),
   serverId:      integer("server_id")
@@ -595,6 +639,8 @@ export type SnapshotInsert = typeof serverSnapshots.$inferInsert
 export type FilterRuleRow  = typeof filterRules.$inferSelect
 export type RecursiveRouteRow = typeof recursiveRoutes.$inferSelect
 export type TrafficSettingsRow = typeof trafficSettings.$inferSelect
+export type TrafficFlowSettingsRow = typeof trafficFlowSettings.$inferSelect
+export type FlowBucketRow = typeof flowBuckets.$inferSelect
 export type ServersApiPingSettingsRow = typeof serversApiPingSettings.$inferSelect
 export type TrafficSampleRow = typeof trafficSamples.$inferSelect
 export type UptimeSettingsRow = typeof uptimeSettings.$inferSelect

backend/src/index.ts

@@ -10,6 +10,7 @@ import execRoutes from "./routes/exec.js"
 import filtersRoutes from "./routes/filters.js"
 import recursiveRoutes from "./routes/recursive-routes.js"
 import trafficRoutes from "./routes/traffic.js"
+import trafficFlowRoutes from "./routes/traffic-flow.js"
 import serversApiPingRoutes from "./routes/servers-api-ping.js"
 import uptimeRoutes from "./routes/uptime.js"
 import networkRoutes from "./routes/network.js"
@@ -27,6 +28,7 @@ import wireguardRoutes from "./routes/wireguard.js"
 import firewallRoutes from "./routes/firewall.js"
 import usersRoutes from "./routes/users.js"
 import { refreshScheduler, stopScheduler } from "./services/scheduler.js"
+import { startTrafficFlowListener, stopTrafficFlowListener } from "./services/traffic-flow-ingest.js"
 
 export async function buildApp(opts?: {
   logger?: boolean
@@ -93,6 +95,7 @@ export async function buildApp(opts?: {
   await app.register(filtersRoutes, { prefix: "/api" })
   await app.register(recursiveRoutes, { prefix: "/api" })
   await app.register(trafficRoutes, { prefix: "/api" })
+  await app.register(trafficFlowRoutes, { prefix: "/api" })
   await app.register(serversApiPingRoutes, { prefix: "/api" })
   await app.register(uptimeRoutes, { prefix: "/api" })
   await app.register(networkRoutes, { prefix: "/api" })
@@ -112,8 +115,10 @@ export async function buildApp(opts?: {
 
   if (opts?.startScheduler !== false) {
     refreshScheduler()
+    startTrafficFlowListener()
     app.addHook("onClose", async () => {
       stopScheduler()
+      stopTrafficFlowListener()
     })
   }
 

backend/src/routes/traffic-flow.ts

@@ -0,0 +1,99 @@
+import type { FastifyPluginAsyncZod } from "@fastify/type-provider-zod"
+import type { FastifyReply, FastifyRequest } from "fastify"
+import {
+  trafficFlowOverlayRequestSchema,
+  trafficFlowSettingsPatchSchema,
+} from "@mmapp/contracts/traffic-flow"
+import {
+  ensureHostKeys,
+  getTrafficFlowSettingsRow,
+  toTrafficFlowSettingsDto,
+  updateTrafficFlowSettings,
+} from "../services/traffic-flow-settings.js"
+import {
+  getFlowListenerState,
+  listFlowTalkers,
+  startTrafficFlowListener,
+} from "../services/traffic-flow-ingest.js"
+import { applyFlowOverlay } from "../services/traffic-flow-overlay.js"
+import { listTrafficFlowHostFiles } from "../services/traffic-flow-host-files.js"
+
+function rangeToMinutes(range: string | undefined): number {
+  switch ((range ?? "5m").toLowerCase()) {
+    case "5m": return 5
+    case "15m": return 15
+    case "1h": return 60
+    case "4h": return 240
+    case "24h": return 1440
+    default: return 5
+  }
+}
+
+async function sendFlowTalkers(req: FastifyRequest, reply: FastifyReply) {
+  const q = req.query as { range?: string }
+  return reply.send(listFlowTalkers(rangeToMinutes(q.range)))
+}
+
+function requestPublicHost(req: FastifyRequest): string {
+  const forwarded = req.headers["x-forwarded-host"]
+  const raw = Array.isArray(forwarded) ? forwarded[0] : forwarded
+  return raw || req.hostname || ""
+}
+
+async function applyOverlayHandler(req: FastifyRequest, reply: FastifyReply) {
+  const parsed = trafficFlowOverlayRequestSchema.safeParse(req.body ?? {})
+  if (!parsed.success) {
+    return reply.status(400).send({ error: "Некорректное тело запроса", details: parsed.error.flatten() })
+  }
+  try {
+    const result = await applyFlowOverlay(parsed.data.serverId, {
+      publicEndpoint: parsed.data.publicEndpoint,
+      requestHost: requestPublicHost(req),
+    })
+    return reply.send(result)
+  } catch (e) {
+    const status = (e as { statusCode?: number }).statusCode ?? 502
+    const msg = e instanceof Error ? e.message : String(e)
+    return reply.status(status).send({ error: msg })
+  }
+}
+
+const trafficFlowRoutes: FastifyPluginAsyncZod = async (app) => {
+  app.get("/traffic/flow/settings", async (_req, reply) => {
+    return reply.send(toTrafficFlowSettingsDto(getFlowListenerState()))
+  })
+
+  app.put("/traffic/flow/settings", async (req, reply) => {
+    const parsed = trafficFlowSettingsPatchSchema.safeParse(req.body ?? {})
+    if (!parsed.success) {
+      return reply.status(400).send({ error: "Некорректное тело запроса", details: parsed.error.flatten() })
+    }
+    updateTrafficFlowSettings(parsed.data)
+    startTrafficFlowListener()
+    return reply.send({ ok: true, settings: toTrafficFlowSettingsDto(getFlowListenerState()) })
+  })
+
+  app.post("/traffic/flow/settings/generate-keys", async (_req, reply) => {
+    const result = ensureHostKeys()
+    return reply.send({
+      ok: true,
+      created: result.created,
+      publicKey: result.publicKey,
+      settings: toTrafficFlowSettingsDto(getFlowListenerState()),
+    })
+  })
+
+  app.get("/traffic/flow/host-files", async (_req, reply) => {
+    const row = getTrafficFlowSettingsRow()
+    if (!row.hostPrivateKey) ensureHostKeys()
+    return reply.send({ files: listTrafficFlowHostFiles() })
+  })
+
+  app.post("/traffic/flow/overlay", applyOverlayHandler)
+  app.post("/traffic/flow-overlay", applyOverlayHandler)
+
+  app.get("/traffic/flow", sendFlowTalkers)
+  app.get("/traffic/flows", sendFlowTalkers)
+}
+
+export default trafficFlowRoutes

backend/src/routes/wireguard.ts

@@ -21,6 +21,12 @@ import {
   getEnabledServerById,
   listWireGuardInterfaces,
 } from "../services/wireguard-live.js"
+import {
+  putIpAddress,
+  putWireguardInterface,
+  putWireguardPeer,
+  toRosBody,
+} from "../services/wireguard-ros.js"
 
 function serverIdParam(v: string): string {
   return decodeURIComponent(v)
@@ -30,14 +36,6 @@ function rosIdParam(v: string): string {
   return decodeURIComponent(v)
 }
 
-function toRosBody(obj: Record<string, string | undefined>): Record<string, string> {
-  const out: Record<string, string> = {}
-  for (const [k, v] of Object.entries(obj)) {
-    if (v !== undefined && v !== "") out[k] = v
-  }
-  return out
-}
-
 function peerToRosBody(p: Omit<WgCreatePeerRequest, "serverId" | "interfaceName"> & { interfaceName: string }) {
   return toRosBody({
     interface: p.interfaceName,
@@ -99,20 +97,17 @@ async function applyParsedConfig(
     comment: parsed.interface.comment,
     disabled: parsed.interface.disabled ? "yes" : undefined,
   })
-  await client.put("/interface/wireguard", ifaceBody)
+  await putWireguardInterface(client, ifaceBody)
 
   if (parsed.interface.address) {
-    await client.put("/ip/address", {
-      address: parsed.interface.address,
-      interface: name,
-    })
+    await putIpAddress(client, parsed.interface.address, name)
   }
 
   let peersCreated = 0
   for (const p of parsed.peers) {
     if (!p.publicKey) continue
-    await client.put(
-      "/interface/wireguard/peers",
+    await putWireguardPeer(
+      client,
       peerToRosBody({
         interfaceName: name,
         publicKey: p.publicKey,
@@ -164,30 +159,21 @@ const wireguardRoutes: FastifyPluginAsyncZod = async (app) => {
 
     const client = MikrotikClient.fromServer(server)
     try {
-      await client.put(
-        "/interface/wireguard",
-        toRosBody({
-          name: body.name,
-          "listen-port": String(body.listenPort),
-          mtu: String(body.mtu),
-          comment: body.comment,
-          "private-key": body.privateKey,
-          disabled: body.disabled ? "yes" : undefined,
-        }),
-      )
+      await putWireguardInterface(client, {
+        name: body.name,
+        "listen-port": String(body.listenPort),
+        mtu: String(body.mtu),
+        comment: body.comment,
+        "private-key": body.privateKey,
+        disabled: body.disabled ? "yes" : undefined,
+      })
 
       if (body.address) {
-        await client.put("/ip/address", {
-          address: body.address,
-          interface: body.name,
-        })
+        await putIpAddress(client, body.address, body.name)
       }
 
       if (body.peer) {
-        await client.put(
-          "/interface/wireguard/peers",
-          peerToRosBody({ ...body.peer, interfaceName: body.name }),
-        )
+        await putWireguardPeer(client, peerToRosBody({ ...body.peer, interfaceName: body.name }))
       }
 
       const list = await listWireGuardInterfaces({
@@ -255,7 +241,7 @@ const wireguardRoutes: FastifyPluginAsyncZod = async (app) => {
     if (!server) return reply.status(404).send({ error: "Сервер не найден" })
     const client = MikrotikClient.fromServer(server)
     try {
-      await client.put("/interface/wireguard/peers", peerToRosBody(body))
+      await putWireguardPeer(client, peerToRosBody(body))
       return reply.status(201).send({ ok: true })
     } catch (e) {
       const msg = e instanceof Error ? e.message : String(e)

backend/src/services/traffic-flow-host-files.ts

@@ -0,0 +1,115 @@
+import { generateNativeConf } from "./wireguard-config.js"
+import { getTrafficFlowSettingsRow, listHostPeers } from "./traffic-flow-settings.js"
+import type { TrafficFlowHostFile } from "@mmapp/contracts/traffic-flow"
+
+const COMPOSE_DIR = "/opt/cdn-mm"
+
+export function buildHostWgQuickConf(): string {
+  const row = getTrafficFlowSettingsRow()
+  const peers = listHostPeers()
+  return generateNativeConf({
+    name: "wg-flow",
+    mtu: 1420,
+    privateKey: row.hostPrivateKey || undefined,
+    address: `${row.collectorIp}/24`,
+    comment: "MikrotikManager traffic-flow collector",
+    peers: peers.map((p) => ({
+      publicKey: p.publicKey,
+      allowedIps: p.allowedIps,
+      comment: p.name,
+      endpoint: p.endpoint,
+      persistentKeepalive: p.endpoint ? 25 : undefined,
+    })),
+  })
+}
+
+export function buildHostComposeOverride(): string {
+  const row = getTrafficFlowSettingsRow()
+  return [
+    "# Docker Compose merge для /opt/cdn-mm",
+    "# Не править docker-compose.yml. Traefik не трогать.",
+    "# Сначала: wg-quick up wg-flow  (адрес " + row.collectorIp + ")",
+    "# затем: docker compose up -d backend",
+    "# Docker userland-proxy может SNAT UDP source в 172.x — ingest сопоставит единственный JH.",
+    "",
+    "services:",
+    "  backend:",
+    "    environment:",
+    "      FLOW_LISTEN_HOST: \"0.0.0.0\"",
+    "    ports:",
+    `      - "${row.collectorIp}:${row.flowListenPort}:${row.flowListenPort}/udp"`,
+    "",
+  ].join("\n")
+}
+
+export function buildHostLinuxInstallSh(): string {
+  const row = getTrafficFlowSettingsRow()
+  const conf = buildHostWgQuickConf().replace(/\s+$/, "") + "\n"
+  const override = buildHostComposeOverride()
+  const collector = row.collectorIp
+  const flowPort = row.flowListenPort
+
+  return `#!/usr/bin/env bash
+# WG-клиент на хосте /opt/cdn-mm → JH:13232, IPFIX в контейнер backend.
+# Запуск: sudo bash install-wg-flow.sh
+set -euo pipefail
+
+if [[ \${EUID:-$(id -u)} -ne 0 ]]; then
+  echo "Запустите от root: sudo bash $0" >&2
+  exit 1
+fi
+
+COLLECTOR_IP="${collector}"
+FLOW_PORT="${flowPort}"
+COMPOSE_DIR="${COMPOSE_DIR}"
+
+if ! command -v wg >/dev/null 2>&1; then
+  apt-get update
+  DEBIAN_FRONTEND=noninteractive apt-get install -y wireguard
+fi
+
+install -d -m 700 /etc/wireguard
+cat > /etc/wireguard/wg-flow.conf <<'WGEOF'
+${conf}WGEOF
+chmod 600 /etc/wireguard/wg-flow.conf
+
+systemctl enable --now wg-quick@wg-flow
+echo "=== wg show wg-flow ==="
+wg show wg-flow
+echo "=== адрес (ожидаем \${COLLECTOR_IP}/24) ==="
+ip -4 addr show dev wg-flow
+
+if [[ ! -d "\$COMPOSE_DIR" ]]; then
+  echo "Нет \$COMPOSE_DIR — положите override.yml туда вручную (вкладка compose)." >&2
+  exit 1
+fi
+
+cat > "\$COMPOSE_DIR/docker-compose.override.yml" <<'OVEOF'
+${override}OVEOF
+
+cd "\$COMPOSE_DIR"
+docker compose up -d backend
+
+echo "=== UDP \${FLOW_PORT} на хосте ==="
+ss -ulnp | grep -E "\${FLOW_PORT}" || true
+echo "=== PortBindings mmapp-backend ==="
+docker inspect -f '{{json .HostConfig.PortBindings}}' mmapp-backend
+echo "=== handshake (keepalive 25s к JH:13232) ==="
+wg show wg-flow
+
+# ufw: исходящий WG не открывать; 4739 на WAN не публиковать
+if command -v ufw >/dev/null 2>&1; then
+  ufw deny "\${FLOW_PORT}/udp" comment 'ipfix-not-public' || true
+fi
+
+echo "Готово. Traefik не трогали. UDP \${FLOW_PORT} только на \${COLLECTOR_IP}, не на 0.0.0.0."
+`
+}
+
+export function listTrafficFlowHostFiles(): TrafficFlowHostFile[] {
+  return [
+    { id: "linux", label: "Linux", filename: "install-wg-flow.sh", code: buildHostLinuxInstallSh() },
+    { id: "wg-quick", label: "wg-flow.conf", filename: "wg-flow.conf", code: buildHostWgQuickConf() },
+    { id: "compose", label: "compose", filename: "docker-compose.override.yml", code: buildHostComposeOverride() },
+  ]
+}

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

@@ -0,0 +1,274 @@
+import { createSocket, type Socket } from "node:dgram"
+import { desc, eq, gte, sql } from "drizzle-orm"
+import { db } from "../db/index.js"
+import { flowBuckets, servers } from "../db/schema.js"
+import type { FlowStatsDto, FlowTalkerDto } from "@mmapp/contracts/traffic-flow"
+import { parseFlowPacket, protoName, type ParsedFlow } from "./traffic-flow-parse.js"
+import { pickServerIdForExporter } from "./traffic-flow-map-exporter.js"
+import {
+  getTrafficFlowSettingsRow,
+  listHostPeers,
+  recordFlowListenerError,
+  recordFlowPacket,
+} from "./traffic-flow-settings.js"
+
+export interface FlowListenerState {
+  bound: boolean
+  address: string | null
+}
+
+let socket: Socket | null = null
+let state: FlowListenerState = { bound: false, address: null }
+const pending = new Map<string, {
+  serverId: number
+  bucketAt: string
+  flow: ParsedFlow
+  bytes: number
+  packets: number
+}>()
+let flushTimer: ReturnType<typeof setInterval> | null = null
+
+export function getFlowListenerState(): FlowListenerState {
+  return state
+}
+
+function minuteBucketIso(at = Date.now()): string {
+  const d = new Date(at)
+  d.setSeconds(0, 0)
+  return d.toISOString()
+}
+
+function resolveServerId(exporterIp: string): number | null {
+  const settings = getTrafficFlowSettingsRow()
+  const rows = db.select({
+    id: servers.id,
+    host: servers.host,
+    mgmtTunnelIp: servers.mgmtTunnelIp,
+  }).from(servers).all()
+  const byTunnelIp = new Map<string, number>()
+  const hostIps = new Map<string, number>()
+  for (const row of rows) {
+    if (row.mgmtTunnelIp) byTunnelIp.set(row.mgmtTunnelIp, row.id)
+    if (/^\d{1,3}(?:\.\d{1,3}){3}$/.test(row.host)) hostIps.set(row.host, row.id)
+  }
+  return pickServerIdForExporter({
+    exporterIp,
+    overlayPrefix: settings.prefix,
+    byTunnelIp,
+    peers: listHostPeers(),
+    hostIps,
+  })
+}
+
+function queueFlows(exporterIp: string, flows: ParsedFlow[]): boolean {
+  const serverId = resolveServerId(exporterIp)
+  if (serverId == null) return false
+  const bucketAt = minuteBucketIso()
+  for (const flow of flows) {
+    const key = `${serverId}\0${bucketAt}\0${flow.src}\0${flow.dst}\0${flow.proto}\0${flow.srcPort}\0${flow.dstPort}`
+    const prev = pending.get(key)
+    if (prev) {
+      prev.bytes += flow.bytes
+      prev.packets += flow.packets
+    } else {
+      pending.set(key, {
+        serverId,
+        bucketAt,
+        flow: { ...flow },
+        bytes: flow.bytes,
+        packets: flow.packets,
+      })
+    }
+  }
+  return true
+}
+
+function flushPending() {
+  if (pending.size === 0) return
+  const settings = getTrafficFlowSettingsRow()
+  const topN = Math.max(20, settings.topN)
+  const cutoff = new Date(Date.now() - settings.retentionHours * 3600_000).toISOString()
+  const rows = [...pending.values()]
+  pending.clear()
+
+  for (const row of rows) {
+    try {
+      db.insert(flowBuckets).values({
+        serverId: row.serverId,
+        bucketAt: row.bucketAt,
+        src: row.flow.src || "0.0.0.0",
+        dst: row.flow.dst || "0.0.0.0",
+        proto: row.flow.proto,
+        srcPort: row.flow.srcPort,
+        dstPort: row.flow.dstPort,
+        bytes: row.bytes,
+        packets: row.packets,
+        inIface: row.flow.inIface,
+      }).onConflictDoUpdate({
+        target: [
+          flowBuckets.serverId,
+          flowBuckets.bucketAt,
+          flowBuckets.src,
+          flowBuckets.dst,
+          flowBuckets.proto,
+          flowBuckets.srcPort,
+          flowBuckets.dstPort,
+        ],
+        set: {
+          bytes: sql`${flowBuckets.bytes} + excluded.bytes`,
+          packets: sql`${flowBuckets.packets} + excluded.packets`,
+        },
+      }).run()
+    } catch {
+      // ignore single-row failures
+    }
+  }
+
+  db.delete(flowBuckets).where(sql`${flowBuckets.bucketAt} < ${cutoff}`).run()
+
+  const latest = db.select({ bucketAt: flowBuckets.bucketAt }).from(flowBuckets)
+    .orderBy(desc(flowBuckets.bucketAt)).limit(1).all()[0]?.bucketAt
+  if (!latest) return
+  const latestRows = db.select().from(flowBuckets).where(eq(flowBuckets.bucketAt, latest)).all()
+  const byServer = new Map<number, typeof latestRows>()
+  for (const r of latestRows) {
+    const list = byServer.get(r.serverId) ?? []
+    list.push(r)
+    byServer.set(r.serverId, list)
+  }
+  for (const list of byServer.values()) {
+    if (list.length <= topN) continue
+    list.sort((a, b) => b.bytes - a.bytes)
+    for (const d of list.slice(topN)) {
+      db.delete(flowBuckets).where(eq(flowBuckets.id, d.id)).run()
+    }
+  }
+}
+
+function onMessage(msg: Buffer, rinfo: { address: string }) {
+  try {
+    const flows = parseFlowPacket(msg, rinfo.address)
+    recordFlowPacket(rinfo.address)
+    if (!flows.length) return
+    if (!queueFlows(rinfo.address, flows)) {
+      recordFlowListenerError(
+        `IPFIX от ${rinfo.address}: нет jump-host с адресом wg-flow. Docker SNAT (172.x) при нескольких JH не различим.`,
+      )
+      return
+    }
+    recordFlowListenerError("")
+  } catch (e) {
+    recordFlowListenerError(e instanceof Error ? e.message : String(e))
+  }
+}
+
+export function stopTrafficFlowListener() {
+  if (flushTimer) {
+    clearInterval(flushTimer)
+    flushTimer = null
+  }
+  flushPending()
+  if (socket) {
+    try { socket.close() } catch { /* ignore */ }
+    socket = null
+  }
+  state = { bound: false, address: null }
+}
+
+export function startTrafficFlowListener() {
+  stopTrafficFlowListener()
+  const settings = getTrafficFlowSettingsRow()
+  if (!settings.enabled) {
+    state = { bound: false, address: null }
+    return
+  }
+  const host = process.env.FLOW_LISTEN_HOST?.trim() || settings.collectorIp || "127.0.0.1"
+  const port = settings.flowListenPort
+  const sock = createSocket("udp4")
+  sock.on("error", (err) => {
+    recordFlowListenerError(err.message)
+    state = { bound: false, address: null }
+  })
+  sock.on("message", onMessage)
+  sock.bind(port, host, () => {
+    state = { bound: true, address: `${host}:${port}` }
+    recordFlowListenerError("")
+  })
+  socket = sock
+  flushTimer = setInterval(flushPending, 15_000)
+}
+
+export function listFlowTalkers(minutes = 5): FlowStatsDto {
+  const settings = getTrafficFlowSettingsRow()
+  const rangeStart = new Date(Date.now() - minutes * 60_000).toISOString()
+  const rows = db.select().from(flowBuckets).where(gte(flowBuckets.bucketAt, rangeStart)).all()
+  const serverRows = db.select().from(servers).all()
+  const nameById = new Map(serverRows.map((s) => [s.id, s.name || s.host]))
+  const agg = new Map<string, FlowTalkerDto & { rawBytes: number }>()
+  const protoBytes = new Map<number, number>()
+  const srcs = new Set<string>()
+  const dsts = new Set<string>()
+  const exporters = new Set<number>()
+  let totalBytes = 0
+  for (const r of rows) {
+    const key = `${r.serverId}|${r.src}|${r.dst}|${r.proto}|${r.srcPort}|${r.dstPort}`
+    const prev = agg.get(key)
+    const bytes = r.bytes
+    totalBytes += bytes
+    srcs.add(r.src)
+    dsts.add(r.dst)
+    exporters.add(r.serverId)
+    protoBytes.set(r.proto, (protoBytes.get(r.proto) ?? 0) + bytes)
+    if (prev) {
+      prev.rawBytes += bytes
+      prev.bytes += bytes
+      prev.packets += r.packets
+    } else {
+      agg.set(key, {
+        serverId: String(r.serverId),
+        serverName: nameById.get(r.serverId) ?? String(r.serverId),
+        src: r.src,
+        dst: r.dst,
+        proto: r.proto,
+        protoName: protoName(r.proto),
+        srcPort: r.srcPort,
+        dstPort: r.dstPort,
+        bytes,
+        packets: r.packets,
+        bps: 0,
+        inIface: r.inIface,
+        rawBytes: bytes,
+      })
+    }
+  }
+  const windowSec = Math.max(60, minutes * 60)
+  const talkers = [...agg.values()]
+    .map((t) => ({ ...t, bps: (t.rawBytes * 8) / windowSec }))
+    .sort((a, b) => b.bytes - a.bytes)
+    .slice(0, settings.topN)
+    .map(({ rawBytes: _raw, ...rest }) => rest)
+  let topProto = "—"
+  let topProtoBytes = 0
+  for (const [p, b] of protoBytes) {
+    if (b > topProtoBytes) {
+      topProtoBytes = b
+      topProto = protoName(p)
+    }
+  }
+  return {
+    exportersOnline: exporters.size,
+    bytesPerMin: minutes > 0 ? totalBytes / minutes : totalBytes,
+    uniqueSrc: srcs.size,
+    uniqueDst: dsts.size,
+    topProto,
+    talkers,
+    lastExporterIp: settings.lastExporterIp ?? null,
+    lastError: settings.lastError || null,
+    packetsReceived: settings.packetsReceived,
+  }
+}
+
+export function ingestParsedFlowsForTests(exporterIp: string, flows: ParsedFlow[]) {
+  queueFlows(exporterIp, flows)
+  flushPending()
+}

backend/src/services/traffic-flow-map-exporter.test.ts

@@ -0,0 +1,54 @@
+import assert from "node:assert/strict"
+import {
+  bareIpv4,
+  ipInCidr,
+  isNatMasqueradeExporter,
+  normalizeExporterIp,
+  pickServerIdForExporter,
+} from "./traffic-flow-map-exporter.js"
+
+assert.equal(normalizeExporterIp("::ffff:172.18.0.2"), "172.18.0.2")
+assert.equal(bareIpv4("10.255.254.3/32"), "10.255.254.3")
+assert.equal(ipInCidr("10.255.254.3", "10.255.254.0/24"), true)
+assert.equal(ipInCidr("172.18.0.2", "10.255.254.0/24"), false)
+assert.equal(isNatMasqueradeExporter("172.18.0.2", "10.255.254.0/24"), true)
+assert.equal(isNatMasqueradeExporter("10.255.254.3", "10.255.254.0/24"), false)
+assert.equal(isNatMasqueradeExporter("10.0.0.12", "10.255.254.0/24"), true)
+
+const byTunnel = new Map([["10.255.254.3", 7]])
+assert.equal(pickServerIdForExporter({
+  exporterIp: "10.255.254.3",
+  overlayPrefix: "10.255.254.0/24",
+  byTunnelIp: byTunnel,
+  peers: [],
+  hostIps: new Map(),
+}), 7)
+
+assert.equal(pickServerIdForExporter({
+  exporterIp: "172.18.0.2",
+  overlayPrefix: "10.255.254.0/24",
+  byTunnelIp: byTunnel,
+  peers: [{ serverId: 7, address: "10.255.254.3", allowedIps: ["10.255.254.3/32"] }],
+  hostIps: new Map(),
+}), 7)
+
+assert.equal(pickServerIdForExporter({
+  exporterIp: "172.18.0.2",
+  overlayPrefix: "10.255.254.0/24",
+  byTunnelIp: new Map([["10.255.254.3", 7], ["10.255.254.4", 8]]),
+  peers: [
+    { serverId: 7, address: "10.255.254.3", allowedIps: ["10.255.254.3/32"] },
+    { serverId: 8, address: "10.255.254.4", allowedIps: ["10.255.254.4/32"] },
+  ],
+  hostIps: new Map(),
+}), null)
+
+assert.equal(pickServerIdForExporter({
+  exporterIp: "94.142.140.141",
+  overlayPrefix: "10.255.254.0/24",
+  byTunnelIp: byTunnel,
+  peers: [],
+  hostIps: new Map([["94.142.140.141", 7]]),
+}), 7)
+
+console.log("traffic-flow-map-exporter.test.ts: ok")

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

@@ -0,0 +1,82 @@
+export interface OverlayPeerRef {
+  serverId: number
+  address: string
+  allowedIps: string[]
+}
+
+export function normalizeExporterIp(ip: string): string {
+  const trimmed = ip.trim()
+  if (trimmed.toLowerCase().startsWith("::ffff:")) return trimmed.slice(7)
+  return trimmed
+}
+
+export function bareIpv4(value: string): string {
+  const raw = normalizeExporterIp(value).split("/")[0]?.trim() ?? ""
+  return raw
+}
+
+function ipv4ToInt(ip: string): number | null {
+  const parts = ip.split(".")
+  if (parts.length !== 4) return null
+  const n = parts.map((x) => Number(x))
+  if (n.some((x) => !Number.isInteger(x) || x < 0 || x > 255)) return null
+  return ((n[0]! << 24) | (n[1]! << 16) | (n[2]! << 8) | n[3]!) >>> 0
+}
+
+export function ipInCidr(ip: string, cidr: string): boolean {
+  const host = bareIpv4(ip)
+  const [base, bitsRaw] = cidr.split("/")
+  const bits = Number(bitsRaw ?? 32)
+  const a = ipv4ToInt(host)
+  const b = ipv4ToInt(bareIpv4(base ?? ""))
+  if (a == null || b == null || !Number.isFinite(bits) || bits < 0 || bits > 32) return false
+  const mask = bits === 0 ? 0 : (0xffffffff << (32 - bits)) >>> 0
+  return (a & mask) === (b & mask)
+}
+
+/** Docker userland-proxy / bridge SNAT, не адрес из оверлея wg-flow. */
+export function isNatMasqueradeExporter(ip: string, overlayPrefix: string): boolean {
+  const host = bareIpv4(ip)
+  if (!host) return false
+  if (ipInCidr(host, overlayPrefix)) return false
+  return ipInCidr(host, "10.0.0.0/8")
+    || ipInCidr(host, "172.16.0.0/12")
+    || ipInCidr(host, "192.168.0.0/16")
+    || ipInCidr(host, "127.0.0.0/8")
+}
+
+export function pickServerIdForExporter(opts: {
+  exporterIp: string
+  overlayPrefix: string
+  byTunnelIp: Map<string, number>
+  peers: OverlayPeerRef[]
+  hostIps: Map<string, number>
+}): number | null {
+  const exporter = bareIpv4(opts.exporterIp)
+  if (!exporter) return null
+
+  const exact = opts.byTunnelIp.get(exporter)
+  if (exact != null) return exact
+
+  for (const [ip, id] of opts.byTunnelIp) {
+    if (bareIpv4(ip) === exporter) return id
+  }
+
+  for (const peer of opts.peers) {
+    if (bareIpv4(peer.address) === exporter) return peer.serverId
+    if (peer.allowedIps.some((cidr) => ipInCidr(exporter, cidr) || bareIpv4(cidr) === exporter)) {
+      return peer.serverId
+    }
+  }
+
+  const byHost = opts.hostIps.get(exporter)
+  if (byHost != null) return byHost
+
+  if (!isNatMasqueradeExporter(exporter, opts.overlayPrefix)) return null
+
+  const tunnelIds = [...new Set(opts.byTunnelIp.values())]
+  if (tunnelIds.length === 1) return tunnelIds[0] ?? null
+  const peerIds = [...new Set(opts.peers.map((p) => p.serverId))]
+  if (peerIds.length === 1) return peerIds[0] ?? null
+  return null
+}

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

@@ -0,0 +1,281 @@
+import { eq } from "drizzle-orm"
+import { db } from "../db/index.js"
+import { servers } from "../db/schema.js"
+import type { TrafficFlowOverlayResult } from "@mmapp/contracts/traffic-flow"
+import { encodeRosId, MikrotikClient, MikrotikError } from "./mikrotik.js"
+import { getEnabledServerById, listWireGuardInterfaces } from "./wireguard-live.js"
+import {
+  asRosArray,
+  patchRosPath,
+  putIpAddress,
+  putWireguardInterface,
+  putWireguardPeer,
+  rosRowId,
+  toRosBody,
+} from "./wireguard-ros.js"
+import {
+  ensureHostKeys,
+  getTrafficFlowSettingsRow,
+  upsertHostPeer,
+} from "./traffic-flow-settings.js"
+import { listTrafficFlowHostFiles } from "./traffic-flow-host-files.js"
+
+const IFACE_NAME = "wg-flow"
+const JH_LISTEN_PORT = 13232
+const WG_INPUT_COMMENT = "mm-wg-flow"
+
+export function allocateOverlayAddress(prefix: string, collectorIp: string, serverId: number, taken: Set<string>): string {
+  const [base] = prefix.split("/")
+  const parts = (base ?? "10.255.254.0").split(".").map((n) => Number.parseInt(n, 10))
+  const a = parts[0] || 10
+  const b = parts[1] || 255
+  const c = parts[2] || 254
+  const preferredLast = 2 + ((serverId - 1) % 250)
+  const candidates = [preferredLast, ...Array.from({ length: 253 }, (_, i) => 2 + ((preferredLast - 2 + i) % 253))]
+  for (const last of candidates) {
+    const ip = `${a}.${b}.${c}.${last}`
+    if (ip === collectorIp) continue
+    if (taken.has(ip)) continue
+    return ip
+  }
+  throw new Error("Нет свободных адресов в префиксе wg-flow")
+}
+
+function linuxPeerBlock(publicKey: string, address: string, comment: string, endpoint: string): string {
+  return [
+    `[Peer]`,
+    `PublicKey = ${publicKey}`,
+    `AllowedIPs = ${address}/32`,
+    `Endpoint = ${endpoint}:${JH_LISTEN_PORT}`,
+    `PersistentKeepalive = 25`,
+    comment ? `# ${comment}` : "",
+  ].filter(Boolean).join("\n")
+}
+
+async function findIface(client: MikrotikClient, name: string): Promise<Record<string, unknown> | undefined> {
+  const list = asRosArray<Record<string, unknown>>(await client.get("/interface/wireguard"))
+  return list.find((i) => String(i.name ?? "") === name)
+}
+
+async function findPeer(
+  client: MikrotikClient,
+  iface: string,
+  publicKey: string,
+): Promise<Record<string, unknown> | undefined> {
+  const list = asRosArray<Record<string, unknown>>(await client.get("/interface/wireguard/peers"))
+  return list.find((p) =>
+    String(p.interface ?? "") === iface && String(p["public-key"] ?? "") === publicKey,
+  )
+}
+
+async function findAddress(client: MikrotikClient, iface: string): Promise<Record<string, unknown> | undefined> {
+  const list = asRosArray<Record<string, unknown>>(await client.get("/ip/address"))
+  return list.find((a) => String(a.interface ?? "") === iface)
+}
+
+async function findRoute(client: MikrotikClient, dst: string): Promise<Record<string, unknown> | undefined> {
+  const list = asRosArray<Record<string, unknown>>(await client.get("/ip/route"))
+  return list.find((r) => String(r["dst-address"] ?? "") === dst)
+}
+
+async function ensureWgInputAccept(client: MikrotikClient, listenPort: number): Promise<boolean> {
+  const rules = asRosArray<Record<string, unknown>>(await client.get("/ip/firewall/filter"))
+  const existing = rules.find((r) => String(r.comment ?? "") === WG_INPUT_COMMENT)
+  if (existing) return false
+  await client.put("/ip/firewall/filter", toRosBody({
+    chain: "input",
+    protocol: "udp",
+    "dst-port": String(listenPort),
+    action: "accept",
+    comment: WG_INPUT_COMMENT,
+  }))
+  return true
+}
+
+async function ensureTrafficFlow(
+  client: MikrotikClient,
+  collectorIp: string,
+  port: number,
+  srcAddress: string,
+): Promise<void> {
+  const body = toRosBody({
+    enabled: "yes",
+    interfaces: "all",
+    "active-flow-timeout": "1m",
+    "inactive-flow-timeout": "15s",
+  })
+  const rows = asRosArray<Record<string, unknown>>(await client.get("/ip/traffic-flow"))
+  const id = rows[0] ? rosRowId(rows[0]) : ""
+  if (id) {
+    await patchRosPath(client, `/ip/traffic-flow/${encodeRosId(id)}`, body)
+  } else {
+    await client.post("/ip/traffic-flow/set", body)
+  }
+
+  const targets = asRosArray<Record<string, unknown>>(await client.get("/ip/traffic-flow/target"))
+  const existing = targets.find((t) => String(t["dst-address"] ?? "") === collectorIp)
+  const targetBody = toRosBody({
+    "dst-address": collectorIp,
+    "src-address": srcAddress,
+    port: String(port),
+    version: "ipfix",
+  })
+  if (existing) {
+    const targetId = rosRowId(existing)
+    if (targetId) await patchRosPath(client, `/ip/traffic-flow/target/${encodeRosId(targetId)}`, targetBody)
+    return
+  }
+  await client.put("/ip/traffic-flow/target", targetBody)
+}
+
+export function usablePublicHost(raw: string | undefined): string {
+  if (!raw) return ""
+  const host = raw.split(",")[0]?.trim().replace(/^\[/, "").replace(/\]:\d+$/, "").split(":")[0]?.trim() ?? ""
+  const lower = host.toLowerCase()
+  if (!host) return ""
+  if (lower === "localhost" || lower === "127.0.0.1" || lower === "::1" || lower === "0.0.0.0") return ""
+  if (lower.endsWith(".local") || lower.endsWith(".internal") || lower.endsWith(".lan")) return ""
+  if (/^(10\.|192\.168\.|172\.(1[6-9]|2\d|3[01])\.)/.test(host)) return ""
+  return host
+}
+
+export async function applyFlowOverlay(
+  serverIdRaw: string | number,
+  opts?: { publicEndpoint?: string; requestHost?: string },
+): Promise<TrafficFlowOverlayResult> {
+  const steps: string[] = []
+  const keys = ensureHostKeys()
+  const settings = getTrafficFlowSettingsRow()
+  const hostPublicKey = settings.hostPublicKey || keys.publicKey
+  if (!hostPublicKey) {
+    throw Object.assign(new Error("Не удалось создать ключи хоста MM"), { statusCode: 500 })
+  }
+
+  const server = getEnabledServerById(String(serverIdRaw))
+  if (!server || !server.enabled) {
+    throw Object.assign(new Error("Сервер не найден или выключен"), { statusCode: 404 })
+  }
+
+  const endpointHost = (opts?.publicEndpoint?.trim() || server.host.trim()).trim()
+  if (!endpointHost) {
+    throw Object.assign(new Error("Укажите публичный IP или DNS jump-host"), { statusCode: 400 })
+  }
+  const peerEndpoint = `${endpointHost}:${JH_LISTEN_PORT}`
+
+  const taken = new Set(
+    db.select({ ip: servers.mgmtTunnelIp }).from(servers).all()
+      .map((r) => r.ip)
+      .filter(Boolean),
+  )
+  const address = server.mgmtTunnelIp || allocateOverlayAddress(settings.prefix, settings.collectorIp, server.id, taken)
+  const client = MikrotikClient.fromServer(server)
+
+  try {
+    let iface = await findIface(client, IFACE_NAME)
+    if (!iface) {
+      await putWireguardInterface(client, {
+        name: IFACE_NAME,
+        "listen-port": String(JH_LISTEN_PORT),
+        mtu: "1420",
+        comment: "MikrotikManager traffic-flow overlay",
+      })
+      steps.push(`Создан интерфейс ${IFACE_NAME}`)
+      iface = await findIface(client, IFACE_NAME)
+    } else {
+      steps.push(`Интерфейс ${IFACE_NAME} уже есть`)
+    }
+
+    const addrRow = await findAddress(client, IFACE_NAME)
+    const mask = (settings.prefix.split("/")[1] || "24").replace(/\D/g, "") || "24"
+    const cidr = `${address}/${mask}`
+    if (!addrRow) {
+      await putIpAddress(client, cidr, IFACE_NAME)
+      steps.push(`Адрес ${cidr}`)
+    } else {
+      steps.push(`Адрес на ${IFACE_NAME} уже назначен`)
+    }
+
+    const peer = await findPeer(client, IFACE_NAME, hostPublicKey)
+    const peerBody = {
+      interface: IFACE_NAME,
+      "public-key": hostPublicKey,
+      "allowed-address": `${settings.collectorIp}/32`,
+      comment: "MM traffic-flow collector",
+      name: "mm-collector",
+    }
+    if (!peer) {
+      await putWireguardPeer(client, peerBody)
+      steps.push("Добавлен пир на pubkey хоста MM (сервер, без endpoint)")
+    } else {
+      const id = rosRowId(peer)
+      const hadEndpoint = Boolean(String(peer["endpoint-address"] ?? "").trim())
+      if (hadEndpoint && id) {
+        await client.delete(`/interface/wireguard/peers/${encodeURIComponent(id)}`)
+        await putWireguardPeer(client, peerBody)
+        steps.push("Пир пересоздан как сервер (endpoint снят)")
+      } else if (id) {
+        await patchRosPath(client, `/interface/wireguard/peers/${encodeURIComponent(id)}`, peerBody)
+        steps.push("Пир хоста MM обновлён")
+      }
+    }
+
+    const routeDst = `${settings.collectorIp}/32`
+    const route = await findRoute(client, routeDst)
+    if (!route) {
+      await client.put("/ip/route", toRosBody({
+        "dst-address": routeDst,
+        gateway: IFACE_NAME,
+        comment: "MM traffic-flow collector",
+      }))
+      steps.push(`Маршрут ${routeDst} через ${IFACE_NAME}`)
+    } else {
+      steps.push("Маршрут до collector уже есть")
+    }
+
+    if (await ensureWgInputAccept(client, JH_LISTEN_PORT)) {
+      steps.push(`Firewall input accept UDP ${JH_LISTEN_PORT}`)
+    } else {
+      steps.push("Firewall input WG уже есть")
+    }
+
+    await ensureTrafficFlow(client, settings.collectorIp, settings.flowListenPort, address)
+    steps.push(`Traffic Flow → ${settings.collectorIp}:${settings.flowListenPort} ipfix (src ${address})`)
+
+    const listed = await listWireGuardInterfaces({ serverId: String(server.id), includePrivateKey: false })
+    const created = listed.interfaces.find((i) => i.name === IFACE_NAME)
+    const publicKey = created?.publicKey ?? ""
+    if (!publicKey) {
+      throw new Error("Не удалось прочитать public-key интерфейса wg-flow")
+    }
+
+    db.update(servers).set({
+      mgmtTunnelIp: address,
+      updatedAt: new Date().toISOString(),
+    }).where(eq(servers.id, server.id)).run()
+
+    upsertHostPeer({
+      serverId: server.id,
+      name: server.name || server.host,
+      publicKey,
+      allowedIps: [`${address}/32`],
+      address,
+      endpoint: peerEndpoint,
+    })
+
+    return {
+      ok: true,
+      serverId: server.id,
+      interfaceName: IFACE_NAME,
+      address,
+      publicKey,
+      linuxPeerBlock: linuxPeerBlock(publicKey, address, server.name || server.host, endpointHost),
+      trafficFlow: true,
+      steps,
+      hostFiles: listTrafficFlowHostFiles(),
+    }
+  } catch (e) {
+    const msg = e instanceof MikrotikError ? e.message : e instanceof Error ? e.message : String(e)
+    const err = Object.assign(new Error(`RouterOS: ${msg}`), { statusCode: 502 })
+    throw err
+  }
+}

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

@@ -0,0 +1,69 @@
+import assert from "node:assert/strict"
+import { parseFlowPacket, protoName, resetFlowTemplatesForTests } from "./traffic-flow-parse.js"
+import { allocateOverlayAddress } from "./traffic-flow-overlay.js"
+
+function netflowV5One(): Buffer {
+  const buf = Buffer.alloc(24 + 48)
+  buf.writeUInt16BE(5, 0)
+  buf.writeUInt16BE(1, 2)
+  buf[24] = 10; buf[25] = 1; buf[26] = 1; buf[27] = 8
+  buf[28] = 8; buf[29] = 8; buf[30] = 8; buf[31] = 8
+  buf.writeUInt16BE(1, 24 + 12)
+  buf.writeUInt32BE(10, 24 + 16)
+  buf.writeUInt32BE(1500, 24 + 20)
+  buf.writeUInt16BE(443, 24 + 32)
+  buf.writeUInt16BE(443, 24 + 34)
+  buf.writeUInt8(6, 24 + 38)
+  return buf
+}
+
+resetFlowTemplatesForTests()
+const flows = parseFlowPacket(netflowV5One(), "10.255.254.5")
+assert.equal(flows.length, 1)
+assert.equal(flows[0]?.src, "10.1.1.8")
+assert.equal(flows[0]?.dst, "8.8.8.8")
+assert.equal(flows[0]?.proto, 6)
+assert.equal(flows[0]?.bytes, 1500)
+assert.equal(protoName(6), "TCP")
+assert.equal(parseFlowPacket(Buffer.from([0, 1]), "1.1.1.1").length, 0)
+
+const taken = new Set(["10.255.254.2"])
+assert.equal(allocateOverlayAddress("10.255.254.0/24", "10.255.254.1", 1, taken), "10.255.254.3")
+assert.equal(allocateOverlayAddress("10.255.254.0/24", "10.255.254.1", 2, new Set()), "10.255.254.3")
+
+import { usablePublicHost } from "./traffic-flow-overlay.js"
+assert.equal(usablePublicHost("localhost:8000"), "")
+assert.equal(usablePublicHost("127.0.0.1"), "")
+assert.equal(usablePublicHost("192.168.1.10"), "")
+assert.equal(usablePublicHost("mm.example.com:443"), "mm.example.com")
+assert.equal(usablePublicHost("203.0.113.10"), "203.0.113.10")
+
+resetFlowTemplatesForTests()
+{
+  const tpl = Buffer.alloc(16 + 16 + 20)
+  tpl.writeUInt16BE(10, 0)
+  tpl.writeUInt16BE(tpl.length, 2)
+  tpl.writeUInt16BE(2, 16)
+  tpl.writeUInt16BE(16, 18)
+  tpl.writeUInt16BE(256, 20)
+  tpl.writeUInt16BE(2, 22)
+  tpl.writeUInt16BE(8, 24)
+  tpl.writeUInt16BE(4, 26)
+  tpl.writeUInt16BE(12, 28)
+  tpl.writeUInt16BE(4, 30)
+  const data = Buffer.alloc(16 + 12)
+  data.writeUInt16BE(10, 0)
+  data.writeUInt16BE(data.length, 2)
+  data.writeUInt16BE(256, 16)
+  data.writeUInt16BE(12, 18)
+  data[20] = 10; data[21] = 1; data[22] = 1; data[23] = 8
+  data[24] = 8; data[25] = 8; data[26] = 8; data[27] = 8
+  const fromTpl = parseFlowPacket(tpl, "172.18.0.2")
+  assert.equal(fromTpl.length, 0)
+  const fromData = parseFlowPacket(data, "172.18.0.2")
+  assert.equal(fromData.length, 1)
+  assert.equal(fromData[0]?.src, "10.1.1.8")
+  assert.equal(fromData[0]?.dst, "8.8.8.8")
+}
+
+console.log("traffic-flow-parse.test.ts: ok")

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

@@ -0,0 +1,299 @@
+export interface ParsedFlow {
+  src: string
+  dst: string
+  proto: number
+  srcPort: number
+  dstPort: number
+  bytes: number
+  packets: number
+  inIface: string
+}
+
+interface FieldSpec {
+  type: number
+  length: number
+}
+
+interface Template {
+  fields: FieldSpec[]
+}
+
+const templatesByExporter = new Map<string, Map<number, Template>>()
+
+function ipv4(buf: Buffer, offset: number): string {
+  return `${buf[offset]}.${buf[offset + 1]}.${buf[offset + 2]}.${buf[offset + 3]}`
+}
+
+function ipv6(buf: Buffer, offset: number): string {
+  const parts: string[] = []
+  for (let i = 0; i < 8; i++) parts.push(buf.readUInt16BE(offset + i * 2).toString(16))
+  return parts.join(":")
+}
+
+const VAR_LEN = 0xffff
+
+function consumeField(
+  buf: Buffer,
+  off: number,
+  length: number,
+  limit: number,
+): { data: Buffer; next: number } | null {
+  if (length === VAR_LEN) {
+    if (off >= limit) return null
+    const first = buf[off]!
+    if (first < 255) {
+      const end = off + 1 + first
+      if (end > limit) return null
+      return { data: buf.subarray(off + 1, end), next: end }
+    }
+    if (off + 3 > limit) return null
+    const len = buf.readUInt16BE(off + 1)
+    const end = off + 3 + len
+    if (end > limit) return null
+    return { data: buf.subarray(off + 3, end), next: end }
+  }
+  const end = off + length
+  if (end > limit) return null
+  return { data: buf.subarray(off, end), next: end }
+}
+
+function fixedRecordSize(fields: FieldSpec[]): number | null {
+  let n = 0
+  for (const f of fields) {
+    if (f.length === VAR_LEN) return null
+    n += f.length
+  }
+  return n
+}
+
+function readUint(buf: Buffer, offset: number, length: number): number {
+  if (length === 1) return buf.readUInt8(offset)
+  if (length === 2) return buf.readUInt16BE(offset)
+  if (length === 4) return buf.readUInt32BE(offset)
+  if (length === 8) {
+    const big = buf.readBigUInt64BE(offset)
+    const n = Number(big)
+    return Number.isFinite(n) ? n : 0
+  }
+  let v = 0
+  for (let i = 0; i < length; i++) v = (v << 8) + buf[offset + i]
+  return v >>> 0
+}
+
+function parseNetflowV5(buf: Buffer): ParsedFlow[] {
+  if (buf.length < 24) return []
+  const count = buf.readUInt16BE(2)
+  const out: ParsedFlow[] = []
+  let off = 24
+  for (let i = 0; i < count && off + 48 <= buf.length; i++) {
+    out.push({
+      src: ipv4(buf, off),
+      dst: ipv4(buf, off + 4),
+      packets: buf.readUInt32BE(off + 16),
+      bytes: buf.readUInt32BE(off + 20),
+      srcPort: buf.readUInt16BE(off + 32),
+      dstPort: buf.readUInt16BE(off + 34),
+      proto: buf.readUInt8(off + 38),
+      inIface: String(buf.readUInt16BE(off + 12)),
+    })
+    off += 48
+  }
+  return out
+}
+
+function parseIpfixTemplates(exporter: string, buf: Buffer, setStart: number, setEnd: number, setId: number) {
+  let off = setStart + 4
+  const map = templatesByExporter.get(exporter) ?? new Map<number, Template>()
+  while (off + 4 <= setEnd) {
+    const templateId = buf.readUInt16BE(off)
+    const fieldCount = buf.readUInt16BE(off + 2)
+    off += 4
+    if (setId === 3) {
+      // options template: skip scope count
+      if (off + 2 > setEnd) break
+      off += 2
+    }
+    const fields: FieldSpec[] = []
+    for (let i = 0; i < fieldCount && off + 4 <= setEnd; i++) {
+      const type = buf.readUInt16BE(off)
+      const length = buf.readUInt16BE(off + 2)
+      off += 4
+      if (type & 0x8000) {
+        if (off + 4 > setEnd) break
+        off += 4
+      }
+      fields.push({ type: type & 0x7fff, length })
+    }
+    if (templateId >= 256) map.set(templateId, { fields })
+  }
+  templatesByExporter.set(exporter, map)
+}
+
+function recordFromFields(
+  fields: FieldSpec[],
+  buf: Buffer,
+  offset: number,
+  limit: number,
+): { flow: ParsedFlow; next: number } | null {
+  let off = offset
+  let src = ""
+  let dst = ""
+  let proto = 0
+  let srcPort = 0
+  let dstPort = 0
+  let bytes = 0
+  let packets = 0
+  let inIface = ""
+  for (const f of fields) {
+    const field = consumeField(buf, off, f.length, limit)
+    if (!field) return null
+    const { data } = field
+    switch (f.type) {
+      case 8:
+        if (data.length === 4) src = ipv4(data, 0)
+        break
+      case 12:
+        if (data.length === 4) dst = ipv4(data, 0)
+        break
+      case 27:
+        if (data.length === 16 && !src) src = ipv6(data, 0)
+        break
+      case 28:
+        if (data.length === 16 && !dst) dst = ipv6(data, 0)
+        break
+      case 225:
+        if (data.length === 4 && !src) src = ipv4(data, 0)
+        break
+      case 226:
+        if (data.length === 4 && !dst) dst = ipv4(data, 0)
+        break
+      case 4:
+        proto = readUint(data, 0, data.length)
+        break
+      case 7:
+        srcPort = readUint(data, 0, data.length)
+        break
+      case 11:
+        dstPort = readUint(data, 0, data.length)
+        break
+      case 1:
+        bytes = readUint(data, 0, data.length)
+        break
+      case 2:
+        packets = readUint(data, 0, data.length)
+        break
+      case 85:
+        if (!bytes) bytes = readUint(data, 0, data.length)
+        break
+      case 86:
+        if (!packets) packets = readUint(data, 0, data.length)
+        break
+      case 10:
+        inIface = String(readUint(data, 0, data.length))
+        break
+      default:
+        break
+    }
+    off = field.next
+  }
+  return { flow: { src, dst, proto, srcPort, dstPort, bytes, packets, inIface }, next: off }
+}
+
+function parseDataRecords(
+  tpl: Template,
+  buf: Buffer,
+  recOff: number,
+  setEnd: number,
+  out: ParsedFlow[],
+) {
+  const size = fixedRecordSize(tpl.fields)
+  while (recOff + 1 < setEnd) {
+    if (size != null && recOff + size > setEnd) break
+    const parsed = recordFromFields(tpl.fields, buf, recOff, setEnd)
+    if (!parsed) break
+    if (parsed.flow.src || parsed.flow.dst) out.push(parsed.flow)
+    if (parsed.next <= recOff) break
+    recOff = parsed.next
+  }
+}
+
+function parseIpfix(buf: Buffer, exporter: string): ParsedFlow[] {
+  if (buf.length < 16) return []
+  const total = buf.readUInt16BE(2)
+  const end = Math.min(buf.length, total)
+  let off = 16
+  const out: ParsedFlow[] = []
+  while (off + 4 <= end) {
+    const setId = buf.readUInt16BE(off)
+    const setLen = buf.readUInt16BE(off + 2)
+    if (setLen < 4 || off + setLen > end) break
+    const setEnd = off + setLen
+    if (setId === 2 || setId === 3) {
+      parseIpfixTemplates(exporter, buf, off, setEnd, setId)
+    } else if (setId >= 256) {
+      const tpl = templatesByExporter.get(exporter)?.get(setId)
+      if (tpl) parseDataRecords(tpl, buf, off + 4, setEnd, out)
+    }
+    off = setEnd
+  }
+  return out
+}
+
+function parseNetflowV9(buf: Buffer, exporter: string): ParsedFlow[] {
+  if (buf.length < 20) return []
+  const count = buf.readUInt16BE(2)
+  let off = 20
+  const out: ParsedFlow[] = []
+  const map = templatesByExporter.get(exporter) ?? new Map<number, Template>()
+  for (let s = 0; s < count && off + 4 <= buf.length; s++) {
+    const setId = buf.readUInt16BE(off)
+    const setLen = buf.readUInt16BE(off + 2)
+    if (setLen < 4 || off + setLen > buf.length) break
+    const setEnd = off + setLen
+    if (setId === 0) {
+      let tOff = off + 4
+      while (tOff + 4 <= setEnd) {
+        const templateId = buf.readUInt16BE(tOff)
+        const fieldCount = buf.readUInt16BE(tOff + 2)
+        tOff += 4
+        const fields: FieldSpec[] = []
+        for (let i = 0; i < fieldCount && tOff + 4 <= setEnd; i++) {
+          fields.push({ type: buf.readUInt16BE(tOff), length: buf.readUInt16BE(tOff + 2) })
+          tOff += 4
+        }
+        if (templateId >= 256) map.set(templateId, { fields })
+      }
+      templatesByExporter.set(exporter, map)
+    } else if (setId >= 256) {
+      const tpl = map.get(setId)
+      if (tpl) parseDataRecords(tpl, buf, off + 4, setEnd, out)
+    }
+    off = setEnd
+  }
+  return out
+}
+
+export function parseFlowPacket(buf: Buffer, exporterIp: string): ParsedFlow[] {
+  if (buf.length < 2) return []
+  const version = buf.readUInt16BE(0)
+  if (version === 5) return parseNetflowV5(buf)
+  if (version === 9) return parseNetflowV9(buf, exporterIp)
+  if (version === 10) return parseIpfix(buf, exporterIp)
+  return []
+}
+
+export function protoName(proto: number): string {
+  switch (proto) {
+    case 1: return "ICMP"
+    case 6: return "TCP"
+    case 17: return "UDP"
+    case 47: return "GRE"
+    case 50: return "ESP"
+    case 89: return "OSPF"
+    default: return String(proto)
+  }
+}
+
+export function resetFlowTemplatesForTests() {
+  templatesByExporter.clear()
+}

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

@@ -0,0 +1,126 @@
+import { eq } from "drizzle-orm"
+import { db } from "../db/index.js"
+import { trafficFlowSettings } from "../db/schema.js"
+import type { FlowHostPeer, TrafficFlowSettingsDto, TrafficFlowSettingsPatch } from "@mmapp/contracts/traffic-flow"
+import { generateWireGuardKeyPair } from "./wg-keys.js"
+
+function nowIso() {
+  return new Date().toISOString()
+}
+
+function parsePeers(raw: string): FlowHostPeer[] {
+  try {
+    const parsed = JSON.parse(raw) as unknown
+    if (!Array.isArray(parsed)) return []
+    return parsed.filter((p): p is FlowHostPeer =>
+      p != null && typeof p === "object" && typeof (p as FlowHostPeer).publicKey === "string",
+    )
+  } catch {
+    return []
+  }
+}
+
+export function getTrafficFlowSettingsRow() {
+  const row = db.select().from(trafficFlowSettings).where(eq(trafficFlowSettings.id, 1)).limit(1).all()[0]
+  if (row) return row
+  const now = nowIso()
+  db.insert(trafficFlowSettings).values({
+    id: 1,
+    enabled: false,
+    collectorIp: "10.255.254.1",
+    flowListenPort: 4739,
+    wgListenPort: 51821,
+    prefix: "10.255.254.0/24",
+    createdAt: now,
+    updatedAt: now,
+  }).run()
+  return db.select().from(trafficFlowSettings).where(eq(trafficFlowSettings.id, 1)).limit(1).all()[0]
+}
+
+export function toTrafficFlowSettingsDto(
+  listener: { bound: boolean; address: string | null },
+): TrafficFlowSettingsDto {
+  const row = getTrafficFlowSettingsRow()
+  return {
+    enabled: row.enabled,
+    collectorIp: row.collectorIp,
+    flowListenPort: row.flowListenPort,
+    wgListenPort: row.wgListenPort,
+    prefix: row.prefix,
+    publicEndpoint: row.publicEndpoint,
+    hostPublicKey: row.hostPublicKey,
+    hasHostPrivateKey: Boolean(row.hostPrivateKey),
+    hubServerId: row.hubServerId ?? null,
+    retentionHours: row.retentionHours,
+    topN: row.topN,
+    lastDatagramAt: row.lastDatagramAt ?? null,
+    lastExporterIp: row.lastExporterIp ?? null,
+    lastError: row.lastError || null,
+    packetsReceived: row.packetsReceived,
+    listenerBound: listener.bound,
+    listenerAddress: listener.address,
+    peers: parsePeers(row.peersJson),
+  }
+}
+
+export function updateTrafficFlowSettings(patch: TrafficFlowSettingsPatch) {
+  const row = getTrafficFlowSettingsRow()
+  db.update(trafficFlowSettings).set({
+    enabled: patch.enabled ?? row.enabled,
+    collectorIp: patch.collectorIp ?? row.collectorIp,
+    flowListenPort: patch.flowListenPort ?? row.flowListenPort,
+    wgListenPort: patch.wgListenPort ?? row.wgListenPort,
+    prefix: patch.prefix ?? row.prefix,
+    publicEndpoint: patch.publicEndpoint ?? row.publicEndpoint,
+    hubServerId: patch.hubServerId === undefined ? row.hubServerId : patch.hubServerId,
+    retentionHours: patch.retentionHours ?? row.retentionHours,
+    topN: patch.topN ?? row.topN,
+    updatedAt: nowIso(),
+  }).where(eq(trafficFlowSettings.id, 1)).run()
+  return getTrafficFlowSettingsRow()
+}
+
+export function ensureHostKeys(): { publicKey: string; created: boolean } {
+  const row = getTrafficFlowSettingsRow()
+  if (row.hostPublicKey && row.hostPrivateKey) {
+    return { publicKey: row.hostPublicKey, created: false }
+  }
+  const keys = generateWireGuardKeyPair()
+  db.update(trafficFlowSettings).set({
+    hostPublicKey: keys.publicKey,
+    hostPrivateKey: keys.privateKey,
+    updatedAt: nowIso(),
+  }).where(eq(trafficFlowSettings.id, 1)).run()
+  return { publicKey: keys.publicKey, created: true }
+}
+
+export function upsertHostPeer(peer: FlowHostPeer) {
+  const row = getTrafficFlowSettingsRow()
+  const peers = parsePeers(row.peersJson).filter((p) => p.serverId !== peer.serverId)
+  peers.push(peer)
+  db.update(trafficFlowSettings).set({
+    peersJson: JSON.stringify(peers),
+    updatedAt: nowIso(),
+  }).where(eq(trafficFlowSettings.id, 1)).run()
+}
+
+export function recordFlowPacket(exporterIp: string) {
+  const row = getTrafficFlowSettingsRow()
+  db.update(trafficFlowSettings).set({
+    lastDatagramAt: nowIso(),
+    lastExporterIp: exporterIp,
+    packetsReceived: row.packetsReceived + 1,
+    updatedAt: nowIso(),
+  }).where(eq(trafficFlowSettings.id, 1)).run()
+}
+
+export function recordFlowListenerError(message: string) {
+  db.update(trafficFlowSettings).set({
+    lastError: message,
+    updatedAt: nowIso(),
+  }).where(eq(trafficFlowSettings.id, 1)).run()
+}
+
+export function listHostPeers(): FlowHostPeer[] {
+  return parsePeers(getTrafficFlowSettingsRow().peersJson)
+}

backend/src/services/wg-keys.ts

@@ -0,0 +1,12 @@
+import { generateKeyPairSync } from "node:crypto"
+
+/** WireGuard Curve25519 keypair as RouterOS/wg-quick base64 (32 bytes). */
+export function generateWireGuardKeyPair(): { publicKey: string; privateKey: string } {
+  const { publicKey, privateKey } = generateKeyPairSync("x25519")
+  const pubDer = publicKey.export({ type: "spki", format: "der" })
+  const privDer = privateKey.export({ type: "pkcs8", format: "der" })
+  return {
+    publicKey: Buffer.from(pubDer.subarray(-32)).toString("base64"),
+    privateKey: Buffer.from(privDer.subarray(-32)).toString("base64"),
+  }
+}

backend/src/services/wireguard-config.ts

@@ -35,7 +35,7 @@ export type WgParsedConfig = {
 
 export type WgExportIface = {
   name: string
-  listenPort: number
+  listenPort?: number
   mtu: number
   comment?: string
   enabled?: boolean
@@ -258,7 +258,7 @@ export function generateNativeConf(iface: WgExportIface, opts?: { includePrivate
     lines.push(`# PrivateKey = <заполните приватный ключ с роутера>`)
   }
   if (iface.address) lines.push(`Address = ${iface.address}`)
-  lines.push(`ListenPort = ${iface.listenPort}`)
+  if (iface.listenPort) lines.push(`ListenPort = ${iface.listenPort}`)
   if (iface.mtu) lines.push(`MTU = ${iface.mtu}`)
   lines.push(``)
 
@@ -312,7 +312,7 @@ export function generateMikrotikRsc(iface: WgExportIface): string {
   lines.push(``)
   lines.push(`/interface wireguard add \\`)
   lines.push(`  name=${iface.name} \\`)
-  lines.push(`  listen-port=${iface.listenPort} \\`)
+  lines.push(`  listen-port=${iface.listenPort ?? 13231} \\`)
   lines.push(`  mtu=${iface.mtu} \\`)
   if (iface.privateKey) lines.push(`  private-key="${iface.privateKey}" \\`)
   if (iface.comment) lines.push(`  comment="${iface.comment.replace(/"/g, '\\"')}" \\`)

backend/src/services/wireguard-ros.ts

@@ -0,0 +1,50 @@
+import type { MikrotikClient } from "./mikrotik.js"
+
+/** Общие PUT iface / peer / address для `/wireguard` и traffic-flow overlay. */
+export function toRosBody(obj: Record<string, string | undefined>): Record<string, string> {
+  const out: Record<string, string> = {}
+  for (const [k, v] of Object.entries(obj)) {
+    if (v !== undefined && v !== "") out[k] = v
+  }
+  return out
+}
+
+export function asRosArray<T>(raw: unknown): T[] {
+  if (Array.isArray(raw)) return raw as T[]
+  if (raw && typeof raw === "object") return [raw as T]
+  return []
+}
+
+export function rosRowId(row: Record<string, unknown>): string {
+  return String(row[".id"] ?? row.id ?? "")
+}
+
+export async function putWireguardInterface(
+  client: MikrotikClient,
+  fields: Record<string, string | undefined>,
+): Promise<void> {
+  await client.put("/interface/wireguard", toRosBody(fields))
+}
+
+export async function putIpAddress(
+  client: MikrotikClient,
+  address: string,
+  iface: string,
+): Promise<void> {
+  await client.put("/ip/address", { address, interface: iface })
+}
+
+export async function putWireguardPeer(
+  client: MikrotikClient,
+  fields: Record<string, string | undefined>,
+): Promise<void> {
+  await client.put("/interface/wireguard/peers", toRosBody(fields))
+}
+
+export async function patchRosPath(
+  client: MikrotikClient,
+  path: string,
+  fields: Record<string, string | undefined>,
+): Promise<void> {
+  await client.patch(path, toRosBody(fields))
+}

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

@@ -0,0 +1,113 @@
+"use client"
+
+import { useMemo } from "react"
+import { type ColumnDef, getCoreRowModel, useReactTable } from "@tanstack/react-table"
+import type { FlowTalkerDto } from "@mmapp/contracts/traffic-flow"
+import { DataGridShell } from "@/components/data-grids/shared/data-grid-shell"
+import {
+  DATA_GRID_CELL_PAD,
+  DATA_GRID_CELL_PAD_FIRST,
+  DATA_GRID_CELL_PAD_LAST,
+} from "@/components/data-grids/shared/data-grid-layout"
+import { fmtRate } from "@/lib/fmt-rate"
+import { cn } from "@/lib/utils"
+
+function formatBytes(n: number): string {
+  if (n >= 1_000_000_000) return `${(n / 1_000_000_000).toFixed(2)} ГБ`
+  if (n >= 1_000_000) return `${(n / 1_000_000).toFixed(1)} МБ`
+  if (n >= 1000) return `${(n / 1000).toFixed(1)} КБ`
+  return `${n} Б`
+}
+
+function TrafficFlowsDataGrid({
+  rows,
+  emptyHint,
+}: {
+  rows: FlowTalkerDto[]
+  emptyHint?: string
+}) {
+  const columns = useMemo<ColumnDef<FlowTalkerDto>[]>(
+    () => [
+      {
+        id: "server",
+        accessorKey: "serverName",
+        header: () => <span className="text-xs font-medium text-muted-foreground">JH</span>,
+        cell: ({ row }) => <span className="text-sm font-medium">{row.original.serverName}</span>,
+        meta: { headerClassName: DATA_GRID_CELL_PAD_FIRST, cellClassName: DATA_GRID_CELL_PAD_FIRST },
+      },
+      {
+        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>
+        ),
+        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>
+        ),
+        meta: { headerClassName: DATA_GRID_CELL_PAD, cellClassName: DATA_GRID_CELL_PAD },
+      },
+      {
+        id: "proto",
+        accessorKey: "protoName",
+        header: () => <span className="text-xs font-medium text-muted-foreground">Proto</span>,
+        cell: ({ row }) => <span className="text-xs">{row.original.protoName}</span>,
+        meta: { headerClassName: DATA_GRID_CELL_PAD, cellClassName: DATA_GRID_CELL_PAD },
+      },
+      {
+        id: "rate",
+        accessorFn: (r) => r.bps,
+        header: () => <span className="text-xs font-medium text-muted-foreground">Скорость</span>,
+        cell: ({ row }) => <span className="text-xs tabular-nums">{fmtRate(row.original.bps / 1_000_000)}</span>,
+        meta: { headerClassName: DATA_GRID_CELL_PAD, cellClassName: DATA_GRID_CELL_PAD },
+      },
+      {
+        id: "bytes",
+        accessorKey: "bytes",
+        header: () => <span className="text-xs font-medium text-muted-foreground">Байты</span>,
+        cell: ({ row }) => <span className="text-xs tabular-nums">{formatBytes(row.original.bytes)}</span>,
+        meta: { headerClassName: DATA_GRID_CELL_PAD, cellClassName: DATA_GRID_CELL_PAD },
+      },
+      {
+        id: "iface",
+        accessorKey: "inIface",
+        header: () => <span className="text-xs font-medium text-muted-foreground">Iface</span>,
+        cell: ({ row }) => <span className="font-mono text-xs text-muted-foreground">{row.original.inIface || "—"}</span>,
+        meta: { headerClassName: DATA_GRID_CELL_PAD_LAST, cellClassName: cn(DATA_GRID_CELL_PAD_LAST) },
+      },
+    ],
+    [],
+  )
+
+  const table = useReactTable({
+    data: rows,
+    columns,
+    getCoreRowModel: getCoreRowModel(),
+    getRowId: (row, i) => `${row.serverId}-${row.src}-${row.dst}-${row.proto}-${row.srcPort}-${row.dstPort}-${i}`,
+  })
+
+  return (
+    <DataGridShell
+      table={table}
+      recordCount={rows.length}
+      emptyMessage={
+        emptyHint
+        || "Пока нет IPFIX. Поднимите wg-flow на хосте MM и подключите jump-host одним кликом."
+      }
+    />
+  )
+}
+
+export { TrafficFlowsDataGrid }

components/reui-kit/code-export-sheet.tsx

@@ -60,10 +60,10 @@ function highlightLine(line: string): string {
   return "text-foreground/90"
 }
 
-function CodeBlock({ code }: { code: string }) {
+function CodeBlock({ code, className }: { code: string; className?: string }) {
   const lines = code.length ? code.split("\n") : [""]
   return (
-    <pre className="px-4 py-3.5 text-[12px] font-mono leading-[1.65] whitespace-pre-wrap break-all select-all">
+    <pre className={cn("px-4 py-3.5 text-[12px] font-mono leading-[1.65] whitespace-pre-wrap break-all select-all", className)}>
       {lines.map((line, i) => (
         <span key={i} className={cn("block", highlightLine(line))}>
           {line || " "}
@@ -281,4 +281,4 @@ function CodeExportSheet({
   )
 }
 
-export { CodeExportSheet, downloadText }
+export { CodeExportSheet, CodeBlock, downloadText }

components/traffic/flow-overlay-sheet.tsx

@@ -0,0 +1,229 @@
+"use client"
+
+import { useEffect, useMemo, useState } from "react"
+import { toast } from "sonner"
+import { FormField } from "@/components/form-kit"
+import { Alert, AlertDescription, AlertTitle } from "@/components/reui/alert"
+import { Frame, FramePanel } from "@/components/reui/frame"
+import { CodeBlock, downloadText } from "@/components/reui-kit/code-export-sheet"
+import { Button } from "@/components/ui/button"
+import { Input } from "@/components/ui/input"
+import { ScrollArea } from "@/components/ui/scroll-area"
+import {
+  Select,
+  SelectContent,
+  SelectItem,
+  SelectTrigger,
+  SelectValue,
+} from "@/components/ui/select"
+import { Tabs, TabsList, TabsTrigger } from "@/components/ui/tabs"
+import {
+  Sheet, SheetContent, SheetHeader, SheetTitle,
+  SheetDescription, SheetFooter, SheetClose,
+} from "@/components/ui/sheet"
+import { applyTrafficFlowOverlay } from "@/shared/api/traffic-flow"
+import type { ServerRead } from "@mmapp/contracts/servers"
+import type { TrafficFlowOverlayResult } from "@mmapp/contracts/traffic-flow"
+import {
+  CheckIcon,
+  CopyIcon,
+  DownloadIcon,
+  InfoIcon,
+} from "lucide-react"
+
+function FlowOverlaySheet({
+  open,
+  onOpenChange,
+  servers,
+  backendUrl,
+  onDone,
+}: {
+  open: boolean
+  onOpenChange: (v: boolean) => void
+  servers: ServerRead[]
+  backendUrl: string
+  onDone?: (result: TrafficFlowOverlayResult) => void
+}) {
+  const jumpHosts = useMemo(
+    () => servers.filter((s) => s.enabled && s.type === "jump-host"),
+    [servers],
+  )
+  const [serverId, setServerId] = useState("")
+  const [endpoint, setEndpoint] = useState("")
+  const [busy, setBusy] = useState(false)
+  const [result, setResult] = useState<TrafficFlowOverlayResult | null>(null)
+  const [copied, setCopied] = useState(false)
+  const [tab, setTab] = useState("linux")
+
+  useEffect(() => {
+    if (!open) return
+    setResult(null)
+    setCopied(false)
+    setTab("linux")
+    const first = jumpHosts[0]
+    const nextId = first ? String(first.id) : ""
+    setServerId(nextId)
+    setEndpoint(first?.host ?? "")
+  }, [open, jumpHosts])
+
+  function handleServerChange(id: string) {
+    setServerId(id)
+    const selected = jumpHosts.find((s) => String(s.id) === id)
+    if (selected) setEndpoint(selected.host)
+  }
+
+  const formats = result?.hostFiles ?? []
+  const active = formats.find((f) => f.id === tab) ?? formats[0]
+  const selectedHost = jumpHosts.find((s) => String(s.id) === serverId)
+  const selectedLabel = selectedHost
+    ? `${selectedHost.name || selectedHost.host} (${selectedHost.host})`
+    : "Выберите сервер…"
+  const canSubmit = Boolean(serverId && endpoint.trim()) && !busy
+
+  async function handleSubmit() {
+    if (!serverId || !endpoint.trim()) return
+    setBusy(true)
+    try {
+      const res = await applyTrafficFlowOverlay(backendUrl, serverId, endpoint.trim())
+      setResult(res)
+      setTab(res.hostFiles[0]?.id ?? "linux")
+      toast.success(`wg-flow на ${res.address}`)
+      onDone?.(res)
+    } catch (e) {
+      toast.error(e instanceof Error ? e.message : "Не удалось подключить JH")
+    } finally {
+      setBusy(false)
+    }
+  }
+
+  function handleCopy() {
+    const code = active?.code ?? ""
+    if (!code) return
+    void navigator.clipboard.writeText(code).then(() => {
+      setCopied(true)
+      toast.success("Скопировано")
+      setTimeout(() => setCopied(false), 1800)
+    })
+  }
+
+  return (
+    <Sheet open={open} onOpenChange={onOpenChange}>
+      <SheetContent
+        side="right"
+        className="flex w-full flex-col gap-0 overflow-hidden p-0 sm:max-w-2xl"
+      >
+        <SheetHeader className="shrink-0 gap-1 border-b px-5 pt-5 pb-4 pr-12">
+          <SheetTitle className="text-base font-semibold tracking-tight">
+            Подключить jump-host
+          </SheetTitle>
+          <SheetDescription>
+            Создаст wg-flow на MikroTik (сервер, listen 13232) и выдаст готовый bash для Linux-хоста Docker MM (клиент).
+          </SheetDescription>
+        </SheetHeader>
+        <div className="flex min-h-0 flex-1 flex-col gap-4 overflow-y-auto px-5 py-4">
+          <FormField label="Jump-host" required>
+            <Select
+              value={serverId || undefined}
+              onValueChange={(v) => handleServerChange(String(v ?? ""))}
+            >
+              <SelectTrigger className="h-9 w-full min-w-0">
+                <SelectValue>{selectedLabel}</SelectValue>
+              </SelectTrigger>
+              <SelectContent align="start" className="min-w-(--anchor-width)">
+                {jumpHosts.map((s) => (
+                  <SelectItem key={s.id} value={String(s.id)}>
+                    {s.name || s.host} ({s.host})
+                  </SelectItem>
+                ))}
+              </SelectContent>
+            </Select>
+          </FormField>
+          <FormField
+            label="Публичный IP или DNS jump-host"
+            required
+            hint="Куда хост MM (wg-quick) стучится по UDP 13232. Не контейнер backend."
+          >
+            <Input
+              className="font-mono"
+              value={endpoint}
+              onChange={(e) => setEndpoint(e.target.value)}
+              placeholder="jh.example.com"
+              autoComplete="off"
+            />
+          </FormField>
+          {result ? (
+            <div className="flex min-h-0 flex-col gap-4">
+              <Alert variant="success">
+                <InfoIcon />
+                <AlertTitle>Linux-хост /opt/cdn-mm</AlertTitle>
+                <AlertDescription>
+                  Скопируйте вкладку Linux и выполните от root. WG — клиент к JH:13232; контейнер слушает только WG-IP:4739, не 0.0.0.0. Ключ не кладите в git.
+                </AlertDescription>
+              </Alert>
+              <ul className="flex flex-col gap-1.5">
+                {result.steps.map((s) => (
+                  <li key={s} className="flex items-start gap-2 text-sm">
+                    <CheckIcon className="mt-0.5 size-3.5 shrink-0 text-success" />
+                    <span>{s}</span>
+                  </li>
+                ))}
+              </ul>
+              {formats.length > 0 && active ? (
+                <div className="flex min-h-0 flex-col gap-3">
+                  <Tabs
+                    value={tab}
+                    onValueChange={(v) => {
+                      setTab(String(v))
+                      setCopied(false)
+                    }}
+                    className="shrink-0 gap-0"
+                  >
+                    <TabsList className="h-9 w-full">
+                      {formats.map((f) => (
+                        <TabsTrigger key={f.id} value={f.id} className="flex-1 px-2 text-xs sm:text-sm">
+                          {f.label}
+                        </TabsTrigger>
+                      ))}
+                    </TabsList>
+                  </Tabs>
+                  <Frame dense className="flex min-h-0 flex-col overflow-hidden">
+                    <FramePanel className="relative flex min-h-0 flex-col overflow-hidden p-0">
+                      <ScrollArea className="h-full min-h-0 max-h-[min(52vh,22rem)]">
+                        <CodeBlock code={active.code} className="whitespace-pre break-normal" />
+                      </ScrollArea>
+                    </FramePanel>
+                  </Frame>
+                  <div className="flex items-center justify-end gap-2">
+                    <Button
+                      type="button"
+                      variant="outline"
+                      size="sm"
+                      onClick={() => downloadText(active.filename, active.code)}
+                    >
+                      <DownloadIcon className="size-3.5" />
+                      Файл
+                    </Button>
+                    <Button type="button" size="sm" onClick={handleCopy}>
+                      {copied ? <CheckIcon className="size-3.5" /> : <CopyIcon className="size-3.5" />}
+                      {copied ? "Скопировано" : "Копировать"}
+                    </Button>
+                  </div>
+                </div>
+              ) : null}
+            </div>
+          ) : null}
+        </div>
+        <SheetFooter className="shrink-0 flex-row items-center justify-between gap-3 border-t px-5 py-3.5 sm:flex-row">
+          <SheetClose render={<Button type="button" variant="outline" className="shrink-0" />}>
+            Закрыть
+          </SheetClose>
+          <Button disabled={!canSubmit} onClick={() => { void handleSubmit() }}>
+            {busy ? "Подключение…" : "Подключить"}
+          </Button>
+        </SheetFooter>
+      </SheetContent>
+    </Sheet>
+  )
+}
+
+export { FlowOverlaySheet }

components/traffic/netflow-settings-panel.tsx

@@ -0,0 +1,224 @@
+"use client"
+
+import { useCallback, useEffect, useState } from "react"
+import { toast } from "sonner"
+import { FormField, FormToggle } from "@/components/form-kit"
+import { Alert, AlertDescription, AlertTitle } from "@/components/reui/alert"
+import { Badge } from "@/components/reui/badge"
+import { OpsPanel } from "@/components/ops-panel"
+import { Button } from "@/components/ui/button"
+import { Input } from "@/components/ui/input"
+import { CodeExportSheet, type CodeExportFormat } from "@/components/reui-kit/code-export-sheet"
+import type { TrafficFlowSettingsDto } from "@mmapp/contracts/traffic-flow"
+import {
+  generateTrafficFlowKeys,
+  getTrafficFlowHostFiles,
+  getTrafficFlowSettings,
+  putTrafficFlowSettings,
+} from "@/shared/api/traffic-flow"
+import { KeyRoundIcon, DownloadIcon, InfoIcon } from "lucide-react"
+
+const HOST_STEPS = [
+  "На хосте Docker (не в контейнере mmapp-backend): apt install wireguard (или эквивалент).",
+  "Скачайте wg-flow.conf и положите в /etc/wireguard/wg-flow.conf.",
+  "wg-quick up wg-flow  (или systemctl enable --now wg-quick@wg-flow).",
+  "Firewall: разрешите UDP listen WireGuard. UDP 4739 наружу не открывайте.",
+  "В docker-compose у backend раскомментируйте bind IPFIX только на адресе wg-flow.",
+  "Проверка: wg show · ss -ulnp | grep 4739 · в этой панели — last datagram.",
+]
+
+function NetflowSettingsPanel({
+  backendUrl,
+  enabled,
+}: {
+  backendUrl: string
+  enabled: boolean
+}) {
+  const [settings, setSettings] = useState<TrafficFlowSettingsDto | null>(null)
+  const [busy, setBusy] = useState(false)
+  const [exportOpen, setExportOpen] = useState(false)
+  const [formats, setFormats] = useState<CodeExportFormat[]>([])
+  const [collectorIp, setCollectorIp] = useState("10.255.254.1")
+  const [flowPort, setFlowPort] = useState("4739")
+  const [wgPort, setWgPort] = useState("51821")
+  const [prefix, setPrefix] = useState("10.255.254.0/24")
+  const [endpoint, setEndpoint] = useState("")
+  const [retention, setRetention] = useState("24")
+  const [topN, setTopN] = useState("200")
+  const [ingestOn, setIngestOn] = useState(false)
+
+  const load = useCallback(async () => {
+    if (!enabled) return
+    const s = await getTrafficFlowSettings(backendUrl)
+    setSettings(s)
+    setCollectorIp(s.collectorIp)
+    setFlowPort(String(s.flowListenPort))
+    setWgPort(String(s.wgListenPort))
+    setPrefix(s.prefix)
+    setEndpoint(s.publicEndpoint)
+    setRetention(String(s.retentionHours))
+    setTopN(String(s.topN))
+    setIngestOn(s.enabled)
+  }, [backendUrl, enabled])
+
+  useEffect(() => {
+    void load().catch((e: unknown) => {
+      toast.error(e instanceof Error ? e.message : "Не удалось загрузить NetFlow")
+    })
+  }, [load])
+
+  async function handleSave() {
+    setBusy(true)
+    try {
+      const res = await putTrafficFlowSettings(backendUrl, {
+        enabled: ingestOn,
+        collectorIp,
+        flowListenPort: Number.parseInt(flowPort, 10) || 4739,
+        wgListenPort: Number.parseInt(wgPort, 10) || 51821,
+        prefix,
+        publicEndpoint: endpoint,
+        retentionHours: Number.parseInt(retention, 10) || 24,
+        topN: Number.parseInt(topN, 10) || 200,
+      })
+      setSettings(res.settings)
+      toast.success("Настройки NetFlow сохранены")
+    } catch (e) {
+      toast.error(e instanceof Error ? e.message : "Не удалось сохранить")
+    } finally {
+      setBusy(false)
+    }
+  }
+
+  async function handleKeys() {
+    setBusy(true)
+    try {
+      const res = await generateTrafficFlowKeys(backendUrl)
+      setSettings(res.settings)
+      toast.success(res.created ? "Ключи хоста созданы" : "Ключи уже есть")
+    } catch (e) {
+      toast.error(e instanceof Error ? e.message : "Не удалось сгенерировать ключи")
+    } finally {
+      setBusy(false)
+    }
+  }
+
+  async function handleExport() {
+    setBusy(true)
+    try {
+      const res = await getTrafficFlowHostFiles(backendUrl)
+      setFormats(res.files.map((f) => ({
+        id: f.id,
+        label: f.label,
+        filename: f.filename,
+        code: f.code,
+      })))
+      setExportOpen(true)
+      await load()
+    } catch (e) {
+      toast.error(e instanceof Error ? e.message : "Не удалось получить файлы")
+    } finally {
+      setBusy(false)
+    }
+  }
+
+  return (
+    <>
+      <OpsPanel
+        title="Traffic Flow / NetFlow (IPFIX)"
+        description="Дополнение к сбору счётчиков REST. Приём только через WireGuard на хосте Docker MM. Preview: https://reui.io/preview/base/settings-16"
+        headerRight={
+          <div className="flex items-center gap-2">
+            {settings?.listenerBound ? (
+              <Badge variant="success">listener {settings.listenerAddress}</Badge>
+            ) : (
+              <Badge variant="secondary">listener выкл</Badge>
+            )}
+          </div>
+        }
+        contentClassName="px-5 py-4 flex flex-col gap-4"
+      >
+        <Alert>
+          <InfoIcon />
+          <AlertTitle>Ключи и UDP 4739</AlertTitle>
+          <AlertDescription>
+            Приватный ключ хранится в SQLite панели, не коммитьте его. Порт IPFIX публикуйте только на адресе wg-flow, не на 0.0.0.0.
+          </AlertDescription>
+        </Alert>
+
+        <div className="flex items-center gap-3">
+          <FormToggle checked={ingestOn} onChange={setIngestOn} />
+          <span className="text-sm">Принимать IPFIX</span>
+        </div>
+
+        <div className="grid grid-cols-1 sm:grid-cols-2 gap-3">
+          <FormField label="Collector IP" hint="Адрес в туннеле, куда JH шлёт flow">
+            <Input className="font-mono" value={collectorIp} onChange={(e) => setCollectorIp(e.target.value)} />
+          </FormField>
+          <FormField label="Префикс overlay">
+            <Input className="font-mono" value={prefix} onChange={(e) => setPrefix(e.target.value)} />
+          </FormField>
+          <FormField label="UDP IPFIX">
+            <Input className="font-mono" value={flowPort} onChange={(e) => setFlowPort(e.target.value)} />
+          </FormField>
+          <FormField label="WG listen">
+            <Input className="font-mono" value={wgPort} onChange={(e) => setWgPort(e.target.value)} />
+          </FormField>
+          <FormField label="Публичный endpoint хоста MM" hint="IP или DNS, который видят JH" required>
+            <Input className="font-mono" value={endpoint} onChange={(e) => setEndpoint(e.target.value)} placeholder="203.0.113.10" />
+          </FormField>
+          <FormField label="Public key хоста">
+            <Input className="font-mono text-xs" readOnly value={settings?.hostPublicKey || "— сгенерируйте ключи —"} />
+          </FormField>
+          <FormField label="Хранение (часов)">
+            <Input value={retention} onChange={(e) => setRetention(e.target.value)} inputMode="numeric" />
+          </FormField>
+          <FormField label="Top-N разговоров">
+            <Input value={topN} onChange={(e) => setTopN(e.target.value)} inputMode="numeric" />
+          </FormField>
+        </div>
+
+        <p className="text-xs text-muted-foreground">
+          Last datagram:{" "}
+          {settings?.lastDatagramAt
+            ? new Date(settings.lastDatagramAt).toLocaleString("ru-RU")
+            : "—"}
+          {settings?.lastExporterIp ? ` · ${settings.lastExporterIp}` : ""}
+          {settings?.lastError ? ` · ${settings.lastError}` : ""}
+        </p>
+
+        <div className="rounded-md border px-4 py-3 flex flex-col gap-2">
+          <p className="text-sm font-medium">Туннель на сервере Docker MM</p>
+          <ol className="text-xs text-muted-foreground flex flex-col gap-1.5 list-decimal pl-4">
+            {HOST_STEPS.map((s) => (
+              <li key={s}>{s}</li>
+            ))}
+          </ol>
+        </div>
+
+        <div className="flex flex-wrap gap-2">
+          <Button size="sm" disabled={busy} onClick={() => { void handleSave() }}>
+            Сохранить NetFlow
+          </Button>
+          <Button size="sm" variant="outline" disabled={busy} onClick={() => { void handleKeys() }}>
+            <KeyRoundIcon className="size-4" />
+            Ключи хоста
+          </Button>
+          <Button size="sm" variant="outline" disabled={busy} onClick={() => { void handleExport() }}>
+            <DownloadIcon className="size-4" />
+            wg-quick / compose / firewall
+          </Button>
+        </div>
+      </OpsPanel>
+
+      <CodeExportSheet
+        open={exportOpen}
+        onClose={() => setExportOpen(false)}
+        title="Файлы для хоста Docker MM"
+        description="wg-quick, фрагмент compose и firewall. Хост, не контейнер backend."
+        formats={formats}
+      />
+    </>
+  )
+}
+
+export { NetflowSettingsPanel }

deploy/docker-compose.cdn-mm.yml

@@ -134,6 +134,10 @@ services:
       AUTH_JWT_SECRET: ${AUTH_JWT_SECRET:?set AUTH_JWT_SECRET in .env}
       AUTH_ISSUER: ${AUTH_ISSUER:-https://auth.shnt.top}
       AUTH_PORTAL_URL: ${AUTH_PORTAL_URL:-https://auth.shnt.top}
+      # IPFIX: внутри контейнера слушать все iface; на хосте bind только WG-IP после wg-quick@wg-flow
+      FLOW_LISTEN_HOST: "0.0.0.0"
+    # ports:
+    #   - "10.255.254.1:4739:4739/udp"
     volumes:
       - ./data/mm:/app/data
     networks:

deploy/docker-compose.yml

@@ -5,8 +5,12 @@ services:
     restart: unless-stopped
     ports:
       - "8000:8000"
+      # IPFIX: только на WG-адресе хоста после `wg-quick up wg-flow`, не 0.0.0.0
+      # - "10.255.254.1:4739:4739/udp"
     environment:
       CORS_ORIGIN: ${CORS_ORIGIN:-http://localhost:3000}
+      # Внутри контейнера слушаем все интерфейсы; на хосте UDP 4739 публикуется только на WG-IP
+      FLOW_LISTEN_HOST: "0.0.0.0"
     volumes:
       - /opt/mmapp/data:/app/data
     labels:

packages/contracts/package.json

@@ -41,6 +41,10 @@
     "./users": {
       "types": "./dist/users.d.ts",
       "default": "./dist/users.js"
+    },
+    "./traffic-flow": {
+      "types": "./dist/traffic-flow.d.ts",
+      "default": "./dist/traffic-flow.js"
     }
   },
   "dependencies": {

packages/contracts/src/index.ts

@@ -5,3 +5,4 @@ export * from "./certificates.js"
 export * from "./backups.js"
 export * from "./wireguard.js"
 export * from "./users.js"
+export * from "./traffic-flow.js"

packages/contracts/src/traffic-flow.ts

@@ -0,0 +1,102 @@
+import { z } from "zod"
+
+export const flowHostPeerSchema = z.object({
+  serverId: z.number().int().positive(),
+  name: z.string(),
+  publicKey: z.string().min(1),
+  allowedIps: z.array(z.string().min(1)).min(1),
+  address: z.string().min(1),
+  endpoint: z.string().optional(),
+})
+
+export const trafficFlowSettingsDtoSchema = z.object({
+  enabled: z.boolean(),
+  collectorIp: z.string().min(1),
+  flowListenPort: z.number().int().positive(),
+  wgListenPort: z.number().int().positive(),
+  prefix: z.string().min(1),
+  publicEndpoint: z.string(),
+  hostPublicKey: z.string(),
+  hasHostPrivateKey: z.boolean(),
+  hubServerId: z.number().int().positive().nullable(),
+  retentionHours: z.number().int().positive(),
+  topN: z.number().int().positive(),
+  lastDatagramAt: z.string().nullable(),
+  lastExporterIp: z.string().nullable(),
+  lastError: z.string().nullable(),
+  packetsReceived: z.number().int().nonnegative(),
+  listenerBound: z.boolean(),
+  listenerAddress: z.string().nullable(),
+  peers: z.array(flowHostPeerSchema),
+})
+
+export const trafficFlowSettingsPatchSchema = z.object({
+  enabled: z.boolean().optional(),
+  collectorIp: z.string().min(1).optional(),
+  flowListenPort: z.number().int().positive().optional(),
+  wgListenPort: z.number().int().positive().optional(),
+  prefix: z.string().min(1).optional(),
+  publicEndpoint: z.string().optional(),
+  hubServerId: z.number().int().positive().nullable().optional(),
+  retentionHours: z.number().int().positive().optional(),
+  topN: z.number().int().positive().max(1000).optional(),
+})
+
+export const trafficFlowOverlayRequestSchema = z.object({
+  serverId: z.union([z.string(), z.number()]),
+  publicEndpoint: z.string().optional(),
+})
+
+export const trafficFlowHostFileSchema = z.object({
+  id: z.string(),
+  label: z.string(),
+  filename: z.string(),
+  code: z.string(),
+})
+
+export const trafficFlowOverlayResultSchema = z.object({
+  ok: z.boolean(),
+  serverId: z.number().int(),
+  interfaceName: z.string(),
+  address: z.string(),
+  publicKey: z.string(),
+  linuxPeerBlock: z.string(),
+  trafficFlow: z.boolean(),
+  steps: z.array(z.string()),
+  hostFiles: z.array(trafficFlowHostFileSchema),
+})
+
+export const flowTalkerDtoSchema = z.object({
+  serverId: z.string(),
+  serverName: z.string(),
+  src: z.string(),
+  dst: z.string(),
+  proto: z.number().int(),
+  protoName: z.string(),
+  srcPort: z.number().int(),
+  dstPort: z.number().int(),
+  bytes: z.number().nonnegative(),
+  packets: z.number().nonnegative(),
+  bps: z.number().nonnegative(),
+  inIface: z.string(),
+})
+
+export const flowStatsDtoSchema = z.object({
+  exportersOnline: z.number().int().nonnegative(),
+  bytesPerMin: z.number().nonnegative(),
+  uniqueSrc: z.number().int().nonnegative(),
+  uniqueDst: z.number().int().nonnegative(),
+  topProto: z.string(),
+  talkers: z.array(flowTalkerDtoSchema),
+  lastExporterIp: z.string().nullable().optional(),
+  lastError: z.string().nullable().optional(),
+  packetsReceived: z.number().int().nonnegative().optional(),
+})
+
+export type FlowHostPeer = z.infer<typeof flowHostPeerSchema>
+export type TrafficFlowSettingsDto = z.infer<typeof trafficFlowSettingsDtoSchema>
+export type TrafficFlowSettingsPatch = z.infer<typeof trafficFlowSettingsPatchSchema>
+export type TrafficFlowOverlayResult = z.infer<typeof trafficFlowOverlayResultSchema>
+export type TrafficFlowHostFile = z.infer<typeof trafficFlowHostFileSchema>
+export type FlowTalkerDto = z.infer<typeof flowTalkerDtoSchema>
+export type FlowStatsDto = z.infer<typeof flowStatsDtoSchema>

shared/api/traffic-flow.ts

@@ -0,0 +1,52 @@
+import type {
+  FlowStatsDto,
+  TrafficFlowHostFile,
+  TrafficFlowOverlayResult,
+  TrafficFlowSettingsDto,
+  TrafficFlowSettingsPatch,
+} from "@mmapp/contracts/traffic-flow"
+import { requestJson } from "@/shared/api/http-client"
+
+export type { TrafficFlowHostFile }
+
+export async function getTrafficFlowSettings(baseUrl: string): Promise<TrafficFlowSettingsDto> {
+  return requestJson<TrafficFlowSettingsDto>(baseUrl, "/api/traffic/flow/settings")
+}
+
+export async function putTrafficFlowSettings(
+  baseUrl: string,
+  patch: TrafficFlowSettingsPatch,
+): Promise<{ ok: boolean; settings: TrafficFlowSettingsDto }> {
+  return requestJson(baseUrl, "/api/traffic/flow/settings", {
+    method: "PUT",
+    body: JSON.stringify(patch),
+  })
+}
+
+export async function generateTrafficFlowKeys(baseUrl: string): Promise<{
+  ok: boolean
+  created: boolean
+  publicKey: string
+  settings: TrafficFlowSettingsDto
+}> {
+  return requestJson(baseUrl, "/api/traffic/flow/settings/generate-keys", { method: "POST" })
+}
+
+export async function getTrafficFlowHostFiles(baseUrl: string): Promise<{ files: TrafficFlowHostFile[] }> {
+  return requestJson(baseUrl, "/api/traffic/flow/host-files")
+}
+
+export async function applyTrafficFlowOverlay(
+  baseUrl: string,
+  serverId: string | number,
+  publicEndpoint?: string,
+): Promise<TrafficFlowOverlayResult> {
+  return requestJson(baseUrl, "/api/traffic/flow-overlay", {
+    method: "POST",
+    body: JSON.stringify({ serverId, publicEndpoint }),
+  })
+}
+
+export async function getTrafficFlows(baseUrl: string, range = "5m"): Promise<FlowStatsDto> {
+  return requestJson<FlowStatsDto>(baseUrl, `/api/traffic/flows?range=${encodeURIComponent(range)}`)
+}