fix(wireguard): безопасно повторить чтение зависимостей
This commit is contained in:
1 parent
3d5b7f0ea6
commit
78b268cc39
4 files changed
+132
-5
No files matched your search
@@ -35,7 +35,7 @@
|
||||
"test:exec": "tsx src/services/mikrotik-logs.test.ts && tsx src/services/terminal-diagnostics.test.ts && tsx src/routes/exec.test.ts && tsx src/plugins/auth.exec.test.ts",
|
||||
"test:auth": "tsx --test src/modules/mcp/routing-manual-inverse-history.test.ts && tsx src/db/application-writer-profile-registry-structure.test.ts && tsx src/db/application-writer-profile-registry-read.test.ts && tsx src/db/application-writer-profile-registry-read-parity.test.ts && tsx src/db/application-writer-profile-format.test.ts && tsx src/plugins/auth-path.test.ts && tsx src/lib/permissions.test.ts && tsx src/plugins/auth-provenance.test.ts && tsx src/services/bandwidth-rest-speed-original.test.ts && tsx src/plugins/auth.smoke.test.ts && tsx src/plugins/auth.ospf-optimizer.test.ts && npm run test:mcp",
|
||||
"test:mcp": "tsx --test src/services/server-poll-error-provenance.test.ts && tsx --test src/modules/mcp/plan-expiry.test.ts src/modules/mcp/result-budget.test.ts src/modules/mcp/result-delivery.test.ts src/modules/mcp/wire-admission.test.ts src/modules/mcp/flow-envelope-allowance.test.ts src/modules/mcp/sdk-terminal-budget.test.ts src/modules/mcp/inflight-owner.test.ts && npm run test:database-source-binding && tsx --test src/modules/mcp/integration-fixture-expiry.test.ts && npm run test:mcp:operation-origin && npm run test:mcp:service-context && npm run test:mcp:operation-locks && tsx --test src/modules/mcp/routing-community-public.test.ts src/modules/mcp/management-transport.test.ts src/modules/mcp/routing-community-context-model.test.ts src/modules/mcp/routing-community-context-native.test.ts ../packages/contracts/src/mcp-routing.test.ts && tsx src/modules/mcp/human-artifact-receipts.test.ts && tsx src/modules/mcp/chr-overlay-readiness.test.ts && tsx src/modules/mcp/chr-recovery-source.test.ts && tsx src/modules/mcp/chr-native-failure.test.ts && tsx src/modules/mcp/ospf-owner-graph.test.ts && tsx src/modules/mcp/routing-measurement-address.test.ts && tsx src/modules/mcp/routing-observation-source.test.ts && tsx src/services/traffic-flow-talkers-model.test.ts && tsx src/services/bgp-prefix-model.test.ts && tsx src/services/vrf-members.test.ts && tsx src/services/routing-gateway-reference.test.ts && tsx src/services/bgp-export-model.test.ts && tsx src/services/traffic-history-model.test.ts && tsx src/services/internet-path-snapshot.test.ts && tsx src/services/internet-path-observation.test.ts && tsx src/modules/mcp/coverage.test.ts && tsx src/modules/mcp/policy.test.ts && tsx src/modules/mcp/events-batch-model.test.ts && tsx src/modules/mcp/operation-revisions.test.ts && tsx src/modules/mcp/native-create-identity.test.ts && tsx src/modules/mcp/ros-step-verification.test.ts && tsx src/modules/mcp/ros-prefix.test.ts && tsx src/modules/mcp/native-conservation-http.test.ts && tsx src/modules/mcp/security.test.ts && tsx src/modules/mcp/state-file-publish.test.ts && tsx src/modules/mcp/routing-model.test.ts && tsx src/modules/mcp/routing-observations-model.test.ts && tsx src/modules/mcp/ospf-health.test.ts && tsx src/modules/mcp/ecmp-preflight.test.ts && tsx src/modules/mcp/recursive-recovery.test.ts && tsx src/modules/mcp/control-path-model.test.ts && tsx src/modules/mcp/routing-community-scoped-model.test.ts && tsx src/modules/mcp/routing-community-scoped-native.test.ts && tsx src/modules/mcp/path-verification.test.ts && tsx src/modules/mcp/database-maintenance.test.ts && tsx src/routes/mcp.test.ts && tsx ../lib/auth-mcp.test.ts && tsx --test src/modules/mcp/routing-controller-fence.test.ts && tsx src/test/chr-vrf-companion-proof.test.ts && tsx src/test/chr-vrf-companion-telemetry.test.ts && tsx src/test/chr-vrf-companion-address-space.test.ts && tsx src/modules/mcp/flow-facts-rebuild-actions.test.ts && tsx src/modules/mcp/ipsec-peer-create-dependencies.test.ts && tsx src/services/ipsec-certificate-update-model.test.ts && tsx src/modules/mcp/ipsec-native-dispatch.test.ts && tsx src/modules/mcp/ipsec-user-update-actions.test.ts && tsx src/modules/mcp/ipsec-server-delete-actions.test.ts && tsx src/services/ipsec-proof-telemetry.test.ts && tsx src/services/vrf-production-admission.test.ts && tsx src/modules/mcp/flow-host-keys-actions.test.ts && tsx src/services/vrf-production-read.test.ts && tsx src/modules/mcp/ipsec-user-create-actions.test.ts && tsx src/modules/mcp/flow-settings-actions.test.ts && tsx src/services/vrf-production-ticket-store.test.ts && npm run test:mcp:routing-baseline",
|
||||
"test:wireguard": "tsx src/services/mikrotik-response-bounds.test.ts && tsx src/services/wireguard-interface-update.test.ts && tsx src/services/wireguard-peer-update.test.ts && npx tsx src/services/wireguard-config.test.ts && npx tsx src/services/wireguard-ipv6.test.ts && npx tsx --test src/services/wireguard-import.test.ts src/services/wireguard-import-apply.test.ts src/services/wireguard-import-revision.test.ts src/services/wireguard-delete.test.ts && npx tsx src/routes/wireguard-import.test.ts && tsx src/services/wireguard-interface-create.test.ts && tsx src/services/wireguard-peer-create.test.ts && tsx src/services/wireguard-import-plan.test.ts && tsx src/services/wireguard-import-execution.test.ts && tsx src/modules/mcp/wireguard-import-actions.test.ts && tsx --test src/services/wireguard-export.test.ts",
|
||||
"test:wireguard": "tsx src/services/mikrotik-response-bounds.test.ts && tsx --test src/services/wireguard-dependency-read.test.ts && tsx src/services/wireguard-interface-update.test.ts && tsx src/services/wireguard-peer-update.test.ts && npx tsx src/services/wireguard-config.test.ts && npx tsx src/services/wireguard-ipv6.test.ts && npx tsx --test src/services/wireguard-import.test.ts src/services/wireguard-import-apply.test.ts src/services/wireguard-import-revision.test.ts src/services/wireguard-delete.test.ts && npx tsx src/routes/wireguard-import.test.ts && tsx src/services/wireguard-interface-create.test.ts && tsx src/services/wireguard-peer-create.test.ts && tsx src/services/wireguard-import-plan.test.ts && tsx src/services/wireguard-import-execution.test.ts && tsx src/modules/mcp/wireguard-import-actions.test.ts && tsx --test src/services/wireguard-export.test.ts",
|
||||
"test:traffic-rate": "tsx src/services/traffic-rate.test.ts && tsx src/services/traffic-observation.test.ts && tsx src/services/traffic-freshness.test.ts && tsx src/services/traffic-live.test.ts && tsx src/services/traffic-observation-sql.test.ts && tsx src/services/traffic-history-sql.test.ts && tsx src/services/traffic-live-sql.test.ts && tsx src/services/traffic-volume-sql.test.ts && tsx src/services/traffic-binding-status.test.ts && tsx src/services/traffic-status-sql.test.ts && tsx src/services/traffic-gap.test.ts && tsx src/services/traffic-identity.test.ts && tsx src/services/traffic-membership.test.ts && tsx src/services/traffic-aggregate-peak.test.ts && tsx src/services/traffic-users-cache.test.ts && tsx src/services/traffic-collection-lock.test.ts && tsx src/services/traffic-concurrency-sql.test.ts",
|
||||
"test:traffic-flow": "tsx --test src/services/traffic-flow-heartbeat.test.ts src/services/traffic-flow-heartbeat-wiring.test.ts src/services/traffic-flow-heartbeat-receiver.test.ts src/services/traffic-flow-ingest-quality.test.ts src/services/traffic-flow-quality-contract.test.ts && tsx src/services/traffic-flow-catalog-read.test.ts && tsx src/routes/traffic-flow-query.test.ts && tsx src/services/traffic-flow-month.test.ts && tsx src/services/traffic-flow-parse.test.ts && tsx src/services/traffic-flow-ipv6-pipeline.test.ts && tsx src/services/traffic-flow-map-exporter.test.ts && tsx src/services/traffic-flow-ifaces.test.ts && tsx src/services/traffic-flow-ifindex.test.ts && tsx src/services/traffic-flow-dedup.test.ts && tsx src/services/traffic-flow-planes.test.ts && tsx src/services/traffic-flow-ip.test.ts && tsx src/services/traffic-flow-dest.test.ts && tsx src/services/traffic-flow-classify.test.ts && tsx src/services/traffic-flow-ripe.test.ts && tsx src/services/traffic-flow-brands.test.ts && tsx src/services/traffic-flow-ingest.test.ts && tsx src/services/traffic-flow-analytics.test.ts && tsx src/services/traffic-flow-map-hops.test.ts && tsx src/services/traffic-flow-purge.test.ts && tsx src/services/geoip-download.test.ts && tsx src/services/traffic-flow-geoip.test.ts && tsx src/services/traffic-flow-fact-accumulator.test.ts && tsx src/services/traffic-flow-facts.test.ts && tsx src/services/traffic-flow-facts-filter.test.ts && tsx src/services/traffic-flow-facts-rebuild.test.ts && tsx src/services/statistics-aggregate.test.ts && tsx src/services/geoip-manual-update.test.ts && tsx src/services/traffic-flow-overlay-model.test.ts && npm run test:traffic-flow:ownership && tsx src/services/traffic-flow-worker-platform.test.ts && tsx src/services/traffic-flow-template-cache.test.ts && tsx --test src/services/traffic-flow-template-lifetime.test.ts && tsx src/services/traffic-flow-protocol-scope.test.ts && npm run test:traffic-flow:read-budget && npm run test:traffic-flow:resource-profile",
|
||||
"test:users": "tsx ../packages/contracts/src/users.test.ts && tsx src/modules/users/iface-type.test.ts && tsx src/modules/users/bindings.test.ts && tsx src/modules/users/interface-inventory.test.ts && tsx src/routes/users-catalog.test.ts",
|
||||
|
||||
@@ -58,6 +58,13 @@ export interface MikrotikConnectParams {
|
||||
|
||||
// ── low-level HTTP helpers ────────────────────────────────────────────────────
|
||||
|
||||
const preHeaderGetResets = new WeakSet<Error>()
|
||||
|
||||
/** Transport provenance for scoped GET callers. The original error stays intact. */
|
||||
export function isRouterOsPreHeaderGetReset(error: unknown): boolean {
|
||||
return error instanceof Error && preHeaderGetResets.has(error)
|
||||
}
|
||||
|
||||
function rosRequest(
|
||||
params: MikrotikConnectParams,
|
||||
path: string,
|
||||
@@ -86,12 +93,15 @@ function rosRequest(
|
||||
options.agent = params.useSsl ? rosHttpsAgent : rosHttpAgent
|
||||
|
||||
let req: ReturnType<typeof lib.request> | undefined
|
||||
let headersReceived = false, timedOut = false
|
||||
const timer = setTimeout(() => {
|
||||
timedOut = true
|
||||
req?.destroy(new Error(`Connection to ${params.host}:${params.port} timed out after ${timeoutMs / 1000}s`))
|
||||
}, timeoutMs)
|
||||
|
||||
try {
|
||||
req = lib.request(options, (res) => {
|
||||
headersReceived = true
|
||||
watchRosResponseFailure(res, error => { clearTimeout(timer); reject(error) })
|
||||
let body = "", bytes = 0, failed = false
|
||||
const failSize = () => {
|
||||
@@ -128,8 +138,10 @@ function rosRequest(
|
||||
})
|
||||
})
|
||||
|
||||
req.once("information", () => { headersReceived = true })
|
||||
req.on("error", (err) => {
|
||||
clearTimeout(timer)
|
||||
if (!headersReceived && !timedOut && !signal?.aborted && (err as NodeJS.ErrnoException).code === "ECONNRESET") preHeaderGetResets.add(err)
|
||||
reject(err)
|
||||
})
|
||||
|
||||
|
||||
@@ -0,0 +1,90 @@
|
||||
import assert from "node:assert/strict"
|
||||
import { EventEmitter } from "node:events"
|
||||
import http from "node:http"
|
||||
import { test, type TestContext } from "node:test"
|
||||
import { isRouterOsPreHeaderGetReset, MikrotikClient } from "./mikrotik.js"
|
||||
import { createWireguardDependencyReader } from "./wireguard-interface-delete.js"
|
||||
|
||||
type Outcome = "reset" | "late-reset" | "informational-reset" | "body-reset" | "ok" | "refused" | "invalid-json"
|
||||
const reset = () => Object.assign(new Error("fixture reset"), { code: "ECONNRESET" })
|
||||
|
||||
function fixture(t: TestContext, outcomes: Outcome[]) {
|
||||
const paths: string[] = [], timeouts: number[] = []
|
||||
t.mock.method(http, "request", (options: http.RequestOptions, receive: (res: unknown) => void) => {
|
||||
assert.equal(options.method, "GET")
|
||||
paths.push(options.path!)
|
||||
const request = new EventEmitter()
|
||||
const outcome = outcomes.shift()
|
||||
assert(outcome)
|
||||
return Object.assign(request, {
|
||||
end() {
|
||||
if (outcome === "reset") return request.emit("error", reset())
|
||||
if (outcome === "informational-reset") {
|
||||
request.emit("information", { statusCode: 103 })
|
||||
return request.emit("error", reset())
|
||||
}
|
||||
const response = Object.assign(new EventEmitter(), { headers: {}, statusCode: outcome === "refused" ? 429 : 200 })
|
||||
receive(response)
|
||||
if (outcome === "late-reset") return request.emit("error", reset())
|
||||
if (outcome === "body-reset") return response.emit("error", reset())
|
||||
response.emit("data", Buffer.from(outcome === "invalid-json" ? "!" : "[]"))
|
||||
response.emit("end")
|
||||
},
|
||||
})
|
||||
})
|
||||
const client = new MikrotikClient({ host: "fixture.invalid", port: 80, useSsl: false, verifySsl: false, username: "fixture", password: "fixture" })
|
||||
return {
|
||||
paths, timeouts,
|
||||
client: { get<T>(path: string, timeout?: number, signal?: AbortSignal, maxBytes?: number) {
|
||||
assert.equal(maxBytes, 1_048_576)
|
||||
timeouts.push(timeout!)
|
||||
return client.get<T>(path, timeout, signal, maxBytes)
|
||||
} },
|
||||
}
|
||||
}
|
||||
|
||||
test("dependency reset retries the same GET with fresh authority, reduced time, and one capture slot", async t => {
|
||||
const { client, paths, timeouts } = fixture(t, ["reset", "ok", "reset"])
|
||||
let authorized = 0, clock = 0
|
||||
const read = createWireguardDependencyReader(client, async () => { if (++authorized === 2) clock += 120 }, () => clock)
|
||||
assert.deepEqual(await read("/interface"), [])
|
||||
await assert.rejects(read("/routing/ospf/neighbor"), { code: "STATE_UNAVAILABLE" })
|
||||
assert.equal(authorized, 3)
|
||||
assert.deepEqual(paths, ["/rest/interface", "/rest/interface", "/rest/routing/ospf/neighbor"])
|
||||
assert.deepEqual(timeouts, [10_000, 9_880, 10_000])
|
||||
})
|
||||
|
||||
test("post-header reset, refusal, malformed body and unproven reset cannot replay", async t => {
|
||||
const { client, paths } = fixture(t, ["late-reset", "informational-reset", "body-reset", "refused", "invalid-json"])
|
||||
const read = createWireguardDependencyReader(client, async () => {})
|
||||
for (let index = 0; index < 5; index++) await assert.rejects(read("/interface"), { code: "STATE_UNAVAILABLE" })
|
||||
assert.equal(paths.length, 5)
|
||||
const synthetic = reset()
|
||||
assert.equal(isRouterOsPreHeaderGetReset(synthetic), false)
|
||||
let calls = 0
|
||||
const unknown = createWireguardDependencyReader({ async get() { calls++; throw synthetic } }, async () => {})
|
||||
await assert.rejects(unknown("/interface"), { code: "STATE_UNAVAILABLE" })
|
||||
assert.equal(calls, 1)
|
||||
})
|
||||
|
||||
test("base IPv4 and IPv6 address reads never retry or use the dependency slot", async t => {
|
||||
const { client, paths } = fixture(t, ["reset", "reset", "reset", "ok"])
|
||||
const read = createWireguardDependencyReader(client, async () => {})
|
||||
await assert.rejects(read("/ip/address"), { code: "STATE_UNAVAILABLE" })
|
||||
await assert.rejects(read("/ipv6/address"), { code: "STATE_UNAVAILABLE" })
|
||||
assert.deepEqual(await read("/interface"), [])
|
||||
assert.deepEqual(paths, ["/rest/ip/address", "/rest/ipv6/address", "/rest/interface", "/rest/interface"])
|
||||
})
|
||||
|
||||
test("retry authority denial and deadline expiry stop before transport", async t => {
|
||||
const { client, paths } = fixture(t, ["reset", "reset"])
|
||||
const denied = new Error("authority denied")
|
||||
let authorized = 0
|
||||
const read = createWireguardDependencyReader(client, async () => { if (++authorized === 2) throw denied })
|
||||
await assert.rejects(read("/interface"), error => error === denied)
|
||||
let clock = 0
|
||||
const expired = createWireguardDependencyReader(client, async () => { if (++authorized === 4) clock = 10_000 }, () => clock)
|
||||
await assert.rejects(expired("/interface"), { code: "STATE_UNAVAILABLE" })
|
||||
assert.equal(authorized, 4)
|
||||
assert.equal(paths.length, 2)
|
||||
})
|
||||
@@ -6,7 +6,7 @@ import { fingerprint, McpError, privateFingerprinter } from "../modules/mcp/secu
|
||||
import { readWireguardPeerDeletionState, wireguardPeerDeletionTarget, wireguardConfigurationRows, type WireguardPeerDeletionState } from "./wireguard-peer-delete.js"
|
||||
import { greStableRows, assertGreDeleteDependencies } from "./gre-mutation-state.js"
|
||||
import { observedBoolean } from "./ospf-observed-values.js"
|
||||
import { MikrotikClient } from "./mikrotik.js"
|
||||
import { isRouterOsPreHeaderGetReset, MikrotikClient } from "./mikrotik.js"
|
||||
import { withApplicationMutationLock, withConfigMutationLock } from "./config-mutation-lock.js"
|
||||
import { executeGuardedApplicationWrite } from "./config-mutation-policy.js"
|
||||
import { captureAndAppendRevision } from "./config-revisions.js"
|
||||
@@ -45,13 +45,38 @@ function nativeRows(raw: unknown, required: string[] = []): Row[] {
|
||||
if (new Set(raw.map(row => row[".id"])).size !== raw.length) throw fail("STATE_UNAVAILABLE", "Повторяющиеся ID в каталоге WireGuard")
|
||||
return raw as Row[]
|
||||
}
|
||||
|
||||
/** Share one retry across this dependency capture and renew source authority before using it. */
|
||||
export function createWireguardDependencyReader(client: Pick<MikrotikClient, "get">, authorizeSource: () => Promise<unknown>, now = () => performance.now()) {
|
||||
const timeoutMs = 10_000, maxBytes = 1_048_576
|
||||
const unavailable = () => fail("STATE_UNAVAILABLE", "Не прочитан обязательный источник зависимостей WireGuard")
|
||||
let retried = false
|
||||
return async (path: string): Promise<unknown> => {
|
||||
await authorizeSource()
|
||||
const deadline = now() + timeoutMs
|
||||
try {
|
||||
return await client.get(path, timeoutMs, undefined, maxBytes)
|
||||
} catch (error) {
|
||||
if (retried || !wireguardInterfaceDependencyPaths.includes(path) || !isRouterOsPreHeaderGetReset(error) || now() >= deadline) throw unavailable()
|
||||
}
|
||||
retried = true
|
||||
await authorizeSource()
|
||||
const remaining = Math.floor(deadline - now())
|
||||
if (remaining < 1) throw unavailable()
|
||||
try {
|
||||
return await client.get(path, remaining, undefined, maxBytes)
|
||||
} catch {
|
||||
throw unavailable()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
export async function readWireguardInterfaceDeleteState(server: ServerRow, id: string, authorize: () => Promise<unknown> = async () => {}): Promise<WireguardInterfaceDeleteState> {
|
||||
const base = await readWireguardPeerDeletionState(server, authorize), client = MikrotikClient.fromServer(server)
|
||||
const readDependency = createWireguardDependencyReader(client, () => source(server, authorize))
|
||||
let totalRows = base.interfaces.length + base.peers.length, bytes = Buffer.byteLength(JSON.stringify([base.interfaces, base.peers]))
|
||||
const read = async (path: string, required: string[] = []) => {
|
||||
await source(server, authorize)
|
||||
let raw: unknown
|
||||
try { raw = await client.get<unknown>(path, 10000, undefined, 1048576) } catch { throw fail("STATE_UNAVAILABLE", "Не прочитан обязательный источник зависимостей WireGuard") }
|
||||
const raw = await readDependency(path)
|
||||
bytes += Buffer.byteLength(JSON.stringify(raw)); const rows = nativeRows(raw, required); totalRows += rows.length
|
||||
if (bytes > 1048576 || totalRows > 10000) throw fail("STATE_UNAVAILABLE", "Превышен предел зависимостей WireGuard: 10000 строк / 1 МиБ")
|
||||
return rows
|
||||
|
||||
Reference in new issue
Block a user