diff --git a/.gitattributes b/.gitattributes new file mode 100644 index 00000000..16ac5ebe --- /dev/null +++ b/.gitattributes @@ -0,0 +1,2 @@ +deploy/netflow-agent/*.sh text eol=lf +deploy/netflow-agent/*.py text eol=lf diff --git a/.gitea/workflows/docker.yml b/.gitea/workflows/docker.yml index 65fadfa3..11d8f720 100644 --- a/.gitea/workflows/docker.yml +++ b/.gitea/workflows/docker.yml @@ -155,6 +155,8 @@ jobs: cp -R packages/contracts/src "$STAGING/packages/contracts/" cp backend/package.json backend/tsconfig.json "$STAGING/backend/" cp -R backend/src "$STAGING/backend/" + mkdir -p "$STAGING/deploy/netflow-agent" + cp deploy/netflow-agent/install.sh "$STAGING/deploy/netflow-agent/" if [ -d backend/drizzle ]; then cp -R backend/drizzle "$STAGING/backend/" fi @@ -329,12 +331,69 @@ jobs: org.opencontainers.image.revision=${{ gitea.sha }} org.opencontainers.image.created=${{ gitea.event.head_commit.timestamp }} + netflow-agent-image: + needs: prepare-release + runs-on: ubuntu-latest + steps: + - name: Checkout + uses: actions/checkout@v4 + + - name: Configure netflow-agent image name + shell: bash + run: | + set -euo pipefail + owner=$(echo "${{ gitea.repository }}" | cut -d/ -f1 | tr '[:upper:]' '[:lower:]') + name=$(echo "${{ gitea.repository }}" | cut -d/ -f2 | tr '[:upper:]' '[:lower:]') + stem=$(printf '%s' "$name" | sed -E 's/-[0-9]+$//') + echo "IMAGE_NAME=${{ env.REGISTRY }}/${owner}/${stem}-netflow-agent" >> "${GITHUB_ENV}" + + - name: Test controller and installer + run: | + python3 -m unittest discover -s deploy/netflow-agent -p "test_*.py" -v + bash -n deploy/netflow-agent/install.sh + + - name: Set up Docker Buildx + uses: docker/setup-buildx-action@v3 + + - name: Log in to Gitea Registry + uses: docker/login-action@v3 + with: + registry: git.shx.one + username: ${{ gitea.actor }} + password: ${{ secrets.ACTIONS_PAT }} + + - name: Verify image with real WireGuard + run: | + docker build --target test -t mm-netflow-agent-test deploy/netflow-agent + docker run --rm --cap-drop ALL --cap-add NET_ADMIN --read-only --tmpfs /tmp -e FLOW_AGENT_KERNEL_TEST=1 --entrypoint python3 mm-netflow-agent-test -m unittest discover -s /app -p "test_agent.py" -v + + - name: Build and push netflow-agent image + uses: docker/build-push-action@v5 + with: + context: deploy/netflow-agent + file: deploy/netflow-agent/Dockerfile + platforms: linux/amd64 + push: true + tags: | + ${{ env.IMAGE_NAME }}:latest + ${{ env.IMAGE_NAME }}:${{ gitea.sha }} + ${{ env.IMAGE_NAME }}:${{ needs.prepare-release.outputs.version }} + cache-from: type=gha,scope=netflow-agent + cache-to: type=gha,mode=max,scope=netflow-agent + labels: | + org.opencontainers.image.title=mmapp-netflow-agent + org.opencontainers.image.description=MikroTik Manager local WireGuard controller + org.opencontainers.image.version=${{ needs.prepare-release.outputs.version }} + org.opencontainers.image.revision=${{ gitea.sha }} + org.opencontainers.image.created=${{ gitea.event.head_commit.timestamp }} + publish-release: needs: - prepare-release - backend-image - frontend-image - updater-image + - netflow-agent-image if: needs.prepare-release.outputs.should_release == 'true' runs-on: ubuntu-latest steps: @@ -420,6 +479,7 @@ jobs: - prepare-release - backend-image - frontend-image + - netflow-agent-image runs-on: ubuntu-latest env: WEBHOOK_URL: ${{ secrets.DEPLOY_WEBHOOK_URL }} diff --git a/.gitignore b/.gitignore index fed7b35f..af2e1542 100644 --- a/.gitignore +++ b/.gitignore @@ -62,3 +62,11 @@ opencode.json # Local codegraph process .codegraph/daemon.pid + +# Python controller tests +__pycache__/ +*.pyc + +# NetFlow agent host state (when Compose is installed in a checkout) +/data/netflow-wireguard/ +/data/netflow-agent-socket/ diff --git a/README.md b/README.md index 69b66289..61f365ba 100644 --- a/README.md +++ b/README.md @@ -568,3 +568,7 @@ docker exec mmapp-postgres pg_dump -Fc -U mmapp mmapp > mikrotik-manager.dump 10. Проверить `curl` на `:8000/health` и `:3000/`. 11. В UI: режим live и URL бэкенда; сверить с `CORS_ORIGIN`. 12. Просмотреть `docker logs mmapp-updater`; настроить регулярный backup тома `backend-data`. + +### Автоматизация NetFlow через Docker-агент + +[Установка, схема взаимодействия и production-состав](docs/netflow-agent.md). diff --git a/backend/.env.example b/backend/.env.example index 40a3b942..ad027360 100644 --- a/backend/.env.example +++ b/backend/.env.example @@ -17,3 +17,5 @@ AUTH_REQUIRED=false AUTH_JWT_SECRET=dev-secret-change-me AUTH_ISSUER=https://auth.shnt.top AUTH_PORTAL_URL=http://localhost:5175 +# Optional: enabled by the NetFlow agent Compose installer. Linux Unix socket only. +# FLOW_AGENT_SOCKET=/run/mm-netflow-host/agent.sock diff --git a/backend/Dockerfile b/backend/Dockerfile index 9ed93083..95fb213e 100644 --- a/backend/Dockerfile +++ b/backend/Dockerfile @@ -57,6 +57,7 @@ COPY --from=build /app/packages/contracts ./packages/contracts COPY --from=build /app/backend/dist ./backend/dist COPY --from=build /app/backend/package.json ./backend/package.json COPY --from=build /app/backend/drizzle ./backend/drizzle +COPY deploy/netflow-agent/install.sh ./deploy/netflow-agent/install.sh # Nested install from lockfile (dotenv etc.) — ESM resolves from /app/backend/dist → ../node_modules COPY --from=build /app/backend/node_modules ./backend/node_modules EXPOSE 8000 diff --git a/backend/package.json b/backend/package.json index ce7928dc..4a10047b 100644 --- a/backend/package.json +++ b/backend/package.json @@ -33,7 +33,7 @@ "test": "node scripts/test.mjs", "test:evolution": "tsx src/services/evobgp-catalog.test.ts && tsx src/routes/evobgp-catalog.test.ts && tsx src/routes/evobgp-settings.test.ts && tsx src/plugins/auth.evolution-settings.test.ts", "test:netflow-purge": "tsx src/db/operation-lock.test.ts && tsx src/services/traffic-flow-purge-execution.test.ts && tsx src/routes/traffic-flow-purge.test.ts", - "test:netflow-settings": "tsx src/services/traffic-flow-host-file-builders.test.ts && tsx src/services/traffic-flow-key-execution.test.ts && tsx src/routes/traffic-flow-settings.test.ts && tsx src/plugins/auth.netflow-settings.test.ts", + "test:netflow-settings": "tsx src/services/traffic-flow-host-file-builders.test.ts && tsx src/services/traffic-flow-key-execution.test.ts && tsx src/routes/traffic-flow-settings.test.ts && tsx src/plugins/auth.netflow-settings.test.ts && tsx src/services/traffic-flow-agent-install.test.ts && tsx src/services/traffic-flow-agent-client.test.ts && tsx src/services/traffic-flow-operation-queue.test.ts && tsx src/routes/traffic-flow-host.test.ts", "test:geoip": "tsx src/services/geoip-download.test.ts && tsx src/services/traffic-flow-geoip.test.ts && tsx src/plugins/auth.geoip.test.ts", "test:ospf-optimizer": "tsx src/services/ospf-optimizer-identity.test.ts && tsx src/services/ospf-optimizer-metrics.test.ts && tsx src/services/ospf-optimizer-plan.test.ts && tsx src/services/ospf-effective-disabled.test.ts && tsx src/routes/ospf-optimizer.test.ts", "test:containers": "tsx src/services/containers-live.test.ts && tsx src/services/containers-state.test.ts && tsx src/services/container-removal.test.ts && tsx src/routes/containers-state.test.ts", diff --git a/backend/src/index.ts b/backend/src/index.ts index 8880a0ce..4921b63e 100644 --- a/backend/src/index.ts +++ b/backend/src/index.ts @@ -39,6 +39,8 @@ import usersRoutes from "./routes/users.js" import statisticsRoutes from "./routes/statistics.js" import { refreshScheduler, stopScheduler } from "./services/scheduler.js" import { getFlowWorkerHealth, startTrafficFlowListener, stopTrafficFlowListener } from "./services/traffic-flow-ingest.js" +import { agentSocket } from "./services/traffic-flow-agent-client.js" +import { reconcileFlowHost, withFlowConfigurationLock } from "./services/traffic-flow-host-agent.js" import { initGeoip } from "./services/traffic-flow-geoip.js" const eventLoopDelay = monitorEventLoopDelay({ resolution: 20 }) @@ -147,6 +149,10 @@ export async function buildApp(opts?: { await refreshScheduler() await initGeoip() await startTrafficFlowListener() + if (agentSocket()) { + try { await withFlowConfigurationLock(reconcileFlowHost) } + catch { app.log.warn("NetFlow: агент хоста не синхронизирован; проверьте состояние в настройках NetFlow") } + } app.addHook("onClose", async () => { stopScheduler() stopTrafficFlowListener() diff --git a/backend/src/routes/traffic-flow-host.test.ts b/backend/src/routes/traffic-flow-host.test.ts new file mode 100644 index 00000000..0cb86ef4 --- /dev/null +++ b/backend/src/routes/traffic-flow-host.test.ts @@ -0,0 +1,46 @@ +import assert from "node:assert/strict" +import Fastify from "fastify" +import { SignJWT } from "jose" +import { createTrafficFlowHostRoutes } from "./traffic-flow-host.js" +import { emptyHostStatus } from "../services/traffic-flow-agent-client.js" + +process.env.AUTH_REQUIRED = "true" +process.env.AUTH_JWT_SECRET = "flow-host-test-secret" +process.env.AUTH_ISSUER = "https://auth.test.local" +process.env.NODE_ENV = "test" +const { default: authPlugin, requireAuth } = await import("../plugins/auth.js") +const app = Fastify() +await app.register(authPlugin) +app.addHook("preHandler", requireAuth) +let applied = 0, fail = false +let status = emptyHostStatus(false) +await app.register(createTrafficFlowHostRoutes({ + status: async () => status, + reconcile: async () => { applied++; if (fail) throw Error("Туннель не изменён: конфликт адресов"); return status }, +}), { prefix: "/api" }) +async function token(permission: string) { + return new SignJWT({ sub: "fixture", apps: ["mm"], permissions: [permission] }).setProtectedHeader({ alg: "HS256" }) + .setIssuer("https://auth.test.local").setExpirationTime("1h").sign(new TextEncoder().encode("flow-host-test-secret")) +} +for (const permission of ["mm:dashboard:read", "mm:traffic:read", "mm:traffic:write", "mm:traffic:admin"]) { + const headers = { authorization: "Bearer " + await token(permission) } + assert.equal((await app.inject({ method: "GET", url: "/api/traffic/flow/host-status", headers })).statusCode, permission === "mm:dashboard:read" ? 403 : 200) + const before = applied + const response = await app.inject({ method: "POST", url: "/api/traffic/flow/host-reconcile", headers }) + const allowed = permission.endsWith(":write") || permission.endsWith(":admin") + assert.equal(response.statusCode, allowed ? 409 : 403) + assert.equal(applied, before + Number(allowed)) +} +assert.equal((await app.inject({ method: "GET", url: "/api/traffic/flow/host-status" })).statusCode, 401) +const headers = { authorization: "Bearer " + await token("mm:traffic:write") } +status = { ...emptyHostStatus(true), reachable: true, interfaceUp: true, configMatches: true } +let response = await app.inject({ method: "POST", url: "/api/traffic/flow/host-reconcile", headers }) +assert.equal(response.statusCode, 200); assert.equal(response.json().hostStatus.configMatches, true) +fail = true +response = await app.inject({ method: "POST", url: "/api/traffic/flow/host-reconcile", headers }) +assert.equal(response.statusCode, 502); assert.match(response.json().error, /конфликт адресов/) +status = emptyHostStatus(true, "offline") +response = await app.inject({ method: "GET", url: "/api/traffic/flow/host-status", headers }) +assert.equal(response.json().configMatches, null); assert.equal(response.json().reachable, false) +await app.close() +console.log("NetFlow host HTTP/JWT: read/write permissions, manual mode, reconcile success/conflict and offline status") diff --git a/backend/src/routes/traffic-flow-host.ts b/backend/src/routes/traffic-flow-host.ts new file mode 100644 index 00000000..11c4270a --- /dev/null +++ b/backend/src/routes/traffic-flow-host.ts @@ -0,0 +1,20 @@ +import type { FastifyPluginAsyncZod } from "@fastify/type-provider-zod" +import type { TrafficFlowHostStatus } from "@mmapp/contracts/traffic-flow" + +export function createTrafficFlowHostRoutes(deps: { + status: () => Promise + reconcile: () => Promise +}): FastifyPluginAsyncZod { + return async app => { + app.get("/traffic/flow/host-status", async () => deps.status()) + app.post("/traffic/flow/host-reconcile", async (_req, reply) => { + try { + const hostStatus = await deps.reconcile() + if (!hostStatus.configured) return reply.status(409).send({ error: "Автоматизация хоста NetFlow не настроена" }) + return { ok: true, hostStatus } + } catch (error) { + return reply.status(502).send({ error: error instanceof Error ? error.message : "Настройки хоста не применены" }) + } + }) + } +} diff --git a/backend/src/routes/traffic-flow.ts b/backend/src/routes/traffic-flow.ts index 9a409ce2..cbf6d44d 100644 --- a/backend/src/routes/traffic-flow.ts +++ b/backend/src/routes/traffic-flow.ts @@ -8,6 +8,9 @@ import { } from "@mmapp/contracts/traffic-flow" import { createTrafficFlowPurgeRoutes } from "./traffic-flow-purge.js" import { createTrafficFlowSettingsRoutes } from "./traffic-flow-settings.js" +import { createTrafficFlowHostRoutes } from "./traffic-flow-host.js" +import { getFlowHostStatus, reconcileFlowHost, withFlowConfigurationLock } from "../services/traffic-flow-host-agent.js" +import { agentSocket } from "../services/traffic-flow-agent-client.js" import { ensureHostKeys, getTrafficFlowSettingsRow, @@ -150,11 +153,21 @@ function sleep(ms: number, signal: AbortSignal): Promise { } const trafficFlowRoutes: FastifyPluginAsyncZod = async (app) => { + await app.register(createTrafficFlowHostRoutes({ + status: getFlowHostStatus, + reconcile: () => withFlowConfigurationLock(reconcileFlowHost), + })) await app.register(createTrafficFlowSettingsRoutes({ read: () => toTrafficFlowSettingsDto(getFlowListenerState()), - update: async patch => { await updateTrafficFlowSettings(patch) }, + update: async patch => { await withFlowConfigurationLock(async () => { + const current = await getTrafficFlowSettingsRow() + if (agentSocket() && (["collectorIp", "prefix", "flowListenPort", "wgListenPort"] as const).some(key => patch[key] !== undefined && patch[key] !== current[key])) { + throw Object.assign(new Error("Адрес, префикс и порты закреплены установкой агента; повторите установщик хоста"), { statusCode: 409 }) + } + await updateTrafficFlowSettings(patch) + }) }, startListener: startTrafficFlowListener, - ensureKeys: ensureHostKeys, + ensureKeys: () => withFlowConfigurationLock(ensureHostKeys), hasPrivateKey: async () => Boolean((await getTrafficFlowSettingsRow()).hostPrivateKey), hostFiles: listTrafficFlowHostFiles, })) diff --git a/backend/src/services/traffic-flow-agent-client.test.ts b/backend/src/services/traffic-flow-agent-client.test.ts new file mode 100644 index 00000000..79d16d50 --- /dev/null +++ b/backend/src/services/traffic-flow-agent-client.test.ts @@ -0,0 +1,42 @@ +import assert from "node:assert/strict" +import { createServer } from "node:http" +import { randomUUID } from "node:crypto" +import { tmpdir } from "node:os" +import { join } from "node:path" +import { callFlowAgent, emptyHostStatus, type FlowAgentSnapshot } from "./traffic-flow-agent-client.js" + +const socket = process.platform === "win32" ? `\\\\.\\pipe\\flow-agent-${randomUUID()}` : join(tmpdir(), `flow-agent-${randomUUID()}.sock`) +const snapshot: FlowAgentSnapshot = { + collectorIp: "10.255.254.1", prefix: "10.255.254.0/24", flowListenPort: 4739, wgListenPort: 51821, + hostPublicKey: "public-fixture", hostPrivateKey: "private-fixture", peers: [], +} +let mode = "ok" +const server = createServer((req, res) => { + let body = "" + req.on("data", chunk => { body += chunk }) + req.on("end", () => { + assert.equal(req.url, "/check"); assert.equal(req.method, "POST") + assert.deepEqual(JSON.parse(body), snapshot) + if (mode === "conflict") { res.writeHead(409); res.end(JSON.stringify({ error: "Конфликт адресов" })); return } + if (mode === "invalid") { res.end("not-json"); return } + if (mode === "oversize") { res.end("x".repeat(300_000)); return } + res.end(JSON.stringify({ + interfaceUp: true, configMatches: true, publicKey: "public-fixture", listenPort: 51821, + addresses: ["10.255.254.1/24"], peers: [{ publicKey: "router-fixture", allowedIps: ["10.255.254.2/32"], lastHandshakeAt: null, receivedBytes: 0, sentBytes: 0 }], + privateKey: "must-not-be-exposed", + })) + }) +}) +await new Promise((resolve, reject) => { server.once("error", reject); server.listen(socket, resolve) }) +try { + const status = await callFlowAgent("/check", snapshot, socket) + assert.equal(status.reachable, true); assert.equal(status.configMatches, true) + assert.equal(JSON.stringify(status).includes("must-not-be-exposed"), false) + assert.equal(status.peers[0].lastHandshakeAt, null) + mode = "conflict"; await assert.rejects(callFlowAgent("/check", snapshot, socket), /Конфликт адресов/) + mode = "invalid"; await assert.rejects(callFlowAgent("/check", snapshot, socket), /некорректный ответ/) + mode = "oversize"; await assert.rejects(callFlowAgent("/check", snapshot, socket), /некорректный ответ/) + await assert.rejects(callFlowAgent("/status", undefined, ""), /не настроена/) + assert.equal(emptyHostStatus(true, "offline").configMatches, null) +} finally { await new Promise(resolve => server.close(() => resolve())) } +console.log("NetFlow agent IPC: request, validated status, secret stripping, conflict, invalid/oversize response, manual mode") diff --git a/backend/src/services/traffic-flow-agent-client.ts b/backend/src/services/traffic-flow-agent-client.ts new file mode 100644 index 00000000..d187ecbf --- /dev/null +++ b/backend/src/services/traffic-flow-agent-client.ts @@ -0,0 +1,45 @@ +import { request } from "node:http" +import { trafficFlowHostStatusSchema, type TrafficFlowHostStatus } from "@mmapp/contracts/traffic-flow" + +export type FlowAgentSnapshot = { + collectorIp: string; prefix: string; flowListenPort: number; wgListenPort: number + hostPublicKey: string; hostPrivateKey: string + peers: { serverId: number; publicKey: string; allowedIps: string[]; endpoint?: string }[] +} + +export const agentSocket = () => process.env.FLOW_AGENT_SOCKET?.trim() || "" +export const emptyHostStatus = (configured: boolean, error: string | null = null): TrafficFlowHostStatus => ({ + configured, reachable: false, interfaceUp: false, configMatches: null, + publicKey: "", listenPort: 0, addresses: [], peers: [], error, +}) + +export function callFlowAgent(path: "/status" | "/check" | "/apply", snapshot?: FlowAgentSnapshot, socketPath = agentSocket()): Promise { + if (!socketPath) return Promise.reject(new Error("Автоматизация хоста NetFlow не настроена")) + const body = snapshot ? JSON.stringify(snapshot) : undefined + return new Promise((resolve, reject) => { + const fail = () => reject(new Error("Агент NetFlow недоступен или вернул некорректный ответ; проверьте контейнер mmapp-netflow-agent")) + const req = request({ socketPath, path, method: body ? "POST" : "GET", headers: body ? { "content-type": "application/json", "content-length": Buffer.byteLength(body) } : {} }, res => { + let text = "" + res.setEncoding("utf8") + res.on("data", (chunk: string) => { text += chunk; if (text.length > 262144) req.destroy() }) + res.on("error", fail) + res.on("end", () => { + try { + const raw: unknown = JSON.parse(text) + if (res.statusCode !== 200) { + // The local controller deliberately emits sanitized errors only. + const error = raw as { error?: unknown } + reject(new Error(typeof error.error === "string" ? error.error.slice(0, 500) : "Агент NetFlow отклонил настройки")) + return + } + resolve(trafficFlowHostStatusSchema.parse({ ...(raw as object), configured: true, reachable: true, error: null })) + } catch { fail() } + }) + }) + // An absolute deadline also covers a peer that keeps sending small chunks. + const timer = setTimeout(() => req.destroy(), path === "/apply" ? 90_000 : 30_000) + req.on("close", () => clearTimeout(timer)) + req.on("error", fail) + req.end(body) + }) +} diff --git a/backend/src/services/traffic-flow-agent-install.test.ts b/backend/src/services/traffic-flow-agent-install.test.ts new file mode 100644 index 00000000..0f278e01 --- /dev/null +++ b/backend/src/services/traffic-flow-agent-install.test.ts @@ -0,0 +1,18 @@ +import assert from "node:assert/strict" +import { readFile } from "node:fs/promises" +import { buildFlowAgentInstaller } from "./traffic-flow-agent-install.js" +import { buildTrafficFlowHostFiles } from "./traffic-flow-host-file-builders.js" +const files = buildTrafficFlowHostFiles({ collectorIp: "10.255.254.1", prefix: "10.255.254.0/24", flowListenPort: 4740, + wgListenPort: 51821, hostPrivateKey: Buffer.alloc(32, 1).toString("base64"), peers: [] }) +const conf = files.find(file => file.id === "wg-quick")!.code +const source = await readFile(new URL("../../../deploy/netflow-agent/install.sh", import.meta.url), "utf8") +const script = buildFlowAgentInstaller(conf, 4740, source) +assert.ok(script.startsWith("#!/usr/bin/env bash\n")) +assert.ok(script.includes(conf.trim() + "\nMM_CONFIG_EOF\n")) +assert.ok(script.includes('"${1:-/opt/cdn-mm}" 4740')) +assert.ok(script.includes("wg-quick@wg-flow.service")) +assert.ok(script.includes("docker-compose.netflow-agent.yml")) +assert.ok(!script.includes("\r")) +assert.throws(() => buildFlowAgentInstaller("MM_CONFIG_EOF\n", 4739, source)) +assert.throws(() => buildFlowAgentInstaller(conf, 0, source)) +console.log("NetFlow agent installer export: validated snapshot, custom port/path, embedded installer and LF-only output") diff --git a/backend/src/services/traffic-flow-agent-install.ts b/backend/src/services/traffic-flow-agent-install.ts new file mode 100644 index 00000000..0ee06645 --- /dev/null +++ b/backend/src/services/traffic-flow-agent-install.ts @@ -0,0 +1,8 @@ +/** Call only with a configuration already validated by buildTrafficFlowHostFiles. */ +export function buildFlowAgentInstaller(conf: string, flowPort: number, installer: string): string { + if (!Number.isInteger(flowPort) || flowPort < 1 || flowPort > 65535 || conf.split(/\r?\n/).includes("MM_CONFIG_EOF")) { + throw new Error("Некорректная конфигурация установщика NetFlow") + } + const clean = (text: string) => text.replaceAll("\r\n", "\n").replace(/\s*$/, "") + "\n" + return "#!/usr/bin/env bash\nset -euo pipefail\numask 077\nEXPORT_DIR=$(mktemp -d)\ntrap 'rm -rf -- \"$EXPORT_DIR\"' EXIT\ncat > \"$EXPORT_DIR/wg-flow.conf\" <<'MM_CONFIG_EOF'\n" + clean(conf) + "MM_CONFIG_EOF\ncat > \"$EXPORT_DIR/install.sh\" <<'MM_INSTALL_EOF'\n" + clean(installer) + "MM_INSTALL_EOF\nbash \"$EXPORT_DIR/install.sh\" \"$EXPORT_DIR/wg-flow.conf\" \"${1:-/opt/cdn-mm}\" " + flowPort + "\n" +} diff --git a/backend/src/services/traffic-flow-host-agent.ts b/backend/src/services/traffic-flow-host-agent.ts new file mode 100644 index 00000000..4770bd86 --- /dev/null +++ b/backend/src/services/traffic-flow-host-agent.ts @@ -0,0 +1,46 @@ +import { withAdvisoryLock } from "../db/index.js" +import { parseJsonArray } from "../db/json.js" +import { getTrafficFlowSettingsRow, invalidateTrafficFlowSettingsCache } from "./traffic-flow-settings.js" +import { agentSocket, callFlowAgent, emptyHostStatus, type FlowAgentSnapshot } from "./traffic-flow-agent-client.js" +import { FlowConfigurationQueue } from "./traffic-flow-operation-queue.js" + +// Session lock: RouterOS/network operations use other pooled connections. Never lock the settings row here. +const queue = new FlowConfigurationQueue() +export const withFlowConfigurationLock = (fn: () => Promise) => queue.run(() => withAdvisoryLock(7340291, async () => { + invalidateTrafficFlowSettingsCache() + return fn() +})) + +async function snapshot(): Promise { + const row = await getTrafficFlowSettingsRow() + return { + collectorIp: row.collectorIp, prefix: row.prefix, flowListenPort: row.flowListenPort, wgListenPort: row.wgListenPort, + hostPublicKey: row.hostPublicKey, hostPrivateKey: row.hostPrivateKey, + peers: (parseJsonArray(row.peersJson) as FlowAgentSnapshot["peers"]).map(p => ({ + serverId: p.serverId, publicKey: p.publicKey, allowedIps: p.allowedIps, ...(p.endpoint ? { endpoint: p.endpoint } : {}), + })), + } +} + +export async function getFlowHostStatus() { + if (!agentSocket()) return emptyHostStatus(false) + try { return await callFlowAgent("/check", await snapshot()) } + catch (error) { + const message = error instanceof Error ? error.message : "Проверка агента не выполнена" + // Preserve live measurements on a settings conflict; never claim configuration agreement. + try { return { ...await callFlowAgent("/status"), configMatches: false, error: message } } + catch { return emptyHostStatus(true, message) } + } +} + +export async function preflightFlowHost() { + if (!agentSocket()) return + try { await callFlowAgent("/check", await snapshot()) } + catch (error) { throw Object.assign(new Error(error instanceof Error ? error.message : "Агент NetFlow не готов"), { statusCode: 409 }) } +} + +// Caller holds withFlowConfigurationLock. This is also used after saving a newly discovered router key. +export async function reconcileFlowHost() { + if (!agentSocket()) return emptyHostStatus(false) + return callFlowAgent("/apply", await snapshot()) +} diff --git a/backend/src/services/traffic-flow-host-files.ts b/backend/src/services/traffic-flow-host-files.ts index 9a77564b..3a0909d9 100644 --- a/backend/src/services/traffic-flow-host-files.ts +++ b/backend/src/services/traffic-flow-host-files.ts @@ -1,8 +1,15 @@ import { getTrafficFlowSettingsRow } from "./traffic-flow-settings.js" import { parseJsonArray } from "../db/json.js" import { buildTrafficFlowHostFiles } from "./traffic-flow-host-file-builders.js" +import { readFile } from "node:fs/promises" +import { buildFlowAgentInstaller } from "./traffic-flow-agent-install.js" export async function listTrafficFlowHostFiles() { const row = await getTrafficFlowSettingsRow() - return buildTrafficFlowHostFiles({ ...row, peers: parseJsonArray(row.peersJson) }) + const files = buildTrafficFlowHostFiles({ ...row, peers: parseJsonArray(row.peersJson) }) + // The same source is used by a repository install and by the downloadable self-contained installer. + const installer = await readFile(new URL("../../../deploy/netflow-agent/install.sh", import.meta.url), "utf8") + const conf = files.find(file => file.id === "wg-quick")!.code + const code = buildFlowAgentInstaller(conf, row.flowListenPort, installer) + return [{ id: "agent", label: "Агент Docker", filename: "install-netflow-agent.sh", code }, ...files] } diff --git a/backend/src/services/traffic-flow-operation-queue.test.ts b/backend/src/services/traffic-flow-operation-queue.test.ts new file mode 100644 index 00000000..fc751cab --- /dev/null +++ b/backend/src/services/traffic-flow-operation-queue.test.ts @@ -0,0 +1,21 @@ +import assert from "node:assert/strict" +import { FlowConfigurationQueue } from "./traffic-flow-operation-queue.js" +const queue = new FlowConfigurationQueue() +let release!: () => void, active = 0, maximum = 0 +const hold = new Promise(resolve => { release = resolve }) +const events: number[] = [] +const tasks = Array.from({ length: 32 }, (_, i) => queue.run(async () => { + active++; maximum = Math.max(active, maximum); events.push(i) + try { if (i === 0) await hold; if (i === 1) throw Error("fixture failure"); return i } + finally { active-- } +})) +const done = Promise.allSettled(tasks) +await new Promise(resolve => setImmediate(resolve)) +assert.deepEqual(events, [0]); assert.equal(active, 1) +release() +const results = await done +assert.equal(maximum, 1); assert.equal(results[1].status, "rejected") +assert.equal(results.filter(result => result.status === "fulfilled").length, 31) +assert.deepEqual(events, Array.from({ length: 32 }, (_, i) => i)) +assert.equal(await queue.run(async () => "recovered"), "recovered") +console.log("NetFlow queue: 32 concurrent operations use one lock waiter, failures release the queue") diff --git a/backend/src/services/traffic-flow-operation-queue.ts b/backend/src/services/traffic-flow-operation-queue.ts new file mode 100644 index 00000000..b4ee7e12 --- /dev/null +++ b/backend/src/services/traffic-flow-operation-queue.ts @@ -0,0 +1,12 @@ +/** Only one local operation may wait for the PostgreSQL advisory lock. + * Otherwise queued requests could exhaust the pool needed by its current owner. + */ +export class FlowConfigurationQueue { + private tail: Promise = Promise.resolve() + + run(operation: () => Promise): Promise { + const result = this.tail.then(operation) + this.tail = result.then(() => undefined, () => undefined) + return result + } +} diff --git a/backend/src/services/traffic-flow-overlay.ts b/backend/src/services/traffic-flow-overlay.ts index 8b01155f..a6df6afa 100644 --- a/backend/src/services/traffic-flow-overlay.ts +++ b/backend/src/services/traffic-flow-overlay.ts @@ -22,6 +22,8 @@ import { import { refreshFlowExporterMap, startTrafficFlowListener } from "./traffic-flow-ingest.js" import { listTrafficFlowHostFiles } from "./traffic-flow-host-files.js" import { invalidateFlowCatalogCache } from "./traffic-flow-topology.js" +import { agentSocket } from "./traffic-flow-agent-client.js" +import { getFlowHostStatus, preflightFlowHost, reconcileFlowHost, withFlowConfigurationLock } from "./traffic-flow-host-agent.js" const IFACE_NAME = "wg-flow" const JH_LISTEN_PORT = 13232 @@ -202,8 +204,16 @@ export function usablePublicHost(raw: string | undefined): string { export async function applyFlowOverlay( serverIdRaw: string | number, opts?: { publicEndpoint?: string; requestHost?: string; disableGreFastPath?: boolean }, +): Promise { + return withFlowConfigurationLock(() => applyFlowOverlayLocked(serverIdRaw, opts)) +} + +async function applyFlowOverlayLocked( + serverIdRaw: string | number, + opts?: { publicEndpoint?: string; requestHost?: string; disableGreFastPath?: boolean }, ): Promise { const steps: string[] = [] + await preflightFlowHost() const keys = await ensureHostKeys() const settings = await getTrafficFlowSettingsRow() const hostPublicKey = settings.hostPublicKey || keys.publicKey @@ -338,8 +348,21 @@ export async function applyFlowOverlay( await refreshFlowExporterMap() steps.push("Коллектор IPFIX на MM включён") + let hostStatus = undefined + let hostError: string | undefined + if (agentSocket()) { + try { + hostStatus = await reconcileFlowHost() + steps.push("Пир применён на Linux-хосте без перезапуска WireGuard") + } catch (error) { + hostError = error instanceof Error ? error.message : "Настройки хоста не применены" + hostStatus = await getFlowHostStatus() + steps.push("MikroTik настроен; требуется повторное применение настроек хоста") + } + } + return { - ok: true, + ok: !hostError, serverId: server.id, interfaceName: IFACE_NAME, address, @@ -347,11 +370,14 @@ export async function applyFlowOverlay( linuxPeerBlock: linuxPeerBlock(publicKey, address, server.name || server.host, endpointHost), trafficFlow: true, steps, - hostFiles: await listTrafficFlowHostFiles(), + hostFiles: agentSocket() ? [] : await listTrafficFlowHostFiles(), + stage: hostError ? "host-pending" : agentSocket() ? "applied" : "manual", + ...(hostStatus ? { hostStatus } : {}), + ...(hostError ? { hostError } : {}), } } 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 }) + const err = Object.assign(new Error(`Настройка MikroTik могла быть выполнена частично: ${msg}. Проверьте состояние и повторите подключение.`), { statusCode: 502 }) throw err } } diff --git a/components/traffic/flow-host-panel.tsx b/components/traffic/flow-host-panel.tsx new file mode 100644 index 00000000..6e54ba14 --- /dev/null +++ b/components/traffic/flow-host-panel.tsx @@ -0,0 +1,63 @@ +"use client" + +import { useCallback, useEffect, useRef, useState } from "react" +import { can } from "@/lib/auth" +import { Button } from "@/components/ui/button" +import { Alert, AlertTitle, AlertDescription } from "@/components/reui/alert" +import { AlertCircleIcon } from "lucide-react" +import { getTrafficFlowHostStatus, reconcileTrafficFlowHost } from "@/shared/api/traffic-flow" +import type { TrafficFlowHostStatus } from "@mmapp/contracts/traffic-flow" + +export function FlowHostPanel({ backendUrl, enabled = true, onConfigured }: { + backendUrl: string; enabled?: boolean; onConfigured?: (configured: boolean) => void +}) { + const [status, setStatus] = useState(null) + const [receivedAt, setReceivedAt] = useState(null) + const [error, setError] = useState(null) + const [operationError, setOperationError] = useState(null) + const [busy, setBusy] = useState(false) + const operation = useRef(null) + const [revision, setRevision] = useState(0) + const accept = useCallback((value: TrafficFlowHostStatus) => { + setStatus(value); setReceivedAt(Date.now()); setError(null); onConfigured?.(value.configured) + }, [onConfigured]) + useEffect(() => () => { operation.current?.abort() }, []) + useEffect(() => { + if (!enabled || !can("mm:traffic:read") || busy) return + const controller = new AbortController() + let timer: ReturnType | undefined + async function refresh() { + try { const value = await getTrafficFlowHostStatus(backendUrl, { signal: controller.signal }); if (!controller.signal.aborted) accept(value) } + catch (reason) { if (!controller.signal.aborted) setError(reason instanceof Error ? reason.message : "Статус хоста не получен") } + finally { if (!controller.signal.aborted) timer = setTimeout(() => { void refresh() }, 10_000) } + } + void refresh() + return () => { controller.abort(); if (timer) clearTimeout(timer) } + }, [accept, backendUrl, enabled, busy, revision]) + + async function apply() { + if (operation.current) return + const controller = new AbortController(); operation.current = controller + setBusy(true); setOperationError(null) + try { const value = await reconcileTrafficFlowHost(backendUrl, { signal: controller.signal }); if (!controller.signal.aborted) accept(value) } + catch (reason) { if (!controller.signal.aborted) setOperationError(reason instanceof Error ? reason.message : "Применение не подтверждено; обновите состояние хоста") } + finally { operation.current = null; if (!controller.signal.aborted) setBusy(false) } + } + if (!enabled || !can("mm:traffic:read")) return null + return
+

