From 23144bcdbab817f080758e9d94919fba90276d4a Mon Sep 17 00:00:00 2001 From: Brooke Date: Mon, 17 Aug 2026 22:16:58 -0700 Subject: [PATCH] add dm alert for record creation bursts --- .env.example | 5 +++ README.md | 5 +++ server/src/index.ts | 15 ++++++++- server/src/notifier.ts | 72 +++++++++++++++++++++++++++------------- server/src/sync.ts | 72 ++++++++++++++++++++++++++++++++++++++++ server/test/sync.test.ts | 60 +++++++++++++++++++++++++++++++++ 6 files changed, 205 insertions(+), 24 deletions(-) diff --git a/.env.example b/.env.example index c5f93a0..dd917d0 100644 --- a/.env.example +++ b/.env.example @@ -31,5 +31,10 @@ NOTIFY_APP_PASSWORD= NOTIFY_RECIPIENT= DASHBOARD_URL=http://localhost:5173 +# With DMs enabled, also alert when one account creates this many records within the +# window (spam bursts). Defaults shown; set ACTIVITY_ALERT_CREATES=0 to disable. +# ACTIVITY_ALERT_CREATES=500 +# ACTIVITY_ALERT_WINDOW_MINUTES=60 + # Audit log location (defaults to server/audit.log) # AUDIT_LOG_PATH=/var/log/pds-operator/audit.log diff --git a/README.md b/README.md index e8fa173..0cb04b2 100644 --- a/README.md +++ b/README.md @@ -86,6 +86,11 @@ password created **with** DM access, recipient handle, and the dashboard URL for links. The recipient has to accept DMs from the sender (follow them, or allow DMs from everyone). Unset = disabled. +With DMs enabled you also get an alert when a single account creates a burst of records +(likely spam). The default fires at 500 record creations in 60 minutes, tune with +`ACTIVITY_ALERT_CREATES` and `ACTIVITY_ALERT_WINDOW_MINUTES`, disable with +`ACTIVITY_ALERT_CREATES=0`. At most one alert per account per window. + ## Dev ``` diff --git a/server/src/index.ts b/server/src/index.ts index 2e55344..b4fb37e 100644 --- a/server/src/index.ts +++ b/server/src/index.ts @@ -125,8 +125,21 @@ for (const l of labelers) { .catch((err) => app.log.warn({ err, labeler: l.did }, "label names fetch failed")); } +// DM when one account creates this many records within the window; 0 disables. +// Needs the notifier (NOTIFY_* vars) to have somewhere to send the alert. +const alertCreates = Number(process.env.ACTIVITY_ALERT_CREATES ?? 500); +const alertWindowMinutes = Number(process.env.ACTIVITY_ALERT_WINDOW_MINUTES ?? 60); +const activityAlert = + notifier && Number.isFinite(alertCreates) && alertCreates > 0 && + Number.isFinite(alertWindowMinutes) && alertWindowMinutes > 0 + ? { creates: alertCreates, windowMinutes: alertWindowMinutes } + : null; +if (notifier && !activityAlert) { + app.log.info("record creation burst alerts disabled (ACTIVITY_ALERT_CREATES=0)"); +} + const db = openDb(); -const syncer = new Syncer(db, pds, labelers, app.log, notifier, labelNames); +const syncer = new Syncer(db, pds, labelers, app.log, notifier, labelNames, activityAlert); syncer.start(); for (const sig of ["SIGTERM", "SIGINT"] as const) { diff --git a/server/src/notifier.ts b/server/src/notifier.ts index f8addbc..930a93c 100644 --- a/server/src/notifier.ts +++ b/server/src/notifier.ts @@ -162,35 +162,61 @@ export class BskyDmNotifier { return this.convoId; } - async notifyFlags(account: { did: string; handle: string }, labels: string[]): Promise { - try { - const convoId = await this.ensureConvo(); + /** Send "@handle" plus a dashboard deep link as a rich-text DM. */ + private async sendAccountAlert( + account: { did: string; handle: string }, + prefix: string, + suffix: string, + ): Promise { + const convoId = await this.ensureConvo(); + + // rich text with byte-offset facets: mention the account, link to the dashboard + let text = ""; + const facets: Facet[] = []; + const plain = (s: string) => { + text += s; + }; + const faceted = (s: string, feature: Record) => { + const byteStart = Buffer.byteLength(text); + text += s; + facets.push({ index: { byteStart, byteEnd: Buffer.byteLength(text) }, features: [feature] }); + }; - // rich text with byte-offset facets: mention the account, link to the dashboard - let text = ""; - const facets: Facet[] = []; - const plain = (s: string) => { - text += s; - }; - const faceted = (s: string, feature: Record) => { - const byteStart = Buffer.byteLength(text); - text += s; - facets.push({ index: { byteStart, byteEnd: Buffer.byteLength(text) }, features: [feature] }); - }; + const link = `${this.cfg.dashboardUrl}/?q=${encodeURIComponent(account.handle)}`; + plain(prefix); + faceted(`@${account.handle}`, { $type: "app.bsky.richtext.facet#mention", did: account.did }); + plain(`${suffix}\n`); + faceted(link, { $type: "app.bsky.richtext.facet#link", uri: link }); - const link = `${this.cfg.dashboardUrl}/?q=${encodeURIComponent(account.handle)}`; - plain("⚠️ "); - faceted(`@${account.handle}`, { $type: "app.bsky.richtext.facet#mention", did: account.did }); - plain(` flagged: ${labels.join(", ")}\n`); - faceted(link, { $type: "app.bsky.richtext.facet#link", uri: link }); + await this.chatXrpc("chat.bsky.convo.sendMessage", { + method: "POST", + body: JSON.stringify({ convoId, message: { text, facets } }), + }); + } - await this.chatXrpc("chat.bsky.convo.sendMessage", { - method: "POST", - body: JSON.stringify({ convoId, message: { text, facets } }), - }); + async notifyFlags(account: { did: string; handle: string }, labels: string[]): Promise { + try { + await this.sendAccountAlert(account, "⚠️ ", ` flagged: ${labels.join(", ")}`); this.log.info({ account: account.handle, labels }, "flag notification sent"); } catch (err) { this.log.error({ err, account: account.handle }, "flag notification failed"); } } + + async notifyActivitySpike( + account: { did: string; handle: string }, + creates: number, + windowMinutes: number, + ): Promise { + try { + await this.sendAccountAlert( + account, + "🚨 ", + ` created ${creates.toLocaleString()} records in the last ${windowMinutes} minutes`, + ); + this.log.info({ account: account.handle, creates }, "activity spike notification sent"); + } catch (err) { + this.log.error({ err, account: account.handle }, "activity spike notification failed"); + } + } } diff --git a/server/src/sync.ts b/server/src/sync.ts index 21b3fde..8305c7b 100644 --- a/server/src/sync.ts +++ b/server/src/sync.ts @@ -79,6 +79,12 @@ function attachHeartbeat(ws: WebSocket, intervalMs = 30_000) { const ACCOUNT_STREAM_CURSOR_KEY = "account_stream_cursor"; +/** DM an alert when one account makes this many record creations within the window. */ +export interface ActivityAlertConfig { + creates: number; + windowMinutes: number; +} + export function accountStreamUrl(hostname: string, cursor: string | null): string { return ( `wss://${hostname}/xrpc/com.atproto.sync.subscribeRepos` + @@ -101,6 +107,9 @@ export class Syncer { private timers: NodeJS.Timeout[] = []; // latest firehose seq, persisted on the flush tick — a write per frame would hammer SQLite private accountCursor: number | null = null; + // record creations per account, bucketed by minute, for burst alerts + private createBursts = new Map>(); + private burstAlertedAt = new Map(); constructor( private db: Db, @@ -109,6 +118,7 @@ export class Syncer { private log: FastifyBaseLogger, private notifier: BskyDmNotifier | null = null, private labelNames: Map> = new Map(), + private activityAlert: ActivityAlertConfig | null = null, ) {} private isWatched(labeler: WatchedLabeler, val: string): boolean { @@ -134,6 +144,7 @@ export class Syncer { try { this.flushActivity(); this.flushCursor(); + this.pruneCreateBursts(); } catch (err) { this.log.error({ err }, "activity flush failed"); } @@ -331,6 +342,12 @@ export class Syncer { const at = Number.isNaN(eventTime.getTime()) ? new Date() : eventTime; const key = `${at.toISOString().slice(0, 13)}|${did}`; this.activity.set(key, (this.activity.get(key) ?? 0) + 1); + if (this.activityAlert && this.notifier) { + const creates = Array.isArray(body.ops) + ? body.ops.filter((op: { action?: string }) => op?.action === "create").length + : 0; + if (creates > 0) this.trackCreateBurst(did, creates, at.getTime()); + } } } catch { // ignore undecodable frames @@ -432,6 +449,61 @@ export class Syncer { this.accountCursor = null; } + // ---- create-burst alerts ------------------------------------------------- + + /** + * Sliding window of record creations per account, bucketed by minute. Crossing the + * threshold DMs an alert, at most once per window per account. Buckets key on the + * event's own time and stale events are ignored outright, so cursor-replayed + * history after downtime can't fake a burst. + */ + private trackCreateBurst(did: string, creates: number, eventMs: number) { + const alert = this.activityAlert!; + const windowMs = alert.windowMinutes * 60_000; + const now = Date.now(); + if (eventMs <= now - windowMs) return; + const cutoff = Math.floor((now - windowMs) / 60_000); + let buckets = this.createBursts.get(did); + if (!buckets) { + buckets = new Map(); + this.createBursts.set(did, buckets); + } + const minute = Math.floor(eventMs / 60_000); + buckets.set(minute, (buckets.get(minute) ?? 0) + creates); + let total = 0; + for (const [m, n] of buckets) { + if (m <= cutoff) buckets.delete(m); + else total += n; + } + if (total < alert.creates) return; + if (now - (this.burstAlertedAt.get(did) ?? 0) < windowMs) return; + this.burstAlertedAt.set(did, now); + const row = this.db.prepare("SELECT handle FROM accounts WHERE did = ?").get(did) as + | { handle: string } + | undefined; + this.log.warn({ did, creates: total }, "record creation burst"); + void this.notifier?.notifyActivitySpike( + { did, handle: row?.handle ?? did }, + total, + alert.windowMinutes, + ); + } + + /** Drop burst state that has aged out of the window, so idle accounts don't linger. */ + private pruneCreateBursts() { + if (!this.activityAlert) return; + const windowMs = this.activityAlert.windowMinutes * 60_000; + const now = Date.now(); + const cutoff = Math.floor((now - windowMs) / 60_000); + for (const [did, buckets] of this.createBursts) { + for (const m of buckets.keys()) if (m <= cutoff) buckets.delete(m); + if (buckets.size === 0) this.createBursts.delete(did); + } + for (const [did, at] of this.burstAlertedAt) { + if (now - at >= windowMs) this.burstAlertedAt.delete(did); + } + } + // ---- live label streams -------------------------------------------------- private connectLabelStream(labeler: WatchedLabeler) { diff --git a/server/test/sync.test.ts b/server/test/sync.test.ts index 5d1a268..193dc5a 100644 --- a/server/test/sync.test.ts +++ b/server/test/sync.test.ts @@ -4,9 +4,30 @@ import type { FastifyBaseLogger } from "fastify"; import { Syncer, accountStreamUrl } from "../src/sync.js"; import { openDb, setSyncState, getSyncState } from "../src/db.js"; import type { PdsClient } from "../src/pdsClient.js"; +import type { BskyDmNotifier } from "../src/notifier.js"; const stubLogger = { info() {}, warn() {}, error() {} } as unknown as FastifyBaseLogger; +function burstSyncer(threshold: number) { + const calls: { did: string; creates: number; windowMinutes: number }[] = []; + const notifier = { + notifyActivitySpike: (account: { did: string }, creates: number, windowMinutes: number) => { + calls.push({ did: account.did, creates, windowMinutes }); + return Promise.resolve(); + }, + } as unknown as BskyDmNotifier; + const syncer = new Syncer( + openDb(":memory:"), + { hostname: "pds.test" } as unknown as PdsClient, + [], + stubLogger, + notifier, + new Map(), + { creates: threshold, windowMinutes: 60 }, + ); + return { syncer: syncer as any, calls }; +} + test("flushDirty re-queues dids when the flush fails", async () => { const pds = { hostname: "pds.test", @@ -57,6 +78,45 @@ test("account cursor persists on flush, not per frame", () => { assert.equal(getSyncState(db, "account_stream_cursor"), "99"); }); +test("create burst alerts once when the threshold is crossed", () => { + const { syncer, calls } = burstSyncer(5); + const now = Date.now(); + syncer.trackCreateBurst("did:plc:spam", 3, now); + assert.equal(calls.length, 0, "below threshold must not alert"); + syncer.trackCreateBurst("did:plc:spam", 3, now); + assert.equal(calls.length, 1); + assert.equal(calls[0].creates, 6); + assert.equal(calls[0].windowMinutes, 60); + // still inside the window: more creates must not re-alert + syncer.trackCreateBurst("did:plc:spam", 50, now); + assert.equal(calls.length, 1, "one alert per account per window"); + // a different account keeps its own counter + syncer.trackCreateBurst("did:plc:other", 2, now); + assert.equal(calls.length, 1); +}); + +test("create bursts ignore replayed history and age out", () => { + const { syncer, calls } = burstSyncer(5); + const now = Date.now(); + // events older than the window (e.g. cursor replay after downtime) don't count + syncer.trackCreateBurst("did:plc:x", 100, now - 61 * 60_000); + assert.equal(calls.length, 0); + assert.equal(syncer.createBursts.size, 0); + // in-window creates below threshold stay buffered, then prune drops aged buckets + syncer.trackCreateBurst("did:plc:x", 3, now - 30 * 60_000); + assert.equal(syncer.createBursts.size, 1); + syncer.pruneCreateBursts(); + assert.equal(syncer.createBursts.size, 1, "in-window buckets survive the prune"); + for (const buckets of syncer.createBursts.values()) { + for (const m of [...buckets.keys()]) { + buckets.set(m - 61, buckets.get(m)); + buckets.delete(m); + } + } + syncer.pruneCreateBursts(); + assert.equal(syncer.createBursts.size, 0, "aged buckets are dropped"); +}); + test("account stream url resumes from the persisted cursor", () => { const db = openDb(":memory:"); assert.equal( -- knotmirror 0.0.1