diff --git a/configs/notifications/notifications-config.yml b/configs/notifications/notifications-config.yml index 40d330427..e25331436 100644 --- a/configs/notifications/notifications-config.yml +++ b/configs/notifications/notifications-config.yml @@ -274,3 +274,13 @@ events: torrent_blocker.report: telegram: true webhook: true + + # ════════════════════════════════════════════════════════════════════════════ + # ABUSE BLOCKER EVENTS + # ════════════════════════════════════════════════════════════════════════════ + + # Triggered for alert, temporary-block, repeat-block, and disable incidents. + # Suspicious reports remain database-only. + abuse_blocker.report: + telegram: true + webhook: true diff --git a/libs/contract/api/controllers/node-plugins.ts b/libs/contract/api/controllers/node-plugins.ts index d521f818c..1163a53bb 100644 --- a/libs/contract/api/controllers/node-plugins.ts +++ b/libs/contract/api/controllers/node-plugins.ts @@ -2,6 +2,7 @@ export const NODE_PLUGINS_CONTROLLER = 'node-plugins' as const; const ACTIONS_ROUTE = 'actions' as const; const TORRENT_BLOCKER_ROUTE = 'torrent-blocker' as const; +const ABUSE_BLOCKER_ROUTE = 'abuse-blocker' as const; export const NODE_PLUGINS_ROUTES = { GET_ALL: '', // get @@ -22,4 +23,11 @@ export const NODE_PLUGINS_ROUTES = { GET_REPORTS_STATS: `${TORRENT_BLOCKER_ROUTE}/stats`, TRUNCATE_REPORTS: `${TORRENT_BLOCKER_ROUTE}/truncate`, }, + ABUSE_BLOCKER: { + GET_REPORTS: `${ABUSE_BLOCKER_ROUTE}`, + GET_REPORTS_STATS: `${ABUSE_BLOCKER_ROUTE}/stats`, + GET_REVIEW_QUEUE: `${ABUSE_BLOCKER_ROUTE}/review`, + REVIEW: `${ABUSE_BLOCKER_ROUTE}/review/:userUuid`, + TRUNCATE_REPORTS: `${ABUSE_BLOCKER_ROUTE}/truncate`, + }, } as const; diff --git a/libs/contract/api/routes.ts b/libs/contract/api/routes.ts index 50644e122..fb2359fe0 100644 --- a/libs/contract/api/routes.ts +++ b/libs/contract/api/routes.ts @@ -421,6 +421,14 @@ export const REST_API = { GET_REPORTS_STATS: `${ROOT}/${CONTROLLERS.NODE_PLUGINS_CONTROLLER}/${CONTROLLERS.NODE_PLUGINS_ROUTES.TORRENT_BLOCKER.GET_REPORTS_STATS}`, TRUNCATE_REPORTS: `${ROOT}/${CONTROLLERS.NODE_PLUGINS_CONTROLLER}/${CONTROLLERS.NODE_PLUGINS_ROUTES.TORRENT_BLOCKER.TRUNCATE_REPORTS}`, }, + ABUSE_BLOCKER: { + GET_REPORTS: `${ROOT}/${CONTROLLERS.NODE_PLUGINS_CONTROLLER}/${CONTROLLERS.NODE_PLUGINS_ROUTES.ABUSE_BLOCKER.GET_REPORTS}`, + GET_REPORTS_STATS: `${ROOT}/${CONTROLLERS.NODE_PLUGINS_CONTROLLER}/${CONTROLLERS.NODE_PLUGINS_ROUTES.ABUSE_BLOCKER.GET_REPORTS_STATS}`, + GET_REVIEW_QUEUE: `${ROOT}/${CONTROLLERS.NODE_PLUGINS_CONTROLLER}/${CONTROLLERS.NODE_PLUGINS_ROUTES.ABUSE_BLOCKER.GET_REVIEW_QUEUE}`, + REVIEW: (userUuid: string) => + `${ROOT}/${CONTROLLERS.NODE_PLUGINS_CONTROLLER}/${CONTROLLERS.NODE_PLUGINS_ROUTES.ABUSE_BLOCKER.REVIEW.replace(':userUuid', userUuid)}`, + TRUNCATE_REPORTS: `${ROOT}/${CONTROLLERS.NODE_PLUGINS_CONTROLLER}/${CONTROLLERS.NODE_PLUGINS_ROUTES.ABUSE_BLOCKER.TRUNCATE_REPORTS}`, + }, }, BANDWIDTH_STATS: { NODES: { diff --git a/libs/contract/commands/node-plugins/abuse-blocker/get-abuse-blocker-reports.command.ts b/libs/contract/commands/node-plugins/abuse-blocker/get-abuse-blocker-reports.command.ts new file mode 100644 index 000000000..a84678d30 --- /dev/null +++ b/libs/contract/commands/node-plugins/abuse-blocker/get-abuse-blocker-reports.command.ts @@ -0,0 +1,42 @@ +import { z } from 'zod'; + +import { NODE_PLUGINS_ROUTES, REST_API } from '../../../api'; +import { getEndpointDetails } from '../../../constants'; +import { AbuseBlockerStoredReportSchema } from '../../../models'; + +export namespace GetAbuseBlockerReportsCommand { + export const url = REST_API.NODE_PLUGINS.ABUSE_BLOCKER.GET_REPORTS; + export const TSQ_url = url; + export const endpointDetails = getEndpointDetails( + NODE_PLUGINS_ROUTES.ABUSE_BLOCKER.GET_REPORTS, + 'get', + 'Get Abuse Blocker Reports', + { scope: 'abuse-blocker-reports', kind: 'read' }, + ); + + export const RequestQuerySchema = z + .object({ + start: z.coerce.number().int().min(0).default(0), + size: z.coerce.number().int().min(1).max(500).default(50), + userId: z.coerce.number().int().positive().optional(), + nodeUuid: z.uuid().optional(), + severity: z.enum(['suspicious', 'alert', 'blocked']).optional(), + rule: z.enum(['horizontal_scan', 'destination_sweep']).optional(), + action: z.enum(['none', 'initial_block', 'repeat_block', 'disabled']).optional(), + dateFrom: z.coerce.date().optional(), + dateTo: z.coerce.date().optional(), + }) + .refine((query) => !query.dateFrom || !query.dateTo || query.dateFrom <= query.dateTo, { + message: 'dateFrom must be before or equal to dateTo.', + path: ['dateTo'], + }); + export const ResponseSchema = z.object({ + response: z.object({ + records: z.array(AbuseBlockerStoredReportSchema), + total: z.number(), + }), + }); + + export type RequestQuery = z.infer; + export type Response = z.infer; +} diff --git a/libs/contract/commands/node-plugins/abuse-blocker/get-abuse-blocker-review-queue.command.ts b/libs/contract/commands/node-plugins/abuse-blocker/get-abuse-blocker-review-queue.command.ts new file mode 100644 index 000000000..a5af53d1e --- /dev/null +++ b/libs/contract/commands/node-plugins/abuse-blocker/get-abuse-blocker-review-queue.command.ts @@ -0,0 +1,24 @@ +import { z } from 'zod'; + +import { NODE_PLUGINS_ROUTES, REST_API } from '../../../api'; +import { getEndpointDetails } from '../../../constants'; +import { AbuseBlockerReviewStateSchema } from '../../../models'; + +export namespace GetAbuseBlockerReviewQueueCommand { + export const url = REST_API.NODE_PLUGINS.ABUSE_BLOCKER.GET_REVIEW_QUEUE; + export const endpointDetails = getEndpointDetails( + NODE_PLUGINS_ROUTES.ABUSE_BLOCKER.GET_REVIEW_QUEUE, + 'get', + 'Get Abuse Blocker Manual Review Queue', + { scope: 'abuse-blocker-reports', kind: 'read' }, + ); + export const RequestQuerySchema = z.object({ + start: z.coerce.number().int().min(0).default(0), + size: z.coerce.number().int().min(1).max(500).default(50), + }); + export const ResponseSchema = z.object({ + response: z.object({ records: z.array(AbuseBlockerReviewStateSchema), total: z.number() }), + }); + export type RequestQuery = z.infer; + export type Response = z.infer; +} diff --git a/libs/contract/commands/node-plugins/abuse-blocker/get-abuse-blocker-stats.command.ts b/libs/contract/commands/node-plugins/abuse-blocker/get-abuse-blocker-stats.command.ts new file mode 100644 index 000000000..0a8117a6e --- /dev/null +++ b/libs/contract/commands/node-plugins/abuse-blocker/get-abuse-blocker-stats.command.ts @@ -0,0 +1,40 @@ +import { z } from 'zod'; + +import { NODE_PLUGINS_ROUTES, REST_API } from '../../../api'; +import { getEndpointDetails } from '../../../constants'; + +export namespace GetAbuseBlockerStatsCommand { + export const url = REST_API.NODE_PLUGINS.ABUSE_BLOCKER.GET_REPORTS_STATS; + export const endpointDetails = getEndpointDetails( + NODE_PLUGINS_ROUTES.ABUSE_BLOCKER.GET_REPORTS_STATS, + 'get', + 'Get Abuse Blocker Statistics', + { scope: 'abuse-blocker-reports', kind: 'read' }, + ); + export const ResponseSchema = z.object({ + response: z.object({ + totalReports: z.number(), + reportsLast24Hours: z.number(), + distinctUsers: z.number(), + distinctNodes: z.number(), + manualReviewRequired: z.number(), + bySeverity: z.object({ + suspicious: z.number(), + alert: z.number(), + blocked: z.number(), + }), + topUsers: z.array( + z.object({ userId: z.number(), username: z.string(), total: z.number() }), + ), + topNodes: z.array( + z.object({ + uuid: z.uuid(), + name: z.string(), + countryCode: z.string(), + total: z.number(), + }), + ), + }), + }); + export type Response = z.infer; +} diff --git a/libs/contract/commands/node-plugins/abuse-blocker/index.ts b/libs/contract/commands/node-plugins/abuse-blocker/index.ts new file mode 100644 index 000000000..9ed5255c3 --- /dev/null +++ b/libs/contract/commands/node-plugins/abuse-blocker/index.ts @@ -0,0 +1,5 @@ +export * from './get-abuse-blocker-reports.command'; +export * from './get-abuse-blocker-review-queue.command'; +export * from './get-abuse-blocker-stats.command'; +export * from './review-abuse-blocker-user.command'; +export * from './truncate-abuse-blocker-reports.command'; diff --git a/libs/contract/commands/node-plugins/abuse-blocker/review-abuse-blocker-user.command.ts b/libs/contract/commands/node-plugins/abuse-blocker/review-abuse-blocker-user.command.ts new file mode 100644 index 000000000..a1f19ac9d --- /dev/null +++ b/libs/contract/commands/node-plugins/abuse-blocker/review-abuse-blocker-user.command.ts @@ -0,0 +1,24 @@ +import { z } from 'zod'; + +import { NODE_PLUGINS_ROUTES, REST_API } from '../../../api'; +import { getEndpointDetails } from '../../../constants'; +import { AbuseBlockerReviewStateSchema } from '../../../models'; + +export namespace ReviewAbuseBlockerUserCommand { + export const url = REST_API.NODE_PLUGINS.ABUSE_BLOCKER.REVIEW; + export const TSQ_url = url(':userUuid'); + export const endpointDetails = getEndpointDetails( + NODE_PLUGINS_ROUTES.ABUSE_BLOCKER.REVIEW, + 'post', + 'Resolve Abuse Blocker Manual Review', + { scope: 'abuse-blocker-reports', kind: 'write' }, + ); + export const RequestParamSchema = z.object({ + userUuid: z.uuid().describe('User VLESS UUID used by the current Users model.'), + }); + export const RequestBodySchema = z.object({ action: z.enum(['enable', 'keep_disabled']) }); + export const ResponseSchema = z.object({ response: AbuseBlockerReviewStateSchema }); + export type RequestParam = z.infer; + export type RequestBody = z.infer; + export type Response = z.infer; +} diff --git a/libs/contract/commands/node-plugins/abuse-blocker/truncate-abuse-blocker-reports.command.ts b/libs/contract/commands/node-plugins/abuse-blocker/truncate-abuse-blocker-reports.command.ts new file mode 100644 index 000000000..f0b23d741 --- /dev/null +++ b/libs/contract/commands/node-plugins/abuse-blocker/truncate-abuse-blocker-reports.command.ts @@ -0,0 +1,12 @@ +import { NODE_PLUGINS_ROUTES, REST_API } from '../../../api'; +import { getEndpointDetails } from '../../../constants'; + +export namespace TruncateAbuseBlockerReportsCommand { + export const url = REST_API.NODE_PLUGINS.ABUSE_BLOCKER.TRUNCATE_REPORTS; + export const endpointDetails = getEndpointDetails( + NODE_PLUGINS_ROUTES.ABUSE_BLOCKER.TRUNCATE_REPORTS, + 'delete', + 'Truncate Abuse Blocker Reports', + { scope: 'abuse-blocker-reports', kind: 'write' }, + ); +} diff --git a/libs/contract/commands/node-plugins/index.ts b/libs/contract/commands/node-plugins/index.ts index 5db89d774..8a2efbccf 100644 --- a/libs/contract/commands/node-plugins/index.ts +++ b/libs/contract/commands/node-plugins/index.ts @@ -1,4 +1,5 @@ export * from './actions'; +export * from './abuse-blocker'; export * from './create-node-plugin.command'; export * from './delete-node-plugin.command'; export * from './executor.command'; diff --git a/libs/contract/constants/events/events.ts b/libs/contract/constants/events/events.ts index 0b2c730d3..0c59abea9 100644 --- a/libs/contract/constants/events/events.ts +++ b/libs/contract/constants/events/events.ts @@ -71,6 +71,9 @@ export const EVENTS = { TORRENT_BLOCKER: { REPORT: 'torrent_blocker.report', }, + ABUSE_BLOCKER: { + REPORT: 'abuse_blocker.report', + }, CATCH_ALL_USER_EVENTS: 'user.*', CATCH_ALL_USER_HWID_DEVICES_EVENTS: 'user_hwid_devices.*', CATCH_ALL_NODE_EVENTS: 'node.*', @@ -78,6 +81,7 @@ export const EVENTS = { CATCH_ALL_ERRORS_EVENTS: 'errors.*', CATCH_ALL_CRM_EVENTS: 'crm.*', CATCH_ALL_TORRENT_BLOCKER_EVENTS: 'torrent_blocker.*', + CATCH_ALL_ABUSE_BLOCKER_EVENTS: 'abuse_blocker.*', } as const; export type TNodeEvents = (typeof EVENTS.NODE)[keyof typeof EVENTS.NODE]; @@ -90,6 +94,7 @@ export type TUserHwidDevicesEvents = export type TTorrentBlockerEvents = (typeof EVENTS.TORRENT_BLOCKER)[keyof typeof EVENTS.TORRENT_BLOCKER]; +export type TAbuseBlockerEvents = (typeof EVENTS.ABUSE_BLOCKER)[keyof typeof EVENTS.ABUSE_BLOCKER]; export type TAllEvents = | TUserEvents @@ -98,7 +103,8 @@ export type TAllEvents = | TErrorsEvents | TCRMEvents | TUserHwidDevicesEvents - | TTorrentBlockerEvents; + | TTorrentBlockerEvents + | TAbuseBlockerEvents; export type TAllEventChannels = 'telegram' | 'webhook'; export const EVENTS_SCOPES = { @@ -109,6 +115,7 @@ export const EVENTS_SCOPES = { ERRORS: 'errors', CRM: 'crm', TORRENT_BLOCKER: 'torrent_blocker', + ABUSE_BLOCKER: 'abuse_blocker', } as const; export type TEventsScope = (typeof EVENTS_SCOPES)[keyof typeof EVENTS_SCOPES]; diff --git a/libs/contract/models/abuse-blocker-report.schema.ts b/libs/contract/models/abuse-blocker-report.schema.ts new file mode 100644 index 000000000..7db2599e7 --- /dev/null +++ b/libs/contract/models/abuse-blocker-report.schema.ts @@ -0,0 +1,44 @@ +import { z } from 'zod'; + +import { NodesSchema } from './nodes.schema'; + +export const AbuseBlockerStoredReportSchema = z.object({ + eventId: z.uuid(), + userId: z.number(), + nodeId: z.number(), + severity: z.enum(['suspicious', 'alert', 'blocked']), + score: z.number(), + sourceIp: z.string(), + action: z.enum(['none', 'initial_block', 'repeat_block', 'disabled']), + detectedAt: z.coerce.date(), + report: z.unknown(), + createdAt: z.coerce.date(), + updatedAt: z.coerce.date(), + user: z.object({ + username: z.string(), + vlessUuid: z.uuid(), + status: z.string(), + }), + node: NodesSchema.pick({ + uuid: true, + name: true, + countryCode: true, + }), +}); + +export const AbuseBlockerReviewStateSchema = z.object({ + userId: z.number(), + strikeLevel: z.number(), + lastBlockingIncidentAt: z.coerce.date().nullable(), + manualReviewRequired: z.boolean(), + disabledByPlugin: z.boolean(), + reviewRequestedAt: z.coerce.date().nullable(), + reviewedAt: z.coerce.date().nullable(), + reviewAction: z.enum(['enable', 'keep_disabled']).nullable(), + updatedAt: z.coerce.date(), + user: z.object({ + username: z.string(), + vlessUuid: z.uuid(), + status: z.string(), + }), +}); diff --git a/libs/contract/models/index.ts b/libs/contract/models/index.ts index d5bff94bb..ec855c944 100644 --- a/libs/contract/models/index.ts +++ b/libs/contract/models/index.ts @@ -38,3 +38,4 @@ export * from './webhook'; export * from './xray-json-advanced'; export * from './path-params.schema'; export * from './host-mapper'; +export * from './abuse-blocker-report.schema'; diff --git a/libs/contract/models/webhook/webhook.schema.ts b/libs/contract/models/webhook/webhook.schema.ts index a407bac8e..0cc167aa2 100644 --- a/libs/contract/models/webhook/webhook.schema.ts +++ b/libs/contract/models/webhook/webhook.schema.ts @@ -156,6 +156,21 @@ export const RemnawaveWebhookTorrentBlockerEvents = z.object({ }), }), }); +export const RemnawaveWebhookAbuseBlockerEvents = z.object({ + scope: z.literal(EVENTS_SCOPES.ABUSE_BLOCKER), + event: z.enum(toZodEnum(EVENTS.ABUSE_BLOCKER)), + timestamp: z + .string() + .datetime() + .transform((str) => new Date(str)), + data: z.object({ + node: NodesSchema, + user: ExtendedUsersSchema, + report: z.unknown(), + backendAction: z.enum(['none', 'initial_block', 'repeat_block', 'disabled']), + strikeLevel: z.number(), + }), +}); export const RemnawaveWebhookEventSchema = z.discriminatedUnion('scope', [ RemnawaveWebhookUserEvents, RemnawaveWebhookUserHwidDevicesEvents, @@ -164,6 +179,7 @@ export const RemnawaveWebhookEventSchema = z.discriminatedUnion('scope', [ RemnawaveWebhookErrorsEvents, RemnawaveWebhookCrmEvents, RemnawaveWebhookTorrentBlockerEvents, + RemnawaveWebhookAbuseBlockerEvents, ]); export type TRemnawaveWebhookEvent = z.infer; @@ -179,3 +195,4 @@ export type TRemnawaveWebhookUserHwidDevicesEvent = z.infer< export type TRemnawaveWebhookTorrentBlockerEvent = z.infer< typeof RemnawaveWebhookTorrentBlockerEvents >; +export type TRemnawaveWebhookAbuseBlockerEvent = z.infer; diff --git a/prisma/migrations/20260815110000_abuse_blocker/migration.sql b/prisma/migrations/20260815110000_abuse_blocker/migration.sql new file mode 100644 index 000000000..ead63abb3 --- /dev/null +++ b/prisma/migrations/20260815110000_abuse_blocker/migration.sql @@ -0,0 +1,51 @@ +CREATE TABLE "abuse_blocker_reports" ( + "event_id" UUID NOT NULL, + "user_id" BIGINT NOT NULL, + "node_id" BIGINT NOT NULL, + "severity" VARCHAR(16) NOT NULL, + "score" INTEGER NOT NULL, + "source_ip" VARCHAR(45) NOT NULL, + "action" VARCHAR(32) NOT NULL, + "detected_at" TIMESTAMP(3) NOT NULL, + "report" JSONB NOT NULL, + "created_at" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP, + "updated_at" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP, + + CONSTRAINT "abuse_blocker_reports_pkey" PRIMARY KEY ("event_id") +); + +CREATE TABLE "abuse_blocker_user_state" ( + "user_id" BIGINT NOT NULL, + "strike_level" INTEGER NOT NULL DEFAULT 0, + "last_blocking_incident_at" TIMESTAMP(3), + "manual_review_required" BOOLEAN NOT NULL DEFAULT false, + "disabled_by_plugin" BOOLEAN NOT NULL DEFAULT false, + "review_requested_at" TIMESTAMP(3), + "reviewed_at" TIMESTAMP(3), + "review_action" VARCHAR(20), + "created_at" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP, + "updated_at" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP, + + CONSTRAINT "abuse_blocker_user_state_pkey" PRIMARY KEY ("user_id") +); + +CREATE INDEX "abuse_blocker_reports_user_id_detected_at_idx" + ON "abuse_blocker_reports"("user_id", "detected_at"); +CREATE INDEX "abuse_blocker_reports_node_id_detected_at_idx" + ON "abuse_blocker_reports"("node_id", "detected_at"); +CREATE INDEX "abuse_blocker_reports_severity_detected_at_idx" + ON "abuse_blocker_reports"("severity", "detected_at"); +CREATE INDEX "abuse_blocker_reports_action_detected_at_idx" + ON "abuse_blocker_reports"("action", "detected_at"); +CREATE INDEX "abuse_blocker_user_state_manual_review_required_updated_at_idx" + ON "abuse_blocker_user_state"("manual_review_required", "updated_at"); + +ALTER TABLE "abuse_blocker_reports" + ADD CONSTRAINT "abuse_blocker_reports_user_id_fkey" + FOREIGN KEY ("user_id") REFERENCES "users"("id") ON DELETE CASCADE ON UPDATE CASCADE; +ALTER TABLE "abuse_blocker_reports" + ADD CONSTRAINT "abuse_blocker_reports_node_id_fkey" + FOREIGN KEY ("node_id") REFERENCES "nodes"("id") ON DELETE CASCADE ON UPDATE CASCADE; +ALTER TABLE "abuse_blocker_user_state" + ADD CONSTRAINT "abuse_blocker_user_state_user_id_fkey" + FOREIGN KEY ("user_id") REFERENCES "users"("id") ON DELETE CASCADE ON UPDATE CASCADE; diff --git a/prisma/schema.prisma b/prisma/schema.prisma index 383929a55..999eef5fa 100644 --- a/prisma/schema.prisma +++ b/prisma/schema.prisma @@ -75,6 +75,8 @@ model Users { userSubscriptionRequestHistory UserSubscriptionRequestHistory[] userMetas UserMeta[] torrentBlockerReports TorrentBlockerReports[] + abuseBlockerReports AbuseBlockerReports[] + abuseBlockerState AbuseBlockerUserState? traffic UserTraffic? externalSquad ExternalSquads? @relation(fields: [externalSquadUuid], references: [uuid], onDelete: SetNull) @@ -210,6 +212,7 @@ model Nodes { connectedUsers UserTraffic[] nodeMetas NodeMeta[] torrentBlockerReports TorrentBlockerReports[] + abuseBlockerReports AbuseBlockerReports[] activeConfigProfile ConfigProfiles? @relation(fields: [activeConfigProfileUuid], references: [uuid], onDelete: SetNull) provider InfraProviders? @relation(fields: [providerUuid], references: [uuid], onDelete: SetNull) @@ -646,3 +649,46 @@ model TorrentBlockerReports { @@map("torrent_blocker_reports") } + +model AbuseBlockerReports { + eventId String @id @map("event_id") @db.Uuid + userId BigInt @map("user_id") + nodeId BigInt @map("node_id") + severity String @map("severity") @db.VarChar(16) + score Int @map("score") + sourceIp String @map("source_ip") @db.VarChar(45) + action String @map("action") @db.VarChar(32) + detectedAt DateTime @map("detected_at") + report Json @map("report") + + createdAt DateTime @default(now()) @map("created_at") + updatedAt DateTime @default(now()) @updatedAt @map("updated_at") + + user Users @relation(fields: [userId], references: [id], onDelete: Cascade) + node Nodes @relation(fields: [nodeId], references: [id], onDelete: Cascade) + + @@index([userId, detectedAt]) + @@index([nodeId, detectedAt]) + @@index([severity, detectedAt]) + @@index([action, detectedAt]) + @@map("abuse_blocker_reports") +} + +model AbuseBlockerUserState { + userId BigInt @id @map("user_id") + strikeLevel Int @default(0) @map("strike_level") + lastBlockingIncidentAt DateTime? @map("last_blocking_incident_at") + manualReviewRequired Boolean @default(false) @map("manual_review_required") + disabledByPlugin Boolean @default(false) @map("disabled_by_plugin") + reviewRequestedAt DateTime? @map("review_requested_at") + reviewedAt DateTime? @map("reviewed_at") + reviewAction String? @map("review_action") @db.VarChar(20) + + createdAt DateTime @default(now()) @map("created_at") + updatedAt DateTime @default(now()) @updatedAt @map("updated_at") + + user Users @relation(fields: [userId], references: [id], onDelete: Cascade) + + @@index([manualReviewRequired, updatedAt]) + @@map("abuse_blocker_user_state") +} diff --git a/src/common/axios/axios.service.ts b/src/common/axios/axios.service.ts index 65cd2b129..2b822e53c 100644 --- a/src/common/axios/axios.service.ts +++ b/src/common/axios/axios.service.ts @@ -18,6 +18,7 @@ import { AddUsersCommand, BlockIpsCommand, CollectReportsCommand, + CollectAbuseBlockerReportsCommand, DropIpsCommand, DropUsersConnectionsCommand, GetCombinedStatsCommand, @@ -27,6 +28,7 @@ import { GetUsersIpListCommand, GetUsersStatsCommand, RecreateTablesCommand, + RefreshAbuseBlockCommand, RemoveUserCommand, RemoveUsersCommand, StartXrayCommand, @@ -433,6 +435,31 @@ export class AxiosService { }); } + public async collectAbuseBlockerReports( + opts: INodeConnectionOpts, + ): Promise> { + return this.request({ + label: 'COLLECT ABUSE BLOCKER REPORTS', + path: CollectAbuseBlockerReportsCommand.url, + opts, + logAxiosError: false, + timeout: 20_000, + }); + } + + public async refreshAbuseBlock( + data: RefreshAbuseBlockCommand.Request, + opts: INodeConnectionOpts, + ): Promise> { + return this.request({ + label: 'REFRESH ABUSE BLOCK', + path: RefreshAbuseBlockCommand.url, + opts, + data, + timeout: 10_000, + }); + } + public async blockIps( data: BlockIpsCommand.Request, opts: INodeConnectionOpts, diff --git a/src/common/config/app-config/config.schema.ts b/src/common/config/app-config/config.schema.ts index f76195442..ba52bbdbf 100644 --- a/src/common/config/app-config/config.schema.ts +++ b/src/common/config/app-config/config.schema.ts @@ -54,6 +54,7 @@ export const configSchema = z TELEGRAM_NOTIFY_CRM: z.string().optional(), TELEGRAM_NOTIFY_SERVICE: z.string().optional(), TELEGRAM_NOTIFY_TBLOCKER: z.string().optional(), + TELEGRAM_NOTIFY_ABUSE_BLOCKER: z.string().optional(), FRONT_END_DOMAIN: z.string(), PANEL_DOMAIN: z.string().optional(), diff --git a/src/common/config/app-config/notifications.config.ts b/src/common/config/app-config/notifications.config.ts index 057802ac1..7e30cd7d7 100644 --- a/src/common/config/app-config/notifications.config.ts +++ b/src/common/config/app-config/notifications.config.ts @@ -17,6 +17,7 @@ const ALL_EVENTS = [ ...Object.values(EVENTS.ERRORS), ...Object.values(EVENTS.CRM), ...Object.values(EVENTS.TORRENT_BLOCKER), + ...Object.values(EVENTS.ABUSE_BLOCKER), ] as const; const eventConfigSchema = z.object({ diff --git a/src/common/utils/startup-app/docs.ts b/src/common/utils/startup-app/docs.ts index 29f1b53f7..5d04b7b61 100644 --- a/src/common/utils/startup-app/docs.ts +++ b/src/common/utils/startup-app/docs.ts @@ -18,6 +18,7 @@ import { RemnawaveWebhookUserEventsDto, RemnawaveWebhookUserHwidDevicesEventsDto, RemnawaveWebhookTorrentBlockerEventsDto, + RemnawaveWebhookAbuseBlockerEventsDto, RemnawaveNotFoundErrorDto, RemnawaveBadRequestErrorDto, RemnawaveInternalServerErrorDto, @@ -115,6 +116,7 @@ export async function getDocs(app: INestApplication) { RemnawaveWebhookErrorsEventsDto, RemnawaveWebhookCrmEventsDto, RemnawaveWebhookTorrentBlockerEventsDto, + RemnawaveWebhookAbuseBlockerEventsDto, RemnawaveNotFoundErrorDto, RemnawaveBadRequestErrorDto, RemnawaveInternalServerErrorDto, diff --git a/src/common/utils/startup-app/extra-models.ts b/src/common/utils/startup-app/extra-models.ts index 6c4848cba..90d5b7e7a 100644 --- a/src/common/utils/startup-app/extra-models.ts +++ b/src/common/utils/startup-app/extra-models.ts @@ -10,6 +10,7 @@ import { RemnawaveWebhookUserEvents, RemnawaveWebhookUserHwidDevicesEvents, RemnawaveWebhookTorrentBlockerEvents, + RemnawaveWebhookAbuseBlockerEvents, RemnawaveUserUsageStreamMessageSchema, RemnawaveSubscriptionRequestStreamMessageSchema, RemnawaveNodeConnectionsStreamMessageSchema, @@ -26,6 +27,9 @@ export class RemnawaveWebhookCrmEventsDto extends createZodDto(RemnawaveWebhookC export class RemnawaveWebhookTorrentBlockerEventsDto extends createZodDto( RemnawaveWebhookTorrentBlockerEvents, ) {} +export class RemnawaveWebhookAbuseBlockerEventsDto extends createZodDto( + RemnawaveWebhookAbuseBlockerEvents, +) {} export class RemnawaveUserUsageStreamMessageDto extends createZodDto( RemnawaveUserUsageStreamMessageSchema, diff --git a/src/common/utils/startup-app/gh-actions-docs.ts b/src/common/utils/startup-app/gh-actions-docs.ts index 9362628d6..68b4fccb6 100644 --- a/src/common/utils/startup-app/gh-actions-docs.ts +++ b/src/common/utils/startup-app/gh-actions-docs.ts @@ -16,6 +16,7 @@ import { RemnawaveWebhookUserEventsDto, RemnawaveWebhookUserHwidDevicesEventsDto, RemnawaveWebhookTorrentBlockerEventsDto, + RemnawaveWebhookAbuseBlockerEventsDto, RemnawaveInternalServerErrorDto, RemnawaveValidationErrorDto, RemnawaveBadRequestErrorDto, @@ -113,6 +114,7 @@ export async function ghActionsDocs(app: INestApplication) { RemnawaveWebhookErrorsEventsDto, RemnawaveWebhookCrmEventsDto, RemnawaveWebhookTorrentBlockerEventsDto, + RemnawaveWebhookAbuseBlockerEventsDto, RemnawaveInternalServerErrorDto, RemnawaveValidationErrorDto, RemnawaveBadRequestErrorDto, diff --git a/src/integration-modules/notifications/interfaces/abuse-blocker.event.interface.ts b/src/integration-modules/notifications/interfaces/abuse-blocker.event.interface.ts new file mode 100644 index 000000000..7892a64b3 --- /dev/null +++ b/src/integration-modules/notifications/interfaces/abuse-blocker.event.interface.ts @@ -0,0 +1,21 @@ +import type { AbuseBlockerReportModel } from '@remnawave/node-contract'; + +import { TAbuseBlockerEvents } from '@libs/contracts/constants'; + +import { NodesEntity } from '@modules/nodes/entities/nodes.entity'; +import { UserEntity } from '@modules/users/entities'; + +type AbuseBlockerBackendAction = 'none' | 'initial_block' | 'repeat_block' | 'disabled'; + +export class AbuseBlockerEvent { + constructor( + public readonly data: { + node: NodesEntity; + user: UserEntity; + report: AbuseBlockerReportModel; + backendAction: AbuseBlockerBackendAction; + strikeLevel: number; + }, + public readonly eventName: TAbuseBlockerEvents, + ) {} +} diff --git a/src/integration-modules/notifications/interfaces/index.ts b/src/integration-modules/notifications/interfaces/index.ts index bfb437b0b..471273b81 100644 --- a/src/integration-modules/notifications/interfaces/index.ts +++ b/src/integration-modules/notifications/interfaces/index.ts @@ -5,3 +5,4 @@ export * from './service.event.interface'; export * from './torrent-blocker.event.interface'; export * from './user-hwid-device.event.interface'; export * from './user.event.interface'; +export * from './abuse-blocker.event.interface'; diff --git a/src/integration-modules/notifications/telegram-bot/events/abuse-blocker/abuse-blocker.events.templates.ts b/src/integration-modules/notifications/telegram-bot/events/abuse-blocker/abuse-blocker.events.templates.ts new file mode 100644 index 000000000..135833223 --- /dev/null +++ b/src/integration-modules/notifications/telegram-bot/events/abuse-blocker/abuse-blocker.events.templates.ts @@ -0,0 +1,55 @@ +import { EVENTS, TAbuseBlockerEvents } from '@libs/contracts/constants'; + +import { AbuseBlockerEvent } from '@integration-modules/notifications/interfaces'; + +import { PANEL_URLS } from '@queue/notifications/telegram-bot-logger/enums'; +import { IInlineKeyboard } from '@queue/notifications/telegram-bot-logger/interfaces/inline-keyboard.interface'; + +export type AbuseBlockerEventsTemplate = ( + event: AbuseBlockerEvent, + panelDomain: string | undefined, +) => { message: string; keyboard?: IInlineKeyboard[] }; + +export const ABUSE_BLOCKER_EVENTS_TEMPLATES: Record< + TAbuseBlockerEvents, + AbuseBlockerEventsTemplate +> = { + [EVENTS.ABUSE_BLOCKER.REPORT]: (event, panelDomain) => { + const { report, backendAction, strikeLevel } = event.data; + const rules = report.detections.map((item) => item.rule).join(', '); + const lines = [ + `🚨 #abuseBlocker #${event.data.user.username}`, + `🖥 ${event.data.node.name} (${event.data.node.address})`, + `🤖 ${event.data.user.username} (${event.data.user.id})`, + '', + '
', + `Severity: ${report.severity}`, + `Score: ${report.score.before} + ${report.score.delta} = ${report.score.after}`, + `Rules: ${rules}`, + `Source: ${report.sourceIp}`, + `Destination: ${report.destinationIp}:${report.destinationPort}`, + `Backend action: ${backendAction}`, + `Strike: ${strikeLevel}`, + '
', + ]; + return { + message: lines.join('\n'), + keyboard: buildUserKeyboard(event.data.user.id.toString(), panelDomain), + }; + }, +}; + +const buildUserKeyboard = ( + userId: string, + panelDomain: string | undefined, +): IInlineKeyboard[] | undefined => + panelDomain + ? [ + { + url: PANEL_URLS.USER(panelDomain, userId), + text: 'View user', + customEmoji: '5282843764451195532', + style: 'primary' as const, + }, + ] + : undefined; diff --git a/src/integration-modules/notifications/telegram-bot/events/abuse-blocker/abuse-blocker.events.ts b/src/integration-modules/notifications/telegram-bot/events/abuse-blocker/abuse-blocker.events.ts new file mode 100644 index 000000000..1d993d9d7 --- /dev/null +++ b/src/integration-modules/notifications/telegram-bot/events/abuse-blocker/abuse-blocker.events.ts @@ -0,0 +1,60 @@ +import { Injectable, Logger, OnApplicationBootstrap } from '@nestjs/common'; +import { EventEmitter2 } from '@nestjs/event-emitter'; + +import { TypedConfigService } from '@common/config/app-config'; +import { NotificationsConfigService } from '@common/config/common-config'; +import { TAbuseBlockerEvents } from '@libs/contracts/constants'; + +import { AbuseBlockerEvent } from '@integration-modules/notifications/interfaces'; + +import { TelegramBotLoggerQueueService } from '@queue/notifications/telegram-bot-logger'; + +import { + ABUSE_BLOCKER_EVENTS_TEMPLATES, + AbuseBlockerEventsTemplate, +} from './abuse-blocker.events.templates'; + +@Injectable() +export class AbuseBlockerEvents implements OnApplicationBootstrap { + private readonly logger = new Logger(AbuseBlockerEvents.name); + private readonly chatId: string | undefined; + private readonly threadId: string | undefined; + private readonly panelDomain: string | undefined; + + constructor( + private readonly eventEmitter: EventEmitter2, + private readonly notificationsConfig: NotificationsConfigService, + private readonly telegramQueue: TelegramBotLoggerQueueService, + configService: TypedConfigService, + ) { + this.panelDomain = configService.get('PANEL_DOMAIN'); + const target = configService.get('TELEGRAM_NOTIFY_ABUSE_BLOCKER'); + if (target) [this.chatId, this.threadId] = target.split(':'); + } + + onApplicationBootstrap(): void { + if (!this.chatId) return; + for (const [eventName, template] of Object.entries(ABUSE_BLOCKER_EVENTS_TEMPLATES)) { + if (!this.notificationsConfig.isEnabled(eventName as TAbuseBlockerEvents, 'telegram')) { + this.logger.debug(`Event "${eventName}" is not enabled for Telegram`); + continue; + } + this.eventEmitter.on(eventName, (event: AbuseBlockerEvent) => + this.handleEvent(event, template), + ); + } + } + + private async handleEvent( + event: AbuseBlockerEvent, + template: AbuseBlockerEventsTemplate, + ): Promise { + const message = template(event, this.panelDomain); + await this.telegramQueue.addJobToSendTelegramMessage({ + message: message.message, + chatId: this.chatId!, + threadId: this.threadId, + keyboard: message.keyboard, + }); + } +} diff --git a/src/integration-modules/notifications/telegram-bot/events/index.ts b/src/integration-modules/notifications/telegram-bot/events/index.ts index 2d39bc1c3..6deef2466 100644 --- a/src/integration-modules/notifications/telegram-bot/events/index.ts +++ b/src/integration-modules/notifications/telegram-bot/events/index.ts @@ -5,9 +5,11 @@ import { TorrentBlockerEvents } from './torrent-blocker/torrent-blocker.events'; import { UsersEvents } from './users/users.events'; export const TELEGRAM_BOT_EVENTS = [ + AbuseBlockerEvents, UsersEvents, NodesEvents, ServiceEvents, CrmEvents, TorrentBlockerEvents, ]; +import { AbuseBlockerEvents } from './abuse-blocker/abuse-blocker.events'; diff --git a/src/integration-modules/notifications/webhook-module/events/webhook.events.ts b/src/integration-modules/notifications/webhook-module/events/webhook.events.ts index bd84df951..ae55ce8f9 100644 --- a/src/integration-modules/notifications/webhook-module/events/webhook.events.ts +++ b/src/integration-modules/notifications/webhook-module/events/webhook.events.ts @@ -17,6 +17,7 @@ import { CrmEvent, UserHwidDeviceEvent, TorrentBlockerEvent, + AbuseBlockerEvent, } from '@integration-modules/notifications/interfaces'; import { BaseUserHwidDevicesResponseModel } from '@modules/hwid-user-devices/models'; @@ -258,6 +259,34 @@ export class WebhookEvents { } } + @OnEvent(EVENTS.CATCH_ALL_ABUSE_BLOCKER_EVENTS) + async onCatchAllAbuseBlockerEvents(event: AbuseBlockerEvent): Promise { + try { + if (!this.notificationsConfig.isEnabled(event.eventName, 'webhook')) return; + + const payload = { + scope: EVENTS_SCOPES.ABUSE_BLOCKER, + event: event.eventName, + timestamp: dayjs().toISOString(), + data: { + ...event.data, + node: new NodeResponseModel( + event.data.node, + await this.getNodesSystemInfo(event.data.node.uuid), + ), + user: new GetFullUserResponseModel(event.data.user, this.subPublicDomain), + }, + }; + const { json } = serialize(payload); + await this.webhookLoggerQueueService.sendWebhooks( + { payload: JSON.stringify(json), timestamp: payload.timestamp }, + [...this.webhookUrls, ...this.notificationsConfig.getWebhookUrls(event.eventName)], + ); + } catch (error) { + this.logger.error(`Error sending Abuse Blocker webhook event: ${error}`); + } + } + private async getNodesSystemInfo(uuid: string): Promise { const [info, stats, onlineUsers, xrayUptime, versions] = await Promise.all([ this.rawCacheService.get(CACHE_KEYS.NODE_SYSTEM_INFO(uuid)), diff --git a/src/modules/node-plugins/abuse-blocker.controller.ts b/src/modules/node-plugins/abuse-blocker.controller.ts new file mode 100644 index 000000000..49dc4b45e --- /dev/null +++ b/src/modules/node-plugins/abuse-blocker.controller.ts @@ -0,0 +1,99 @@ +import { CONTROLLERS_INFO, NODE_PLUGINS_CONTROLLER } from '@contract/api'; +import { ROLE } from '@contract/constants'; + +import { Body, Controller, HttpStatus, Param, Query, UseFilters, UseGuards } from '@nestjs/common'; +import { ApiBearerAuth, ApiTags } from '@nestjs/swagger'; + +import { Endpoint } from '@common/decorators/base-endpoint'; +import { Roles } from '@common/decorators/roles/roles'; +import { ApiScopeResource } from '@common/decorators/scopes'; +import { HttpExceptionFilter } from '@common/exception/http-exception.filter'; +import { JwtDefaultGuard } from '@common/guards/jwt-guards/def-jwt-guard'; +import { RolesGuard } from '@common/guards/roles'; +import { ScopesGuard } from '@common/guards/scopes'; +import { errorHandler } from '@common/helpers/error-handler.helper'; +import { + GetAbuseBlockerReportsCommand, + GetAbuseBlockerReviewQueueCommand, + GetAbuseBlockerStatsCommand, + ReviewAbuseBlockerUserCommand, + TruncateAbuseBlockerReportsCommand, +} from '@libs/contracts/commands'; + +import { AbuseBlockerService } from './abuse-blocker.service'; +import { + GetAbuseBlockerReportsQueryDto, + GetAbuseBlockerReportsResponseDto, + GetAbuseBlockerReviewQueueQueryDto, + GetAbuseBlockerReviewQueueResponseDto, + GetAbuseBlockerStatsResponseDto, + ReviewAbuseBlockerUserBodyDto, + ReviewAbuseBlockerUserParamDto, + ReviewAbuseBlockerUserResponseDto, +} from './dtos/node-plugins.dtos'; + +@ApiBearerAuth('Authorization') +@ApiScopeResource(CONTROLLERS_INFO.NODE_PLUGINS.resource) +@ApiTags(CONTROLLERS_INFO.NODE_PLUGINS.tag) +@Roles(ROLE.ADMIN, ROLE.API) +@UseGuards(JwtDefaultGuard, RolesGuard, ScopesGuard) +@UseFilters(HttpExceptionFilter) +@Controller(NODE_PLUGINS_CONTROLLER) +export class AbuseBlockerController { + constructor(private readonly abuseBlockerService: AbuseBlockerService) {} + + @Endpoint({ + type: GetAbuseBlockerReportsResponseDto, + command: GetAbuseBlockerReportsCommand, + httpCode: HttpStatus.OK, + }) + async getReports( + @Query() query: GetAbuseBlockerReportsQueryDto, + ): Promise { + return { response: errorHandler(await this.abuseBlockerService.getReports(query)) }; + } + + @Endpoint({ + type: GetAbuseBlockerStatsResponseDto, + command: GetAbuseBlockerStatsCommand, + httpCode: HttpStatus.OK, + }) + async getStats(): Promise { + return { response: errorHandler(await this.abuseBlockerService.getStats()) }; + } + + @Endpoint({ + type: GetAbuseBlockerReviewQueueResponseDto, + command: GetAbuseBlockerReviewQueueCommand, + httpCode: HttpStatus.OK, + }) + async getReviewQueue( + @Query() query: GetAbuseBlockerReviewQueueQueryDto, + ): Promise { + return { response: errorHandler(await this.abuseBlockerService.getReviewQueue(query)) }; + } + + @Endpoint({ + type: ReviewAbuseBlockerUserResponseDto, + command: ReviewAbuseBlockerUserCommand, + httpCode: HttpStatus.OK, + }) + async review( + @Param() param: ReviewAbuseBlockerUserParamDto, + @Body() body: ReviewAbuseBlockerUserBodyDto, + ): Promise { + return { + response: errorHandler( + await this.abuseBlockerService.review(param.userUuid, body.action), + ), + }; + } + + @Endpoint({ + command: TruncateAbuseBlockerReportsCommand, + httpCode: HttpStatus.NO_CONTENT, + }) + async truncateReports(): Promise { + errorHandler(await this.abuseBlockerService.truncateReports()); + } +} diff --git a/src/modules/node-plugins/abuse-blocker.service.ts b/src/modules/node-plugins/abuse-blocker.service.ts new file mode 100644 index 000000000..bacc805cc --- /dev/null +++ b/src/modules/node-plugins/abuse-blocker.service.ts @@ -0,0 +1,195 @@ +import { Injectable, Logger } from '@nestjs/common'; +import { QueryBus } from '@nestjs/cqrs'; +import { EventEmitter2 } from '@nestjs/event-emitter'; + +import type { AbuseBlockerReportModel } from '@remnawave/node-contract'; + +import { INodeConnectionOpts } from '@common/axios'; +import { AxiosService } from '@common/axios/axios.service'; +import { fail, ok, TResult } from '@common/types'; +import { + GetAbuseBlockerReportsCommand, + GetAbuseBlockerReviewQueueCommand, + GetAbuseBlockerStatsCommand, + ReviewAbuseBlockerUserCommand, +} from '@libs/contracts/commands'; +import { ERRORS, EVENTS } from '@libs/contracts/constants'; + +import { AbuseBlockerEvent } from '@integration-modules/notifications/interfaces'; + +import { GetNodeByUuidQuery } from '@modules/nodes/queries/get-node-by-uuid'; +import { GetUserByUniqueFieldQuery } from '@modules/users/queries/get-user-by-unique-field'; +import { UsersService } from '@modules/users/users.service'; + +import { AbuseBlockerRepository } from './repositories/abuse-blocker.repository'; + +@Injectable() +export class AbuseBlockerService { + private readonly logger = new Logger(AbuseBlockerService.name); + + constructor( + private readonly repository: AbuseBlockerRepository, + private readonly axios: AxiosService, + private readonly usersService: UsersService, + private readonly queryBus: QueryBus, + private readonly eventEmitter: EventEmitter2, + ) {} + + async processReport( + nodeUuid: string, + connectionOpts: INodeConnectionOpts, + report: AbuseBlockerReportModel, + ): Promise { + try { + const userId = BigInt(report.userId); + const [userResult, nodeResult] = await Promise.all([ + this.queryBus.execute( + new GetUserByUniqueFieldQuery({ id: userId }, { activeInternalSquads: true }), + ), + this.queryBus.execute(new GetNodeByUuidQuery(nodeUuid)), + ]); + if (!userResult.isOk || !nodeResult.isOk) { + this.logger.warn( + `Ignoring abuse report ${report.eventId}: user or node was not found.`, + ); + return; + } + + const user = userResult.response; + const node = nodeResult.response; + let eventUser = user; + const processed = await this.repository.processReport(user.id, node.id, report); + if (!processed.created) return; + + if (processed.action === 'repeat_block') { + const refreshed = await this.axios.refreshAbuseBlock( + { ip: report.sourceIp, timeout: report.policy.repeatBlockSeconds }, + connectionOpts, + ); + if (!refreshed.isOk || !refreshed.response.accepted) { + this.logger.error( + `Failed to refresh abuse block for ${report.sourceIp} on ${nodeUuid}.`, + ); + } + } else if (processed.action === 'disabled') { + const disabled = await this.usersService.disableUser(Number(user.id)); + if (disabled.isOk) { + eventUser = disabled.response; + await this.repository.markDisabledByPlugin(user.id, true); + } else { + this.logger.warn( + `Failed to disable abuse offender ${user.id}: ${disabled.message}`, + ); + } + } + + if (processed.notify) { + this.eventEmitter.emit( + EVENTS.ABUSE_BLOCKER.REPORT, + new AbuseBlockerEvent( + { + node, + user: eventUser, + report, + backendAction: processed.action, + strikeLevel: processed.strikeLevel, + }, + EVENTS.ABUSE_BLOCKER.REPORT, + ), + ); + } + } catch (error) { + this.logger.error(`Failed to process abuse report ${report.eventId}: ${error}`); + } + } + + async getReports( + query: GetAbuseBlockerReportsCommand.RequestQuery, + ): Promise> { + try { + const result = await this.repository.getReports(query); + return ok({ + total: result.total, + records: result.records.map((record) => ({ + ...record, + userId: Number(record.userId), + nodeId: Number(record.nodeId), + severity: record.severity as 'suspicious' | 'alert' | 'blocked', + action: record.action as 'none' | 'initial_block' | 'repeat_block' | 'disabled', + })), + }); + } catch (error) { + this.logger.error(error); + return fail(ERRORS.INTERNAL_SERVER_ERROR); + } + } + + async getStats(): Promise> { + try { + return ok(await this.repository.getStats()); + } catch (error) { + this.logger.error(error); + return fail(ERRORS.INTERNAL_SERVER_ERROR); + } + } + + async getReviewQueue( + query: GetAbuseBlockerReviewQueueCommand.RequestQuery, + ): Promise> { + try { + const result = await this.repository.getReviewQueue(query.start, query.size); + return ok({ + total: result.total, + records: result.records.map((record) => ({ + ...record, + userId: Number(record.userId), + reviewAction: record.reviewAction as 'enable' | 'keep_disabled' | null, + })), + }); + } catch (error) { + this.logger.error(error); + return fail(ERRORS.INTERNAL_SERVER_ERROR); + } + } + + async review( + userUuid: string, + action: ReviewAbuseBlockerUserCommand.RequestBody['action'], + ): Promise> { + try { + const current = await this.repository.findReviewStateByUserUuid(userUuid); + if (!current) return fail(ERRORS.USER_NOT_FOUND); + + if (action === 'enable' && current.user.status !== 'ACTIVE') { + const enabled = await this.usersService.enableUser(Number(current.userId)); + if (!enabled.isOk) return fail(ERRORS.INTERNAL_SERVER_ERROR); + } else if (action === 'keep_disabled' && current.user.status !== 'DISABLED') { + const disabled = await this.usersService.disableUser(Number(current.userId)); + if (!disabled.isOk) return fail(ERRORS.INTERNAL_SERVER_ERROR); + } + + const state = await this.repository.resolveReview(userUuid, action); + if (!state) return fail(ERRORS.USER_NOT_FOUND); + const updated = await this.repository.findReviewState(state.userId); + if (!updated) return fail(ERRORS.USER_NOT_FOUND); + return ok({ + ...updated, + userId: Number(updated.userId), + reviewAction: updated.reviewAction as 'enable' | 'keep_disabled' | null, + }); + } catch (error) { + this.logger.error(error); + return fail(ERRORS.INTERNAL_SERVER_ERROR); + } + } + + async truncateReports(): Promise> { + try { + await this.repository.truncateReports(); + return ok(true); + } catch (error) { + this.logger.error(error); + return fail(ERRORS.INTERNAL_SERVER_ERROR); + } + } +} diff --git a/src/modules/node-plugins/commands/index.ts b/src/modules/node-plugins/commands/index.ts index 2ce492e5a..312e0bddd 100644 --- a/src/modules/node-plugins/commands/index.ts +++ b/src/modules/node-plugins/commands/index.ts @@ -1,3 +1,4 @@ import { CreateTorrentReportHandler } from './create-torrent-report'; +import { ProcessAbuseReportHandler } from './process-abuse-report'; -export const COMMANDS = [CreateTorrentReportHandler]; +export const COMMANDS = [CreateTorrentReportHandler, ProcessAbuseReportHandler]; diff --git a/src/modules/node-plugins/commands/process-abuse-report/index.ts b/src/modules/node-plugins/commands/process-abuse-report/index.ts new file mode 100644 index 000000000..7ac6f605b --- /dev/null +++ b/src/modules/node-plugins/commands/process-abuse-report/index.ts @@ -0,0 +1,2 @@ +export * from './process-abuse-report.command'; +export * from './process-abuse-report.handler'; diff --git a/src/modules/node-plugins/commands/process-abuse-report/process-abuse-report.command.ts b/src/modules/node-plugins/commands/process-abuse-report/process-abuse-report.command.ts new file mode 100644 index 000000000..49d072d19 --- /dev/null +++ b/src/modules/node-plugins/commands/process-abuse-report/process-abuse-report.command.ts @@ -0,0 +1,15 @@ +import { Command } from '@nestjs/cqrs'; + +import type { AbuseBlockerReportModel } from '@remnawave/node-contract'; + +import { INodeConnectionOpts } from '@common/axios'; + +export class ProcessAbuseReportCommand extends Command { + constructor( + public readonly nodeUuid: string, + public readonly connectionOpts: INodeConnectionOpts, + public readonly report: AbuseBlockerReportModel, + ) { + super(); + } +} diff --git a/src/modules/node-plugins/commands/process-abuse-report/process-abuse-report.handler.ts b/src/modules/node-plugins/commands/process-abuse-report/process-abuse-report.handler.ts new file mode 100644 index 000000000..4c8bbd7e2 --- /dev/null +++ b/src/modules/node-plugins/commands/process-abuse-report/process-abuse-report.handler.ts @@ -0,0 +1,17 @@ +import { CommandHandler, ICommandHandler } from '@nestjs/cqrs'; + +import { AbuseBlockerService } from '../../abuse-blocker.service'; +import { ProcessAbuseReportCommand } from './process-abuse-report.command'; + +@CommandHandler(ProcessAbuseReportCommand) +export class ProcessAbuseReportHandler implements ICommandHandler { + constructor(private readonly abuseBlockerService: AbuseBlockerService) {} + + async execute(command: ProcessAbuseReportCommand): Promise { + await this.abuseBlockerService.processReport( + command.nodeUuid, + command.connectionOpts, + command.report, + ); + } +} diff --git a/src/modules/node-plugins/constants/default-subpage-config.ts b/src/modules/node-plugins/constants/default-subpage-config.ts index 2039fc115..9839dad07 100644 --- a/src/modules/node-plugins/constants/default-subpage-config.ts +++ b/src/modules/node-plugins/constants/default-subpage-config.ts @@ -15,6 +15,9 @@ export const EXAMPLE_NODE_PLUGIN_CONFIG = { ip: [], }, }, + abuseBlocker: { + enabled: false, + }, connectionDrop: { enabled: false, whitelistIps: [], diff --git a/src/modules/node-plugins/dtos/node-plugins.dtos.ts b/src/modules/node-plugins/dtos/node-plugins.dtos.ts index 3449e4b0c..726107e5a 100644 --- a/src/modules/node-plugins/dtos/node-plugins.dtos.ts +++ b/src/modules/node-plugins/dtos/node-plugins.dtos.ts @@ -1,6 +1,9 @@ import { createZodDto } from 'nestjs-zod'; import { + GetAbuseBlockerReportsCommand, + GetAbuseBlockerReviewQueueCommand, + GetAbuseBlockerStatsCommand, GetNodePluginsCommand, UpdateNodePluginCommand, GetNodePluginCommand, @@ -9,6 +12,7 @@ import { ReorderNodePluginCommand, CloneNodePluginCommand, PluginExecutorCommand, + ReviewAbuseBlockerUserCommand, } from '@libs/contracts/commands'; import { GetTorrentBlockerReportsCommand } from '@libs/contracts/commands/node-plugins/torrent-blocker'; import { GetTorrentBlockerReportsStatsCommand } from '@libs/contracts/commands/node-plugins/torrent-blocker/get-torrent-blocker-reports-stats.command'; @@ -65,3 +69,28 @@ export class GetTorrentBlockerReportsResponseDto extends createZodDto( export class GetTorrentBlockerReportsStatsResponseDto extends createZodDto( GetTorrentBlockerReportsStatsCommand.ResponseSchema, ) {} // TORRENT_BLOCKER_REPORT_STATS + +export class GetAbuseBlockerReportsQueryDto extends createZodDto( + GetAbuseBlockerReportsCommand.RequestQuerySchema, +) {} +export class GetAbuseBlockerReportsResponseDto extends createZodDto( + GetAbuseBlockerReportsCommand.ResponseSchema, +) {} +export class GetAbuseBlockerStatsResponseDto extends createZodDto( + GetAbuseBlockerStatsCommand.ResponseSchema, +) {} +export class GetAbuseBlockerReviewQueueQueryDto extends createZodDto( + GetAbuseBlockerReviewQueueCommand.RequestQuerySchema, +) {} +export class GetAbuseBlockerReviewQueueResponseDto extends createZodDto( + GetAbuseBlockerReviewQueueCommand.ResponseSchema, +) {} +export class ReviewAbuseBlockerUserParamDto extends createZodDto( + ReviewAbuseBlockerUserCommand.RequestParamSchema, +) {} +export class ReviewAbuseBlockerUserBodyDto extends createZodDto( + ReviewAbuseBlockerUserCommand.RequestBodySchema, +) {} +export class ReviewAbuseBlockerUserResponseDto extends createZodDto( + ReviewAbuseBlockerUserCommand.ResponseSchema, +) {} diff --git a/src/modules/node-plugins/node-plugins.module.ts b/src/modules/node-plugins/node-plugins.module.ts index a39d26e29..46c03a5d9 100644 --- a/src/modules/node-plugins/node-plugins.module.ts +++ b/src/modules/node-plugins/node-plugins.module.ts @@ -1,21 +1,28 @@ import { Module } from '@nestjs/common'; import { CqrsModule } from '@nestjs/cqrs'; +import { UsersModule } from '@modules/users/users.module'; + +import { AbuseBlockerController } from './abuse-blocker.controller'; +import { AbuseBlockerService } from './abuse-blocker.service'; import { COMMANDS } from './commands'; import { NodePluginController } from './node-plugins.controller'; import { NodePluginConverter } from './node-plugins.converter'; import { NodePluginService } from './node-plugins.service'; import { QUERIES } from './queries'; +import { AbuseBlockerRepository } from './repositories/abuse-blocker.repository'; import { NodePluginRepository } from './repositories/node-plugins.repository'; import { TorrentBlockerReportsRepository } from './repositories/torrent-blocker-report.repository'; import { TorrentBlockerReportConverter } from './torrent-blocker-report.converter'; import { TorrentBlockerReportsController } from './torrent-blocker-reports.controller'; @Module({ - imports: [CqrsModule], - controllers: [TorrentBlockerReportsController, NodePluginController], + imports: [CqrsModule, UsersModule], + controllers: [AbuseBlockerController, TorrentBlockerReportsController, NodePluginController], providers: [ NodePluginService, + AbuseBlockerService, + AbuseBlockerRepository, NodePluginRepository, NodePluginConverter, TorrentBlockerReportsRepository, diff --git a/src/modules/node-plugins/repositories/abuse-blocker.repository.ts b/src/modules/node-plugins/repositories/abuse-blocker.repository.ts new file mode 100644 index 000000000..348970628 --- /dev/null +++ b/src/modules/node-plugins/repositories/abuse-blocker.repository.ts @@ -0,0 +1,340 @@ +import { Transactional } from '@nestjs-cls/transactional'; +import { TransactionHost } from '@nestjs-cls/transactional'; +import { TransactionalAdapterPrisma } from '@nestjs-cls/transactional-adapter-prisma'; +import { Prisma } from '@prisma/client'; + +import { Injectable } from '@nestjs/common'; + +import type { AbuseBlockerReportModel } from '@remnawave/node-contract'; + +import { GetAbuseBlockerReportsCommand } from '@libs/contracts/commands'; + +export type AbuseBlockerBackendAction = 'none' | 'initial_block' | 'repeat_block' | 'disabled'; + +export interface IProcessedAbuseReport { + action: AbuseBlockerBackendAction; + created: boolean; + notify: boolean; + strikeLevel: number; +} + +export const decideAbuseEscalation = ( + state: { + strikeLevel: number; + lastBlockingIncidentAt: Date | null; + manualReviewRequired: boolean; + } | null, + now: Date, + repeatWindowSeconds: number, +): Pick => { + if (state?.manualReviewRequired && state.strikeLevel >= 3) { + return { action: 'none', notify: false, strikeLevel: 3 }; + } + + const isFreshChain = + !!state?.lastBlockingIncidentAt && + now.getTime() - state.lastBlockingIncidentAt.getTime() <= repeatWindowSeconds * 1000; + if (!isFreshChain) return { action: 'initial_block', notify: true, strikeLevel: 1 }; + if ((state?.strikeLevel ?? 0) === 1) { + return { action: 'repeat_block', notify: true, strikeLevel: 2 }; + } + if ((state?.strikeLevel ?? 0) === 2) { + return { action: 'disabled', notify: true, strikeLevel: 3 }; + } + return { action: 'none', notify: false, strikeLevel: 3 }; +}; + +const asJson = (value: unknown): Prisma.InputJsonValue => + JSON.parse(JSON.stringify(value)) as Prisma.InputJsonValue; + +@Injectable() +export class AbuseBlockerRepository { + constructor(private readonly prisma: TransactionHost) {} + + @Transactional() + async processReport( + userId: bigint, + nodeId: bigint, + report: AbuseBlockerReportModel, + ): Promise { + if (report.severity === 'blocked') { + await this.prisma.tx.$executeRaw` + SELECT pg_advisory_xact_lock(${userId}) + `; + } + + const existing = await this.prisma.tx.abuseBlockerReports.findUnique({ + where: { eventId: report.eventId }, + }); + if (existing) { + await this.prisma.tx.abuseBlockerReports.update({ + where: { eventId: report.eventId }, + data: { + report: asJson(report), + severity: report.severity, + score: report.score.after, + }, + }); + const state = await this.prisma.tx.abuseBlockerUserState.findUnique({ + where: { userId }, + }); + return { + action: existing.action as AbuseBlockerBackendAction, + created: false, + notify: false, + strikeLevel: state?.strikeLevel ?? 0, + }; + } + + let action: AbuseBlockerBackendAction = 'none'; + let strikeLevel = 0; + let notify = report.severity !== 'suspicious'; + + if (report.severity === 'blocked') { + const now = new Date(); + const state = await this.prisma.tx.abuseBlockerUserState.findUnique({ + where: { userId }, + }); + ({ action, notify, strikeLevel } = decideAbuseEscalation( + state, + now, + report.policy.repeatWindowSeconds, + )); + + await this.prisma.tx.abuseBlockerUserState.upsert({ + where: { userId }, + create: { + userId, + strikeLevel, + lastBlockingIncidentAt: now, + manualReviewRequired: strikeLevel === 3, + reviewRequestedAt: strikeLevel === 3 ? now : null, + }, + update: { + strikeLevel, + lastBlockingIncidentAt: now, + manualReviewRequired: strikeLevel === 3, + reviewRequestedAt: strikeLevel === 3 ? now : null, + }, + }); + } + + await this.prisma.tx.abuseBlockerReports.create({ + data: { + eventId: report.eventId, + userId, + nodeId, + severity: report.severity, + score: report.score.after, + sourceIp: report.sourceIp, + action, + detectedAt: report.detectedAt, + report: asJson(report), + }, + }); + + return { action, created: true, notify, strikeLevel }; + } + + async markDisabledByPlugin(userId: bigint, disabled: boolean): Promise { + await this.prisma.tx.abuseBlockerUserState.update({ + where: { userId }, + data: { disabledByPlugin: disabled }, + }); + } + + async getReports(query: GetAbuseBlockerReportsCommand.RequestQuery) { + const where: Prisma.AbuseBlockerReportsWhereInput = { + userId: query.userId === undefined ? undefined : BigInt(query.userId), + node: query.nodeUuid ? { uuid: query.nodeUuid } : undefined, + severity: query.severity, + action: query.action, + detectedAt: + query.dateFrom || query.dateTo + ? { gte: query.dateFrom, lte: query.dateTo } + : undefined, + report: query.rule + ? { path: ['detections'], array_contains: [{ rule: query.rule }] } + : undefined, + }; + + const [records, total] = await Promise.all([ + this.prisma.tx.abuseBlockerReports.findMany({ + where, + include: { + user: { select: { username: true, vlessUuid: true, status: true } }, + node: { select: { uuid: true, name: true, countryCode: true } }, + }, + orderBy: { detectedAt: 'desc' }, + skip: query.start, + take: query.size, + }), + this.prisma.tx.abuseBlockerReports.count({ where }), + ]); + + return { records, total }; + } + + async getStats() { + const [counts] = await this.prisma.tx.$queryRaw< + Array<{ + total: bigint; + last24h: bigint; + users: bigint; + nodes: bigint; + suspicious: bigint; + alert: bigint; + blocked: bigint; + }> + >` + SELECT + COUNT(*) AS total, + COUNT(*) FILTER (WHERE detected_at > now() - interval '24 hours') AS last24h, + COUNT(DISTINCT user_id) AS users, + COUNT(DISTINCT node_id) AS nodes, + COUNT(*) FILTER (WHERE severity = 'suspicious') AS suspicious, + COUNT(*) FILTER (WHERE severity = 'alert') AS alert, + COUNT(*) FILTER (WHERE severity = 'blocked') AS blocked + FROM abuse_blocker_reports + `; + const [manualReviewRequired, topUserGroups, topNodeGroups] = await Promise.all([ + this.prisma.tx.abuseBlockerUserState.count({ + where: { manualReviewRequired: true }, + }), + this.prisma.tx.abuseBlockerReports.groupBy({ + by: ['userId'], + _count: { _all: true }, + orderBy: { _count: { userId: 'desc' } }, + take: 50, + }), + this.prisma.tx.abuseBlockerReports.groupBy({ + by: ['nodeId'], + _count: { _all: true }, + orderBy: { _count: { nodeId: 'desc' } }, + take: 50, + }), + ]); + const [users, nodes] = await Promise.all([ + this.prisma.tx.users.findMany({ + where: { id: { in: topUserGroups.map((item) => item.userId) } }, + select: { id: true, username: true }, + }), + this.prisma.tx.nodes.findMany({ + where: { id: { in: topNodeGroups.map((item) => item.nodeId) } }, + select: { id: true, uuid: true, name: true, countryCode: true }, + }), + ]); + const userMap = new Map(users.map((user) => [user.id, user])); + const nodeMap = new Map(nodes.map((node) => [node.id, node])); + + return { + totalReports: Number(counts?.total ?? 0), + reportsLast24Hours: Number(counts?.last24h ?? 0), + distinctUsers: Number(counts?.users ?? 0), + distinctNodes: Number(counts?.nodes ?? 0), + manualReviewRequired, + bySeverity: { + suspicious: Number(counts?.suspicious ?? 0), + alert: Number(counts?.alert ?? 0), + blocked: Number(counts?.blocked ?? 0), + }, + topUsers: topUserGroups.flatMap((item) => { + const user = userMap.get(item.userId); + return user + ? [ + { + userId: Number(user.id), + username: user.username, + total: item._count._all, + }, + ] + : []; + }), + topNodes: topNodeGroups.flatMap((item) => { + const node = nodeMap.get(item.nodeId); + return node + ? [ + { + uuid: node.uuid, + name: node.name, + countryCode: node.countryCode, + total: item._count._all, + }, + ] + : []; + }), + }; + } + + async getReviewQueue(start: number, size: number) { + const where = { manualReviewRequired: true } as const; + const [records, total] = await Promise.all([ + this.prisma.tx.abuseBlockerUserState.findMany({ + where, + include: { + user: { select: { username: true, vlessUuid: true, status: true } }, + }, + orderBy: { reviewRequestedAt: 'asc' }, + skip: start, + take: size, + }), + this.prisma.tx.abuseBlockerUserState.count({ where }), + ]); + return { records, total }; + } + + @Transactional() + async resolveReview(userUuid: string, action: 'enable' | 'keep_disabled') { + const user = await this.prisma.tx.users.findFirst({ where: { vlessUuid: userUuid } }); + if (!user) return null; + + await this.prisma.tx.$executeRaw` + SELECT pg_advisory_xact_lock(${user.id}) + `; + const now = new Date(); + return this.prisma.tx.abuseBlockerUserState.upsert({ + where: { userId: user.id }, + create: { + userId: user.id, + strikeLevel: 0, + manualReviewRequired: false, + disabledByPlugin: false, + reviewedAt: now, + reviewAction: action, + }, + update: { + strikeLevel: 0, + lastBlockingIncidentAt: null, + manualReviewRequired: false, + disabledByPlugin: false, + reviewedAt: now, + reviewAction: action, + }, + include: { + user: { select: { username: true, vlessUuid: true, status: true } }, + }, + }); + } + + async findReviewStateByUserUuid(userUuid: string) { + return this.prisma.tx.abuseBlockerUserState.findFirst({ + where: { user: { vlessUuid: userUuid } }, + include: { + user: { select: { username: true, vlessUuid: true, status: true } }, + }, + }); + } + + async findReviewState(userId: bigint) { + return this.prisma.tx.abuseBlockerUserState.findUnique({ + where: { userId }, + include: { + user: { select: { username: true, vlessUuid: true, status: true } }, + }, + }); + } + + async truncateReports(): Promise { + await this.prisma.tx.abuseBlockerReports.deleteMany(); + } +} diff --git a/src/modules/node-plugins/utils/order-node-plugins.util.ts b/src/modules/node-plugins/utils/order-node-plugins.util.ts index b8c9144ab..4c087be1f 100644 --- a/src/modules/node-plugins/utils/order-node-plugins.util.ts +++ b/src/modules/node-plugins/utils/order-node-plugins.util.ts @@ -1,13 +1,22 @@ import { TNodePlugin } from 'libs/node-plugins'; export const orderNodePluginsConfig = (config: TNodePlugin) => { - const { sharedLists, ingressFilter, torrentBlocker, connectionDrop, egressFilter, ...rest } = - config; + const configWithAbuse = config as TNodePlugin & { abuseBlocker?: unknown }; + const { + sharedLists, + ingressFilter, + torrentBlocker, + abuseBlocker, + connectionDrop, + egressFilter, + ...rest + } = configWithAbuse; return { ingressFilter, egressFilter, torrentBlocker, + abuseBlocker, connectionDrop, sharedLists, ...rest, diff --git a/src/modules/users/users.module.ts b/src/modules/users/users.module.ts index bf879accd..c6590b3ae 100644 --- a/src/modules/users/users.module.ts +++ b/src/modules/users/users.module.ts @@ -11,6 +11,6 @@ import { UsersService } from './users.service'; imports: [CqrsModule], controllers: [UsersController, UsersBulkActionsController], providers: [UsersRepository, UserConverter, UsersService, ...QUERIES, ...COMMANDS], - exports: [], + exports: [UsersService], }) export class UsersModule {} diff --git a/src/queue/_nodes/processors/node-health-check.processor.ts b/src/queue/_nodes/processors/node-health-check.processor.ts index 2ff9fd4f6..04bfca383 100644 --- a/src/queue/_nodes/processors/node-health-check.processor.ts +++ b/src/queue/_nodes/processors/node-health-check.processor.ts @@ -119,14 +119,17 @@ export class NodeHealthCheckQueueProcessor extends WorkerHost { }, ]); - const reports = stats.plugins.torrentBlocker.reportsCount; - if (reports !== undefined && reports > 0) { + const torrentReports = stats.plugins.torrentBlocker.reportsCount ?? 0; + const abuseReports = stats.plugins.abuseBlocker?.reportsCount ?? 0; + if (torrentReports + abuseReports > 0) { await this.nodesQueuesService.collectReports({ nodeUuid, connectionOpts, }); - this.logger.log(`Node ${nodeUuid} has ${reports} reports, collecting reports...`); + this.logger.log( + `Node ${nodeUuid} has ${torrentReports} torrent and ${abuseReports} abuse reports, collecting...`, + ); } if (!isConnected) { diff --git a/src/queue/_nodes/processors/node-plugins.processor.ts b/src/queue/_nodes/processors/node-plugins.processor.ts index b49dd4ee0..8d3db1acf 100644 --- a/src/queue/_nodes/processors/node-plugins.processor.ts +++ b/src/queue/_nodes/processors/node-plugins.processor.ts @@ -2,12 +2,13 @@ import { Job } from 'bullmq'; import { Processor, WorkerHost } from '@nestjs/bullmq'; import { Logger } from '@nestjs/common'; -import { QueryBus } from '@nestjs/cqrs'; +import { CommandBus, QueryBus } from '@nestjs/cqrs'; import { INodeConnectionOpts } from '@common/axios'; import { AxiosService } from '@common/axios/axios.service'; import { EVENTS } from '@libs/contracts/constants/events/events'; +import { ProcessAbuseReportCommand } from '@modules/node-plugins/commands/process-abuse-report'; import { GetPluginByUuidQuery } from '@modules/node-plugins/queries/get-plugin-by-uuid'; import { GetNodeByUuidQuery } from '@modules/nodes/queries/get-node-by-uuid'; @@ -27,6 +28,7 @@ export class NodePluginsProcessor extends WorkerHost { private readonly axios: AxiosService, private readonly queryBus: QueryBus, private readonly usersQueuesService: UsersQueuesService, + private readonly commandBus: CommandBus, ) { super(); this.CONCURRENCY = 20; @@ -142,33 +144,44 @@ export class NodePluginsProcessor extends WorkerHost { try { const { nodeUuid, connectionOpts } = job.data; - const response = await this.axios.collectTorrentBlockerReports(connectionOpts); - - if (!response.isOk) { - this.logger.error(`Failed to collect reports: ${response.message}`); - - return { - success: false, - nodeUuid, - collectedReports: [], - }; + const [torrentResponse, abuseResponse] = await Promise.all([ + this.axios.collectTorrentBlockerReports(connectionOpts), + this.axios.collectAbuseBlockerReports(connectionOpts), + ]); + let success = true; + + if (!torrentResponse.isOk) { + success = false; + this.logger.error(`Failed to collect torrent reports: ${torrentResponse.message}`); + } else { + for (const report of torrentResponse.response.reports) { + await this.usersQueuesService.fireTorrentBlockerEvent({ + id: report.actionReport.userId, + event: EVENTS.TORRENT_BLOCKER.REPORT, + nodeUuid, + report, + }); + } } - const { response: collectedReports } = response; - - for (const report of collectedReports.reports) { - await this.usersQueuesService.fireTorrentBlockerEvent({ - id: report.actionReport.userId, - event: EVENTS.TORRENT_BLOCKER.REPORT, - nodeUuid, - report, - }); + if (!abuseResponse.isOk) { + success = false; + this.logger.error(`Failed to collect abuse reports: ${abuseResponse.message}`); + } else { + for (const report of abuseResponse.response.reports) { + await this.commandBus.execute( + new ProcessAbuseReportCommand(nodeUuid, connectionOpts, report), + ); + } } return { - success: true, + success, nodeUuid, - collectedReports, + collectedReports: torrentResponse.isOk ? torrentResponse.response : { reports: [] }, + collectedAbuseReports: abuseResponse.isOk + ? abuseResponse.response + : { reports: [] }, }; } catch (error) { this.logger.error(`Failed to collect reports: ${error}`); diff --git a/tests/abuse-blocker/abuse-blocker.runtime.test.ts b/tests/abuse-blocker/abuse-blocker.runtime.test.ts new file mode 100644 index 000000000..c61d621b0 --- /dev/null +++ b/tests/abuse-blocker/abuse-blocker.runtime.test.ts @@ -0,0 +1,235 @@ +import assert from 'node:assert/strict'; +import { describe, it } from 'node:test'; + +import type { AbuseBlockerReportModel } from '@remnawave/node-contract'; + +import { AbuseBlockerService } from '../../src/modules/node-plugins/abuse-blocker.service'; +import { decideAbuseEscalation } from '../../src/modules/node-plugins/repositories/abuse-blocker.repository'; + +const report = { + eventId: '00000000-0000-4000-8000-000000000001', + userId: '42', + sourceIp: '198.51.100.10', + policy: { repeatBlockSeconds: 3600, repeatWindowSeconds: 604800 }, +} as AbuseBlockerReportModel; + +describe('Abuse Blocker escalation', () => { + const now = new Date('2026-08-15T00:00:00.000Z'); + + it('progresses from 10-minute Node block to one-hour refresh and disable', () => { + const first = decideAbuseEscalation(null, now, 604800); + const second = decideAbuseEscalation( + { + strikeLevel: first.strikeLevel, + lastBlockingIncidentAt: now, + manualReviewRequired: false, + }, + new Date(now.getTime() + 1000), + 604800, + ); + const third = decideAbuseEscalation( + { + strikeLevel: second.strikeLevel, + lastBlockingIncidentAt: new Date(now.getTime() + 1000), + manualReviewRequired: false, + }, + new Date(now.getTime() + 2000), + 604800, + ); + + assert.deepEqual(first, { action: 'initial_block', notify: true, strikeLevel: 1 }); + assert.deepEqual(second, { action: 'repeat_block', notify: true, strikeLevel: 2 }); + assert.deepEqual(third, { action: 'disabled', notify: true, strikeLevel: 3 }); + }); + + it('starts a fresh chain after seven days', () => { + assert.deepEqual( + decideAbuseEscalation( + { + strikeLevel: 2, + lastBlockingIncidentAt: now, + manualReviewRequired: false, + }, + new Date(now.getTime() + 604_800_001), + 604800, + ), + { action: 'initial_block', notify: true, strikeLevel: 1 }, + ); + }); + + it('stores later incidents without repeating disable or notifications', () => { + assert.deepEqual( + decideAbuseEscalation( + { + strikeLevel: 3, + lastBlockingIncidentAt: now, + manualReviewRequired: true, + }, + new Date(now.getTime() + 604_800_001), + 604800, + ), + { action: 'none', notify: false, strikeLevel: 3 }, + ); + }); +}); + +const createService = (processed: { + action: 'none' | 'initial_block' | 'repeat_block' | 'disabled'; + created: boolean; + notify: boolean; + strikeLevel: number; +}) => { + const calls = { disable: 0, emit: 0, mark: 0, refresh: 0 }; + const user = { id: 42n, status: 'ACTIVE' }; + const node = { id: 7n, uuid: '00000000-0000-4000-8000-000000000007' }; + const repository = { + processReport: async () => processed, + markDisabledByPlugin: async () => { + calls.mark += 1; + }, + }; + const service = new AbuseBlockerService( + repository as never, + { + refreshAbuseBlock: async () => { + calls.refresh += 1; + return { isOk: true, response: { accepted: true } }; + }, + } as never, + { + disableUser: async () => { + calls.disable += 1; + return { isOk: true, response: user }; + }, + } as never, + { + execute: async (query: object) => ({ + isOk: true, + response: query.constructor.name.includes('User') ? user : node, + }), + } as never, + { + emit: () => { + calls.emit += 1; + }, + } as never, + ); + return { calls, service }; +}; + +describe('AbuseBlockerService report idempotency and actions', () => { + it('keeps suspicious reports database-only', async () => { + const { calls, service } = createService({ + action: 'none', + created: true, + notify: false, + strikeLevel: 0, + }); + + await service.processReport('00000000-0000-4000-8000-000000000007', {} as never, report); + assert.deepEqual(calls, { disable: 0, emit: 0, mark: 0, refresh: 0 }); + }); + + it('does not repeat escalation for an existing eventId', async () => { + const { calls, service } = createService({ + action: 'repeat_block', + created: false, + notify: false, + strikeLevel: 2, + }); + + await service.processReport('00000000-0000-4000-8000-000000000007', {} as never, report); + assert.deepEqual(calls, { disable: 0, emit: 0, mark: 0, refresh: 0 }); + }); + + it('refreshes the Node block once for a repeat offender', async () => { + const { calls, service } = createService({ + action: 'repeat_block', + created: true, + notify: true, + strikeLevel: 2, + }); + + await service.processReport('00000000-0000-4000-8000-000000000007', {} as never, report); + assert.equal(calls.refresh, 1); + assert.equal(calls.emit, 1); + assert.equal(calls.disable, 0); + }); + + it('uses the standard user service and marks plugin disablement on strike three', async () => { + const { calls, service } = createService({ + action: 'disabled', + created: true, + notify: true, + strikeLevel: 3, + }); + + await service.processReport('00000000-0000-4000-8000-000000000007', {} as never, report); + assert.equal(calls.disable, 1); + assert.equal(calls.mark, 1); + assert.equal(calls.emit, 1); + }); +}); + +describe('Abuse Blocker manual review', () => { + const createReviewService = (status: 'ACTIVE' | 'DISABLED') => { + const calls = { disable: 0, enable: 0 }; + const state = { + userId: 42n, + strikeLevel: 0, + lastBlockingIncidentAt: null, + manualReviewRequired: false, + disabledByPlugin: false, + reviewRequestedAt: null, + reviewedAt: new Date(), + reviewAction: null, + updatedAt: new Date(), + user: { + username: 'test', + vlessUuid: '00000000-0000-4000-8000-000000000042', + status, + }, + }; + const service = new AbuseBlockerService( + { + findReviewStateByUserUuid: async () => state, + resolveReview: async (_uuid: string, action: string) => ({ + ...state, + reviewAction: action, + }), + findReviewState: async () => state, + } as never, + {} as never, + { + enableUser: async () => { + calls.enable += 1; + return { isOk: true }; + }, + disableUser: async () => { + calls.disable += 1; + return { isOk: true }; + }, + } as never, + {} as never, + {} as never, + ); + return { calls, service }; + }; + + it('enables a disabled user and keeps reports untouched', async () => { + const { calls, service } = createReviewService('DISABLED'); + const result = await service.review('00000000-0000-4000-8000-000000000042', 'enable'); + assert.equal(result.isOk, true); + assert.deepEqual(calls, { disable: 0, enable: 1 }); + }); + + it('idempotently keeps an already disabled user disabled', async () => { + const { calls, service } = createReviewService('DISABLED'); + const result = await service.review( + '00000000-0000-4000-8000-000000000042', + 'keep_disabled', + ); + assert.equal(result.isOk, true); + assert.deepEqual(calls, { disable: 0, enable: 0 }); + }); +});