WireGuard на сервере MM

+

{!status ? "Получение состояния агента…" : !status.configured ? "Ручной режим. Установите агент из раздела «Файлы хоста»." : !status.reachable ? "Агент недоступен" : status.configMatches ? "Настройки применены" : "Настройки требуют проверки или применения"}

+ {error && receivedAt ?

Показано последнее состояние: {new Date(receivedAt).toLocaleString("ru-RU")}

: null} + {status?.configured && status.reachable ? <> +

Интерфейс: {status.interfaceUp ? status.addresses.join(", ") : "не поднят"} · UDP {status.listenPort}

+
    + {status.peers.map(peer =>
  • {peer.allowedIps.join(", ") || "Без адреса"} · {peer.lastHandshakeAt ? "Handshake: " + new Date(peer.lastHandshakeAt * 1000).toLocaleString("ru-RU") + ((receivedAt ?? 0) / 1000 - peer.lastHandshakeAt > 180 ? " (давно)" : "") : "Ожидание handshake"}
    • )} +
+ : null} + {operationError || error || status?.error ? Состояние хоста требует внимания{operationError || error || status?.error} : null} +
+ + {status?.configured && can("mm:traffic:write") ? : null} +
+
+} diff --git a/components/traffic/flow-overlay-sheet.tsx b/components/traffic/flow-overlay-sheet.tsx index 9d35bdd0..3fe8f5e5 100644 --- a/components/traffic/flow-overlay-sheet.tsx +++ b/components/traffic/flow-overlay-sheet.tsx @@ -22,6 +22,7 @@ import { Sheet, SheetContent, SheetHeader, SheetTitle, SheetDescription, SheetFooter, SheetClose, } from "@/components/ui/sheet" +import { FlowHostPanel } from "./flow-host-panel" import { applyTrafficFlowOverlay } from "@/shared/api/traffic-flow" import type { ServerRead } from "@mmapp/contracts/servers" import type { TrafficFlowOverlayResult } from "@mmapp/contracts/traffic-flow" @@ -95,7 +96,8 @@ function FlowOverlaySheet({ }) setResult(res) setTab(res.hostFiles[0]?.id ?? "linux") - toast.success(`wg-flow на ${res.address}`) + if (res.ok) toast.success(`wg-flow на ${res.address}`) + else toast.warning(res.hostError || "MikroTik настроен; требуется применение на хосте") onDone?.(res) } catch (e) { toast.error(e instanceof Error ? e.message : "Не удалось подключить JH") @@ -125,7 +127,7 @@ function FlowOverlaySheet({ Подключить jump-host - Создаст wg-flow на MikroTik (сервер, listen 13232) и выдаст готовый bash для Linux-хоста Docker MM (клиент). + Настроит wg-flow и IPFIX на MikroTik. Если агент Docker установлен, автоматически применит пир на Linux-хосте MM.
@@ -170,13 +172,14 @@ function FlowOverlaySheet({ {result ? (
- + - Linux-хост /opt/cdn-mm + {result.stage === "applied" ? "Настройки применены" : result.stage === "host-pending" ? "MikroTik настроен, хост ожидает применения" : "Установка агента на Linux-хосте"} - Скопируйте вкладку Linux и выполните от root. WG — клиент к JH:13232; контейнер слушает только WG-IP:4739, не 0.0.0.0. Ключ не кладите в git. + {result.hostError || (result.stage === "applied" ? "Пир добавлен без перезапуска WireGuard. Ожидайте handshake и поступления IPFIX в панели NetFlow." : "Выполните установщик из вкладки «Агент Docker» от root на сервере MM. Он однократно пересоздаст backend. Экспорт содержит приватный ключ — храните его как секрет.")} +
    {result.steps.map((s) => (
  • diff --git a/components/traffic/netflow-settings-panel.tsx b/components/traffic/netflow-settings-panel.tsx index 0ef13bff..1093fafd 100644 --- a/components/traffic/netflow-settings-panel.tsx +++ b/components/traffic/netflow-settings-panel.tsx @@ -13,6 +13,7 @@ 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 { GeoipSettingsSection } from "./geoip-settings-section" +import { FlowHostPanel } from "./flow-host-panel" import { generateTrafficFlowKeys, getTrafficFlowHostFiles, @@ -24,11 +25,11 @@ import { formatFlowPurgeResult, NetflowPurgeConfirm } from "@/components/traffic import { KeyRoundIcon, DownloadIcon, InfoIcon, AlertCircleIcon } 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).", + "Создайте ключи хоста, откройте «Файлы хоста» и скачайте вкладку «Агент Docker».", + "Выполните sudo bash install-netflow-agent.sh /opt/cdn-mm на хосте Docker. При установке backend будет пересоздан.", + "Обновите статус хоста. Дальнейшие подключения автоматически применяются агентом.", "Firewall: разрешите UDP listen WireGuard. UDP 4739 наружу не открывайте.", - "В docker-compose у backend раскомментируйте bind IPFIX только на адресе wg-flow.", + "Установщик публикует IPFIX только на адресе wg-flow. Ручные файлы доступны в остальных вкладках экспорта.", "Проверка: wg show · ss -ulnp | grep 4739 · в этой панели — last datagram.", ] @@ -43,6 +44,7 @@ function NetflowSettingsContent({ enabled: boolean }) { const [settings,setSettings] = useState(null) + const [automatic,setAutomatic] = useState(false) const [draft,setDraft] = useState(emptyNetflowDraft) const dirty = useRef(new Set()) const [dirtyCount,setDirtyCount] = useState(0) @@ -147,16 +149,16 @@ function NetflowSettingsContent({
    - change("collectorIp", e.target.value)} /> + change("collectorIp", e.target.value)} /> - change("prefix", e.target.value)} /> + change("prefix", e.target.value)} /> - change("flowListenPort", e.target.value)} /> + change("flowListenPort", e.target.value)} /> - change("wgListenPort", e.target.value)} /> + change("wgListenPort", e.target.value)} /> change("publicEndpoint", e.target.value)} placeholder="203.0.113.10" /> @@ -201,14 +203,15 @@ function NetflowSettingsContent({ {settings?.lastError ? ` · ${settings.lastError}` : ""}

    -
    + + {automatic ?

    Адрес, префикс и порты закреплены установкой агента. Для их изменения повторите установщик.

    :

    Туннель на сервере Docker MM

      {HOST_STEPS.map((s) => (
    1. {s.replaceAll("4739", settings ? String(settings.flowListenPort) : "IPFIX")}
      2. ))}
    -
    +
    }