From a2d4194990127b484940dc1b291b5ad39c511412 Mon Sep 17 00:00:00 2001 From: Maxime Beauchamp <15185355+baktun14@users.noreply.github.com> Date: Sun, 20 Sep 2026 14:51:38 +0400 Subject: [PATCH 1/5] feat(deployment): add behavioural workload signals with replay tooling Two shape signals now run over the evidence rows a probe writes: one for a workload that holds the accelerator while carrying nothing on disk, one for a workload with nothing established in either direction. Both are recorded on the row and counted, neither changes a verdict, and both stay off until the flag is set. A console command re-scores recorded evidence, or a corpus of fixture shapes, and reports what each agreement length would have reached. It reads and never writes. --- apps/api/src/app/console.ts | 22 ++ .../workload-abuse/config/env.config.spec.ts | 20 ++ .../src/workload-abuse/config/env.config.ts | 21 +- .../workload-abuse.controller.spec.ts | 20 +- .../controllers/workload-abuse.controller.ts | 12 +- .../accel-without-artifacts.spec.ts | 89 ++++++++ .../accel-without-artifacts.ts | 27 +++ .../evaluate-behavioural-signals.spec.ts | 53 +++++ .../evaluate-behavioural-signals.ts | 16 ++ .../network-isolation.spec.ts | 73 +++++++ .../behavioural-signals/network-isolation.ts | 25 +++ .../lib/behavioural-signals/types.ts | 28 +++ .../workload-probe-evidence.repository.ts | 15 +- .../behavioural-signal-replay.service.spec.ts | 147 +++++++++++++ .../behavioural-signal-replay.service.ts | 193 ++++++++++++++++++ .../accelerator-bound-no-artifacts.json | 44 ++++ .../fixtures/cpu-only-workload.json | 13 ++ .../fixtures/idle-shell-host.json | 15 ++ .../fixtures/index.ts | 16 ++ .../fixtures/inference-with-weights.json | 40 ++++ .../fixtures/packaged-runtime.json | 31 +++ .../probe-evidence.service.spec.ts | 76 ++++++- .../probe-evidence/probe-evidence.service.ts | 39 ++++ ...be-trial-deployment.handler.integration.ts | 45 +++- .../probe-trial-deployment.handler.spec.ts | 31 ++- .../probe-trial-deployment.handler.ts | 13 ++ .../workload-abuse-instrumentation.service.ts | 8 + 27 files changed, 1120 insertions(+), 12 deletions(-) create mode 100644 apps/api/src/workload-abuse/lib/behavioural-signals/accel-without-artifacts.spec.ts create mode 100644 apps/api/src/workload-abuse/lib/behavioural-signals/accel-without-artifacts.ts create mode 100644 apps/api/src/workload-abuse/lib/behavioural-signals/evaluate-behavioural-signals.spec.ts create mode 100644 apps/api/src/workload-abuse/lib/behavioural-signals/evaluate-behavioural-signals.ts create mode 100644 apps/api/src/workload-abuse/lib/behavioural-signals/network-isolation.spec.ts create mode 100644 apps/api/src/workload-abuse/lib/behavioural-signals/network-isolation.ts create mode 100644 apps/api/src/workload-abuse/lib/behavioural-signals/types.ts create mode 100644 apps/api/src/workload-abuse/services/behavioural-signal-replay/behavioural-signal-replay.service.spec.ts create mode 100644 apps/api/src/workload-abuse/services/behavioural-signal-replay/behavioural-signal-replay.service.ts create mode 100644 apps/api/src/workload-abuse/services/behavioural-signal-replay/fixtures/accelerator-bound-no-artifacts.json create mode 100644 apps/api/src/workload-abuse/services/behavioural-signal-replay/fixtures/cpu-only-workload.json create mode 100644 apps/api/src/workload-abuse/services/behavioural-signal-replay/fixtures/idle-shell-host.json create mode 100644 apps/api/src/workload-abuse/services/behavioural-signal-replay/fixtures/index.ts create mode 100644 apps/api/src/workload-abuse/services/behavioural-signal-replay/fixtures/inference-with-weights.json create mode 100644 apps/api/src/workload-abuse/services/behavioural-signal-replay/fixtures/packaged-runtime.json diff --git a/apps/api/src/app/console.ts b/apps/api/src/app/console.ts index 5052d5c7e0..e27dd5d442 100644 --- a/apps/api/src/app/console.ts +++ b/apps/api/src/app/console.ts @@ -104,6 +104,28 @@ program }); }); +program + .command("replay-behavioural-signals") + .description("Re-score recorded workload probe evidence against the behavioural signals, writing nothing") + .option("-s, --since ", "Start of the evidence window", value => z.coerce.date().parse(value)) + .option("-u, --until ", "End of the evidence window", value => z.coerce.date().parse(value)) + .option("-f, --fixture ", "Replay a fixture file or a directory of fixture files") + .option("-b, --bundled-fixtures", "Include the fixture corpus shipped with the repo", false) + .option("--accel-min-vram-mb ", "Override the accelerator memory threshold", value => z.number({ coerce: true }).parse(value)) + .option("--artifact-min-mb ", "Override the artifact size threshold", value => z.number({ coerce: true }).parse(value)) + .action(async (options, command) => { + await executeCliHandler(command.name(), async () => { + return container.resolve(WorkloadAbuseController).replayBehaviouralSignals({ + since: options.since, + until: options.until, + fixturePaths: options.fixture, + bundledFixtures: options.bundledFixtures, + accelMinVramMb: options.accelMinVramMb, + artifactMinMb: options.artifactMinMb + }); + }); + }); + program .command("cleanup-provider-deployments") .description("Close trial deployments for a provider") diff --git a/apps/api/src/workload-abuse/config/env.config.spec.ts b/apps/api/src/workload-abuse/config/env.config.spec.ts index 7c9e20d7cd..21f33c3b73 100644 --- a/apps/api/src/workload-abuse/config/env.config.spec.ts +++ b/apps/api/src/workload-abuse/config/env.config.spec.ts @@ -27,6 +27,26 @@ describe("workload abuse env config", () => { ]); }); + it("defaults behavioural signals to off with no relay endpoints", () => { + const config = envSchema.parse({}); + + expect(config.WORKLOAD_ABUSE_BEHAVIOURAL_SIGNALS_ENABLED).toBe(false); + expect(config.WORKLOAD_ABUSE_SIGNAL_RELAY_ENDPOINTS).toEqual([]); + expect(config.WORKLOAD_ABUSE_SIGNAL_ACCEL_MIN_VRAM_MB).toBe(1024); + expect(config.WORKLOAD_ABUSE_SIGNAL_ARTIFACT_MIN_MB).toBe(256); + }); + + it("parses the relay endpoint list", () => { + const config = envSchema.parse({ WORKLOAD_ABUSE_SIGNAL_RELAY_ENDPOINTS: JSON.stringify(["10.0.0.7", "10.0.0.8:443"]) }); + + expect(config.WORKLOAD_ABUSE_SIGNAL_RELAY_ENDPOINTS).toEqual(["10.0.0.7", "10.0.0.8:443"]); + }); + + it("rejects a relay endpoint list that is not a JSON array of strings", () => { + expect(() => envSchema.parse({ WORKLOAD_ABUSE_SIGNAL_RELAY_ENDPOINTS: "10.0.0.7" })).toThrow(/ip or ip:port/); + expect(() => envSchema.parse({ WORKLOAD_ABUSE_SIGNAL_RELAY_ENDPOINTS: JSON.stringify([443]) })).toThrow(/ip or ip:port/); + }); + it("rejects a signature document that is not JSON", () => { expect(() => envSchema.parse({ WORKLOAD_ABUSE_SIGNATURES: "not json" })).toThrow(/valid signature document/); }); diff --git a/apps/api/src/workload-abuse/config/env.config.ts b/apps/api/src/workload-abuse/config/env.config.ts index 1f7b9529a7..2478e9d731 100644 --- a/apps/api/src/workload-abuse/config/env.config.ts +++ b/apps/api/src/workload-abuse/config/env.config.ts @@ -69,6 +69,17 @@ function blankToUndefined(value: unknown): unknown { return typeof value === "string" && value.trim() === "" ? undefined : value; } +function parseRelayEndpoints(raw: string, ctx: z.RefinementCtx): string[] { + const parsed = z.array(z.string().min(1)).safeParse(parseJson(raw)); + + if (!parsed.success) { + ctx.addIssue({ code: z.ZodIssueCode.custom, message: "WORKLOAD_ABUSE_SIGNAL_RELAY_ENDPOINTS must be a JSON array of ip or ip:port strings" }); + return z.NEVER; + } + + return parsed.data; +} + export const envSchema = z.object({ WORKLOAD_ABUSE_PROBE_ENABLED: z .enum(["true", "false"]) @@ -99,7 +110,15 @@ export const envSchema = z.object({ /** A domain with an account older than this predates the attack, so it is somebody's real domain. */ WORKLOAD_ABUSE_DOMAIN_BLOCK_MIN_ACCOUNT_AGE_DAYS: z.number({ coerce: true }).int().positive().default(30), /** Evidence rows feed the behavioural replay, so they must outlive its 30-day window with margin. */ - WORKLOAD_ABUSE_EVIDENCE_RETENTION_DAYS: z.number({ coerce: true }).int().positive().default(90) + WORKLOAD_ABUSE_EVIDENCE_RETENTION_DAYS: z.number({ coerce: true }).int().positive().default(90), + WORKLOAD_ABUSE_BEHAVIOURAL_SIGNALS_ENABLED: z + .enum(["true", "false"]) + .default("false") + .transform(value => value === "true"), + WORKLOAD_ABUSE_SIGNAL_ACCEL_MIN_VRAM_MB: z.number({ coerce: true }).int().positive().default(1_024), + WORKLOAD_ABUSE_SIGNAL_ARTIFACT_MIN_MB: z.number({ coerce: true }).int().positive().default(256), + /** The list itself lives in Doppler: these are our own endpoints, so a reader learns how the exclusion works but not what it covers. */ + WORKLOAD_ABUSE_SIGNAL_RELAY_ENDPOINTS: z.string().default("[]").transform(parseRelayEndpoints) }); export type WorkloadAbuseConfig = z.infer; diff --git a/apps/api/src/workload-abuse/controllers/workload-abuse.controller.spec.ts b/apps/api/src/workload-abuse/controllers/workload-abuse.controller.spec.ts index 7a79090175..fcb54e4a70 100644 --- a/apps/api/src/workload-abuse/controllers/workload-abuse.controller.spec.ts +++ b/apps/api/src/workload-abuse/controllers/workload-abuse.controller.spec.ts @@ -1,6 +1,10 @@ import { describe, expect, it } from "vitest"; import { mock } from "vitest-mock-extended"; +import type { + BehaviouralReplaySummary, + BehaviouralSignalReplayService +} from "@src/workload-abuse/services/behavioural-signal-replay/behavioural-signal-replay.service"; import type { ProbeEvidenceService } from "@src/workload-abuse/services/probe-evidence/probe-evidence.service"; import type { TrialAbuseEnforcementJobService } from "@src/workload-abuse/services/trial-abuse-enforcement-job/trial-abuse-enforcement-job.service"; import type { TrialWorkloadProbeJobService } from "@src/workload-abuse/services/trial-workload-probe-job/trial-workload-probe-job.service"; @@ -61,14 +65,26 @@ describe(WorkloadAbuseController.name, () => { expect(probeEvidenceService.purgeExpired).not.toHaveBeenCalled(); }); + it("returns the replay of the window it was asked for", async () => { + const { controller, behaviouralSignalReplayService } = setup(); + const summary = mock({ deployments: [] }); + behaviouralSignalReplayService.replay.mockResolvedValue(summary); + const options = { since: new Date("2026-08-01T00:00:00.000Z") }; + + await expect(controller.replayBehaviouralSignals(options)).resolves.toBe(summary); + + expect(behaviouralSignalReplayService.replay).toHaveBeenCalledWith(options); + }); + function setup() { const probeJobService = mock(); probeJobService.reconcile.mockResolvedValue(); const enforcementJobService = mock(); enforcementJobService.reconcile.mockResolvedValue(); const probeEvidenceService = mock(); - const controller = new WorkloadAbuseController(probeJobService, enforcementJobService, probeEvidenceService); + const behaviouralSignalReplayService = mock(); + const controller = new WorkloadAbuseController(probeJobService, enforcementJobService, probeEvidenceService, behaviouralSignalReplayService); - return { controller, probeJobService, enforcementJobService, probeEvidenceService }; + return { controller, probeJobService, enforcementJobService, probeEvidenceService, behaviouralSignalReplayService }; } }); diff --git a/apps/api/src/workload-abuse/controllers/workload-abuse.controller.ts b/apps/api/src/workload-abuse/controllers/workload-abuse.controller.ts index b949c7651c..d62dc5d13d 100644 --- a/apps/api/src/workload-abuse/controllers/workload-abuse.controller.ts +++ b/apps/api/src/workload-abuse/controllers/workload-abuse.controller.ts @@ -1,6 +1,11 @@ import { singleton } from "tsyringe"; import type { DryRunOptions } from "@src/core/types/console"; +import { + type BehaviouralReplayOptions, + type BehaviouralReplaySummary, + BehaviouralSignalReplayService +} from "@src/workload-abuse/services/behavioural-signal-replay/behavioural-signal-replay.service"; import { ProbeEvidenceService } from "@src/workload-abuse/services/probe-evidence/probe-evidence.service"; import { TrialAbuseEnforcementJobService } from "@src/workload-abuse/services/trial-abuse-enforcement-job/trial-abuse-enforcement-job.service"; import { TrialWorkloadProbeJobService } from "@src/workload-abuse/services/trial-workload-probe-job/trial-workload-probe-job.service"; @@ -10,7 +15,8 @@ export class WorkloadAbuseController { constructor( private readonly probeJobService: TrialWorkloadProbeJobService, private readonly enforcementJobService: TrialAbuseEnforcementJobService, - private readonly probeEvidenceService: ProbeEvidenceService + private readonly probeEvidenceService: ProbeEvidenceService, + private readonly behaviouralSignalReplayService: BehaviouralSignalReplayService ) {} /** Each sweep runs whether or not the other fails, and retention runs whether or not the sweeps do, so nothing waits another run. */ @@ -24,4 +30,8 @@ export class WorkloadAbuseController { if (failures.length === 1) throw failures[0]; if (failures.length > 1) throw new AggregateError(failures, "Both the probe sweep and the enforcement sweep failed"); } + + async replayBehaviouralSignals(options: BehaviouralReplayOptions): Promise { + return await this.behaviouralSignalReplayService.replay(options); + } } diff --git a/apps/api/src/workload-abuse/lib/behavioural-signals/accel-without-artifacts.spec.ts b/apps/api/src/workload-abuse/lib/behavioural-signals/accel-without-artifacts.spec.ts new file mode 100644 index 0000000000..137bca35f5 --- /dev/null +++ b/apps/api/src/workload-abuse/lib/behavioural-signals/accel-without-artifacts.spec.ts @@ -0,0 +1,89 @@ +import { describe, expect, it } from "vitest"; + +import type { ProbeEvidenceAccelerator, ProbeEvidenceArtifact } from "@src/workload-abuse/model-schemas"; +import { evaluateAccelWithoutArtifacts } from "./accel-without-artifacts"; +import type { BehaviouralSignalParams, ProbeEvidenceSnapshot } from "./types"; + +describe("evaluateAccelWithoutArtifacts", () => { + it("fires when a process holds the accelerator and nothing large sits on disk", () => { + const { snapshot, params } = setup({ vramMb: 18_000, artifacts: [{ path: "/opt/run", sizeBytes: 4_194_304 }] }); + + expect(evaluateAccelWithoutArtifacts(snapshot, params)).toEqual({ + signal: "accel_without_artifacts", + detail: { heaviestVramMb: 18_000, largestArtifactMb: 4 } + }); + }); + + it("reports the heaviest process when several share the accelerator", () => { + const { snapshot, params } = setup({ + accelerator: [ + { + name: "accelerator-0", + utilPct: 99, + memUsedMb: 20_000, + memTotalMb: 24_576, + processes: [ + { pid: 10, name: "worker", vramMb: 2_000 }, + { pid: 11, name: "worker", vramMb: 16_000 } + ] + } + ], + artifacts: [] + }); + + expect(evaluateAccelWithoutArtifacts(snapshot, params)?.detail).toEqual({ heaviestVramMb: 16_000, largestArtifactMb: 0 }); + }); + + it("stays silent when the workload carries large files", () => { + const { snapshot, params } = setup({ vramMb: 18_000, artifacts: [{ path: "/models/weights", sizeBytes: 8_589_934_592 }] }); + + expect(evaluateAccelWithoutArtifacts(snapshot, params)).toBeNull(); + }); + + it("stays silent when no accelerator was reported", () => { + const { snapshot, params } = setup({ accelerator: null, artifacts: [] }); + + expect(evaluateAccelWithoutArtifacts(snapshot, params)).toBeNull(); + }); + + it("stays silent when accelerator memory stays under the threshold", () => { + const { snapshot, params } = setup({ vramMb: 200, artifacts: [], accelMinVramMb: 1_024 }); + + expect(evaluateAccelWithoutArtifacts(snapshot, params)).toBeNull(); + }); + + it("stays silent when the probe collected no disk section", () => { + const { snapshot, params } = setup({ vramMb: 18_000, artifacts: null }); + + expect(evaluateAccelWithoutArtifacts(snapshot, params)).toBeNull(); + }); + + function setup(input: { + vramMb?: number; + accelerator?: ProbeEvidenceAccelerator[] | null; + artifacts?: ProbeEvidenceArtifact[] | null; + accelMinVramMb?: number; + artifactMinMb?: number; + }) { + const accelerator = + input.accelerator !== undefined + ? input.accelerator + : [ + { + name: "accelerator-0", + utilPct: 98, + memUsedMb: 20_480, + memTotalMb: 24_576, + processes: [{ pid: 1234, name: "worker", vramMb: input.vramMb ?? 18_000 }] + } + ]; + const snapshot: ProbeEvidenceSnapshot = { accelerator, artifacts: input.artifacts ?? null, netShape: null }; + const params: BehaviouralSignalParams = { + accelMinVramMb: input.accelMinVramMb ?? 1_024, + artifactMinMb: input.artifactMinMb ?? 256, + relayEndpoints: [] + }; + + return { snapshot, params }; + } +}); diff --git a/apps/api/src/workload-abuse/lib/behavioural-signals/accel-without-artifacts.ts b/apps/api/src/workload-abuse/lib/behavioural-signals/accel-without-artifacts.ts new file mode 100644 index 0000000000..90a2bee94e --- /dev/null +++ b/apps/api/src/workload-abuse/lib/behavioural-signals/accel-without-artifacts.ts @@ -0,0 +1,27 @@ +import { BEHAVIOURAL_SIGNALS, type BehaviouralFinding, type BehaviouralSignalParams, type ProbeEvidenceSnapshot } from "./types"; + +const BYTES_PER_MB = 1024 * 1024; + +/** A snapshot without a disk section reports nothing rather than an empty disk, so it cannot support this signal. */ +export function evaluateAccelWithoutArtifacts(snapshot: ProbeEvidenceSnapshot, params: BehaviouralSignalParams): BehaviouralFinding | null { + const heaviestVramMb = findHeaviestVramMb(snapshot); + + if (heaviestVramMb < params.accelMinVramMb) return null; + if (!snapshot.artifacts) return null; + + const largestArtifactMb = findLargestArtifactMb(snapshot); + + if (largestArtifactMb >= params.artifactMinMb) return null; + + return { signal: BEHAVIOURAL_SIGNALS.accelWithoutArtifacts, detail: { heaviestVramMb, largestArtifactMb } }; +} + +function findHeaviestVramMb(snapshot: ProbeEvidenceSnapshot): number { + const vramPerProcess = (snapshot.accelerator ?? []).flatMap(accelerator => accelerator.processes.map(process => process.vramMb)); + return vramPerProcess.reduce((heaviest, vramMb) => Math.max(heaviest, vramMb), 0); +} + +function findLargestArtifactMb(snapshot: ProbeEvidenceSnapshot): number { + const sizes = (snapshot.artifacts ?? []).map(artifact => Math.round(artifact.sizeBytes / BYTES_PER_MB)); + return sizes.reduce((largest, sizeMb) => Math.max(largest, sizeMb), 0); +} diff --git a/apps/api/src/workload-abuse/lib/behavioural-signals/evaluate-behavioural-signals.spec.ts b/apps/api/src/workload-abuse/lib/behavioural-signals/evaluate-behavioural-signals.spec.ts new file mode 100644 index 0000000000..f15e0c2339 --- /dev/null +++ b/apps/api/src/workload-abuse/lib/behavioural-signals/evaluate-behavioural-signals.spec.ts @@ -0,0 +1,53 @@ +import { describe, expect, it } from "vitest"; + +import { evaluateBehaviouralSignals, isBehaviouralCandidate } from "./evaluate-behavioural-signals"; +import type { BehaviouralSignalParams, ProbeEvidenceSnapshot } from "./types"; + +describe("evaluateBehaviouralSignals", () => { + it("returns both findings when the snapshot matches both signals", () => { + const { snapshot, params } = setup({ vramMb: 18_000, artifactBytes: 1_024, established: [] }); + + expect(evaluateBehaviouralSignals(snapshot, params).map(finding => finding.signal)).toEqual(["accel_without_artifacts", "network_isolated"]); + }); + + it("returns one finding when the workload carries weights but talks to nobody", () => { + const { snapshot, params } = setup({ vramMb: 18_000, artifactBytes: 8_589_934_592, established: [] }); + + expect(evaluateBehaviouralSignals(snapshot, params).map(finding => finding.signal)).toEqual(["network_isolated"]); + }); + + it("returns nothing when the snapshot matches neither signal", () => { + const { snapshot, params } = setup({ + vramMb: 18_000, + artifactBytes: 8_589_934_592, + established: [{ localPort: 8_080, remoteIp: "203.0.113.9", remotePort: 51_000, count: 2 }] + }); + + expect(evaluateBehaviouralSignals(snapshot, params)).toEqual([]); + }); + + it("treats the two signals together as a candidate", () => { + const { snapshot, params } = setup({ vramMb: 18_000, artifactBytes: 1_024, established: [] }); + + expect(isBehaviouralCandidate(evaluateBehaviouralSignals(snapshot, params))).toBe(true); + }); + + it("treats a single signal as no candidate", () => { + const { snapshot, params } = setup({ vramMb: 18_000, artifactBytes: 8_589_934_592, established: [] }); + + expect(isBehaviouralCandidate(evaluateBehaviouralSignals(snapshot, params))).toBe(false); + }); + + function setup(input: { vramMb: number; artifactBytes: number; established: NonNullable["established"] }) { + const snapshot: ProbeEvidenceSnapshot = { + accelerator: [ + { name: "accelerator-0", utilPct: 97, memUsedMb: 20_480, memTotalMb: 24_576, processes: [{ pid: 1234, name: "worker", vramMb: input.vramMb }] } + ], + artifacts: [{ path: "/opt/payload", sizeBytes: input.artifactBytes }], + netShape: { listenPorts: [8_080], established: input.established } + }; + const params: BehaviouralSignalParams = { accelMinVramMb: 1_024, artifactMinMb: 256, relayEndpoints: [] }; + + return { snapshot, params }; + } +}); diff --git a/apps/api/src/workload-abuse/lib/behavioural-signals/evaluate-behavioural-signals.ts b/apps/api/src/workload-abuse/lib/behavioural-signals/evaluate-behavioural-signals.ts new file mode 100644 index 0000000000..7e78a52bf2 --- /dev/null +++ b/apps/api/src/workload-abuse/lib/behavioural-signals/evaluate-behavioural-signals.ts @@ -0,0 +1,16 @@ +import { evaluateAccelWithoutArtifacts } from "./accel-without-artifacts"; +import { evaluateNetworkIsolation } from "./network-isolation"; +import { BEHAVIOURAL_SIGNALS, type BehaviouralFinding, type BehaviouralSignalParams, type ProbeEvidenceSnapshot } from "./types"; + +export function evaluateBehaviouralSignals(snapshot: ProbeEvidenceSnapshot, params: BehaviouralSignalParams): BehaviouralFinding[] { + const evaluated = [evaluateAccelWithoutArtifacts(snapshot, params), evaluateNetworkIsolation(snapshot, params)]; + + return evaluated.filter((finding): finding is BehaviouralFinding => finding !== null); +} + +/** Either signal alone has a known benign population, so only the conjunction describes the shape worth acting on. */ +export function isBehaviouralCandidate(findings: BehaviouralFinding[]): boolean { + const signals = new Set(findings.map(finding => finding.signal)); + + return signals.has(BEHAVIOURAL_SIGNALS.accelWithoutArtifacts) && signals.has(BEHAVIOURAL_SIGNALS.networkIsolated); +} diff --git a/apps/api/src/workload-abuse/lib/behavioural-signals/network-isolation.spec.ts b/apps/api/src/workload-abuse/lib/behavioural-signals/network-isolation.spec.ts new file mode 100644 index 0000000000..1941453928 --- /dev/null +++ b/apps/api/src/workload-abuse/lib/behavioural-signals/network-isolation.spec.ts @@ -0,0 +1,73 @@ +import { describe, expect, it } from "vitest"; + +import type { ProbeEvidenceNetShape } from "@src/workload-abuse/model-schemas"; +import { evaluateNetworkIsolation } from "./network-isolation"; +import type { BehaviouralSignalParams, ProbeEvidenceSnapshot } from "./types"; + +describe("evaluateNetworkIsolation", () => { + it("fires when nothing is established in either direction", () => { + const { snapshot, params } = setup({ netShape: { listenPorts: [22, 8080], established: [] } }); + + expect(evaluateNetworkIsolation(snapshot, params)).toEqual({ + signal: "network_isolated", + detail: { excludedRelay: 0, listenPorts: 2 } + }); + }); + + it("fires when the only outbound connection goes to a first-party relay", () => { + const { snapshot, params } = setup({ + netShape: { listenPorts: [22], established: [{ localPort: 41_234, remoteIp: "10.0.0.7", remotePort: 443, count: 1 }] }, + relayEndpoints: ["10.0.0.7"] + }); + + expect(evaluateNetworkIsolation(snapshot, params)?.detail).toEqual({ excludedRelay: 1, listenPorts: 1 }); + }); + + it("fires when a relay entry pins the port the connection uses", () => { + const { snapshot, params } = setup({ + netShape: { listenPorts: [], established: [{ localPort: 41_234, remoteIp: "10.0.0.7", remotePort: 443, count: 1 }] }, + relayEndpoints: ["10.0.0.7:443"] + }); + + expect(evaluateNetworkIsolation(snapshot, params)).not.toBeNull(); + }); + + it("fires when the relay answers over IPv6", () => { + const { snapshot, params } = setup({ + netShape: { listenPorts: [], established: [{ localPort: 41_234, remoteIp: "2001:0db8:0000:0000:0000:0000:0000:0001", remotePort: 443, count: 1 }] }, + relayEndpoints: ["2001:0db8:0000:0000:0000:0000:0000:0001"] + }); + + expect(evaluateNetworkIsolation(snapshot, params)).not.toBeNull(); + }); + + it("stays silent when a connection lands on a listening port", () => { + const { snapshot, params } = setup({ + netShape: { listenPorts: [8080], established: [{ localPort: 8080, remoteIp: "203.0.113.9", remotePort: 51_000, count: 3 }] } + }); + + expect(evaluateNetworkIsolation(snapshot, params)).toBeNull(); + }); + + it("stays silent when an outbound connection goes somewhere other than a relay", () => { + const { snapshot, params } = setup({ + netShape: { listenPorts: [22], established: [{ localPort: 41_234, remoteIp: "198.51.100.4", remotePort: 3_333, count: 1 }] }, + relayEndpoints: ["10.0.0.7"] + }); + + expect(evaluateNetworkIsolation(snapshot, params)).toBeNull(); + }); + + it("stays silent when the probe collected no socket section", () => { + const { snapshot, params } = setup({ netShape: null }); + + expect(evaluateNetworkIsolation(snapshot, params)).toBeNull(); + }); + + function setup(input: { netShape: ProbeEvidenceNetShape | null; relayEndpoints?: string[] }) { + const snapshot: ProbeEvidenceSnapshot = { accelerator: null, artifacts: null, netShape: input.netShape }; + const params: BehaviouralSignalParams = { accelMinVramMb: 1_024, artifactMinMb: 256, relayEndpoints: input.relayEndpoints ?? [] }; + + return { snapshot, params }; + } +}); diff --git a/apps/api/src/workload-abuse/lib/behavioural-signals/network-isolation.ts b/apps/api/src/workload-abuse/lib/behavioural-signals/network-isolation.ts new file mode 100644 index 0000000000..0d7cab1817 --- /dev/null +++ b/apps/api/src/workload-abuse/lib/behavioural-signals/network-isolation.ts @@ -0,0 +1,25 @@ +import type { ProbeEvidenceNetShape } from "@src/workload-abuse/model-schemas"; +import { BEHAVIOURAL_SIGNALS, type BehaviouralFinding, type BehaviouralSignalParams, type ProbeEvidenceSnapshot } from "./types"; + +type EstablishedConnection = ProbeEvidenceNetShape["established"][number]; + +export function evaluateNetworkIsolation(snapshot: ProbeEvidenceSnapshot, params: BehaviouralSignalParams): BehaviouralFinding | null { + const netShape = snapshot.netShape; + + if (!netShape) return null; + + const listenPorts = new Set(netShape.listenPorts); + const inbound = netShape.established.filter(connection => listenPorts.has(connection.localPort)); + + if (inbound.length > 0) return null; + + const relayOutbound = netShape.established.filter(connection => isRelayEndpoint(connection, params.relayEndpoints)); + + if (relayOutbound.length < netShape.established.length) return null; + + return { signal: BEHAVIOURAL_SIGNALS.networkIsolated, detail: { excludedRelay: relayOutbound.length, listenPorts: netShape.listenPorts.length } }; +} + +function isRelayEndpoint(connection: EstablishedConnection, relayEndpoints: string[]): boolean { + return relayEndpoints.includes(connection.remoteIp) || relayEndpoints.includes(`${connection.remoteIp}:${connection.remotePort}`); +} diff --git a/apps/api/src/workload-abuse/lib/behavioural-signals/types.ts b/apps/api/src/workload-abuse/lib/behavioural-signals/types.ts new file mode 100644 index 0000000000..6c2f4daf7c --- /dev/null +++ b/apps/api/src/workload-abuse/lib/behavioural-signals/types.ts @@ -0,0 +1,28 @@ +import type { + ProbeEvidenceAccelerator, + ProbeEvidenceArtifact, + ProbeEvidenceBehaviouralFinding, + ProbeEvidenceNetShape +} from "@src/workload-abuse/model-schemas"; + +export const BEHAVIOURAL_SIGNALS = { + accelWithoutArtifacts: "accel_without_artifacts", + networkIsolated: "network_isolated" +} as const; + +export type BehaviouralSignal = (typeof BEHAVIOURAL_SIGNALS)[keyof typeof BEHAVIOURAL_SIGNALS]; + +export type BehaviouralFinding = ProbeEvidenceBehaviouralFinding; + +/** The columns of one evidence row, so the replay and the live probe evaluate the exact same input. */ +export type ProbeEvidenceSnapshot = { + accelerator: ProbeEvidenceAccelerator[] | null; + artifacts: ProbeEvidenceArtifact[] | null; + netShape: ProbeEvidenceNetShape | null; +}; + +export type BehaviouralSignalParams = { + accelMinVramMb: number; + artifactMinMb: number; + relayEndpoints: string[]; +}; diff --git a/apps/api/src/workload-abuse/repositories/workload-probe-evidence/workload-probe-evidence.repository.ts b/apps/api/src/workload-abuse/repositories/workload-probe-evidence/workload-probe-evidence.repository.ts index 8896df592a..b8716af4c5 100644 --- a/apps/api/src/workload-abuse/repositories/workload-probe-evidence/workload-probe-evidence.repository.ts +++ b/apps/api/src/workload-abuse/repositories/workload-probe-evidence/workload-probe-evidence.repository.ts @@ -1,9 +1,10 @@ -import { and, asc, eq, gt, lt, sql } from "drizzle-orm"; +import { and, asc, eq, gt, gte, lt, lte, sql } from "drizzle-orm"; import { singleton } from "tsyringe"; import { type ApiPgDatabase, type ApiPgTables, InjectPg, InjectPgTable } from "@src/core/providers"; import { type AbilityParams, BaseRepository } from "@src/core/repositories/base.repository"; import { TxService } from "@src/core/services"; +import type { ProbeEvidenceBehaviouralFinding } from "@src/workload-abuse/model-schemas"; type Table = ApiPgTables["WorkloadProbeEvidence"]; export type WorkloadProbeEvidenceInput = Partial; @@ -29,6 +30,18 @@ export class WorkloadProbeEvidenceRepository extends BaseRepository { + await this.cursor.update(this.table).set({ behaviouralFindings: findings }).where(eq(this.table.id, id)); + } + + async findCreatedBetween({ since, until }: { since: Date; until: Date }): Promise { + return await this.cursor + .select() + .from(this.table) + .where(and(gte(this.table.createdAt, since), lte(this.table.createdAt, until))) + .orderBy(asc(this.table.createdAt)); + } + async findRecentForDeployment({ walletId, dseq, since }: { walletId: number; dseq: string; since: Date }): Promise { return await this.cursor .select() diff --git a/apps/api/src/workload-abuse/services/behavioural-signal-replay/behavioural-signal-replay.service.spec.ts b/apps/api/src/workload-abuse/services/behavioural-signal-replay/behavioural-signal-replay.service.spec.ts new file mode 100644 index 0000000000..58b65a96e4 --- /dev/null +++ b/apps/api/src/workload-abuse/services/behavioural-signal-replay/behavioural-signal-replay.service.spec.ts @@ -0,0 +1,147 @@ +import { describe, expect, it, vi } from "vitest"; +import { mock } from "vitest-mock-extended"; + +import type { CreateLogger } from "@src/core"; +import type { + WorkloadProbeEvidenceOutput, + WorkloadProbeEvidenceRepository +} from "@src/workload-abuse/repositories/workload-probe-evidence/workload-probe-evidence.repository"; +import type { WorkloadAbuseConfigService } from "@src/workload-abuse/services/workload-abuse-config/workload-abuse-config.service"; +import { BehaviouralSignalReplayService } from "./behavioural-signal-replay.service"; + +import { mockConfigService } from "@test/mocks/config-service.mock"; + +describe(BehaviouralSignalReplayService.name, () => { + it("flags the accelerator-bound shape that carries nothing on disk and talks to nobody", async () => { + const { service } = setup(); + + const summary = await service.replay({ bundledFixtures: true }); + + expect(summary.deployments).toContainEqual({ + source: "fixture", + label: "accelerator-bound-no-artifacts", + probes: 3, + accelFires: 3, + networkFires: 3, + candidateProbes: 3, + longestAgreement: 3 + }); + }); + + it("leaves a workload that serves traffic from large local files alone", async () => { + const { service } = setup(); + + const summary = await service.replay({ bundledFixtures: true }); + const inference = summary.deployments.find(deployment => deployment.label === "inference-with-weights"); + + expect(inference).toMatchObject({ accelFires: 0, networkFires: 0, candidateProbes: 0 }); + }); + + it("records a single signal for a quiet host whose only connection is a first-party relay", async () => { + const { service } = setup({ relayEndpoints: ["10.0.0.7"] }); + + const summary = await service.replay({ bundledFixtures: true }); + const bastion = summary.deployments.find(deployment => deployment.label === "idle-shell-host"); + + expect(bastion).toMatchObject({ accelFires: 0, networkFires: 2, candidateProbes: 0 }); + }); + + it("stops short of a candidate when a packaged runtime fills the disk", async () => { + const { service } = setup(); + + const summary = await service.replay({ bundledFixtures: true }); + const packaged = summary.deployments.find(deployment => deployment.label === "packaged-runtime"); + + expect(packaged).toMatchObject({ accelFires: 0, networkFires: 2, candidateProbes: 0 }); + }); + + it("counts how many deployments each agreement length would reach", async () => { + const { service } = setup(); + + const summary = await service.replay({ bundledFixtures: true }); + + expect(summary.wouldEnforce).toEqual([ + { agreementProbes: 1, deployments: 1 }, + { agreementProbes: 2, deployments: 1 }, + { agreementProbes: 3, deployments: 1 }, + { agreementProbes: 5, deployments: 0 } + ]); + }); + + it("reports how the outcome moves when the thresholds move", async () => { + const { service } = setup(); + + const summary = await service.replay({ bundledFixtures: true }); + + expect(summary.sensitivity).toContainEqual({ accelMinVramMb: 1_024, artifactMinMb: 256, candidateDeployments: 1 }); + expect(summary.sensitivity).toContainEqual({ accelMinVramMb: 1_024, artifactMinMb: 512, candidateDeployments: 1 }); + expect(summary.sensitivity).toContainEqual({ accelMinVramMb: 2_048, artifactMinMb: 128, candidateDeployments: 1 }); + }); + + it("groups recorded evidence by deployment and service", async () => { + const { service, evidenceRepository } = setup({ + rows: [ + createRow({ id: "row-1", service: "web" }), + createRow({ id: "row-2", service: "web" }), + createRow({ id: "row-3", service: "sidecar", accelerator: null }) + ] + }); + + const summary = await service.replay({ since: new Date("2026-08-01T00:00:00.000Z"), until: new Date("2026-09-01T00:00:00.000Z") }); + + expect(evidenceRepository.findCreatedBetween).toHaveBeenCalledWith({ + since: new Date("2026-08-01T00:00:00.000Z"), + until: new Date("2026-09-01T00:00:00.000Z") + }); + expect(summary.deployments).toEqual([ + { source: "database", label: "42/1000001/web", probes: 2, accelFires: 2, networkFires: 2, candidateProbes: 2, longestAgreement: 2 }, + { source: "database", label: "42/1000001/sidecar", probes: 1, accelFires: 0, networkFires: 1, candidateProbes: 0, longestAgreement: 0 } + ]); + }); + + it("reads the window from the request and writes nothing back", async () => { + const { service, evidenceRepository } = setup(); + + const summary = await service.replay({ since: new Date("2026-08-01T00:00:00.000Z"), until: new Date("2026-09-01T00:00:00.000Z") }); + + expect(summary.window).toEqual({ since: "2026-08-01T00:00:00.000Z", until: "2026-09-01T00:00:00.000Z" }); + expect(evidenceRepository.recordBehaviouralFindings).not.toHaveBeenCalled(); + expect(evidenceRepository.insertMany).not.toHaveBeenCalled(); + expect(evidenceRepository.deleteOlderThan).not.toHaveBeenCalled(); + }); + + it("applies the thresholds the operator asked for instead of the configured ones", async () => { + const { service } = setup({ rows: [createRow({ id: "row-1", service: "web" })] }); + + const summary = await service.replay({ since: new Date("2026-08-01T00:00:00.000Z"), accelMinVramMb: 20_000 }); + + expect(summary.params).toMatchObject({ accelMinVramMb: 20_000, artifactMinMb: 256 }); + expect(summary.deployments[0]).toMatchObject({ accelFires: 0 }); + }); + + function createRow(overrides: Partial) { + return mock({ + walletId: 42, + dseq: "1000001", + accelerator: [{ name: "accelerator-0", utilPct: 99, memUsedMb: 20_480, memTotalMb: 24_576, processes: [{ pid: 1234, name: "worker", vramMb: 18_000 }] }], + artifacts: [{ path: "/opt/worker", sizeBytes: 4_194_304 }], + netShape: { listenPorts: [22], established: [] }, + ...overrides + }); + } + + function setup(input?: { rows?: WorkloadProbeEvidenceOutput[]; relayEndpoints?: string[] }) { + const evidenceRepository = mock(); + evidenceRepository.findCreatedBetween.mockResolvedValue(input?.rows ?? []); + const config = mockConfigService({ + WORKLOAD_ABUSE_SIGNAL_ACCEL_MIN_VRAM_MB: 1_024, + WORKLOAD_ABUSE_SIGNAL_ARTIFACT_MIN_MB: 256, + WORKLOAD_ABUSE_SIGNAL_RELAY_ENDPOINTS: input?.relayEndpoints ?? [] + }); + const logger = mock>(); + const createLogger = vi.fn(() => logger); + const service = new BehaviouralSignalReplayService(evidenceRepository, config, createLogger); + + return { service, evidenceRepository, logger }; + } +}); diff --git a/apps/api/src/workload-abuse/services/behavioural-signal-replay/behavioural-signal-replay.service.ts b/apps/api/src/workload-abuse/services/behavioural-signal-replay/behavioural-signal-replay.service.ts new file mode 100644 index 0000000000..14478a08c8 --- /dev/null +++ b/apps/api/src/workload-abuse/services/behavioural-signal-replay/behavioural-signal-replay.service.ts @@ -0,0 +1,193 @@ +import { readdir, readFile } from "node:fs/promises"; +import { join } from "node:path"; +import { inject, singleton } from "tsyringe"; + +import { type CreateLogger, LOGGER_FACTORY } from "@src/core"; +import { evaluateBehaviouralSignals, isBehaviouralCandidate } from "@src/workload-abuse/lib/behavioural-signals/evaluate-behavioural-signals"; +import { BEHAVIOURAL_SIGNALS, type BehaviouralSignalParams, type ProbeEvidenceSnapshot } from "@src/workload-abuse/lib/behavioural-signals/types"; +import { WorkloadProbeEvidenceRepository } from "@src/workload-abuse/repositories/workload-probe-evidence/workload-probe-evidence.repository"; +import { WorkloadAbuseConfigService } from "@src/workload-abuse/services/workload-abuse-config/workload-abuse-config.service"; +import { type BehaviouralReplayFixture, BUNDLED_REPLAY_FIXTURES } from "./fixtures"; + +const DEFAULT_WINDOW_DAYS = 30; +const AGREEMENT_SWEEP = [1, 2, 3, 5]; +const THRESHOLD_SWEEP_FACTORS = [0.5, 1, 2]; + +export type BehaviouralReplayOptions = { + since?: Date; + until?: Date; + fixturePaths?: string[]; + bundledFixtures?: boolean; + accelMinVramMb?: number; + artifactMinMb?: number; +}; + +export type BehaviouralReplayDeployment = { + source: "database" | "fixture"; + label: string; + probes: number; + accelFires: number; + networkFires: number; + candidateProbes: number; + longestAgreement: number; +}; + +export type BehaviouralReplaySummary = { + params: BehaviouralSignalParams; + window: { since: string; until: string } | null; + deployments: BehaviouralReplayDeployment[]; + wouldEnforce: Array<{ agreementProbes: number; deployments: number }>; + sensitivity: Array<{ accelMinVramMb: number; artifactMinMb: number; candidateDeployments: number }>; +}; + +type SnapshotSeries = { source: "database" | "fixture"; label: string; snapshots: ProbeEvidenceSnapshot[] }; + +/** Reads recorded evidence and never writes, so an operator can re-score a window as often as they like. */ +@singleton() +export class BehaviouralSignalReplayService { + private readonly logger: ReturnType; + + constructor( + private readonly evidenceRepository: WorkloadProbeEvidenceRepository, + private readonly config: WorkloadAbuseConfigService, + @inject(LOGGER_FACTORY) createLogger: CreateLogger + ) { + this.logger = createLogger({ context: BehaviouralSignalReplayService.name }); + } + + async replay(options: BehaviouralReplayOptions = {}): Promise { + const params = this.#readParams(options); + const fixtureSeries = await this.#loadFixtures(options); + const usesDatabase = fixtureSeries.length === 0 || Boolean(options.since); + const window = usesDatabase ? this.#resolveWindow(options) : null; + const databaseSeries = window ? await this.#loadFromDatabase(window) : []; + const series = [...databaseSeries, ...fixtureSeries]; + + const deployments = series.map(entry => this.#summarise(entry, params)); + + for (const deployment of deployments) { + this.logger.info({ event: "BEHAVIOURAL_REPLAY_DEPLOYMENT", ...deployment }); + } + + const summary: BehaviouralReplaySummary = { + params, + window: window ? { since: window.since.toISOString(), until: window.until.toISOString() } : null, + deployments, + wouldEnforce: AGREEMENT_SWEEP.map(agreementProbes => ({ + agreementProbes, + deployments: deployments.filter(deployment => deployment.longestAgreement >= agreementProbes).length + })), + sensitivity: this.#sweepThresholds(series, params) + }; + + this.logger.info({ + event: "BEHAVIOURAL_REPLAY_COMPLETED", + window: summary.window, + deployments: deployments.length, + probes: deployments.reduce((total, deployment) => total + deployment.probes, 0), + candidateDeployments: deployments.filter(deployment => deployment.candidateProbes > 0).length, + wouldEnforce: summary.wouldEnforce + }); + + return summary; + } + + #readParams(options: BehaviouralReplayOptions): BehaviouralSignalParams { + return { + accelMinVramMb: options.accelMinVramMb ?? this.config.get("WORKLOAD_ABUSE_SIGNAL_ACCEL_MIN_VRAM_MB"), + artifactMinMb: options.artifactMinMb ?? this.config.get("WORKLOAD_ABUSE_SIGNAL_ARTIFACT_MIN_MB"), + relayEndpoints: this.config.get("WORKLOAD_ABUSE_SIGNAL_RELAY_ENDPOINTS") + }; + } + + #resolveWindow(options: BehaviouralReplayOptions): { since: Date; until: Date } { + const until = options.until ?? new Date(); + const since = options.since ?? new Date(until.getTime() - DEFAULT_WINDOW_DAYS * 24 * 60 * 60 * 1000); + + return { since, until }; + } + + async #loadFromDatabase(window: { since: Date; until: Date }): Promise { + const rows = await this.evidenceRepository.findCreatedBetween(window); + const byDeployment = new Map(); + + for (const row of rows) { + const label = `${row.walletId}/${row.dseq}/${row.service}`; + const series = byDeployment.get(label) ?? { source: "database" as const, label, snapshots: [] }; + series.snapshots.push(row); + byDeployment.set(label, series); + } + + return [...byDeployment.values()]; + } + + async #loadFixtures(options: BehaviouralReplayOptions): Promise { + const bundled = options.bundledFixtures ? BUNDLED_REPLAY_FIXTURES : []; + const loaded: BehaviouralReplayFixture[] = []; + + for (const path of options.fixturePaths ?? []) { + loaded.push(...(await readFixtures(path))); + } + + return [...bundled, ...loaded].map(fixture => ({ source: "fixture" as const, label: fixture.deployment, snapshots: fixture.snapshots })); + } + + #summarise(series: SnapshotSeries, params: BehaviouralSignalParams): BehaviouralReplayDeployment { + const findingsPerSnapshot = series.snapshots.map(snapshot => evaluateBehaviouralSignals(snapshot, params)); + const signalsPerSnapshot = findingsPerSnapshot.map(findings => new Set(findings.map(finding => finding.signal))); + + return { + source: series.source, + label: series.label, + probes: series.snapshots.length, + accelFires: signalsPerSnapshot.filter(signals => signals.has(BEHAVIOURAL_SIGNALS.accelWithoutArtifacts)).length, + networkFires: signalsPerSnapshot.filter(signals => signals.has(BEHAVIOURAL_SIGNALS.networkIsolated)).length, + candidateProbes: findingsPerSnapshot.filter(isBehaviouralCandidate).length, + longestAgreement: findLongestAgreement(findingsPerSnapshot.map(isBehaviouralCandidate)) + }; + } + + #sweepThresholds(series: SnapshotSeries[], params: BehaviouralSignalParams): BehaviouralReplaySummary["sensitivity"] { + return THRESHOLD_SWEEP_FACTORS.flatMap(accelFactor => + THRESHOLD_SWEEP_FACTORS.map(artifactFactor => { + const swept: BehaviouralSignalParams = { + ...params, + accelMinVramMb: Math.round(params.accelMinVramMb * accelFactor), + artifactMinMb: Math.round(params.artifactMinMb * artifactFactor) + }; + + return { + accelMinVramMb: swept.accelMinVramMb, + artifactMinMb: swept.artifactMinMb, + candidateDeployments: series.filter(entry => this.#summarise(entry, swept).candidateProbes > 0).length + }; + }) + ); + } +} + +async function readFixtures(path: string): Promise { + const entries = await readdir(path).catch(() => null); + + if (!entries) return [await readFixtureFile(path)]; + + const files = entries.filter(entry => entry.endsWith(".json")); + + return await Promise.all(files.map(file => readFixtureFile(join(path, file)))); +} + +async function readFixtureFile(path: string): Promise { + return JSON.parse(await readFile(path, "utf8")) as BehaviouralReplayFixture; +} + +function findLongestAgreement(candidates: boolean[]): number { + let longest = 0; + let current = 0; + + for (const candidate of candidates) { + current = candidate ? current + 1 : 0; + longest = Math.max(longest, current); + } + + return longest; +} diff --git a/apps/api/src/workload-abuse/services/behavioural-signal-replay/fixtures/accelerator-bound-no-artifacts.json b/apps/api/src/workload-abuse/services/behavioural-signal-replay/fixtures/accelerator-bound-no-artifacts.json new file mode 100644 index 0000000000..bb6cd6a7d5 --- /dev/null +++ b/apps/api/src/workload-abuse/services/behavioural-signal-replay/fixtures/accelerator-bound-no-artifacts.json @@ -0,0 +1,44 @@ +{ + "deployment": "accelerator-bound-no-artifacts", + "snapshots": [ + { + "accelerator": [ + { + "name": "accelerator-0", + "utilPct": 99, + "memUsedMb": 20480, + "memTotalMb": 24576, + "processes": [{ "pid": 1234, "name": "worker", "vramMb": 18000 }] + } + ], + "artifacts": [{ "path": "/opt/worker", "sizeBytes": 6291456 }], + "netShape": { "listenPorts": [22], "established": [] } + }, + { + "accelerator": [ + { + "name": "accelerator-0", + "utilPct": 98, + "memUsedMb": 20480, + "memTotalMb": 24576, + "processes": [{ "pid": 1234, "name": "worker", "vramMb": 18000 }] + } + ], + "artifacts": [{ "path": "/opt/worker", "sizeBytes": 6291456 }], + "netShape": { "listenPorts": [22], "established": [] } + }, + { + "accelerator": [ + { + "name": "accelerator-0", + "utilPct": 97, + "memUsedMb": 20480, + "memTotalMb": 24576, + "processes": [{ "pid": 1234, "name": "worker", "vramMb": 17800 }] + } + ], + "artifacts": [{ "path": "/opt/worker", "sizeBytes": 6291456 }], + "netShape": { "listenPorts": [22], "established": [] } + } + ] +} diff --git a/apps/api/src/workload-abuse/services/behavioural-signal-replay/fixtures/cpu-only-workload.json b/apps/api/src/workload-abuse/services/behavioural-signal-replay/fixtures/cpu-only-workload.json new file mode 100644 index 0000000000..bb1e3b0eb4 --- /dev/null +++ b/apps/api/src/workload-abuse/services/behavioural-signal-replay/fixtures/cpu-only-workload.json @@ -0,0 +1,13 @@ +{ + "deployment": "cpu-only-workload", + "snapshots": [ + { + "accelerator": null, + "artifacts": [{ "path": "/srv/app", "sizeBytes": 10485760 }], + "netShape": { + "listenPorts": [3000], + "established": [{ "localPort": 46000, "remoteIp": "198.51.100.10", "remotePort": 443, "count": 2 }] + } + } + ] +} diff --git a/apps/api/src/workload-abuse/services/behavioural-signal-replay/fixtures/idle-shell-host.json b/apps/api/src/workload-abuse/services/behavioural-signal-replay/fixtures/idle-shell-host.json new file mode 100644 index 0000000000..1dc6705cd8 --- /dev/null +++ b/apps/api/src/workload-abuse/services/behavioural-signal-replay/fixtures/idle-shell-host.json @@ -0,0 +1,15 @@ +{ + "deployment": "idle-shell-host", + "snapshots": [ + { + "accelerator": null, + "artifacts": [{ "path": "/usr/bin/toolchain", "sizeBytes": 20971520 }], + "netShape": { "listenPorts": [22], "established": [{ "localPort": 40100, "remoteIp": "10.0.0.7", "remotePort": 443, "count": 1 }] } + }, + { + "accelerator": null, + "artifacts": [{ "path": "/usr/bin/toolchain", "sizeBytes": 20971520 }], + "netShape": { "listenPorts": [22], "established": [{ "localPort": 40101, "remoteIp": "10.0.0.7", "remotePort": 443, "count": 1 }] } + } + ] +} diff --git a/apps/api/src/workload-abuse/services/behavioural-signal-replay/fixtures/index.ts b/apps/api/src/workload-abuse/services/behavioural-signal-replay/fixtures/index.ts new file mode 100644 index 0000000000..4db5e2dd79 --- /dev/null +++ b/apps/api/src/workload-abuse/services/behavioural-signal-replay/fixtures/index.ts @@ -0,0 +1,16 @@ +import type { ProbeEvidenceSnapshot } from "@src/workload-abuse/lib/behavioural-signals/types"; +import acceleratorBoundNoArtifacts from "./accelerator-bound-no-artifacts.json"; +import cpuOnlyWorkload from "./cpu-only-workload.json"; +import idleShellHost from "./idle-shell-host.json"; +import inferenceWithWeights from "./inference-with-weights.json"; +import packagedRuntime from "./packaged-runtime.json"; + +export type BehaviouralReplayFixture = { deployment: string; snapshots: ProbeEvidenceSnapshot[] }; + +export const BUNDLED_REPLAY_FIXTURES: BehaviouralReplayFixture[] = [ + acceleratorBoundNoArtifacts, + inferenceWithWeights, + idleShellHost, + packagedRuntime, + cpuOnlyWorkload +]; diff --git a/apps/api/src/workload-abuse/services/behavioural-signal-replay/fixtures/inference-with-weights.json b/apps/api/src/workload-abuse/services/behavioural-signal-replay/fixtures/inference-with-weights.json new file mode 100644 index 0000000000..9aaa908917 --- /dev/null +++ b/apps/api/src/workload-abuse/services/behavioural-signal-replay/fixtures/inference-with-weights.json @@ -0,0 +1,40 @@ +{ + "deployment": "inference-with-weights", + "snapshots": [ + { + "accelerator": [ + { + "name": "accelerator-0", + "utilPct": 64, + "memUsedMb": 14000, + "memTotalMb": 24576, + "processes": [{ "pid": 2001, "name": "runtime", "vramMb": 13800 }] + } + ], + "artifacts": [ + { "path": "/models/a.bin", "sizeBytes": 8589934592 }, + { "path": "/models/b.bin", "sizeBytes": 4294967296 } + ], + "netShape": { + "listenPorts": [8080], + "established": [{ "localPort": 8080, "remoteIp": "203.0.113.20", "remotePort": 54321, "count": 4 }] + } + }, + { + "accelerator": [ + { + "name": "accelerator-0", + "utilPct": 71, + "memUsedMb": 14200, + "memTotalMb": 24576, + "processes": [{ "pid": 2001, "name": "runtime", "vramMb": 13900 }] + } + ], + "artifacts": [{ "path": "/models/a.bin", "sizeBytes": 8589934592 }], + "netShape": { + "listenPorts": [8080], + "established": [{ "localPort": 8080, "remoteIp": "203.0.113.21", "remotePort": 51222, "count": 2 }] + } + } + ] +} diff --git a/apps/api/src/workload-abuse/services/behavioural-signal-replay/fixtures/packaged-runtime.json b/apps/api/src/workload-abuse/services/behavioural-signal-replay/fixtures/packaged-runtime.json new file mode 100644 index 0000000000..f0307858d8 --- /dev/null +++ b/apps/api/src/workload-abuse/services/behavioural-signal-replay/fixtures/packaged-runtime.json @@ -0,0 +1,31 @@ +{ + "deployment": "packaged-runtime", + "snapshots": [ + { + "accelerator": [ + { + "name": "accelerator-0", + "utilPct": 96, + "memUsedMb": 19000, + "memTotalMb": 24576, + "processes": [{ "pid": 3100, "name": "runtime", "vramMb": 17000 }] + } + ], + "artifacts": [{ "path": "/usr/lib/runtime/packages", "sizeBytes": 3221225472 }], + "netShape": { "listenPorts": [22], "established": [] } + }, + { + "accelerator": [ + { + "name": "accelerator-0", + "utilPct": 95, + "memUsedMb": 19000, + "memTotalMb": 24576, + "processes": [{ "pid": 3100, "name": "runtime", "vramMb": 17000 }] + } + ], + "artifacts": [{ "path": "/usr/lib/runtime/packages", "sizeBytes": 3221225472 }], + "netShape": { "listenPorts": [22], "established": [] } + } + ] +} diff --git a/apps/api/src/workload-abuse/services/probe-evidence/probe-evidence.service.spec.ts b/apps/api/src/workload-abuse/services/probe-evidence/probe-evidence.service.spec.ts index be5e213d6e..bc47cf4908 100644 --- a/apps/api/src/workload-abuse/services/probe-evidence/probe-evidence.service.spec.ts +++ b/apps/api/src/workload-abuse/services/probe-evidence/probe-evidence.service.spec.ts @@ -2,7 +2,10 @@ import { describe, expect, it, vi } from "vitest"; import { mock } from "vitest-mock-extended"; import { type CreateLogger } from "@src/core"; -import type { WorkloadProbeEvidenceRepository } from "@src/workload-abuse/repositories/workload-probe-evidence/workload-probe-evidence.repository"; +import type { + WorkloadProbeEvidenceOutput, + WorkloadProbeEvidenceRepository +} from "@src/workload-abuse/repositories/workload-probe-evidence/workload-probe-evidence.repository"; import type { WorkloadAbuseConfigService } from "@src/workload-abuse/services/workload-abuse-config/workload-abuse-config.service"; import type { WorkloadAbuseInstrumentationService } from "@src/workload-abuse/services/workload-abuse-instrumentation/workload-abuse-instrumentation.service"; import { ProbeEvidenceService } from "./probe-evidence.service"; @@ -84,6 +87,54 @@ describe(ProbeEvidenceService.name, () => { expect(logger.warn).toHaveBeenCalledWith(expect.objectContaining({ event: "WORKLOAD_EVIDENCE_WRITE_FAILED", walletId: 42, dseq: "1000001" })); }); + it("records the signals that fire on a stored row", async () => { + const { service, evidenceRepository, instrumentation } = setup({ signalsEnabled: true }); + + const recorded = await service.recordBehaviouralFindings([createEvidenceRow()]); + + expect(evidenceRepository.recordBehaviouralFindings).toHaveBeenCalledWith({ + id: "evidence-1", + findings: [ + { signal: "accel_without_artifacts", detail: { heaviestVramMb: 18_000, largestArtifactMb: 4 } }, + { signal: "network_isolated", detail: { excludedRelay: 0, listenPorts: 1 } } + ] + }); + expect(recorded).toEqual([{ evidenceId: "evidence-1", service: "web", findings: expect.any(Array) }]); + expect(instrumentation.recordBehaviouralFinding).toHaveBeenCalledTimes(2); + }); + + it("evaluates nothing while behavioural signals are disabled", async () => { + const { service, evidenceRepository } = setup({ signalsEnabled: false }); + + await expect(service.recordBehaviouralFindings([createEvidenceRow()])).resolves.toEqual([]); + + expect(evidenceRepository.recordBehaviouralFindings).not.toHaveBeenCalled(); + }); + + it("leaves a row alone when no signal fires", async () => { + const { service, evidenceRepository } = setup({ signalsEnabled: true }); + + const recorded = await service.recordBehaviouralFindings([ + createEvidenceRow({ + accelerator: null, + netShape: { listenPorts: [8_080], established: [{ localPort: 8_080, remoteIp: "203.0.113.5", remotePort: 51_000, count: 1 }] } + }) + ]); + + expect(recorded).toEqual([]); + expect(evidenceRepository.recordBehaviouralFindings).not.toHaveBeenCalled(); + }); + + it("logs and counts a findings write failure without rethrowing", async () => { + const { service, evidenceRepository, instrumentation, logger } = setup({ signalsEnabled: true }); + evidenceRepository.recordBehaviouralFindings.mockRejectedValue(new Error("connection refused")); + + await expect(service.recordBehaviouralFindings([createEvidenceRow()])).resolves.toEqual([]); + + expect(instrumentation.recordEvidenceWriteFailure).toHaveBeenCalledTimes(1); + expect(logger.warn).toHaveBeenCalledWith(expect.objectContaining({ event: "WORKLOAD_EVIDENCE_FINDINGS_WRITE_FAILED", walletId: 42, dseq: "1000001" })); + }); + it("purges rows older than the retention window", async () => { const { service, evidenceRepository } = setup(); @@ -102,12 +153,31 @@ describe(ProbeEvidenceService.name, () => { expect(logger.warn).toHaveBeenCalledWith(expect.objectContaining({ event: "WORKLOAD_EVIDENCE_PURGE_FAILED" })); }); - function setup(input?: { retentionDays?: number }) { + function createEvidenceRow(overrides: Partial = {}) { + return mock({ + id: "evidence-1", + walletId: 42, + dseq: "1000001", + service: "web", + accelerator: [{ name: "accelerator-0", utilPct: 99, memUsedMb: 20_480, memTotalMb: 24_576, processes: [{ pid: 1234, name: "worker", vramMb: 18_000 }] }], + artifacts: [{ path: "/opt/worker", sizeBytes: 4_194_304 }], + netShape: { listenPorts: [22], established: [] }, + ...overrides + }); + } + + function setup(input?: { retentionDays?: number; signalsEnabled?: boolean }) { const evidenceRepository = mock(); const instrumentation = mock(); const logger = mock>(); const createLogger = vi.fn(() => logger); - const config = mockConfigService({ WORKLOAD_ABUSE_EVIDENCE_RETENTION_DAYS: input?.retentionDays ?? 90 }); + const config = mockConfigService({ + WORKLOAD_ABUSE_EVIDENCE_RETENTION_DAYS: input?.retentionDays ?? 90, + WORKLOAD_ABUSE_BEHAVIOURAL_SIGNALS_ENABLED: input?.signalsEnabled ?? false, + WORKLOAD_ABUSE_SIGNAL_ACCEL_MIN_VRAM_MB: 1_024, + WORKLOAD_ABUSE_SIGNAL_ARTIFACT_MIN_MB: 256, + WORKLOAD_ABUSE_SIGNAL_RELAY_ENDPOINTS: [] + }); const service = new ProbeEvidenceService(evidenceRepository, instrumentation, config, createLogger); return { service, evidenceRepository, instrumentation, logger, config }; diff --git a/apps/api/src/workload-abuse/services/probe-evidence/probe-evidence.service.ts b/apps/api/src/workload-abuse/services/probe-evidence/probe-evidence.service.ts index c268079dab..c8e4ed76b6 100644 --- a/apps/api/src/workload-abuse/services/probe-evidence/probe-evidence.service.ts +++ b/apps/api/src/workload-abuse/services/probe-evidence/probe-evidence.service.ts @@ -1,6 +1,8 @@ import { inject, singleton } from "tsyringe"; import { type CreateLogger, LOGGER_FACTORY } from "@src/core"; +import { evaluateBehaviouralSignals } from "@src/workload-abuse/lib/behavioural-signals/evaluate-behavioural-signals"; +import type { BehaviouralFinding, BehaviouralSignalParams } from "@src/workload-abuse/lib/behavioural-signals/types"; import { parseProbeEvidence } from "@src/workload-abuse/lib/probe-evidence/parse-probe-evidence"; import { type WorkloadProbeEvidenceOutput, @@ -10,6 +12,8 @@ import type { ShellEvidence } from "@src/workload-abuse/services/trial-workload- import { WorkloadAbuseConfigService } from "@src/workload-abuse/services/workload-abuse-config/workload-abuse-config.service"; import { WorkloadAbuseInstrumentationService } from "@src/workload-abuse/services/workload-abuse-instrumentation/workload-abuse-instrumentation.service"; +export type RecordedBehaviouralFindings = { evidenceId: string; service: string; findings: BehaviouralFinding[] }; + @singleton() export class ProbeEvidenceService { private readonly logger: ReturnType; @@ -58,6 +62,41 @@ export class ProbeEvidenceService { } } + /** Signals read the stored row rather than the raw output, so a replay over recorded evidence sees exactly what the live probe saw. */ + async recordBehaviouralFindings(rows: WorkloadProbeEvidenceOutput[]): Promise { + if (!this.config.get("WORKLOAD_ABUSE_BEHAVIOURAL_SIGNALS_ENABLED")) return []; + + const params = this.#readSignalParams(); + const recorded: RecordedBehaviouralFindings[] = []; + + for (const row of rows) { + const findings = evaluateBehaviouralSignals(row, params); + + if (!findings.length) continue; + + try { + await this.evidenceRepository.recordBehaviouralFindings({ id: row.id, findings }); + } catch (error) { + this.instrumentation.recordEvidenceWriteFailure("findings"); + this.logger.warn({ event: "WORKLOAD_EVIDENCE_FINDINGS_WRITE_FAILED", error, walletId: row.walletId, dseq: row.dseq }); + continue; + } + + for (const finding of findings) this.instrumentation.recordBehaviouralFinding(finding.signal); + recorded.push({ evidenceId: row.id, service: row.service, findings }); + } + + return recorded; + } + + #readSignalParams(): BehaviouralSignalParams { + return { + accelMinVramMb: this.config.get("WORKLOAD_ABUSE_SIGNAL_ACCEL_MIN_VRAM_MB"), + artifactMinMb: this.config.get("WORKLOAD_ABUSE_SIGNAL_ARTIFACT_MIN_MB"), + relayEndpoints: this.config.get("WORKLOAD_ABUSE_SIGNAL_RELAY_ENDPOINTS") + }; + } + async purgeExpired(): Promise { const retentionDays = this.config.get("WORKLOAD_ABUSE_EVIDENCE_RETENTION_DAYS"); const before = new Date(Date.now() - retentionDays * 24 * 60 * 60 * 1000); diff --git a/apps/api/src/workload-abuse/services/probe-trial-deployment/probe-trial-deployment.handler.integration.ts b/apps/api/src/workload-abuse/services/probe-trial-deployment/probe-trial-deployment.handler.integration.ts index 0a1164ada9..dd46ba05e8 100644 --- a/apps/api/src/workload-abuse/services/probe-trial-deployment/probe-trial-deployment.handler.integration.ts +++ b/apps/api/src/workload-abuse/services/probe-trial-deployment/probe-trial-deployment.handler.integration.ts @@ -19,6 +19,16 @@ const MAX_ATTEMPTS = 30; const ACCELERATED_EVIDENCE = "--accel\nGPU-0001, NVIDIA A100, 95, 20480, 24576\nGPU-0001, 1234, python3, 18000"; +const ISOLATED_ACCELERATED_EVIDENCE = [ + "--accel", + "GPU-0001, NVIDIA A100, 95, 20480, 24576", + "GPU-0001, 1234, python3, 18000", + "--netl", + " 2 listen=22", + "--disk", + "4194304 /opt/worker" +].join("\n"); + const jobWorkers = useJobWorkers(() => [container.resolve(ProbeTrialDeploymentHandler)]); describe(ProbeTrialDeploymentHandler.name, () => { @@ -104,6 +114,34 @@ describe(ProbeTrialDeploymentHandler.name, () => { expect(await findEvidence()).toHaveLength(0); }); + it("records the behavioural signals on the row once the signals are enabled", async () => { + const { probeDeployment, findEvidence } = await setup({ + verdict: "clean", + behaviouralSignalsEnabled: true, + shellEvidence: [{ service: "ssh", provider: "akash1provider", status: "completed", evidence: ISOLATED_ACCELERATED_EVIDENCE }] + }); + + await probeDeployment(); + + const [evidence] = await findEvidence(); + expect(evidence.behaviouralFindings).toEqual([ + { signal: "accel_without_artifacts", detail: { heaviestVramMb: 18000, largestArtifactMb: 4 } }, + { signal: "network_isolated", detail: { excludedRelay: 0, listenPorts: 1 } } + ]); + }); + + it("leaves the row without findings while the signals are disabled", async () => { + const { probeDeployment, findEvidence } = await setup({ + verdict: "clean", + shellEvidence: [{ service: "ssh", provider: "akash1provider", status: "completed", evidence: ISOLATED_ACCELERATED_EVIDENCE }] + }); + + await probeDeployment(); + + const [evidence] = await findEvidence(); + expect(evidence.behaviouralFindings).toBeNull(); + }); + it("records nothing more for a deployment already judged abusive", async () => { const { probeDeployment, findDetections, probe, seedExistingDetection } = await setup({ verdict: "hard" }); await seedExistingDetection(); @@ -120,6 +158,7 @@ describe(ProbeTrialDeploymentHandler.name, () => { isTrialing?: boolean; probeStatus?: ProbeReport["probeStatus"]; shellEvidence?: ProbeReport["shellEvidence"]; + behaviouralSignalsEnabled?: boolean; }) { const { enqueue, startWorkers } = await jobWorkers(); const detectionRepository = container.resolve(WorkloadAbuseDetectionRepository); @@ -129,7 +168,11 @@ describe(ProbeTrialDeploymentHandler.name, () => { const config = container.resolve(WorkloadAbuseConfigService); const readConfig = config.get.bind(config); - vi.spyOn(config, "get").mockImplementation((key => (key === "WORKLOAD_ABUSE_PROBE_ENABLED" ? true : readConfig(key))) as typeof config.get); + const overrides: Record = { + WORKLOAD_ABUSE_PROBE_ENABLED: true, + WORKLOAD_ABUSE_BEHAVIOURAL_SIGNALS_ENABLED: input.behaviouralSignalsEnabled ?? false + }; + vi.spyOn(config, "get").mockImplementation((key => (key in overrides ? overrides[key] : readConfig(key))) as typeof config.get); const probe = vi.spyOn(container.resolve(TrialWorkloadProbeService), "probe").mockResolvedValue({ verdict: input.verdict, diff --git a/apps/api/src/workload-abuse/services/probe-trial-deployment/probe-trial-deployment.handler.spec.ts b/apps/api/src/workload-abuse/services/probe-trial-deployment/probe-trial-deployment.handler.spec.ts index d4a0554a7a..e8278f01b1 100644 --- a/apps/api/src/workload-abuse/services/probe-trial-deployment/probe-trial-deployment.handler.spec.ts +++ b/apps/api/src/workload-abuse/services/probe-trial-deployment/probe-trial-deployment.handler.spec.ts @@ -9,7 +9,7 @@ import type { } from "@src/workload-abuse/repositories/workload-abuse-detection/workload-abuse-detection.repository"; import type { WorkloadProbeEvidenceOutput } from "@src/workload-abuse/repositories/workload-probe-evidence/workload-probe-evidence.repository"; import { EnforceTrialAbuse } from "@src/workload-abuse/services/enforce-trial-abuse/enforce-trial-abuse.handler"; -import type { ProbeEvidenceService } from "@src/workload-abuse/services/probe-evidence/probe-evidence.service"; +import type { ProbeEvidenceService, RecordedBehaviouralFindings } from "@src/workload-abuse/services/probe-evidence/probe-evidence.service"; import type { ProbeReport, TrialWorkloadProbeService } from "@src/workload-abuse/services/trial-workload-probe/trial-workload-probe.service"; import type { TrialWorkloadProbeJobService } from "@src/workload-abuse/services/trial-workload-probe-job/trial-workload-probe-job.service"; import type { WorkloadAbuseConfigService } from "@src/workload-abuse/services/workload-abuse-config/workload-abuse-config.service"; @@ -109,6 +109,28 @@ describe(ProbeTrialDeploymentHandler.name, () => { expect(probeEvidenceService.recordEvidence).not.toHaveBeenCalled(); }); + it("hands the rows it just wrote to the behavioural signals", async () => { + const evidenceRows = [mock({ id: "evidence-1", service: "web" })]; + const { handler, probeEvidenceService } = setup({ report: createReport({ verdict: "clean" }), evidenceRows }); + + await handler.handle(PAYLOAD); + + expect(probeEvidenceService.recordBehaviouralFindings).toHaveBeenCalledWith(evidenceRows); + }); + + it("logs the signals recorded for each service they fired on", async () => { + const { handler, logger } = setup({ + report: createReport({ verdict: "clean" }), + behaviouralFindings: [{ evidenceId: "evidence-1", service: "web", findings: [{ signal: "network_isolated", detail: {} }] }] + }); + + await handler.handle(PAYLOAD); + + expect(logger.info).toHaveBeenCalledWith( + expect.objectContaining({ event: "TRIAL_WORKLOAD_BEHAVIOURAL_FINDING", service: "web", evidenceId: "evidence-1", signals: ["network_isolated"] }) + ); + }); + it("logs what the shell saw for every verdict, cut so the line survives log shipping", async () => { const { handler, logger } = setup({ report: createReport({ verdict: "clean", excerpt: "x".repeat(5_000) }) }); @@ -272,6 +294,8 @@ describe(ProbeTrialDeploymentHandler.name, () => { maxAttempts?: number; existingDetection?: boolean; enforcementMode?: "detect" | "enforce"; + evidenceRows?: WorkloadProbeEvidenceOutput[]; + behaviouralFindings?: RecordedBehaviouralFindings[]; }) { const wallet = input.wallet === undefined ? createUserWallet({ isTrialing: true }) : input.wallet; const userWalletRepository = mock(); @@ -283,9 +307,10 @@ describe(ProbeTrialDeploymentHandler.name, () => { detectionRepository.create.mockResolvedValue(mock({ id: "detection-1" })); detectionRepository.findOneBy.mockResolvedValue(input.existingDetection ? mock({ id: "detection-0" }) : undefined); const probeEvidenceService = mock(); - probeEvidenceService.recordEvidence.mockImplementation(async ({ shellEvidence }) => - shellEvidence.map(entry => mock({ service: entry.service })) + probeEvidenceService.recordEvidence.mockImplementation( + async ({ shellEvidence }) => input.evidenceRows ?? shellEvidence.map(entry => mock({ service: entry.service })) ); + probeEvidenceService.recordBehaviouralFindings.mockResolvedValue(input.behaviouralFindings ?? []); const instrumentation = mock(); const config = mockConfigService({ WORKLOAD_ABUSE_PROBE_ENABLED: input.enabled ?? true, diff --git a/apps/api/src/workload-abuse/services/probe-trial-deployment/probe-trial-deployment.handler.ts b/apps/api/src/workload-abuse/services/probe-trial-deployment/probe-trial-deployment.handler.ts index 3fb280a694..51cea2ad86 100644 --- a/apps/api/src/workload-abuse/services/probe-trial-deployment/probe-trial-deployment.handler.ts +++ b/apps/api/src/workload-abuse/services/probe-trial-deployment/probe-trial-deployment.handler.ts @@ -111,6 +111,19 @@ export class ProbeTrialDeploymentHandler implements JobHandler finding.signal) + }); + } + this.logger.info({ event: "TRIAL_WORKLOAD_PROBED", ...context, diff --git a/apps/api/src/workload-abuse/services/workload-abuse-instrumentation/workload-abuse-instrumentation.service.ts b/apps/api/src/workload-abuse/services/workload-abuse-instrumentation/workload-abuse-instrumentation.service.ts index 8de0d3b7d6..7d677cecd7 100644 --- a/apps/api/src/workload-abuse/services/workload-abuse-instrumentation/workload-abuse-instrumentation.service.ts +++ b/apps/api/src/workload-abuse/services/workload-abuse-instrumentation/workload-abuse-instrumentation.service.ts @@ -17,6 +17,7 @@ export class WorkloadAbuseInstrumentationService { private readonly blockedDomainLookupFailures: Counter; private readonly domainBlocks: Counter; private readonly evidenceWriteFailures: Counter; + private readonly behaviouralFindings: Counter; constructor(metricsService: MetricsService) { this.meter = metricsService.getMeter("workload-abuse", "1.0.0"); @@ -38,6 +39,9 @@ export class WorkloadAbuseInstrumentationService { this.evidenceWriteFailures = metricsService.createCounter(this.meter, "workload_abuse_evidence_write_failures_total", { description: "Probe evidence statements that failed to persist, by operation" }); + this.behaviouralFindings = metricsService.createCounter(this.meter, "workload_abuse_behavioural_findings_total", { + description: "Behavioural shape signals recorded on probe evidence, by signal" + }); } recordProbe(input: { verdict: WorkloadVerdict; probeStatus: string }): void { @@ -63,4 +67,8 @@ export class WorkloadAbuseInstrumentationService { recordEvidenceWriteFailure(operation: EvidenceWriteOperation): void { this.evidenceWriteFailures.add(1, { operation }); } + + recordBehaviouralFinding(signal: string): void { + this.behaviouralFindings.add(1, { signal }); + } } From 08a83ebc78468d369a05f0a6c7a9d11f8a011ac9 Mon Sep 17 00:00:00 2001 From: Maxime Beauchamp <15185355+baktun14@users.noreply.github.com> Date: Sun, 20 Sep 2026 16:03:17 +0400 Subject: [PATCH 2/5] fix(deployment): score behavioural signals only on evidence a shell finished collecting A section that stopped early looks exactly like a section with nothing in it, and both signals read an absence as the thing worth noticing. Snapshots now carry the status of the shell that produced them, and a collection that was cut short supports no signal, in the live path and in the replay. The replay honours a window given by its end alone, skips a fixture file it cannot parse instead of losing the whole run with it, and logs the threshold sweep it computes so an operator running the command actually sees it. Relay addresses configured in the short form of an IPv6 address now match the expanded form the probe decodes. --- .../accel-without-artifacts.spec.ts | 2 +- .../evaluate-behavioural-signals.spec.ts | 14 +++- .../evaluate-behavioural-signals.ts | 5 +- .../network-isolation.spec.ts | 11 ++- .../behavioural-signals/network-isolation.ts | 33 ++++++++- .../lib/behavioural-signals/types.ts | 4 ++ .../behavioural-signal-replay.service.spec.ts | 45 ++++++++++++ .../behavioural-signal-replay.service.ts | 44 ++++++++---- .../accelerator-bound-no-artifacts.json | 69 ++++++++++++++++--- .../fixtures/cpu-only-workload.json | 21 +++++- .../fixtures/idle-shell-host.json | 44 ++++++++++-- .../fixtures/index.ts | 27 +++++++- .../fixtures/inference-with-weights.json | 61 +++++++++++++--- .../fixtures/packaged-runtime.json | 46 +++++++++++-- .../probe-evidence.service.spec.ts | 11 ++- .../workload-abuse-instrumentation.service.ts | 2 +- 16 files changed, 385 insertions(+), 54 deletions(-) diff --git a/apps/api/src/workload-abuse/lib/behavioural-signals/accel-without-artifacts.spec.ts b/apps/api/src/workload-abuse/lib/behavioural-signals/accel-without-artifacts.spec.ts index 137bca35f5..12a7f9837f 100644 --- a/apps/api/src/workload-abuse/lib/behavioural-signals/accel-without-artifacts.spec.ts +++ b/apps/api/src/workload-abuse/lib/behavioural-signals/accel-without-artifacts.spec.ts @@ -77,7 +77,7 @@ describe("evaluateAccelWithoutArtifacts", () => { processes: [{ pid: 1234, name: "worker", vramMb: input.vramMb ?? 18_000 }] } ]; - const snapshot: ProbeEvidenceSnapshot = { accelerator, artifacts: input.artifacts ?? null, netShape: null }; + const snapshot: ProbeEvidenceSnapshot = { shellStatus: "completed", accelerator, artifacts: input.artifacts ?? null, netShape: null }; const params: BehaviouralSignalParams = { accelMinVramMb: input.accelMinVramMb ?? 1_024, artifactMinMb: input.artifactMinMb ?? 256, diff --git a/apps/api/src/workload-abuse/lib/behavioural-signals/evaluate-behavioural-signals.spec.ts b/apps/api/src/workload-abuse/lib/behavioural-signals/evaluate-behavioural-signals.spec.ts index f15e0c2339..b4e500fe6b 100644 --- a/apps/api/src/workload-abuse/lib/behavioural-signals/evaluate-behavioural-signals.spec.ts +++ b/apps/api/src/workload-abuse/lib/behavioural-signals/evaluate-behavioural-signals.spec.ts @@ -26,6 +26,12 @@ describe("evaluateBehaviouralSignals", () => { expect(evaluateBehaviouralSignals(snapshot, params)).toEqual([]); }); + it("returns nothing when the shell that collected the snapshot was cut short", () => { + const { snapshot, params } = setup({ vramMb: 18_000, artifactBytes: 1_024, established: [], shellStatus: "output_capped" }); + + expect(evaluateBehaviouralSignals(snapshot, params)).toEqual([]); + }); + it("treats the two signals together as a candidate", () => { const { snapshot, params } = setup({ vramMb: 18_000, artifactBytes: 1_024, established: [] }); @@ -38,8 +44,14 @@ describe("evaluateBehaviouralSignals", () => { expect(isBehaviouralCandidate(evaluateBehaviouralSignals(snapshot, params))).toBe(false); }); - function setup(input: { vramMb: number; artifactBytes: number; established: NonNullable["established"] }) { + function setup(input: { + vramMb: number; + artifactBytes: number; + established: NonNullable["established"]; + shellStatus?: string; + }) { const snapshot: ProbeEvidenceSnapshot = { + shellStatus: input.shellStatus ?? "completed", accelerator: [ { name: "accelerator-0", utilPct: 97, memUsedMb: 20_480, memTotalMb: 24_576, processes: [{ pid: 1234, name: "worker", vramMb: input.vramMb }] } ], diff --git a/apps/api/src/workload-abuse/lib/behavioural-signals/evaluate-behavioural-signals.ts b/apps/api/src/workload-abuse/lib/behavioural-signals/evaluate-behavioural-signals.ts index 7e78a52bf2..8ecdbc3864 100644 --- a/apps/api/src/workload-abuse/lib/behavioural-signals/evaluate-behavioural-signals.ts +++ b/apps/api/src/workload-abuse/lib/behavioural-signals/evaluate-behavioural-signals.ts @@ -1,8 +1,11 @@ import { evaluateAccelWithoutArtifacts } from "./accel-without-artifacts"; import { evaluateNetworkIsolation } from "./network-isolation"; -import { BEHAVIOURAL_SIGNALS, type BehaviouralFinding, type BehaviouralSignalParams, type ProbeEvidenceSnapshot } from "./types"; +import { BEHAVIOURAL_SIGNALS, type BehaviouralFinding, type BehaviouralSignalParams, COMPLETE_SHELL_STATUS, type ProbeEvidenceSnapshot } from "./types"; +/** A section that stopped early looks the same as a section with nothing in it, so a shell that was cut short supports no signal. */ export function evaluateBehaviouralSignals(snapshot: ProbeEvidenceSnapshot, params: BehaviouralSignalParams): BehaviouralFinding[] { + if (snapshot.shellStatus !== COMPLETE_SHELL_STATUS) return []; + const evaluated = [evaluateAccelWithoutArtifacts(snapshot, params), evaluateNetworkIsolation(snapshot, params)]; return evaluated.filter((finding): finding is BehaviouralFinding => finding !== null); diff --git a/apps/api/src/workload-abuse/lib/behavioural-signals/network-isolation.spec.ts b/apps/api/src/workload-abuse/lib/behavioural-signals/network-isolation.spec.ts index 1941453928..0a8fa623d6 100644 --- a/apps/api/src/workload-abuse/lib/behavioural-signals/network-isolation.spec.ts +++ b/apps/api/src/workload-abuse/lib/behavioural-signals/network-isolation.spec.ts @@ -41,6 +41,15 @@ describe("evaluateNetworkIsolation", () => { expect(evaluateNetworkIsolation(snapshot, params)).not.toBeNull(); }); + it("fires when the relay is configured in the short form of its IPv6 address", () => { + const { snapshot, params } = setup({ + netShape: { listenPorts: [], established: [{ localPort: 41_234, remoteIp: "2001:0db8:0000:0000:0000:0000:0000:0001", remotePort: 443, count: 1 }] }, + relayEndpoints: ["2001:db8::1"] + }); + + expect(evaluateNetworkIsolation(snapshot, params)).not.toBeNull(); + }); + it("stays silent when a connection lands on a listening port", () => { const { snapshot, params } = setup({ netShape: { listenPorts: [8080], established: [{ localPort: 8080, remoteIp: "203.0.113.9", remotePort: 51_000, count: 3 }] } @@ -65,7 +74,7 @@ describe("evaluateNetworkIsolation", () => { }); function setup(input: { netShape: ProbeEvidenceNetShape | null; relayEndpoints?: string[] }) { - const snapshot: ProbeEvidenceSnapshot = { accelerator: null, artifacts: null, netShape: input.netShape }; + const snapshot: ProbeEvidenceSnapshot = { shellStatus: "completed", accelerator: null, artifacts: null, netShape: input.netShape }; const params: BehaviouralSignalParams = { accelMinVramMb: 1_024, artifactMinMb: 256, relayEndpoints: input.relayEndpoints ?? [] }; return { snapshot, params }; diff --git a/apps/api/src/workload-abuse/lib/behavioural-signals/network-isolation.ts b/apps/api/src/workload-abuse/lib/behavioural-signals/network-isolation.ts index 0d7cab1817..3421aafec6 100644 --- a/apps/api/src/workload-abuse/lib/behavioural-signals/network-isolation.ts +++ b/apps/api/src/workload-abuse/lib/behavioural-signals/network-isolation.ts @@ -21,5 +21,36 @@ export function evaluateNetworkIsolation(snapshot: ProbeEvidenceSnapshot, params } function isRelayEndpoint(connection: EstablishedConnection, relayEndpoints: string[]): boolean { - return relayEndpoints.includes(connection.remoteIp) || relayEndpoints.includes(`${connection.remoteIp}:${connection.remotePort}`); + const configured = new Set(relayEndpoints.map(normalizeEndpoint)); + const host = normalizeHost(connection.remoteIp); + + return configured.has(host) || configured.has(`${host}/${connection.remotePort}`); +} + +function normalizeEndpoint(endpoint: string): string { + const bracketed = endpoint.trim().match(/^\[(.+)\]:(\d+)$/); + + if (bracketed) return `${normalizeHost(bracketed[1])}/${bracketed[2]}`; + + const withPort = endpoint.trim().match(/^([^:]+):(\d+)$/); + + if (withPort) return `${normalizeHost(withPort[1])}/${withPort[2]}`; + + return normalizeHost(endpoint); +} + +/** The probe decodes IPv6 from the kernel's fixed width hex, so an address configured in its compressed form has to be expanded to match one. */ +function normalizeHost(host: string): string { + const address = host.trim().toLowerCase(); + + if (!address.includes(":")) return address; + + const [head, tail = ""] = address.split("::"); + const headGroups = head ? head.split(":") : []; + const tailGroups = tail ? tail.split(":") : []; + const groups = address.includes("::") + ? [...headGroups, ...Array(Math.max(8 - headGroups.length - tailGroups.length, 0)).fill("0"), ...tailGroups] + : address.split(":"); + + return groups.map(group => group.padStart(4, "0")).join(":"); } diff --git a/apps/api/src/workload-abuse/lib/behavioural-signals/types.ts b/apps/api/src/workload-abuse/lib/behavioural-signals/types.ts index 6c2f4daf7c..d563ada21c 100644 --- a/apps/api/src/workload-abuse/lib/behavioural-signals/types.ts +++ b/apps/api/src/workload-abuse/lib/behavioural-signals/types.ts @@ -14,8 +14,12 @@ export type BehaviouralSignal = (typeof BEHAVIOURAL_SIGNALS)[keyof typeof BEHAVI export type BehaviouralFinding = ProbeEvidenceBehaviouralFinding; +/** Only a shell session that ran to the end proves an absence, and every signal reads one. */ +export const COMPLETE_SHELL_STATUS = "completed"; + /** The columns of one evidence row, so the replay and the live probe evaluate the exact same input. */ export type ProbeEvidenceSnapshot = { + shellStatus: string; accelerator: ProbeEvidenceAccelerator[] | null; artifacts: ProbeEvidenceArtifact[] | null; netShape: ProbeEvidenceNetShape | null; diff --git a/apps/api/src/workload-abuse/services/behavioural-signal-replay/behavioural-signal-replay.service.spec.ts b/apps/api/src/workload-abuse/services/behavioural-signal-replay/behavioural-signal-replay.service.spec.ts index 58b65a96e4..50360a5b02 100644 --- a/apps/api/src/workload-abuse/services/behavioural-signal-replay/behavioural-signal-replay.service.spec.ts +++ b/apps/api/src/workload-abuse/services/behavioural-signal-replay/behavioural-signal-replay.service.spec.ts @@ -1,3 +1,6 @@ +import { mkdtempSync, writeFileSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; import { describe, expect, it, vi } from "vitest"; import { mock } from "vitest-mock-extended"; @@ -99,6 +102,47 @@ describe(BehaviouralSignalReplayService.name, () => { ]); }); + it("leaves out evidence whose shell was cut short, since its empty sections prove nothing", async () => { + const { service } = setup({ + rows: [createRow({ id: "row-1", service: "web" }), createRow({ id: "row-2", service: "web", shellStatus: "output_capped" })] + }); + + const summary = await service.replay({ since: new Date("2026-08-01T00:00:00.000Z") }); + + expect(summary.deployments).toEqual([ + { source: "database", label: "42/1000001/web", probes: 1, accelFires: 1, networkFires: 1, candidateProbes: 1, longestAgreement: 1 } + ]); + }); + + it("scores the recorded window an operator asks for by its end alone", async () => { + const { service, evidenceRepository } = setup({ rows: [createRow({ id: "row-1", service: "web" })] }); + + const summary = await service.replay({ bundledFixtures: true, until: new Date("2026-09-01T00:00:00.000Z") }); + + expect(evidenceRepository.findCreatedBetween).toHaveBeenCalledWith({ since: expect.any(Date), until: new Date("2026-09-01T00:00:00.000Z") }); + expect(summary.deployments).toContainEqual(expect.objectContaining({ source: "database" })); + }); + + it("skips a fixture file it cannot read and reports every other one", async () => { + const { service, logger } = setup(); + const directory = mkdtempSync(join(tmpdir(), "behavioural-replay-")); + writeFileSync(join(directory, "broken.json"), '{ "deployment": "broken", "snapshots": [{ "netShape": {} }] }'); + + const summary = await service.replay({ bundledFixtures: true, fixturePaths: [directory] }); + + expect(summary.deployments.map(deployment => deployment.label)).not.toContain("broken"); + expect(summary.deployments).not.toHaveLength(0); + expect(logger.warn).toHaveBeenCalledWith(expect.objectContaining({ event: "BEHAVIOURAL_REPLAY_FIXTURE_SKIPPED" })); + }); + + it("logs the threshold sweep with the rest of the report, so a run leaves it behind", async () => { + const { service, logger } = setup(); + + const summary = await service.replay({ bundledFixtures: true }); + + expect(logger.info).toHaveBeenCalledWith(expect.objectContaining({ event: "BEHAVIOURAL_REPLAY_COMPLETED", sensitivity: summary.sensitivity })); + }); + it("reads the window from the request and writes nothing back", async () => { const { service, evidenceRepository } = setup(); @@ -123,6 +167,7 @@ describe(BehaviouralSignalReplayService.name, () => { return mock({ walletId: 42, dseq: "1000001", + shellStatus: "completed", accelerator: [{ name: "accelerator-0", utilPct: 99, memUsedMb: 20_480, memTotalMb: 24_576, processes: [{ pid: 1234, name: "worker", vramMb: 18_000 }] }], artifacts: [{ path: "/opt/worker", sizeBytes: 4_194_304 }], netShape: { listenPorts: [22], established: [] }, diff --git a/apps/api/src/workload-abuse/services/behavioural-signal-replay/behavioural-signal-replay.service.ts b/apps/api/src/workload-abuse/services/behavioural-signal-replay/behavioural-signal-replay.service.ts index 14478a08c8..4167965d54 100644 --- a/apps/api/src/workload-abuse/services/behavioural-signal-replay/behavioural-signal-replay.service.ts +++ b/apps/api/src/workload-abuse/services/behavioural-signal-replay/behavioural-signal-replay.service.ts @@ -4,10 +4,15 @@ import { inject, singleton } from "tsyringe"; import { type CreateLogger, LOGGER_FACTORY } from "@src/core"; import { evaluateBehaviouralSignals, isBehaviouralCandidate } from "@src/workload-abuse/lib/behavioural-signals/evaluate-behavioural-signals"; -import { BEHAVIOURAL_SIGNALS, type BehaviouralSignalParams, type ProbeEvidenceSnapshot } from "@src/workload-abuse/lib/behavioural-signals/types"; +import { + BEHAVIOURAL_SIGNALS, + type BehaviouralSignalParams, + COMPLETE_SHELL_STATUS, + type ProbeEvidenceSnapshot +} from "@src/workload-abuse/lib/behavioural-signals/types"; import { WorkloadProbeEvidenceRepository } from "@src/workload-abuse/repositories/workload-probe-evidence/workload-probe-evidence.repository"; import { WorkloadAbuseConfigService } from "@src/workload-abuse/services/workload-abuse-config/workload-abuse-config.service"; -import { type BehaviouralReplayFixture, BUNDLED_REPLAY_FIXTURES } from "./fixtures"; +import { type BehaviouralReplayFixture, BUNDLED_REPLAY_FIXTURES, parseReplayFixture } from "./fixtures"; const DEFAULT_WINDOW_DAYS = 30; const AGREEMENT_SWEEP = [1, 2, 3, 5]; @@ -58,7 +63,7 @@ export class BehaviouralSignalReplayService { async replay(options: BehaviouralReplayOptions = {}): Promise { const params = this.#readParams(options); const fixtureSeries = await this.#loadFixtures(options); - const usesDatabase = fixtureSeries.length === 0 || Boolean(options.since); + const usesDatabase = fixtureSeries.length === 0 || Boolean(options.since || options.until); const window = usesDatabase ? this.#resolveWindow(options) : null; const databaseSeries = window ? await this.#loadFromDatabase(window) : []; const series = [...databaseSeries, ...fixtureSeries]; @@ -86,7 +91,8 @@ export class BehaviouralSignalReplayService { deployments: deployments.length, probes: deployments.reduce((total, deployment) => total + deployment.probes, 0), candidateDeployments: deployments.filter(deployment => deployment.candidateProbes > 0).length, - wouldEnforce: summary.wouldEnforce + wouldEnforce: summary.wouldEnforce, + sensitivity: summary.sensitivity }); return summary; @@ -112,6 +118,8 @@ export class BehaviouralSignalReplayService { const byDeployment = new Map(); for (const row of rows) { + if (row.shellStatus !== COMPLETE_SHELL_STATUS) continue; + const label = `${row.walletId}/${row.dseq}/${row.service}`; const series = byDeployment.get(label) ?? { source: "database" as const, label, snapshots: [] }; series.snapshots.push(row); @@ -126,12 +134,26 @@ export class BehaviouralSignalReplayService { const loaded: BehaviouralReplayFixture[] = []; for (const path of options.fixturePaths ?? []) { - loaded.push(...(await readFixtures(path))); + for (const file of await listFixtureFiles(path)) { + const fixture = await this.#readFixture(file); + + if (fixture) loaded.push(fixture); + } } return [...bundled, ...loaded].map(fixture => ({ source: "fixture" as const, label: fixture.deployment, snapshots: fixture.snapshots })); } + /** One unreadable file an operator passed in must not cost them the report on everything else. */ + async #readFixture(path: string): Promise { + try { + return parseReplayFixture(JSON.parse(await readFile(path, "utf8"))); + } catch (error) { + this.logger.warn({ event: "BEHAVIOURAL_REPLAY_FIXTURE_SKIPPED", path, error }); + return null; + } + } + #summarise(series: SnapshotSeries, params: BehaviouralSignalParams): BehaviouralReplayDeployment { const findingsPerSnapshot = series.snapshots.map(snapshot => evaluateBehaviouralSignals(snapshot, params)); const signalsPerSnapshot = findingsPerSnapshot.map(findings => new Set(findings.map(finding => finding.signal))); @@ -166,18 +188,12 @@ export class BehaviouralSignalReplayService { } } -async function readFixtures(path: string): Promise { +async function listFixtureFiles(path: string): Promise { const entries = await readdir(path).catch(() => null); - if (!entries) return [await readFixtureFile(path)]; - - const files = entries.filter(entry => entry.endsWith(".json")); - - return await Promise.all(files.map(file => readFixtureFile(join(path, file)))); -} + if (!entries) return [path]; -async function readFixtureFile(path: string): Promise { - return JSON.parse(await readFile(path, "utf8")) as BehaviouralReplayFixture; + return entries.filter(entry => entry.endsWith(".json")).map(entry => join(path, entry)); } function findLongestAgreement(candidates: boolean[]): number { diff --git a/apps/api/src/workload-abuse/services/behavioural-signal-replay/fixtures/accelerator-bound-no-artifacts.json b/apps/api/src/workload-abuse/services/behavioural-signal-replay/fixtures/accelerator-bound-no-artifacts.json index bb6cd6a7d5..eb2738790e 100644 --- a/apps/api/src/workload-abuse/services/behavioural-signal-replay/fixtures/accelerator-bound-no-artifacts.json +++ b/apps/api/src/workload-abuse/services/behavioural-signal-replay/fixtures/accelerator-bound-no-artifacts.json @@ -2,43 +2,94 @@ "deployment": "accelerator-bound-no-artifacts", "snapshots": [ { + "shellStatus": "completed", "accelerator": [ { "name": "accelerator-0", "utilPct": 99, "memUsedMb": 20480, "memTotalMb": 24576, - "processes": [{ "pid": 1234, "name": "worker", "vramMb": 18000 }] + "processes": [ + { + "pid": 1234, + "name": "worker", + "vramMb": 18000 + } + ] } ], - "artifacts": [{ "path": "/opt/worker", "sizeBytes": 6291456 }], - "netShape": { "listenPorts": [22], "established": [] } + "artifacts": [ + { + "path": "/opt/worker", + "sizeBytes": 6291456 + } + ], + "netShape": { + "listenPorts": [ + 22 + ], + "established": [] + } }, { + "shellStatus": "completed", "accelerator": [ { "name": "accelerator-0", "utilPct": 98, "memUsedMb": 20480, "memTotalMb": 24576, - "processes": [{ "pid": 1234, "name": "worker", "vramMb": 18000 }] + "processes": [ + { + "pid": 1234, + "name": "worker", + "vramMb": 18000 + } + ] } ], - "artifacts": [{ "path": "/opt/worker", "sizeBytes": 6291456 }], - "netShape": { "listenPorts": [22], "established": [] } + "artifacts": [ + { + "path": "/opt/worker", + "sizeBytes": 6291456 + } + ], + "netShape": { + "listenPorts": [ + 22 + ], + "established": [] + } }, { + "shellStatus": "completed", "accelerator": [ { "name": "accelerator-0", "utilPct": 97, "memUsedMb": 20480, "memTotalMb": 24576, - "processes": [{ "pid": 1234, "name": "worker", "vramMb": 17800 }] + "processes": [ + { + "pid": 1234, + "name": "worker", + "vramMb": 17800 + } + ] + } + ], + "artifacts": [ + { + "path": "/opt/worker", + "sizeBytes": 6291456 } ], - "artifacts": [{ "path": "/opt/worker", "sizeBytes": 6291456 }], - "netShape": { "listenPorts": [22], "established": [] } + "netShape": { + "listenPorts": [ + 22 + ], + "established": [] + } } ] } diff --git a/apps/api/src/workload-abuse/services/behavioural-signal-replay/fixtures/cpu-only-workload.json b/apps/api/src/workload-abuse/services/behavioural-signal-replay/fixtures/cpu-only-workload.json index bb1e3b0eb4..de16edea5a 100644 --- a/apps/api/src/workload-abuse/services/behavioural-signal-replay/fixtures/cpu-only-workload.json +++ b/apps/api/src/workload-abuse/services/behavioural-signal-replay/fixtures/cpu-only-workload.json @@ -2,11 +2,26 @@ "deployment": "cpu-only-workload", "snapshots": [ { + "shellStatus": "completed", "accelerator": null, - "artifacts": [{ "path": "/srv/app", "sizeBytes": 10485760 }], + "artifacts": [ + { + "path": "/srv/app", + "sizeBytes": 10485760 + } + ], "netShape": { - "listenPorts": [3000], - "established": [{ "localPort": 46000, "remoteIp": "198.51.100.10", "remotePort": 443, "count": 2 }] + "listenPorts": [ + 3000 + ], + "established": [ + { + "localPort": 46000, + "remoteIp": "198.51.100.10", + "remotePort": 443, + "count": 2 + } + ] } } ] diff --git a/apps/api/src/workload-abuse/services/behavioural-signal-replay/fixtures/idle-shell-host.json b/apps/api/src/workload-abuse/services/behavioural-signal-replay/fixtures/idle-shell-host.json index 1dc6705cd8..d0fd91c2b2 100644 --- a/apps/api/src/workload-abuse/services/behavioural-signal-replay/fixtures/idle-shell-host.json +++ b/apps/api/src/workload-abuse/services/behavioural-signal-replay/fixtures/idle-shell-host.json @@ -2,14 +2,50 @@ "deployment": "idle-shell-host", "snapshots": [ { + "shellStatus": "completed", "accelerator": null, - "artifacts": [{ "path": "/usr/bin/toolchain", "sizeBytes": 20971520 }], - "netShape": { "listenPorts": [22], "established": [{ "localPort": 40100, "remoteIp": "10.0.0.7", "remotePort": 443, "count": 1 }] } + "artifacts": [ + { + "path": "/usr/bin/toolchain", + "sizeBytes": 20971520 + } + ], + "netShape": { + "listenPorts": [ + 22 + ], + "established": [ + { + "localPort": 40100, + "remoteIp": "10.0.0.7", + "remotePort": 443, + "count": 1 + } + ] + } }, { + "shellStatus": "completed", "accelerator": null, - "artifacts": [{ "path": "/usr/bin/toolchain", "sizeBytes": 20971520 }], - "netShape": { "listenPorts": [22], "established": [{ "localPort": 40101, "remoteIp": "10.0.0.7", "remotePort": 443, "count": 1 }] } + "artifacts": [ + { + "path": "/usr/bin/toolchain", + "sizeBytes": 20971520 + } + ], + "netShape": { + "listenPorts": [ + 22 + ], + "established": [ + { + "localPort": 40101, + "remoteIp": "10.0.0.7", + "remotePort": 443, + "count": 1 + } + ] + } } ] } diff --git a/apps/api/src/workload-abuse/services/behavioural-signal-replay/fixtures/index.ts b/apps/api/src/workload-abuse/services/behavioural-signal-replay/fixtures/index.ts index 4db5e2dd79..3ec7539ed1 100644 --- a/apps/api/src/workload-abuse/services/behavioural-signal-replay/fixtures/index.ts +++ b/apps/api/src/workload-abuse/services/behavioural-signal-replay/fixtures/index.ts @@ -1,11 +1,34 @@ -import type { ProbeEvidenceSnapshot } from "@src/workload-abuse/lib/behavioural-signals/types"; +import { z } from "zod"; + import acceleratorBoundNoArtifacts from "./accelerator-bound-no-artifacts.json"; import cpuOnlyWorkload from "./cpu-only-workload.json"; import idleShellHost from "./idle-shell-host.json"; import inferenceWithWeights from "./inference-with-weights.json"; import packagedRuntime from "./packaged-runtime.json"; -export type BehaviouralReplayFixture = { deployment: string; snapshots: ProbeEvidenceSnapshot[] }; +const PROCESS_SCHEMA = z.object({ pid: z.number(), name: z.string(), vramMb: z.number() }); + +const SNAPSHOT_SCHEMA = z.object({ + shellStatus: z.string(), + accelerator: z + .array(z.object({ name: z.string(), utilPct: z.number(), memUsedMb: z.number(), memTotalMb: z.number(), processes: z.array(PROCESS_SCHEMA) })) + .nullable(), + artifacts: z.array(z.object({ path: z.string(), sizeBytes: z.number() })).nullable(), + netShape: z + .object({ + listenPorts: z.array(z.number()), + established: z.array(z.object({ localPort: z.number(), remoteIp: z.string(), remotePort: z.number(), count: z.number() })) + }) + .nullable() +}); + +const FIXTURE_SCHEMA = z.object({ deployment: z.string(), snapshots: z.array(SNAPSHOT_SCHEMA) }); + +export type BehaviouralReplayFixture = z.infer; + +export function parseReplayFixture(contents: unknown): BehaviouralReplayFixture { + return FIXTURE_SCHEMA.parse(contents); +} export const BUNDLED_REPLAY_FIXTURES: BehaviouralReplayFixture[] = [ acceleratorBoundNoArtifacts, diff --git a/apps/api/src/workload-abuse/services/behavioural-signal-replay/fixtures/inference-with-weights.json b/apps/api/src/workload-abuse/services/behavioural-signal-replay/fixtures/inference-with-weights.json index 9aaa908917..c3a87ac027 100644 --- a/apps/api/src/workload-abuse/services/behavioural-signal-replay/fixtures/inference-with-weights.json +++ b/apps/api/src/workload-abuse/services/behavioural-signal-replay/fixtures/inference-with-weights.json @@ -2,38 +2,81 @@ "deployment": "inference-with-weights", "snapshots": [ { + "shellStatus": "completed", "accelerator": [ { "name": "accelerator-0", "utilPct": 64, "memUsedMb": 14000, "memTotalMb": 24576, - "processes": [{ "pid": 2001, "name": "runtime", "vramMb": 13800 }] + "processes": [ + { + "pid": 2001, + "name": "runtime", + "vramMb": 13800 + } + ] } ], "artifacts": [ - { "path": "/models/a.bin", "sizeBytes": 8589934592 }, - { "path": "/models/b.bin", "sizeBytes": 4294967296 } + { + "path": "/models/a.bin", + "sizeBytes": 8589934592 + }, + { + "path": "/models/b.bin", + "sizeBytes": 4294967296 + } ], "netShape": { - "listenPorts": [8080], - "established": [{ "localPort": 8080, "remoteIp": "203.0.113.20", "remotePort": 54321, "count": 4 }] + "listenPorts": [ + 8080 + ], + "established": [ + { + "localPort": 8080, + "remoteIp": "203.0.113.20", + "remotePort": 54321, + "count": 4 + } + ] } }, { + "shellStatus": "completed", "accelerator": [ { "name": "accelerator-0", "utilPct": 71, "memUsedMb": 14200, "memTotalMb": 24576, - "processes": [{ "pid": 2001, "name": "runtime", "vramMb": 13900 }] + "processes": [ + { + "pid": 2001, + "name": "runtime", + "vramMb": 13900 + } + ] + } + ], + "artifacts": [ + { + "path": "/models/a.bin", + "sizeBytes": 8589934592 } ], - "artifacts": [{ "path": "/models/a.bin", "sizeBytes": 8589934592 }], "netShape": { - "listenPorts": [8080], - "established": [{ "localPort": 8080, "remoteIp": "203.0.113.21", "remotePort": 51222, "count": 2 }] + "listenPorts": [ + 8080 + ], + "established": [ + { + "localPort": 8080, + "remoteIp": "203.0.113.21", + "remotePort": 51222, + "count": 2 + } + ] } } ] diff --git a/apps/api/src/workload-abuse/services/behavioural-signal-replay/fixtures/packaged-runtime.json b/apps/api/src/workload-abuse/services/behavioural-signal-replay/fixtures/packaged-runtime.json index f0307858d8..360bedef61 100644 --- a/apps/api/src/workload-abuse/services/behavioural-signal-replay/fixtures/packaged-runtime.json +++ b/apps/api/src/workload-abuse/services/behavioural-signal-replay/fixtures/packaged-runtime.json @@ -2,30 +2,64 @@ "deployment": "packaged-runtime", "snapshots": [ { + "shellStatus": "completed", "accelerator": [ { "name": "accelerator-0", "utilPct": 96, "memUsedMb": 19000, "memTotalMb": 24576, - "processes": [{ "pid": 3100, "name": "runtime", "vramMb": 17000 }] + "processes": [ + { + "pid": 3100, + "name": "runtime", + "vramMb": 17000 + } + ] } ], - "artifacts": [{ "path": "/usr/lib/runtime/packages", "sizeBytes": 3221225472 }], - "netShape": { "listenPorts": [22], "established": [] } + "artifacts": [ + { + "path": "/usr/lib/runtime/packages", + "sizeBytes": 3221225472 + } + ], + "netShape": { + "listenPorts": [ + 22 + ], + "established": [] + } }, { + "shellStatus": "completed", "accelerator": [ { "name": "accelerator-0", "utilPct": 95, "memUsedMb": 19000, "memTotalMb": 24576, - "processes": [{ "pid": 3100, "name": "runtime", "vramMb": 17000 }] + "processes": [ + { + "pid": 3100, + "name": "runtime", + "vramMb": 17000 + } + ] + } + ], + "artifacts": [ + { + "path": "/usr/lib/runtime/packages", + "sizeBytes": 3221225472 } ], - "artifacts": [{ "path": "/usr/lib/runtime/packages", "sizeBytes": 3221225472 }], - "netShape": { "listenPorts": [22], "established": [] } + "netShape": { + "listenPorts": [ + 22 + ], + "established": [] + } } ] } diff --git a/apps/api/src/workload-abuse/services/probe-evidence/probe-evidence.service.spec.ts b/apps/api/src/workload-abuse/services/probe-evidence/probe-evidence.service.spec.ts index bc47cf4908..e7b5437583 100644 --- a/apps/api/src/workload-abuse/services/probe-evidence/probe-evidence.service.spec.ts +++ b/apps/api/src/workload-abuse/services/probe-evidence/probe-evidence.service.spec.ts @@ -125,13 +125,21 @@ describe(ProbeEvidenceService.name, () => { expect(evidenceRepository.recordBehaviouralFindings).not.toHaveBeenCalled(); }); + it("records nothing for a row whose shell was cut short", async () => { + const { service, evidenceRepository } = setup({ signalsEnabled: true }); + + await expect(service.recordBehaviouralFindings([createEvidenceRow({ shellStatus: "output_capped" })])).resolves.toEqual([]); + + expect(evidenceRepository.recordBehaviouralFindings).not.toHaveBeenCalled(); + }); + it("logs and counts a findings write failure without rethrowing", async () => { const { service, evidenceRepository, instrumentation, logger } = setup({ signalsEnabled: true }); evidenceRepository.recordBehaviouralFindings.mockRejectedValue(new Error("connection refused")); await expect(service.recordBehaviouralFindings([createEvidenceRow()])).resolves.toEqual([]); - expect(instrumentation.recordEvidenceWriteFailure).toHaveBeenCalledTimes(1); + expect(instrumentation.recordEvidenceWriteFailure).toHaveBeenCalledWith("findings"); expect(logger.warn).toHaveBeenCalledWith(expect.objectContaining({ event: "WORKLOAD_EVIDENCE_FINDINGS_WRITE_FAILED", walletId: 42, dseq: "1000001" })); }); @@ -159,6 +167,7 @@ describe(ProbeEvidenceService.name, () => { walletId: 42, dseq: "1000001", service: "web", + shellStatus: "completed", accelerator: [{ name: "accelerator-0", utilPct: 99, memUsedMb: 20_480, memTotalMb: 24_576, processes: [{ pid: 1234, name: "worker", vramMb: 18_000 }] }], artifacts: [{ path: "/opt/worker", sizeBytes: 4_194_304 }], netShape: { listenPorts: [22], established: [] }, diff --git a/apps/api/src/workload-abuse/services/workload-abuse-instrumentation/workload-abuse-instrumentation.service.ts b/apps/api/src/workload-abuse/services/workload-abuse-instrumentation/workload-abuse-instrumentation.service.ts index 7d677cecd7..7729d65b25 100644 --- a/apps/api/src/workload-abuse/services/workload-abuse-instrumentation/workload-abuse-instrumentation.service.ts +++ b/apps/api/src/workload-abuse/services/workload-abuse-instrumentation/workload-abuse-instrumentation.service.ts @@ -6,7 +6,7 @@ import type { WorkloadVerdict } from "@src/workload-abuse/lib/evidence-scanner/e export type DomainBlockResult = "blocked" | "raced" | "skipped" | "dry_run" | "failed" | "sibling_limit_reached"; -export type EvidenceWriteOperation = "insert" | "purge"; +export type EvidenceWriteOperation = "insert" | "findings" | "purge"; @singleton() export class WorkloadAbuseInstrumentationService { From 898caaf1284f10f2703aa2020087b1e5e00b01c5 Mon Sep 17 00:00:00 2001 From: Maxime Beauchamp <15185355+baktun14@users.noreply.github.com> Date: Sun, 20 Sep 2026 16:20:49 +0400 Subject: [PATCH 3/5] fix(deployment): read connections under the state the collector reported The evidence shape now names connections for what they are and keeps the socket state with each one, so the signals and the replay corpus carry a half open connection as such. --- .../evaluate-behavioural-signals.spec.ts | 16 +++++++-------- .../network-isolation.spec.ts | 20 ++++++++++++------- .../behavioural-signals/network-isolation.ts | 8 ++++---- .../behavioural-signal-replay.service.spec.ts | 2 +- .../accelerator-bound-no-artifacts.json | 6 +++--- .../fixtures/cpu-only-workload.json | 5 +++-- .../fixtures/idle-shell-host.json | 10 ++++++---- .../fixtures/index.ts | 12 +++++++++-- .../fixtures/inference-with-weights.json | 10 ++++++---- .../fixtures/packaged-runtime.json | 4 ++-- .../probe-evidence.service.spec.ts | 4 ++-- 11 files changed, 58 insertions(+), 39 deletions(-) diff --git a/apps/api/src/workload-abuse/lib/behavioural-signals/evaluate-behavioural-signals.spec.ts b/apps/api/src/workload-abuse/lib/behavioural-signals/evaluate-behavioural-signals.spec.ts index b4e500fe6b..4b7fef261d 100644 --- a/apps/api/src/workload-abuse/lib/behavioural-signals/evaluate-behavioural-signals.spec.ts +++ b/apps/api/src/workload-abuse/lib/behavioural-signals/evaluate-behavioural-signals.spec.ts @@ -5,13 +5,13 @@ import type { BehaviouralSignalParams, ProbeEvidenceSnapshot } from "./types"; describe("evaluateBehaviouralSignals", () => { it("returns both findings when the snapshot matches both signals", () => { - const { snapshot, params } = setup({ vramMb: 18_000, artifactBytes: 1_024, established: [] }); + const { snapshot, params } = setup({ vramMb: 18_000, artifactBytes: 1_024, connections: [] }); expect(evaluateBehaviouralSignals(snapshot, params).map(finding => finding.signal)).toEqual(["accel_without_artifacts", "network_isolated"]); }); it("returns one finding when the workload carries weights but talks to nobody", () => { - const { snapshot, params } = setup({ vramMb: 18_000, artifactBytes: 8_589_934_592, established: [] }); + const { snapshot, params } = setup({ vramMb: 18_000, artifactBytes: 8_589_934_592, connections: [] }); expect(evaluateBehaviouralSignals(snapshot, params).map(finding => finding.signal)).toEqual(["network_isolated"]); }); @@ -20,26 +20,26 @@ describe("evaluateBehaviouralSignals", () => { const { snapshot, params } = setup({ vramMb: 18_000, artifactBytes: 8_589_934_592, - established: [{ localPort: 8_080, remoteIp: "203.0.113.9", remotePort: 51_000, count: 2 }] + connections: [{ localPort: 8_080, remoteIp: "203.0.113.9", remotePort: 51_000, count: 2, state: "established" }] }); expect(evaluateBehaviouralSignals(snapshot, params)).toEqual([]); }); it("returns nothing when the shell that collected the snapshot was cut short", () => { - const { snapshot, params } = setup({ vramMb: 18_000, artifactBytes: 1_024, established: [], shellStatus: "output_capped" }); + const { snapshot, params } = setup({ vramMb: 18_000, artifactBytes: 1_024, connections: [], shellStatus: "output_capped" }); expect(evaluateBehaviouralSignals(snapshot, params)).toEqual([]); }); it("treats the two signals together as a candidate", () => { - const { snapshot, params } = setup({ vramMb: 18_000, artifactBytes: 1_024, established: [] }); + const { snapshot, params } = setup({ vramMb: 18_000, artifactBytes: 1_024, connections: [] }); expect(isBehaviouralCandidate(evaluateBehaviouralSignals(snapshot, params))).toBe(true); }); it("treats a single signal as no candidate", () => { - const { snapshot, params } = setup({ vramMb: 18_000, artifactBytes: 8_589_934_592, established: [] }); + const { snapshot, params } = setup({ vramMb: 18_000, artifactBytes: 8_589_934_592, connections: [] }); expect(isBehaviouralCandidate(evaluateBehaviouralSignals(snapshot, params))).toBe(false); }); @@ -47,7 +47,7 @@ describe("evaluateBehaviouralSignals", () => { function setup(input: { vramMb: number; artifactBytes: number; - established: NonNullable["established"]; + connections: NonNullable["connections"]; shellStatus?: string; }) { const snapshot: ProbeEvidenceSnapshot = { @@ -56,7 +56,7 @@ describe("evaluateBehaviouralSignals", () => { { name: "accelerator-0", utilPct: 97, memUsedMb: 20_480, memTotalMb: 24_576, processes: [{ pid: 1234, name: "worker", vramMb: input.vramMb }] } ], artifacts: [{ path: "/opt/payload", sizeBytes: input.artifactBytes }], - netShape: { listenPorts: [8_080], established: input.established } + netShape: { listenPorts: [8_080], connections: input.connections } }; const params: BehaviouralSignalParams = { accelMinVramMb: 1_024, artifactMinMb: 256, relayEndpoints: [] }; diff --git a/apps/api/src/workload-abuse/lib/behavioural-signals/network-isolation.spec.ts b/apps/api/src/workload-abuse/lib/behavioural-signals/network-isolation.spec.ts index 0a8fa623d6..9bfa996722 100644 --- a/apps/api/src/workload-abuse/lib/behavioural-signals/network-isolation.spec.ts +++ b/apps/api/src/workload-abuse/lib/behavioural-signals/network-isolation.spec.ts @@ -6,7 +6,7 @@ import type { BehaviouralSignalParams, ProbeEvidenceSnapshot } from "./types"; describe("evaluateNetworkIsolation", () => { it("fires when nothing is established in either direction", () => { - const { snapshot, params } = setup({ netShape: { listenPorts: [22, 8080], established: [] } }); + const { snapshot, params } = setup({ netShape: { listenPorts: [22, 8080], connections: [] } }); expect(evaluateNetworkIsolation(snapshot, params)).toEqual({ signal: "network_isolated", @@ -16,7 +16,7 @@ describe("evaluateNetworkIsolation", () => { it("fires when the only outbound connection goes to a first-party relay", () => { const { snapshot, params } = setup({ - netShape: { listenPorts: [22], established: [{ localPort: 41_234, remoteIp: "10.0.0.7", remotePort: 443, count: 1 }] }, + netShape: { listenPorts: [22], connections: [{ localPort: 41_234, remoteIp: "10.0.0.7", remotePort: 443, count: 1, state: "established" }] }, relayEndpoints: ["10.0.0.7"] }); @@ -25,7 +25,7 @@ describe("evaluateNetworkIsolation", () => { it("fires when a relay entry pins the port the connection uses", () => { const { snapshot, params } = setup({ - netShape: { listenPorts: [], established: [{ localPort: 41_234, remoteIp: "10.0.0.7", remotePort: 443, count: 1 }] }, + netShape: { listenPorts: [], connections: [{ localPort: 41_234, remoteIp: "10.0.0.7", remotePort: 443, count: 1, state: "established" }] }, relayEndpoints: ["10.0.0.7:443"] }); @@ -34,7 +34,10 @@ describe("evaluateNetworkIsolation", () => { it("fires when the relay answers over IPv6", () => { const { snapshot, params } = setup({ - netShape: { listenPorts: [], established: [{ localPort: 41_234, remoteIp: "2001:0db8:0000:0000:0000:0000:0000:0001", remotePort: 443, count: 1 }] }, + netShape: { + listenPorts: [], + connections: [{ localPort: 41_234, remoteIp: "2001:0db8:0000:0000:0000:0000:0000:0001", remotePort: 443, count: 1, state: "established" }] + }, relayEndpoints: ["2001:0db8:0000:0000:0000:0000:0000:0001"] }); @@ -43,7 +46,10 @@ describe("evaluateNetworkIsolation", () => { it("fires when the relay is configured in the short form of its IPv6 address", () => { const { snapshot, params } = setup({ - netShape: { listenPorts: [], established: [{ localPort: 41_234, remoteIp: "2001:0db8:0000:0000:0000:0000:0000:0001", remotePort: 443, count: 1 }] }, + netShape: { + listenPorts: [], + connections: [{ localPort: 41_234, remoteIp: "2001:0db8:0000:0000:0000:0000:0000:0001", remotePort: 443, count: 1, state: "established" }] + }, relayEndpoints: ["2001:db8::1"] }); @@ -52,7 +58,7 @@ describe("evaluateNetworkIsolation", () => { it("stays silent when a connection lands on a listening port", () => { const { snapshot, params } = setup({ - netShape: { listenPorts: [8080], established: [{ localPort: 8080, remoteIp: "203.0.113.9", remotePort: 51_000, count: 3 }] } + netShape: { listenPorts: [8080], connections: [{ localPort: 8080, remoteIp: "203.0.113.9", remotePort: 51_000, count: 3, state: "established" }] } }); expect(evaluateNetworkIsolation(snapshot, params)).toBeNull(); @@ -60,7 +66,7 @@ describe("evaluateNetworkIsolation", () => { it("stays silent when an outbound connection goes somewhere other than a relay", () => { const { snapshot, params } = setup({ - netShape: { listenPorts: [22], established: [{ localPort: 41_234, remoteIp: "198.51.100.4", remotePort: 3_333, count: 1 }] }, + netShape: { listenPorts: [22], connections: [{ localPort: 41_234, remoteIp: "198.51.100.4", remotePort: 3_333, count: 1, state: "established" }] }, relayEndpoints: ["10.0.0.7"] }); diff --git a/apps/api/src/workload-abuse/lib/behavioural-signals/network-isolation.ts b/apps/api/src/workload-abuse/lib/behavioural-signals/network-isolation.ts index 3421aafec6..d55169bbae 100644 --- a/apps/api/src/workload-abuse/lib/behavioural-signals/network-isolation.ts +++ b/apps/api/src/workload-abuse/lib/behavioural-signals/network-isolation.ts @@ -1,7 +1,7 @@ import type { ProbeEvidenceNetShape } from "@src/workload-abuse/model-schemas"; import { BEHAVIOURAL_SIGNALS, type BehaviouralFinding, type BehaviouralSignalParams, type ProbeEvidenceSnapshot } from "./types"; -type EstablishedConnection = ProbeEvidenceNetShape["established"][number]; +type EstablishedConnection = ProbeEvidenceNetShape["connections"][number]; export function evaluateNetworkIsolation(snapshot: ProbeEvidenceSnapshot, params: BehaviouralSignalParams): BehaviouralFinding | null { const netShape = snapshot.netShape; @@ -9,13 +9,13 @@ export function evaluateNetworkIsolation(snapshot: ProbeEvidenceSnapshot, params if (!netShape) return null; const listenPorts = new Set(netShape.listenPorts); - const inbound = netShape.established.filter(connection => listenPorts.has(connection.localPort)); + const inbound = netShape.connections.filter(connection => listenPorts.has(connection.localPort)); if (inbound.length > 0) return null; - const relayOutbound = netShape.established.filter(connection => isRelayEndpoint(connection, params.relayEndpoints)); + const relayOutbound = netShape.connections.filter(connection => isRelayEndpoint(connection, params.relayEndpoints)); - if (relayOutbound.length < netShape.established.length) return null; + if (relayOutbound.length < netShape.connections.length) return null; return { signal: BEHAVIOURAL_SIGNALS.networkIsolated, detail: { excludedRelay: relayOutbound.length, listenPorts: netShape.listenPorts.length } }; } diff --git a/apps/api/src/workload-abuse/services/behavioural-signal-replay/behavioural-signal-replay.service.spec.ts b/apps/api/src/workload-abuse/services/behavioural-signal-replay/behavioural-signal-replay.service.spec.ts index 50360a5b02..7a3f055422 100644 --- a/apps/api/src/workload-abuse/services/behavioural-signal-replay/behavioural-signal-replay.service.spec.ts +++ b/apps/api/src/workload-abuse/services/behavioural-signal-replay/behavioural-signal-replay.service.spec.ts @@ -170,7 +170,7 @@ describe(BehaviouralSignalReplayService.name, () => { shellStatus: "completed", accelerator: [{ name: "accelerator-0", utilPct: 99, memUsedMb: 20_480, memTotalMb: 24_576, processes: [{ pid: 1234, name: "worker", vramMb: 18_000 }] }], artifacts: [{ path: "/opt/worker", sizeBytes: 4_194_304 }], - netShape: { listenPorts: [22], established: [] }, + netShape: { listenPorts: [22], connections: [] }, ...overrides }); } diff --git a/apps/api/src/workload-abuse/services/behavioural-signal-replay/fixtures/accelerator-bound-no-artifacts.json b/apps/api/src/workload-abuse/services/behavioural-signal-replay/fixtures/accelerator-bound-no-artifacts.json index eb2738790e..97b8a2fa53 100644 --- a/apps/api/src/workload-abuse/services/behavioural-signal-replay/fixtures/accelerator-bound-no-artifacts.json +++ b/apps/api/src/workload-abuse/services/behavioural-signal-replay/fixtures/accelerator-bound-no-artifacts.json @@ -28,7 +28,7 @@ "listenPorts": [ 22 ], - "established": [] + "connections": [] } }, { @@ -58,7 +58,7 @@ "listenPorts": [ 22 ], - "established": [] + "connections": [] } }, { @@ -88,7 +88,7 @@ "listenPorts": [ 22 ], - "established": [] + "connections": [] } } ] diff --git a/apps/api/src/workload-abuse/services/behavioural-signal-replay/fixtures/cpu-only-workload.json b/apps/api/src/workload-abuse/services/behavioural-signal-replay/fixtures/cpu-only-workload.json index de16edea5a..5901eab035 100644 --- a/apps/api/src/workload-abuse/services/behavioural-signal-replay/fixtures/cpu-only-workload.json +++ b/apps/api/src/workload-abuse/services/behavioural-signal-replay/fixtures/cpu-only-workload.json @@ -14,12 +14,13 @@ "listenPorts": [ 3000 ], - "established": [ + "connections": [ { "localPort": 46000, "remoteIp": "198.51.100.10", "remotePort": 443, - "count": 2 + "count": 2, + "state": "established" } ] } diff --git a/apps/api/src/workload-abuse/services/behavioural-signal-replay/fixtures/idle-shell-host.json b/apps/api/src/workload-abuse/services/behavioural-signal-replay/fixtures/idle-shell-host.json index d0fd91c2b2..fe137ce160 100644 --- a/apps/api/src/workload-abuse/services/behavioural-signal-replay/fixtures/idle-shell-host.json +++ b/apps/api/src/workload-abuse/services/behavioural-signal-replay/fixtures/idle-shell-host.json @@ -14,12 +14,13 @@ "listenPorts": [ 22 ], - "established": [ + "connections": [ { "localPort": 40100, "remoteIp": "10.0.0.7", "remotePort": 443, - "count": 1 + "count": 1, + "state": "established" } ] } @@ -37,12 +38,13 @@ "listenPorts": [ 22 ], - "established": [ + "connections": [ { "localPort": 40101, "remoteIp": "10.0.0.7", "remotePort": 443, - "count": 1 + "count": 1, + "state": "established" } ] } diff --git a/apps/api/src/workload-abuse/services/behavioural-signal-replay/fixtures/index.ts b/apps/api/src/workload-abuse/services/behavioural-signal-replay/fixtures/index.ts index 3ec7539ed1..68e1d4be8a 100644 --- a/apps/api/src/workload-abuse/services/behavioural-signal-replay/fixtures/index.ts +++ b/apps/api/src/workload-abuse/services/behavioural-signal-replay/fixtures/index.ts @@ -17,7 +17,15 @@ const SNAPSHOT_SCHEMA = z.object({ netShape: z .object({ listenPorts: z.array(z.number()), - established: z.array(z.object({ localPort: z.number(), remoteIp: z.string(), remotePort: z.number(), count: z.number() })) + connections: z.array( + z.object({ + localPort: z.number(), + remoteIp: z.string(), + remotePort: z.number(), + count: z.number(), + state: z.enum(["established", "connecting"]) + }) + ) }) .nullable() }); @@ -36,4 +44,4 @@ export const BUNDLED_REPLAY_FIXTURES: BehaviouralReplayFixture[] = [ idleShellHost, packagedRuntime, cpuOnlyWorkload -]; +].map(fixture => parseReplayFixture(fixture)); diff --git a/apps/api/src/workload-abuse/services/behavioural-signal-replay/fixtures/inference-with-weights.json b/apps/api/src/workload-abuse/services/behavioural-signal-replay/fixtures/inference-with-weights.json index c3a87ac027..3b8db298e8 100644 --- a/apps/api/src/workload-abuse/services/behavioural-signal-replay/fixtures/inference-with-weights.json +++ b/apps/api/src/workload-abuse/services/behavioural-signal-replay/fixtures/inference-with-weights.json @@ -32,12 +32,13 @@ "listenPorts": [ 8080 ], - "established": [ + "connections": [ { "localPort": 8080, "remoteIp": "203.0.113.20", "remotePort": 54321, - "count": 4 + "count": 4, + "state": "established" } ] } @@ -69,12 +70,13 @@ "listenPorts": [ 8080 ], - "established": [ + "connections": [ { "localPort": 8080, "remoteIp": "203.0.113.21", "remotePort": 51222, - "count": 2 + "count": 2, + "state": "established" } ] } diff --git a/apps/api/src/workload-abuse/services/behavioural-signal-replay/fixtures/packaged-runtime.json b/apps/api/src/workload-abuse/services/behavioural-signal-replay/fixtures/packaged-runtime.json index 360bedef61..c907c9c3c6 100644 --- a/apps/api/src/workload-abuse/services/behavioural-signal-replay/fixtures/packaged-runtime.json +++ b/apps/api/src/workload-abuse/services/behavioural-signal-replay/fixtures/packaged-runtime.json @@ -28,7 +28,7 @@ "listenPorts": [ 22 ], - "established": [] + "connections": [] } }, { @@ -58,7 +58,7 @@ "listenPorts": [ 22 ], - "established": [] + "connections": [] } } ] diff --git a/apps/api/src/workload-abuse/services/probe-evidence/probe-evidence.service.spec.ts b/apps/api/src/workload-abuse/services/probe-evidence/probe-evidence.service.spec.ts index e7b5437583..794a44ba1f 100644 --- a/apps/api/src/workload-abuse/services/probe-evidence/probe-evidence.service.spec.ts +++ b/apps/api/src/workload-abuse/services/probe-evidence/probe-evidence.service.spec.ts @@ -117,7 +117,7 @@ describe(ProbeEvidenceService.name, () => { const recorded = await service.recordBehaviouralFindings([ createEvidenceRow({ accelerator: null, - netShape: { listenPorts: [8_080], established: [{ localPort: 8_080, remoteIp: "203.0.113.5", remotePort: 51_000, count: 1 }] } + netShape: { listenPorts: [8_080], connections: [{ localPort: 8_080, remoteIp: "203.0.113.5", remotePort: 51_000, count: 1, state: "established" }] } }) ]); @@ -170,7 +170,7 @@ describe(ProbeEvidenceService.name, () => { shellStatus: "completed", accelerator: [{ name: "accelerator-0", utilPct: 99, memUsedMb: 20_480, memTotalMb: 24_576, processes: [{ pid: 1234, name: "worker", vramMb: 18_000 }] }], artifacts: [{ path: "/opt/worker", sizeBytes: 4_194_304 }], - netShape: { listenPorts: [22], established: [] }, + netShape: { listenPorts: [22], connections: [] }, ...overrides }); } From b41124bee6597485f45d4b67c1a91778dc8ee9a1 Mon Sep 17 00:00:00 2001 From: Maxime Beauchamp <15185355+baktun14@users.noreply.github.com> Date: Sun, 20 Sep 2026 16:57:14 +0400 Subject: [PATCH 4/5] fix(deployment): keep a blank relay endpoint variable from failing config A variable that is declared but left empty reached the JSON parse and turned every resolution of the workload abuse config into a validation error. It now falls back to the default the same way the sibling list does. A replay that asked for fixtures and could read none of them reports nothing rather than quietly scanning a month of recorded evidence instead. --- apps/api/src/workload-abuse/config/env.config.spec.ts | 6 ++++++ apps/api/src/workload-abuse/config/env.config.ts | 2 +- .../behavioural-signal-replay.service.spec.ts | 9 +++++++++ .../behavioural-signal-replay.service.ts | 3 ++- 4 files changed, 18 insertions(+), 2 deletions(-) diff --git a/apps/api/src/workload-abuse/config/env.config.spec.ts b/apps/api/src/workload-abuse/config/env.config.spec.ts index 21f33c3b73..e6eadace2c 100644 --- a/apps/api/src/workload-abuse/config/env.config.spec.ts +++ b/apps/api/src/workload-abuse/config/env.config.spec.ts @@ -42,6 +42,12 @@ describe("workload abuse env config", () => { expect(config.WORKLOAD_ABUSE_SIGNAL_RELAY_ENDPOINTS).toEqual(["10.0.0.7", "10.0.0.8:443"]); }); + it("falls back to no relay endpoints when the variable is set but blank", () => { + const config = envSchema.parse({ WORKLOAD_ABUSE_SIGNAL_RELAY_ENDPOINTS: " " }); + + expect(config.WORKLOAD_ABUSE_SIGNAL_RELAY_ENDPOINTS).toEqual([]); + }); + it("rejects a relay endpoint list that is not a JSON array of strings", () => { expect(() => envSchema.parse({ WORKLOAD_ABUSE_SIGNAL_RELAY_ENDPOINTS: "10.0.0.7" })).toThrow(/ip or ip:port/); expect(() => envSchema.parse({ WORKLOAD_ABUSE_SIGNAL_RELAY_ENDPOINTS: JSON.stringify([443]) })).toThrow(/ip or ip:port/); diff --git a/apps/api/src/workload-abuse/config/env.config.ts b/apps/api/src/workload-abuse/config/env.config.ts index 2478e9d731..62154a0060 100644 --- a/apps/api/src/workload-abuse/config/env.config.ts +++ b/apps/api/src/workload-abuse/config/env.config.ts @@ -118,7 +118,7 @@ export const envSchema = z.object({ WORKLOAD_ABUSE_SIGNAL_ACCEL_MIN_VRAM_MB: z.number({ coerce: true }).int().positive().default(1_024), WORKLOAD_ABUSE_SIGNAL_ARTIFACT_MIN_MB: z.number({ coerce: true }).int().positive().default(256), /** The list itself lives in Doppler: these are our own endpoints, so a reader learns how the exclusion works but not what it covers. */ - WORKLOAD_ABUSE_SIGNAL_RELAY_ENDPOINTS: z.string().default("[]").transform(parseRelayEndpoints) + WORKLOAD_ABUSE_SIGNAL_RELAY_ENDPOINTS: z.preprocess(blankToUndefined, z.string().default("[]").transform(parseRelayEndpoints)) }); export type WorkloadAbuseConfig = z.infer; diff --git a/apps/api/src/workload-abuse/services/behavioural-signal-replay/behavioural-signal-replay.service.spec.ts b/apps/api/src/workload-abuse/services/behavioural-signal-replay/behavioural-signal-replay.service.spec.ts index 7a3f055422..99ea899c31 100644 --- a/apps/api/src/workload-abuse/services/behavioural-signal-replay/behavioural-signal-replay.service.spec.ts +++ b/apps/api/src/workload-abuse/services/behavioural-signal-replay/behavioural-signal-replay.service.spec.ts @@ -135,6 +135,15 @@ describe(BehaviouralSignalReplayService.name, () => { expect(logger.warn).toHaveBeenCalledWith(expect.objectContaining({ event: "BEHAVIOURAL_REPLAY_FIXTURE_SKIPPED" })); }); + it("reports nothing rather than falling back to the database when every fixture asked for is unreadable", async () => { + const { service, evidenceRepository } = setup({ rows: [createRow({ id: "row-1", service: "web" })] }); + + const summary = await service.replay({ fixturePaths: [join(tmpdir(), "behavioural-replay-missing.json")] }); + + expect(evidenceRepository.findCreatedBetween).not.toHaveBeenCalled(); + expect(summary.deployments).toEqual([]); + }); + it("logs the threshold sweep with the rest of the report, so a run leaves it behind", async () => { const { service, logger } = setup(); diff --git a/apps/api/src/workload-abuse/services/behavioural-signal-replay/behavioural-signal-replay.service.ts b/apps/api/src/workload-abuse/services/behavioural-signal-replay/behavioural-signal-replay.service.ts index 4167965d54..e099cdcc56 100644 --- a/apps/api/src/workload-abuse/services/behavioural-signal-replay/behavioural-signal-replay.service.ts +++ b/apps/api/src/workload-abuse/services/behavioural-signal-replay/behavioural-signal-replay.service.ts @@ -63,7 +63,8 @@ export class BehaviouralSignalReplayService { async replay(options: BehaviouralReplayOptions = {}): Promise { const params = this.#readParams(options); const fixtureSeries = await this.#loadFixtures(options); - const usesDatabase = fixtureSeries.length === 0 || Boolean(options.since || options.until); + const fixturesRequested = Boolean(options.bundledFixtures || options.fixturePaths?.length); + const usesDatabase = !fixturesRequested || Boolean(options.since || options.until); const window = usesDatabase ? this.#resolveWindow(options) : null; const databaseSeries = window ? await this.#loadFromDatabase(window) : []; const series = [...databaseSeries, ...fixtureSeries]; From e6b1c35f85d2137e5e1b430086777cfc11e4567f Mon Sep 17 00:00:00 2001 From: Maxime Beauchamp <15185355+baktun14@users.noreply.github.com> Date: Sun, 20 Sep 2026 21:21:11 +0400 Subject: [PATCH 5/5] fix(deployment): apply defaults to blank settings and compare sizes unrounded A declared but empty environment variable never reaches a Zod default, so a blank value for any of the three new settings failed config resolution at boot. They now go through the same blank-to-undefined preprocessor the endpoint list already used. Artifact sizes were rounded to whole megabytes before the threshold comparison, so a file just under the limit counted as sitting at it. The comparison uses the exact size now and only the reported detail rounds. A replay window that starts after it ends matched no rows and read as a clean report. It fails instead. --- .../workload-abuse/config/env.config.spec.ts | 24 +++++++++++++++++++ .../src/workload-abuse/config/env.config.ts | 15 +++++++----- .../accel-without-artifacts.spec.ts | 12 ++++++++++ .../accel-without-artifacts.ts | 4 ++-- .../behavioural-signal-replay.service.spec.ts | 9 +++++++ .../behavioural-signal-replay.service.ts | 3 +++ 6 files changed, 59 insertions(+), 8 deletions(-) diff --git a/apps/api/src/workload-abuse/config/env.config.spec.ts b/apps/api/src/workload-abuse/config/env.config.spec.ts index e6eadace2c..c4858948a9 100644 --- a/apps/api/src/workload-abuse/config/env.config.spec.ts +++ b/apps/api/src/workload-abuse/config/env.config.spec.ts @@ -72,4 +72,28 @@ describe("workload abuse env config", () => { it("falls back to the default schedule when the initial delay setting is blank", () => { expect(envSchema.parse({ WORKLOAD_ABUSE_PROBE_INITIAL_DELAYS_MIN: " " }).WORKLOAD_ABUSE_PROBE_INITIAL_DELAYS_MIN).toEqual([5, 20, 60]); }); + + it("falls back to the behavioural signal defaults when those variables are set but blank", () => { + const config = envSchema.parse({ + WORKLOAD_ABUSE_BEHAVIOURAL_SIGNALS_ENABLED: "", + WORKLOAD_ABUSE_SIGNAL_ACCEL_MIN_VRAM_MB: "", + WORKLOAD_ABUSE_SIGNAL_ARTIFACT_MIN_MB: " " + }); + + expect(config.WORKLOAD_ABUSE_BEHAVIOURAL_SIGNALS_ENABLED).toBe(false); + expect(config.WORKLOAD_ABUSE_SIGNAL_ACCEL_MIN_VRAM_MB).toBe(1024); + expect(config.WORKLOAD_ABUSE_SIGNAL_ARTIFACT_MIN_MB).toBe(256); + }); + + it("still reads the behavioural signal variables when they carry a value", () => { + const config = envSchema.parse({ + WORKLOAD_ABUSE_BEHAVIOURAL_SIGNALS_ENABLED: "true", + WORKLOAD_ABUSE_SIGNAL_ACCEL_MIN_VRAM_MB: "2048", + WORKLOAD_ABUSE_SIGNAL_ARTIFACT_MIN_MB: "512" + }); + + expect(config.WORKLOAD_ABUSE_BEHAVIOURAL_SIGNALS_ENABLED).toBe(true); + expect(config.WORKLOAD_ABUSE_SIGNAL_ACCEL_MIN_VRAM_MB).toBe(2048); + expect(config.WORKLOAD_ABUSE_SIGNAL_ARTIFACT_MIN_MB).toBe(512); + }); }); diff --git a/apps/api/src/workload-abuse/config/env.config.ts b/apps/api/src/workload-abuse/config/env.config.ts index 62154a0060..6d6fd14e7f 100644 --- a/apps/api/src/workload-abuse/config/env.config.ts +++ b/apps/api/src/workload-abuse/config/env.config.ts @@ -111,12 +111,15 @@ export const envSchema = z.object({ WORKLOAD_ABUSE_DOMAIN_BLOCK_MIN_ACCOUNT_AGE_DAYS: z.number({ coerce: true }).int().positive().default(30), /** Evidence rows feed the behavioural replay, so they must outlive its 30-day window with margin. */ WORKLOAD_ABUSE_EVIDENCE_RETENTION_DAYS: z.number({ coerce: true }).int().positive().default(90), - WORKLOAD_ABUSE_BEHAVIOURAL_SIGNALS_ENABLED: z - .enum(["true", "false"]) - .default("false") - .transform(value => value === "true"), - WORKLOAD_ABUSE_SIGNAL_ACCEL_MIN_VRAM_MB: z.number({ coerce: true }).int().positive().default(1_024), - WORKLOAD_ABUSE_SIGNAL_ARTIFACT_MIN_MB: z.number({ coerce: true }).int().positive().default(256), + WORKLOAD_ABUSE_BEHAVIOURAL_SIGNALS_ENABLED: z.preprocess( + blankToUndefined, + z + .enum(["true", "false"]) + .default("false") + .transform(value => value === "true") + ), + WORKLOAD_ABUSE_SIGNAL_ACCEL_MIN_VRAM_MB: z.preprocess(blankToUndefined, z.number({ coerce: true }).int().positive().default(1_024)), + WORKLOAD_ABUSE_SIGNAL_ARTIFACT_MIN_MB: z.preprocess(blankToUndefined, z.number({ coerce: true }).int().positive().default(256)), /** The list itself lives in Doppler: these are our own endpoints, so a reader learns how the exclusion works but not what it covers. */ WORKLOAD_ABUSE_SIGNAL_RELAY_ENDPOINTS: z.preprocess(blankToUndefined, z.string().default("[]").transform(parseRelayEndpoints)) }); diff --git a/apps/api/src/workload-abuse/lib/behavioural-signals/accel-without-artifacts.spec.ts b/apps/api/src/workload-abuse/lib/behavioural-signals/accel-without-artifacts.spec.ts index 12a7f9837f..c89d102d6b 100644 --- a/apps/api/src/workload-abuse/lib/behavioural-signals/accel-without-artifacts.spec.ts +++ b/apps/api/src/workload-abuse/lib/behavioural-signals/accel-without-artifacts.spec.ts @@ -52,6 +52,18 @@ describe("evaluateAccelWithoutArtifacts", () => { expect(evaluateAccelWithoutArtifacts(snapshot, params)).toBeNull(); }); + it("fires when the largest artifact sits just under the threshold", () => { + const { snapshot, params } = setup({ vramMb: 18_000, artifacts: [{ path: "/opt/run", sizeBytes: 268_016_025 }], artifactMinMb: 256 }); + + expect(evaluateAccelWithoutArtifacts(snapshot, params)?.detail).toEqual({ heaviestVramMb: 18_000, largestArtifactMb: 255 }); + }); + + it("stays silent when the largest artifact reaches the threshold exactly", () => { + const { snapshot, params } = setup({ vramMb: 18_000, artifacts: [{ path: "/opt/run", sizeBytes: 268_435_456 }], artifactMinMb: 256 }); + + expect(evaluateAccelWithoutArtifacts(snapshot, params)).toBeNull(); + }); + it("stays silent when the probe collected no disk section", () => { const { snapshot, params } = setup({ vramMb: 18_000, artifacts: null }); diff --git a/apps/api/src/workload-abuse/lib/behavioural-signals/accel-without-artifacts.ts b/apps/api/src/workload-abuse/lib/behavioural-signals/accel-without-artifacts.ts index 90a2bee94e..8fb89c0646 100644 --- a/apps/api/src/workload-abuse/lib/behavioural-signals/accel-without-artifacts.ts +++ b/apps/api/src/workload-abuse/lib/behavioural-signals/accel-without-artifacts.ts @@ -13,7 +13,7 @@ export function evaluateAccelWithoutArtifacts(snapshot: ProbeEvidenceSnapshot, p if (largestArtifactMb >= params.artifactMinMb) return null; - return { signal: BEHAVIOURAL_SIGNALS.accelWithoutArtifacts, detail: { heaviestVramMb, largestArtifactMb } }; + return { signal: BEHAVIOURAL_SIGNALS.accelWithoutArtifacts, detail: { heaviestVramMb, largestArtifactMb: Math.floor(largestArtifactMb) } }; } function findHeaviestVramMb(snapshot: ProbeEvidenceSnapshot): number { @@ -22,6 +22,6 @@ function findHeaviestVramMb(snapshot: ProbeEvidenceSnapshot): number { } function findLargestArtifactMb(snapshot: ProbeEvidenceSnapshot): number { - const sizes = (snapshot.artifacts ?? []).map(artifact => Math.round(artifact.sizeBytes / BYTES_PER_MB)); + const sizes = (snapshot.artifacts ?? []).map(artifact => artifact.sizeBytes / BYTES_PER_MB); return sizes.reduce((largest, sizeMb) => Math.max(largest, sizeMb), 0); } diff --git a/apps/api/src/workload-abuse/services/behavioural-signal-replay/behavioural-signal-replay.service.spec.ts b/apps/api/src/workload-abuse/services/behavioural-signal-replay/behavioural-signal-replay.service.spec.ts index 99ea899c31..d00d726c31 100644 --- a/apps/api/src/workload-abuse/services/behavioural-signal-replay/behavioural-signal-replay.service.spec.ts +++ b/apps/api/src/workload-abuse/services/behavioural-signal-replay/behavioural-signal-replay.service.spec.ts @@ -163,6 +163,15 @@ describe(BehaviouralSignalReplayService.name, () => { expect(evidenceRepository.deleteOlderThan).not.toHaveBeenCalled(); }); + it("refuses a window that starts after it ends instead of reporting an empty one", async () => { + const { service, evidenceRepository } = setup(); + + await expect(service.replay({ since: new Date("2026-09-01T00:00:00.000Z"), until: new Date("2026-08-01T00:00:00.000Z") })).rejects.toThrow( + /starts after it ends/ + ); + expect(evidenceRepository.findCreatedBetween).not.toHaveBeenCalled(); + }); + it("applies the thresholds the operator asked for instead of the configured ones", async () => { const { service } = setup({ rows: [createRow({ id: "row-1", service: "web" })] }); diff --git a/apps/api/src/workload-abuse/services/behavioural-signal-replay/behavioural-signal-replay.service.ts b/apps/api/src/workload-abuse/services/behavioural-signal-replay/behavioural-signal-replay.service.ts index e099cdcc56..a61c19a088 100644 --- a/apps/api/src/workload-abuse/services/behavioural-signal-replay/behavioural-signal-replay.service.ts +++ b/apps/api/src/workload-abuse/services/behavioural-signal-replay/behavioural-signal-replay.service.ts @@ -107,10 +107,13 @@ export class BehaviouralSignalReplayService { }; } + /** A reversed window matches no row, so without this an operator reads an empty report as a clean one. */ #resolveWindow(options: BehaviouralReplayOptions): { since: Date; until: Date } { const until = options.until ?? new Date(); const since = options.since ?? new Date(until.getTime() - DEFAULT_WINDOW_DAYS * 24 * 60 * 60 * 1000); + if (since > until) throw new Error("Replay window starts after it ends"); + return { since, until }; }