From 7cc16cd83ec01d22e4b5d356a20dff9d8bb01370 Mon Sep 17 00:00:00 2001 From: l0nelynx <113856580+l0nelynx@users.noreply.github.com> Date: Sat, 15 Aug 2026 10:51:37 +0300 Subject: [PATCH 1/3] feat(node-contract): add abuse blocker contracts --- libs/contract/api/controllers/plugin.ts | 5 ++ libs/contract/api/routes.ts | 4 + .../abuse-blocker/collect-reports.schema.ts | 16 ++++ .../commands/plugin/abuse-blocker/index.ts | 2 + .../abuse-blocker/refresh-block.schema.ts | 22 +++++ libs/contract/commands/plugin/index.ts | 1 + .../stats/get-system-stats.command.ts | 15 +++- .../constants/internal/internal.constants.ts | 6 ++ libs/contract/constants/xray/stats.ts | 2 +- .../models/abuse-blocker.report.schema.ts | 89 +++++++++++++++++++ libs/contract/models/index.ts | 1 + libs/contract/tsconfig.json | 15 +--- 12 files changed, 165 insertions(+), 13 deletions(-) create mode 100644 libs/contract/commands/plugin/abuse-blocker/collect-reports.schema.ts create mode 100644 libs/contract/commands/plugin/abuse-blocker/index.ts create mode 100644 libs/contract/commands/plugin/abuse-blocker/refresh-block.schema.ts create mode 100644 libs/contract/models/abuse-blocker.report.schema.ts diff --git a/libs/contract/api/controllers/plugin.ts b/libs/contract/api/controllers/plugin.ts index 58173c4..eae3801 100644 --- a/libs/contract/api/controllers/plugin.ts +++ b/libs/contract/api/controllers/plugin.ts @@ -1,6 +1,7 @@ export const PLUGIN_CONTROLLER = 'plugin' as const; export const TORRENT_BLOCKER_ROUTE = 'torrent-blocker' as const; +export const ABUSE_BLOCKER_ROUTE = 'abuse-blocker' as const; export const NFTABLES_ROUTE = 'nftables' as const; export const PLUGIN_ROUTES = { @@ -9,6 +10,10 @@ export const PLUGIN_ROUTES = { TORRENT_BLOCKER: { COLLECT: `${TORRENT_BLOCKER_ROUTE}/collect`, }, + ABUSE_BLOCKER: { + COLLECT: `${ABUSE_BLOCKER_ROUTE}/collect`, + REFRESH_BLOCK: `${ABUSE_BLOCKER_ROUTE}/refresh-block`, + }, NFTABLES: { UNBLOCK_IPS: `${NFTABLES_ROUTE}/unblock-ips`, BLOCK_IPS: `${NFTABLES_ROUTE}/block-ips`, diff --git a/libs/contract/api/routes.ts b/libs/contract/api/routes.ts index fa5a5e9..fe2adf7 100644 --- a/libs/contract/api/routes.ts +++ b/libs/contract/api/routes.ts @@ -35,6 +35,10 @@ export const REST_API = { TORRENT_BLOCKER: { COLLECT: `${ROOT}/${CONTROLLERS.PLUGIN_CONTROLLER}/${CONTROLLERS.PLUGIN_ROUTES.TORRENT_BLOCKER.COLLECT}`, }, + ABUSE_BLOCKER: { + COLLECT: `${ROOT}/${CONTROLLERS.PLUGIN_CONTROLLER}/${CONTROLLERS.PLUGIN_ROUTES.ABUSE_BLOCKER.COLLECT}`, + REFRESH_BLOCK: `${ROOT}/${CONTROLLERS.PLUGIN_CONTROLLER}/${CONTROLLERS.PLUGIN_ROUTES.ABUSE_BLOCKER.REFRESH_BLOCK}`, + }, NFTABLES: { UNBLOCK_IPS: `${ROOT}/${CONTROLLERS.PLUGIN_CONTROLLER}/${CONTROLLERS.PLUGIN_ROUTES.NFTABLES.UNBLOCK_IPS}`, BLOCK_IPS: `${ROOT}/${CONTROLLERS.PLUGIN_CONTROLLER}/${CONTROLLERS.PLUGIN_ROUTES.NFTABLES.BLOCK_IPS}`, diff --git a/libs/contract/commands/plugin/abuse-blocker/collect-reports.schema.ts b/libs/contract/commands/plugin/abuse-blocker/collect-reports.schema.ts new file mode 100644 index 0000000..37b77c2 --- /dev/null +++ b/libs/contract/commands/plugin/abuse-blocker/collect-reports.schema.ts @@ -0,0 +1,16 @@ +import { z } from 'zod'; + +import { REST_API } from '../../../api'; +import { AbuseBlockerReportSchema } from '../../../models'; + +export namespace CollectAbuseBlockerReportsCommand { + export const url = REST_API.PLUGIN.ABUSE_BLOCKER.COLLECT; + + export const ResponseSchema = z.object({ + response: z.object({ + reports: z.array(AbuseBlockerReportSchema), + }), + }); + + export type Response = z.infer; +} diff --git a/libs/contract/commands/plugin/abuse-blocker/index.ts b/libs/contract/commands/plugin/abuse-blocker/index.ts new file mode 100644 index 0000000..7c9d6e5 --- /dev/null +++ b/libs/contract/commands/plugin/abuse-blocker/index.ts @@ -0,0 +1,2 @@ +export * from './collect-reports.schema'; +export * from './refresh-block.schema'; diff --git a/libs/contract/commands/plugin/abuse-blocker/refresh-block.schema.ts b/libs/contract/commands/plugin/abuse-blocker/refresh-block.schema.ts new file mode 100644 index 0000000..3eb24e7 --- /dev/null +++ b/libs/contract/commands/plugin/abuse-blocker/refresh-block.schema.ts @@ -0,0 +1,22 @@ +import { z } from 'zod'; + +import { REST_API } from '../../../api'; + +export namespace RefreshAbuseBlockCommand { + export const url = REST_API.PLUGIN.ABUSE_BLOCKER.REFRESH_BLOCK; + + export const RequestSchema = z.object({ + ip: z.union([z.ipv4(), z.ipv6()]), + timeout: z.int().min(1).max(2592000), + }); + + export type Request = z.infer; + + export const ResponseSchema = z.object({ + response: z.object({ + accepted: z.boolean(), + }), + }); + + export type Response = z.infer; +} diff --git a/libs/contract/commands/plugin/index.ts b/libs/contract/commands/plugin/index.ts index ad0fd7e..8910470 100644 --- a/libs/contract/commands/plugin/index.ts +++ b/libs/contract/commands/plugin/index.ts @@ -1,3 +1,4 @@ +export * from './abuse-blocker'; export * from './nftables'; export * from './sync.command'; export * from './torrent-blocker'; diff --git a/libs/contract/commands/stats/get-system-stats.command.ts b/libs/contract/commands/stats/get-system-stats.command.ts index 9c11261..2a9e440 100644 --- a/libs/contract/commands/stats/get-system-stats.command.ts +++ b/libs/contract/commands/stats/get-system-stats.command.ts @@ -1,7 +1,7 @@ import { z } from 'zod'; -import { NodeSystemStatsSchema } from '../../models'; import { REST_API } from '../../api'; +import { NodeSystemStatsSchema } from '../../models'; export namespace GetSystemStatsCommand { export const url = REST_API.STATS.GET_SYSTEM_STATS; @@ -22,6 +22,19 @@ export namespace GetSystemStatsCommand { }) .nullable(), plugins: z.object({ + abuseBlocker: z.object({ + available: z.boolean(), + enabled: z.boolean(), + reportsCount: z.number(), + trackedUsers: z.number(), + activeIncidents: z.number(), + coverageMode: z.enum(['full', 'partial']), + skippedWebhookRules: z.number(), + evictedUsers: z.number(), + evictedKeys: z.number(), + droppedReports: z.number(), + lastError: z.string().nullable(), + }), torrentBlocker: z.object({ reportsCount: z.number(), }), diff --git a/libs/contract/constants/internal/internal.constants.ts b/libs/contract/constants/internal/internal.constants.ts index 6b66ad7..bedd4fd 100644 --- a/libs/contract/constants/internal/internal.constants.ts +++ b/libs/contract/constants/internal/internal.constants.ts @@ -3,3 +3,9 @@ export const XRAY_INTERNAL_API_PATH = '/get-config'; export const XRAY_INTERNAL_FULL_PATH = `/${XRAY_INTERNAL_API_CONTROLLER}${XRAY_INTERNAL_API_PATH}`; export const XRAY_INTERNAL_WEBHOOK_PATH = '/webhook'; export const XRAY_INTERNAL_FULL_WEBHOOK_PATH = `/${XRAY_INTERNAL_API_CONTROLLER}${XRAY_INTERNAL_WEBHOOK_PATH}`; +export const XRAY_INTERNAL_TORRENT_WEBHOOK_PATH = '/webhook/torrent'; +export const XRAY_INTERNAL_ABUSE_WEBHOOK_PATH = '/webhook/abuse'; +export const XRAY_INTERNAL_COMBINED_WEBHOOK_PATH = '/webhook/combined'; +export const XRAY_INTERNAL_FULL_TORRENT_WEBHOOK_PATH = `/${XRAY_INTERNAL_API_CONTROLLER}${XRAY_INTERNAL_TORRENT_WEBHOOK_PATH}`; +export const XRAY_INTERNAL_FULL_ABUSE_WEBHOOK_PATH = `/${XRAY_INTERNAL_API_CONTROLLER}${XRAY_INTERNAL_ABUSE_WEBHOOK_PATH}`; +export const XRAY_INTERNAL_FULL_COMBINED_WEBHOOK_PATH = `/${XRAY_INTERNAL_API_CONTROLLER}${XRAY_INTERNAL_COMBINED_WEBHOOK_PATH}`; diff --git a/libs/contract/constants/xray/stats.ts b/libs/contract/constants/xray/stats.ts index 315c8ff..435ac56 100644 --- a/libs/contract/constants/xray/stats.ts +++ b/libs/contract/constants/xray/stats.ts @@ -48,7 +48,7 @@ export const XRAY_TORRENT_BLOCKER_ROUTING_RULES_MODEL = ({ outboundTag: 'RW_TB_OUTBOUND_BLOCK', webhook: { url: webhookUrl, - deduplication: 5, + deduplication: 0, }, }); 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 0000000..32e2dc9 --- /dev/null +++ b/libs/contract/models/abuse-blocker.report.schema.ts @@ -0,0 +1,89 @@ +import { z } from 'zod'; + +import { XrayWebhookSchema } from './xray-webhook.schema'; + +export const AbuseBlockerRuleNameSchema = z.enum(['horizontal_scan', 'destination_sweep']); +export const AbuseBlockerSeveritySchema = z.enum(['suspicious', 'alert', 'blocked']); +export const AbuseBlockerCoverageModeSchema = z.enum(['full', 'partial']); + +export const AbuseBlockerPolicySchema = z.object({ + excludedPorts: z.array(z.int().min(1).max(65535)), + scoreWindowSeconds: z.int().min(1), + incidentCooldownSeconds: z.int().min(0), + suspiciousScore: z.int().min(1), + alertScore: z.int().min(1), + blockScore: z.int().min(1), + initialBlockSeconds: z.int().min(1), + repeatBlockSeconds: z.int().min(1), + repeatWindowSeconds: z.int().min(1), + evidenceLimit: z.int().min(1), + enhancedEvidenceLimit: z.int().min(1), + maxTrackedUsers: z.int().min(1), + maxKeysPerUser: z.int().min(1), + reportBufferSize: z.int().min(1), + horizontalScan: z.object({ + enabled: z.boolean(), + windowSeconds: z.int().min(1), + uniqueDestinations: z.int().min(2), + ipv4Prefix: z.int().min(0).max(32), + ipv6Prefix: z.int().min(0).max(128), + score: z.int().min(1), + }), + destinationSweep: z.object({ + enabled: z.boolean(), + windowSeconds: z.int().min(1), + uniqueDestinations: z.int().min(2), + score: z.int().min(1), + }), +}); + +export const AbuseBlockerReportSchema = z.object({ + eventId: z.uuid(), + userId: z.string().regex(/^\d+$/), + sourceIp: z.union([z.ipv4(), z.ipv6()]), + destinationIp: z.union([z.ipv4(), z.ipv6()]), + destinationPort: z.int().min(1).max(65535), + detectedAt: z.coerce.date(), + detections: z.array( + z.object({ + rule: AbuseBlockerRuleNameSchema, + key: z.string(), + uniqueDestinations: z.int().min(1), + windowSeconds: z.int().min(1), + score: z.int().min(1), + subnet: z.string().nullable(), + }), + ), + score: z.object({ + before: z.int().min(0), + delta: z.int().min(1), + after: z.int().min(1), + windowSeconds: z.int().min(1), + }), + severity: AbuseBlockerSeveritySchema, + evidence: z.array( + z.object({ + destinationIp: z.union([z.ipv4(), z.ipv6()]), + destinationPort: z.int().min(1).max(65535), + lastSeenAt: z.coerce.date(), + }), + ), + actionReport: z.object({ + action: z.enum(['none', 'ip_block']), + blocked: z.boolean(), + blockDuration: z.int().min(0), + willUnblockAt: z.coerce.date().nullable(), + error: z.string().nullable(), + processedAt: z.coerce.date(), + }), + policy: AbuseBlockerPolicySchema, + configFingerprint: z.string().min(1), + coverageMode: AbuseBlockerCoverageModeSchema, + xrayReport: XrayWebhookSchema, +}); + +export type AbuseBlockerRuleName = z.infer; +export type AbuseBlockerSeverity = z.infer; +export type AbuseBlockerCoverageMode = z.infer; +export type AbuseBlockerPolicy = z.infer; +export type AbuseBlockerReportModel = z.infer; diff --git a/libs/contract/models/index.ts b/libs/contract/models/index.ts index 821d616..293e855 100644 --- a/libs/contract/models/index.ts +++ b/libs/contract/models/index.ts @@ -1,4 +1,5 @@ export * from './node-system.schema'; +export * from './abuse-blocker.report.schema'; export * from './torrent-blocker.report.schema'; export * from './xray-webhook.schema'; export * from './node-metadata.schema'; diff --git a/libs/contract/tsconfig.json b/libs/contract/tsconfig.json index 2023c05..197b0f7 100644 --- a/libs/contract/tsconfig.json +++ b/libs/contract/tsconfig.json @@ -10,16 +10,9 @@ "strictPropertyInitialization": false, "skipLibCheck": true, "noEmit": false, - "lib": [ - "es2020" - ], + "lib": ["es2020"], "declaration": true, - "declarationMap": true, + "declarationMap": true }, - "exclude": [ - "scripts/**/*", - "tests/**/*", - "**/*.spec.ts", - "**/*.test.ts" - ] -} \ No newline at end of file + "exclude": ["build/**/*", "scripts/**/*", "tests/**/*", "**/*.spec.ts", "**/*.test.ts"] +} From f4b5e36a5e8bdd1529ec6b66694338e1d50248cf Mon Sep 17 00:00:00 2001 From: l0nelynx <113856580+l0nelynx@users.noreply.github.com> Date: Sat, 15 Aug 2026 10:51:43 +0300 Subject: [PATCH 2/3] feat(plugin): add abuse blocker runtime --- package.json | 2 +- src/common/utils/generate-api-config.ts | 128 +++-- src/main.ts | 19 +- src/modules/_plugin/commands/index.ts | 3 +- .../set-abuse-blocker-coverage/index.ts | 2 + .../set-abuse-blocker-coverage.command.ts | 8 + .../set-abuse-blocker-coverage.handler.ts | 13 + .../_plugin/constants/nfttables.contants.ts | 1 + src/modules/_plugin/dtos/abuse-blocker.dto.ts | 18 + src/modules/_plugin/dtos/index.ts | 1 + .../events/xray-webhook/xray-webhook.event.ts | 7 +- .../xray-webhook/xray-webhook.handler.ts | 195 +++++--- .../_plugin/interfaces/plugins.interface.ts | 1 + .../abuse-blocker-reports.response.model.ts | 9 + src/modules/_plugin/models/index.ts | 1 + src/modules/_plugin/plugin.controller.ts | 27 +- src/modules/_plugin/plugin.service.ts | 44 +- .../get-abuse-blocker-state.handler.ts | 16 + .../get-abuse-blocker-state.query.ts | 1 + .../queries/get-abuse-blocker-state/index.ts | 2 + .../get-abuse-blocker-stats.handler.ts | 16 + .../get-abuse-blocker-stats.query.ts | 1 + .../queries/get-abuse-blocker-stats/index.ts | 2 + src/modules/_plugin/queries/index.ts | 9 +- src/modules/_plugin/services/nft.service.ts | 40 ++ .../_plugin/services/plugin-state.service.ts | 10 +- .../services/states/abuse-blocker.state.ts | 444 ++++++++++++++++++ src/modules/_plugin/services/states/index.ts | 1 + .../services/states/torrent-blocker.state.ts | 9 + src/modules/_plugin/utils/ip-address.utils.ts | 151 ++++++ src/modules/internal/internal.controller.ts | 23 +- .../models/get-system-stats.response.model.ts | 13 + src/modules/stats/stats.service.ts | 11 +- src/modules/xray-core/xray.service.ts | 15 +- 34 files changed, 1137 insertions(+), 106 deletions(-) create mode 100644 src/modules/_plugin/commands/set-abuse-blocker-coverage/index.ts create mode 100644 src/modules/_plugin/commands/set-abuse-blocker-coverage/set-abuse-blocker-coverage.command.ts create mode 100644 src/modules/_plugin/commands/set-abuse-blocker-coverage/set-abuse-blocker-coverage.handler.ts create mode 100644 src/modules/_plugin/dtos/abuse-blocker.dto.ts create mode 100644 src/modules/_plugin/models/abuse-blocker-reports.response.model.ts create mode 100644 src/modules/_plugin/queries/get-abuse-blocker-state/get-abuse-blocker-state.handler.ts create mode 100644 src/modules/_plugin/queries/get-abuse-blocker-state/get-abuse-blocker-state.query.ts create mode 100644 src/modules/_plugin/queries/get-abuse-blocker-state/index.ts create mode 100644 src/modules/_plugin/queries/get-abuse-blocker-stats/get-abuse-blocker-stats.handler.ts create mode 100644 src/modules/_plugin/queries/get-abuse-blocker-stats/get-abuse-blocker-stats.query.ts create mode 100644 src/modules/_plugin/queries/get-abuse-blocker-stats/index.ts create mode 100644 src/modules/_plugin/services/states/abuse-blocker.state.ts create mode 100644 src/modules/_plugin/utils/ip-address.utils.ts diff --git a/package.json b/package.json index 582e133..ef6cc41 100644 --- a/package.json +++ b/package.json @@ -42,7 +42,7 @@ "@nestjs/platform-express": "11.2.0", "@nestjs/schedule": "^6.1.3", "@remnawave/hashed-set": "^0.0.4", - "@remnawave/node-plugins": "0.6.3", + "@remnawave/node-plugins": "0.7.0", "@remnawave/xtls-sdk": "0.16.0", "@remnawave/xtls-sdk-nestjs": "0.6.1", "compression": "^1.8.1", diff --git a/src/common/utils/generate-api-config.ts b/src/common/utils/generate-api-config.ts index aba4905..234b4b8 100644 --- a/src/common/utils/generate-api-config.ts +++ b/src/common/utils/generate-api-config.ts @@ -1,5 +1,10 @@ import { hasCapNetAdmin } from 'sockdestroy'; +import { + XRAY_INTERNAL_FULL_ABUSE_WEBHOOK_PATH, + XRAY_INTERNAL_FULL_COMBINED_WEBHOOK_PATH, + XRAY_INTERNAL_FULL_TORRENT_WEBHOOK_PATH, +} from '@libs/contracts/constants'; import { XRAY_API_INBOUND_MODEL, XRAY_DEFAULT_API_MODEL, @@ -9,19 +14,26 @@ import { XRAY_TORRENT_BLOCKER_OUTBOUND_MODEL, XRAY_TORRENT_BLOCKER_ROUTING_RULES_MODEL, } from '@libs/contracts/constants/xray'; -import { XRAY_INTERNAL_FULL_WEBHOOK_PATH } from '@libs/contracts/constants'; +import type { AbuseBlockerCoverageMode } from '@libs/contracts/models'; import { IPolicyConfig } from './interfaces'; +interface IWebhookConfig { + url: string; + deduplication: number; +} + +interface IRoutingRule { + ruleTag?: string; + outboundTag?: string; + balancerTag?: string; + webhook?: IWebhookConfig; + [key: string]: unknown; +} + interface IRoutingXrayConfig { - rules: { - ruleTag?: string; - webhook?: { - url: string; - deduplication: number; - }; - [key: string]: unknown; - }[]; + domainStrategy?: string; + rules: IRoutingRule[]; } interface IGenerateApiConfigParams { @@ -30,6 +42,9 @@ interface IGenerateApiConfigParams { enabled: boolean; includeRuleTags: Set; }; + abuseBlockerState: { + enabled: boolean; + }; internal: { socketPath: string; token: string; @@ -37,16 +52,26 @@ interface IGenerateApiConfigParams { }; } -export const generateApiConfig = (args: IGenerateApiConfigParams): Record => { - const { config, torrentBlockerState, internal } = args; +interface IGenerateApiConfigResult { + config: Record; + abuseCoverage: { + mode: AbuseBlockerCoverageMode; + skippedWebhookRules: number; + }; +} + +export const generateApiConfig = (args: IGenerateApiConfigParams): IGenerateApiConfigResult => { + const { config, torrentBlockerState, abuseBlockerState, internal } = args; const policyConfig = config.policy as undefined | IPolicyConfig; + const routingConfig = config.routing as Record | undefined; const hasCapNetAdminResult = hasCapNetAdmin(); + const originalOutbounds = Array.isArray(config.outbounds) ? config.outbounds : []; const builtPolicy: IPolicyConfig = { levels: { '0': { - ...(policyConfig?.levels?.['0'] || {}), + ...policyConfig?.levels?.['0'], statsUserUplink: XRAY_DEFAULT_POLICY_MODEL.policy.levels['0'].statsUserUplink, statsUserDownlink: XRAY_DEFAULT_POLICY_MODEL.policy.levels['0'].statsUserDownlink, statsUserOnline: hasCapNetAdminResult, @@ -65,10 +90,10 @@ export const generateApiConfig = (args: IGenerateApiConfigParams): Record 0) { + const defaultRule: IRoutingRule = { + ruleTag: 'RW_ABUSE_DEFAULT', + network: 'tcp', + outboundTag: defaultOutbound.tag, + webhook: { url: abuseUrl, deduplication: 0 }, + }; + if (routing.domainStrategy === 'IPIfNonMatch') { + defaultRule.ip = ['0.0.0.0/0', '::/0']; + } + routing.rules.push(defaultRule); + defaultRuleAdded = true; + } + } + + if (torrentBlockerState.enabled) { + result.outbounds.push(XRAY_TORRENT_BLOCKER_OUTBOUND_MODEL); + routing.rules.splice( + 1, + 0, + XRAY_TORRENT_BLOCKER_ROUTING_RULES_MODEL({ webhookUrl: torrentUrl }), + ); if (torrentBlockerState.includeRuleTags.size > 0) { for (const rule of routing.rules) { if ( - rule.ruleTag && - typeof rule.ruleTag === 'string' && - torrentBlockerState.includeRuleTags.has(rule.ruleTag) + !rule.ruleTag || + typeof rule.ruleTag !== 'string' || + !torrentBlockerState.includeRuleTags.has(rule.ruleTag) ) { - rule.webhook = { - url: webhookUrl, - deduplication: 5, - }; + continue; + } + + if (rule.webhook?.url === abuseUrl) { + rule.webhook = { url: combinedUrl, deduplication: 0 }; + } else if (!rule.webhook) { + rule.webhook = { url: torrentUrl, deduplication: 0 }; } } } } - return result; + return { + config: result, + abuseCoverage: { + mode: + abuseBlockerState.enabled && defaultRuleAdded && skippedWebhookRules === 0 + ? 'full' + : 'partial', + skippedWebhookRules, + }, + }; }; -const buildWebhookUrl = (internal: { socketPath: string; token: string }): string => { - return `@${internal.socketPath}:${XRAY_INTERNAL_FULL_WEBHOOK_PATH}?token=${internal.token}`; -}; +const buildWebhookUrl = (internal: { socketPath: string; token: string }, path: string): string => + `@${internal.socketPath}:${path}?token=${internal.token}`; diff --git a/src/main.ts b/src/main.ts index a523341..efb2a42 100644 --- a/src/main.ts +++ b/src/main.ts @@ -23,7 +23,10 @@ import { getStartMessage } from '@common/utils/get-start-message'; import { isDevelopment } from '@common/utils/is-development'; import { ROOT } from '@libs/contracts/api'; import { + XRAY_INTERNAL_FULL_ABUSE_WEBHOOK_PATH, + XRAY_INTERNAL_FULL_COMBINED_WEBHOOK_PATH, XRAY_INTERNAL_FULL_PATH, + XRAY_INTERNAL_FULL_TORRENT_WEBHOOK_PATH, XRAY_INTERNAL_FULL_WEBHOOK_PATH, } from '@libs/contracts/constants'; @@ -95,7 +98,13 @@ async function bootstrap(): Promise { app.useGlobalFilters(new NotFoundExceptionFilter()); app.setGlobalPrefix(ROOT, { - exclude: [XRAY_INTERNAL_FULL_PATH, XRAY_INTERNAL_FULL_WEBHOOK_PATH], + exclude: [ + XRAY_INTERNAL_FULL_PATH, + XRAY_INTERNAL_FULL_WEBHOOK_PATH, + XRAY_INTERNAL_FULL_TORRENT_WEBHOOK_PATH, + XRAY_INTERNAL_FULL_ABUSE_WEBHOOK_PATH, + XRAY_INTERNAL_FULL_COMBINED_WEBHOOK_PATH, + ], }); app.useGlobalPipes(new ZodValidationPipe()); @@ -110,7 +119,13 @@ async function bootstrap(): Promise { // '/' + REST_API.VISION.BLOCK_IP, '/' + REST_API.VISION.UNBLOCK_IP internalApp.use( - [XRAY_INTERNAL_FULL_PATH, XRAY_INTERNAL_FULL_WEBHOOK_PATH], + [ + XRAY_INTERNAL_FULL_PATH, + XRAY_INTERNAL_FULL_WEBHOOK_PATH, + XRAY_INTERNAL_FULL_TORRENT_WEBHOOK_PATH, + XRAY_INTERNAL_FULL_ABUSE_WEBHOOK_PATH, + XRAY_INTERNAL_FULL_COMBINED_WEBHOOK_PATH, + ], (req, res, next) => { req.url = req.originalUrl; diff --git a/src/modules/_plugin/commands/index.ts b/src/modules/_plugin/commands/index.ts index 0bac247..dfe31ad 100644 --- a/src/modules/_plugin/commands/index.ts +++ b/src/modules/_plugin/commands/index.ts @@ -1,4 +1,5 @@ import { ResetPluginsHandler } from './reset-plugins/reset-plugins.handler'; import { RunPreStartHandler } from './run-pre-start/run-pre-start.handler'; +import { SetAbuseBlockerCoverageHandler } from './set-abuse-blocker-coverage'; -export const COMMANDS = [ResetPluginsHandler, RunPreStartHandler]; +export const COMMANDS = [ResetPluginsHandler, RunPreStartHandler, SetAbuseBlockerCoverageHandler]; diff --git a/src/modules/_plugin/commands/set-abuse-blocker-coverage/index.ts b/src/modules/_plugin/commands/set-abuse-blocker-coverage/index.ts new file mode 100644 index 0000000..a3fe7e2 --- /dev/null +++ b/src/modules/_plugin/commands/set-abuse-blocker-coverage/index.ts @@ -0,0 +1,2 @@ +export * from './set-abuse-blocker-coverage.command'; +export * from './set-abuse-blocker-coverage.handler'; diff --git a/src/modules/_plugin/commands/set-abuse-blocker-coverage/set-abuse-blocker-coverage.command.ts b/src/modules/_plugin/commands/set-abuse-blocker-coverage/set-abuse-blocker-coverage.command.ts new file mode 100644 index 0000000..c53ed39 --- /dev/null +++ b/src/modules/_plugin/commands/set-abuse-blocker-coverage/set-abuse-blocker-coverage.command.ts @@ -0,0 +1,8 @@ +import type { AbuseBlockerCoverageMode } from '@libs/contracts/models'; + +export class SetAbuseBlockerCoverageCommand { + constructor( + public readonly mode: AbuseBlockerCoverageMode, + public readonly skippedWebhookRules: number, + ) {} +} diff --git a/src/modules/_plugin/commands/set-abuse-blocker-coverage/set-abuse-blocker-coverage.handler.ts b/src/modules/_plugin/commands/set-abuse-blocker-coverage/set-abuse-blocker-coverage.handler.ts new file mode 100644 index 0000000..d10038b --- /dev/null +++ b/src/modules/_plugin/commands/set-abuse-blocker-coverage/set-abuse-blocker-coverage.handler.ts @@ -0,0 +1,13 @@ +import { CommandHandler, ICommandHandler } from '@nestjs/cqrs'; + +import { PluginStateService } from '../../services/plugin-state.service'; +import { SetAbuseBlockerCoverageCommand } from './set-abuse-blocker-coverage.command'; + +@CommandHandler(SetAbuseBlockerCoverageCommand) +export class SetAbuseBlockerCoverageHandler implements ICommandHandler { + constructor(private readonly pluginState: PluginStateService) {} + + async execute(command: SetAbuseBlockerCoverageCommand): Promise { + this.pluginState.abuseBlocker.setCoverage(command.mode, command.skippedWebhookRules); + } +} diff --git a/src/modules/_plugin/constants/nfttables.contants.ts b/src/modules/_plugin/constants/nfttables.contants.ts index 03a3f60..f45f69a 100644 --- a/src/modules/_plugin/constants/nfttables.contants.ts +++ b/src/modules/_plugin/constants/nfttables.contants.ts @@ -1,6 +1,7 @@ export const NFT_TABLES_CONSTANTS = { TABLE_NAME: 'remnanode', TORRENT_BLOCKER_SET_NAME: 'torrent-blocker', + ABUSE_BLOCKER_SET_NAME: 'abuse-blocker', INGRESS_FILTER_IP_SET_NAME: 'ingress-filter-ip', EGRESS_FILTER_IP_SET_NAME: 'egress-filter-ip', EGRESS_FILTER_PORT_SET_NAME: 'egress-filter-port', diff --git a/src/modules/_plugin/dtos/abuse-blocker.dto.ts b/src/modules/_plugin/dtos/abuse-blocker.dto.ts new file mode 100644 index 0000000..53ba8b9 --- /dev/null +++ b/src/modules/_plugin/dtos/abuse-blocker.dto.ts @@ -0,0 +1,18 @@ +import { createZodDto } from 'nestjs-zod'; + +import { + CollectAbuseBlockerReportsCommand, + RefreshAbuseBlockCommand, +} from '@libs/contracts/commands/plugin'; + +export class CollectAbuseBlockerReportsResponseDto extends createZodDto( + CollectAbuseBlockerReportsCommand.ResponseSchema, +) {} + +export class RefreshAbuseBlockRequestDto extends createZodDto( + RefreshAbuseBlockCommand.RequestSchema, +) {} + +export class RefreshAbuseBlockResponseDto extends createZodDto( + RefreshAbuseBlockCommand.ResponseSchema, +) {} diff --git a/src/modules/_plugin/dtos/index.ts b/src/modules/_plugin/dtos/index.ts index e095121..fbbf532 100644 --- a/src/modules/_plugin/dtos/index.ts +++ b/src/modules/_plugin/dtos/index.ts @@ -1,3 +1,4 @@ +export * from './abuse-blocker.dto'; export * from './collect-reports.dto'; export * from './nftables.dto'; export * from './sync.dto'; diff --git a/src/modules/_plugin/events/xray-webhook/xray-webhook.event.ts b/src/modules/_plugin/events/xray-webhook/xray-webhook.event.ts index 2471bf5..4794909 100644 --- a/src/modules/_plugin/events/xray-webhook/xray-webhook.event.ts +++ b/src/modules/_plugin/events/xray-webhook/xray-webhook.event.ts @@ -1,3 +1,8 @@ +export type XrayWebhookTarget = 'torrent' | 'abuse' | 'combined'; + export class XrayWebhookEvent { - constructor(public readonly webhook: unknown) {} + constructor( + public readonly webhook: unknown, + public readonly target: XrayWebhookTarget = 'torrent', + ) {} } diff --git a/src/modules/_plugin/events/xray-webhook/xray-webhook.handler.ts b/src/modules/_plugin/events/xray-webhook/xray-webhook.handler.ts index 44d1ade..6fe9e56 100644 --- a/src/modules/_plugin/events/xray-webhook/xray-webhook.handler.ts +++ b/src/modules/_plugin/events/xray-webhook/xray-webhook.handler.ts @@ -1,18 +1,46 @@ -import { isIP } from 'node:net'; +import type { IAbuseBlockerObservation } from '../../services/states/abuse-blocker.state'; + +import { randomUUID } from 'node:crypto'; import { Logger } from '@nestjs/common'; -import { IEventHandler, EventsHandler } from '@nestjs/cqrs'; +import { EventsHandler, IEventHandler } from '@nestjs/cqrs'; import { formatExecutionTime, getTime } from '@common/utils/get-elapsed-time'; -import { TorrentBlockerReportModel, XrayWebhookSchema } from '@libs/contracts/models'; +import { + type AbuseBlockerReportModel, + type XrayWebhookModel, + XrayWebhookSchema, +} from '@libs/contracts/models'; import { NftService } from '../../services/nft.service'; import { PluginStateService } from '../../services/plugin-state.service'; +import { parseNetworkEndpoint } from '../../utils/ip-address.utils'; import { XrayWebhookEvent } from './xray-webhook.event'; -const SOURCE_REGEX = /^(?:(?:tcp|udp):)?(?:\[(.+?)\]|(.+?))(?::(\d+))?$/; const WEBHOOK_TIMEOUT_MS = 5_000; +export const toAbuseBlockerObservation = ( + webhook: XrayWebhookModel, +): IAbuseBlockerObservation | null => { + if (webhook.network.toLowerCase() !== 'tcp') return null; + if (!webhook.email || !/^\d+$/.test(webhook.email)) return null; + + const source = parseNetworkEndpoint(webhook.source); + const destination = [webhook.originalTarget, webhook.routeTarget, webhook.destination] + .map(parseNetworkEndpoint) + .find((candidate) => candidate?.port && candidate.port >= 1 && candidate.port <= 65535); + if (!source || !destination?.port) return null; + + return { + userId: webhook.email, + sourceIp: source.ip, + destinationIp: destination.ip, + destinationPort: destination.port, + timestamp: Number.isFinite(webhook.ts) ? webhook.ts * 1000 : Date.now(), + xrayReport: webhook, + }; +}; + @EventsHandler(XrayWebhookEvent) export class XrayWebhookHandler implements IEventHandler { public readonly logger = new Logger(XrayWebhookHandler.name); @@ -21,86 +49,133 @@ export class XrayWebhookHandler implements IEventHandler { private readonly pluginState: PluginStateService, private readonly nftService: NftService, ) {} - async handle(event: XrayWebhookEvent) { + + async handle(event: XrayWebhookEvent): Promise { const ct = getTime(); try { - if (!this.pluginState.torrentBlocker.isEnabled) return; - const parsed = await XrayWebhookSchema.safeParseAsync(event.webhook); if (!parsed.success) { this.logger.error(`Invalid webhook: ${JSON.stringify(parsed.error)}`); return; } - this.logger.debug(JSON.stringify(parsed.data, null, 2)); + if (event.target === 'torrent' || event.target === 'combined') { + await this.handleTorrentBlocker(parsed.data); + } + if (event.target === 'abuse' || event.target === 'combined') { + await this.handleAbuseBlocker(parsed.data); + } + } catch (error) { + this.logger.error(`Error in XrayWebhookHandler: ${error}`); + } finally { + this.logger.debug(`Webhook handled in: ${formatExecutionTime(ct)}`); + } + } - const webhook = parsed.data; + private async handleTorrentBlocker(webhook: XrayWebhookModel): Promise { + const state = this.pluginState.torrentBlocker; + if (!state.isEnabled || !webhook.email) return; - const ip = this.extractIp(webhook.source); + const source = parseNetworkEndpoint(webhook.source); + if (!source) return; + if (!state.shouldProcess(webhook.email, Date.now())) return; + if (state.isIpIgnored(source.ip) || state.isUserIgnored(webhook.email)) return; - if (!ip || !webhook.email) return; + const blockDuration = state.duration!; + let blocked = false; + try { + await this.nftService.blockIp(source.ip, blockDuration); + blocked = true; + } catch (error) { + this.logger.error(`Failed to block torrent source IP ${source.ip}: ${error}`); + } - const whitelisted = - this.pluginState.torrentBlocker.isIpIgnored(ip) || - this.pluginState.torrentBlocker.isUserIgnored(webhook.email); + const report = { + actionReport: { + blocked, + ip: source.ip, + blockDuration, + willUnblockAt: new Date(Date.now() + blockDuration * 1000), + userId: webhook.email, + processedAt: new Date(), + }, + xrayReport: webhook, + }; + state.addReport(report); + + const webhookUrl = state.getWebhookUrl(); + if (webhookUrl) this.sendWebhook(webhookUrl, report); + } - if (whitelisted) { - return; - } + private async handleAbuseBlocker(webhook: XrayWebhookModel): Promise { + const state = this.pluginState.abuseBlocker; + if (!state.isEnabled) return; - const blockDuration = this.pluginState.torrentBlocker.duration!; + const observation = toAbuseBlockerObservation(webhook); + if (!observation) return; - let blocked = false; + const analysis = state.analyze(observation); + if (!analysis) return; - try { - await this.nftService.blockIp(ip, blockDuration); - blocked = true; + const policy = state.policy; + if (!policy) return; - this.logger.log( - `[TORRENT-BLOCKER] IP: ${ip}, user: ${webhook.email}, blocked: ${blocked}, duration: ${blockDuration}s`, + let blocked = false; + let blockError: string | null = null; + if (analysis.shouldBlock) { + try { + await this.nftService.blockAbuseIp( + observation.sourceIp, + policy.initialBlockSeconds, ); + blocked = true; } catch (error) { - this.logger.error(`Failed to block IP ${ip}: ${error}`); - } - - const report: TorrentBlockerReportModel = { - actionReport: { - blocked, - ip, - blockDuration, - willUnblockAt: new Date(Date.now() + blockDuration * 1000), - userId: webhook.email, - processedAt: new Date(), - }, - xrayReport: webhook, - }; - - this.pluginState.torrentBlocker.addReport(report); - - const webhookUrl = this.pluginState.torrentBlocker.getWebhookUrl(); - - if (webhookUrl) { - this.sendWebhook(webhookUrl, report); + blockError = error instanceof Error ? error.message : String(error); + state.setLastError(error); } - } catch (error) { - this.logger.error(`Error in Event XrayWebhookHandler: ${error}`); - } finally { - this.logger.debug(`Webhook handled in: ${formatExecutionTime(ct)}`); } - } - - private extractIp(source: string | null): string | null { - if (!source) return null; - - const prefixMatch = source.match(SOURCE_REGEX); - const candidate = prefixMatch ? prefixMatch[1] || prefixMatch[2] : source; - - if (isIP(candidate) === 0) return null; - return candidate; + const processedAt = new Date(); + const report: AbuseBlockerReportModel = { + eventId: randomUUID(), + userId: observation.userId, + sourceIp: observation.sourceIp, + destinationIp: observation.destinationIp, + destinationPort: observation.destinationPort, + detectedAt: new Date(observation.timestamp), + detections: analysis.detections, + score: { + before: analysis.scoreBefore, + delta: analysis.scoreDelta, + after: analysis.scoreAfter, + windowSeconds: state.scoreWindowSeconds, + }, + severity: analysis.severity, + evidence: analysis.evidence, + actionReport: { + action: analysis.shouldBlock ? 'ip_block' : 'none', + blocked, + blockDuration: analysis.shouldBlock ? policy.initialBlockSeconds : 0, + willUnblockAt: + analysis.shouldBlock && blocked + ? new Date(processedAt.getTime() + policy.initialBlockSeconds * 1000) + : null, + error: blockError, + processedAt, + }, + policy, + configFingerprint: state.fingerprint, + coverageMode: state.stats.coverageMode, + xrayReport: webhook, + }; + + state.addReport(report); + this.logger.log( + `[ABUSE-BLOCKER] user=${observation.userId}, source=${observation.sourceIp}, score=${analysis.scoreAfter}, severity=${analysis.severity}, blocked=${blocked}`, + ); } - private sendWebhook(url: string, report: TorrentBlockerReportModel): void { + private sendWebhook(url: string, report: unknown): void { fetch(url, { method: 'POST', headers: { 'content-type': 'application/json' }, diff --git a/src/modules/_plugin/interfaces/plugins.interface.ts b/src/modules/_plugin/interfaces/plugins.interface.ts index db3decd..66eba29 100644 --- a/src/modules/_plugin/interfaces/plugins.interface.ts +++ b/src/modules/_plugin/interfaces/plugins.interface.ts @@ -1,4 +1,5 @@ export interface IPlugins { + abuseBlocker: boolean; ingressFilter: boolean; egressFilter: boolean; torrentBlocker: boolean; diff --git a/src/modules/_plugin/models/abuse-blocker-reports.response.model.ts b/src/modules/_plugin/models/abuse-blocker-reports.response.model.ts new file mode 100644 index 0000000..86c2432 --- /dev/null +++ b/src/modules/_plugin/models/abuse-blocker-reports.response.model.ts @@ -0,0 +1,9 @@ +import type { AbuseBlockerReportModel } from '@libs/contracts/models'; + +export class AbuseBlockerReportsResponseModel { + public reports: AbuseBlockerReportModel[]; + + constructor(reports: AbuseBlockerReportModel[]) { + this.reports = reports; + } +} diff --git a/src/modules/_plugin/models/index.ts b/src/modules/_plugin/models/index.ts index d362782..0acd663 100644 --- a/src/modules/_plugin/models/index.ts +++ b/src/modules/_plugin/models/index.ts @@ -1,2 +1,3 @@ +export * from './abuse-blocker-reports.response.model'; export * from './generic.response.model'; export * from './torrent-blocker-reports.response.model'; diff --git a/src/modules/_plugin/plugin.controller.ts b/src/modules/_plugin/plugin.controller.ts index 723bd54..d380649 100644 --- a/src/modules/_plugin/plugin.controller.ts +++ b/src/modules/_plugin/plugin.controller.ts @@ -1,21 +1,24 @@ import { Body, Controller, Post, UseFilters, UseGuards } from '@nestjs/common'; -import { JwtDefaultGuard } from '@common/guards/jwt-guards'; import { HttpExceptionFilter } from '@common/exception'; +import { JwtDefaultGuard } from '@common/guards/jwt-guards'; import { errorHandler } from '@common/helpers'; import { PLUGIN_CONTROLLER, PLUGIN_ROUTES } from '@libs/contracts/api'; import { + CollectAbuseBlockerReportsResponseDto, BlockIpsRequestDto, BlockIpsResponseDto, CollectReportsResponseDto, RecreateTablesResponseDto, + RefreshAbuseBlockRequestDto, + RefreshAbuseBlockResponseDto, UnblockIpsRequestDto, UnblockIpsResponseDto, } from './dtos'; import { SyncRequestDto, SyncResponseDto } from './dtos/sync.dto'; -import { NftService } from './services/nft.service'; import { PluginService } from './plugin.service'; +import { NftService } from './services/nft.service'; @UseFilters(HttpExceptionFilter) @UseGuards(JwtDefaultGuard) @@ -46,6 +49,26 @@ export class PluginController { }; } + @Post(PLUGIN_ROUTES.ABUSE_BLOCKER.COLLECT) + public async collectAbuseBlockerReports(): Promise { + const response = await this.pluginService.collectAbuseBlockerReports(); + const data = errorHandler(response); + + return { response: data }; + } + + @Post(PLUGIN_ROUTES.ABUSE_BLOCKER.REFRESH_BLOCK) + public async refreshAbuseBlock( + @Body() body: RefreshAbuseBlockRequestDto, + ): Promise { + try { + await this.nftService.refreshAbuseIp(body.ip, body.timeout); + return { response: { accepted: true } }; + } catch { + return { response: { accepted: false } }; + } + } + @Post(PLUGIN_ROUTES.NFTABLES.BLOCK_IPS) public async blockIps(@Body() body: BlockIpsRequestDto): Promise { const response = await this.nftService.blockIpsController(body); diff --git a/src/modules/_plugin/plugin.service.ts b/src/modules/_plugin/plugin.service.ts index 0289773..3a9a9e7 100644 --- a/src/modules/_plugin/plugin.service.ts +++ b/src/modules/_plugin/plugin.service.ts @@ -13,7 +13,7 @@ import { RemoveOutboundCommand } from '../handler/commands/remove-outbound/remov import { StopXrayCommand } from '../xray-core/commands/stop-xray'; import { SyncRequestDto } from './dtos'; import { GenericResponseModel } from './models'; -import { TorrentBlockerReportsResponseModel } from './models/torrent-blocker-reports.response.model'; +import { AbuseBlockerReportsResponseModel, TorrentBlockerReportsResponseModel } from './models'; import { NftService } from './services/nft.service'; import { PluginStateService } from './services/plugin-state.service'; @@ -74,6 +74,7 @@ export class PluginService { } const currentTorrentBlocker = this.state.torrentBlocker.isEnabled; + const currentAbuseBlocker = this.state.abuseBlocker.isEnabled; const currentTorrentBlockerIncludeRuleTags = new Set( this.state.torrentBlocker.includeRuleTagsSet, ); @@ -88,6 +89,7 @@ export class PluginService { this.syncConnectionDrop(pluginData, sharedMap); this.syncTorrentBlocker(pluginData, sharedMap); + this.syncAbuseBlocker(pluginData, sharedMap, configHash); this.syncPreStart(pluginData); await this.syncIngressFilter(pluginData, sharedMap); @@ -100,6 +102,7 @@ export class PluginService { const wasEnabled = !!currentTorrentBlocker; const nowEnabled = !!pluginData.torrentBlocker?.enabled; + const abuseNowEnabled = !!pluginData.abuseBlocker?.enabled; if (wasEnabled && !nowEnabled && !pluginData.torrentBlocker?.includeRuleTags) { await this.commandBus.execute( @@ -107,6 +110,7 @@ export class PluginService { ); } else { const needsRestart = + currentAbuseBlocker !== abuseNowEnabled || (wasEnabled && !nowEnabled) || (!wasEnabled && nowEnabled) || (wasEnabled && @@ -217,6 +221,32 @@ export class PluginService { ); } + private syncAbuseBlocker( + pluginData: TNodePlugin, + sharedMap: Map, + configFingerprint: string, + ): void { + const config = pluginData.abuseBlocker; + if (!config?.enabled) return; + if (!this.nftService.isAvailable) return; + + const ignoredSources = this.resolveIpList(config.ignoreLists.sourceIp, sharedMap); + const ignoredDestinations = this.resolveIpList(config.ignoreLists.destinationIp, sharedMap); + const ignoredUsers = config.ignoreLists.userId.map(String); + + this.state.abuseBlocker.configure({ + config, + configFingerprint, + ignoredUsers, + ignoredSources, + ignoredDestinations, + }); + + this.logger.log( + `[PLUGIN] Abuse-Blocker: scoreWindow=${config.scoreWindowSeconds}s, blockScore=${config.blockScore}, ${ignoredUsers.length} ignored users`, + ); + } + private resolveIpList(ips: string[], sharedMap: Map): string[] { return ips.flatMap((ip) => { if (ip.startsWith('ext:')) { @@ -280,4 +310,16 @@ export class PluginService { return { isOk: true, response: new TorrentBlockerReportsResponseModel([]) }; } } + + public async collectAbuseBlockerReports(): Promise< + ICommandResponse + > { + try { + const reports = this.state.abuseBlocker.flushReports(); + return { isOk: true, response: new AbuseBlockerReportsResponseModel(reports) }; + } catch (error) { + this.logger.error(error); + return { isOk: true, response: new AbuseBlockerReportsResponseModel([]) }; + } + } } diff --git a/src/modules/_plugin/queries/get-abuse-blocker-state/get-abuse-blocker-state.handler.ts b/src/modules/_plugin/queries/get-abuse-blocker-state/get-abuse-blocker-state.handler.ts new file mode 100644 index 0000000..c08df8e --- /dev/null +++ b/src/modules/_plugin/queries/get-abuse-blocker-state/get-abuse-blocker-state.handler.ts @@ -0,0 +1,16 @@ +import { Logger } from '@nestjs/common'; +import { IQueryHandler, QueryHandler } from '@nestjs/cqrs'; + +import { PluginStateService } from '../../services/plugin-state.service'; +import { GetAbuseBlockerStateQuery } from './get-abuse-blocker-state.query'; + +@QueryHandler(GetAbuseBlockerStateQuery) +export class GetAbuseBlockerStateHandler implements IQueryHandler { + private readonly logger = new Logger(GetAbuseBlockerStateHandler.name); + + constructor(private readonly pluginState: PluginStateService) {} + + async execute(): Promise<{ enabled: boolean }> { + return { enabled: this.pluginState.abuseBlocker.isEnabled }; + } +} diff --git a/src/modules/_plugin/queries/get-abuse-blocker-state/get-abuse-blocker-state.query.ts b/src/modules/_plugin/queries/get-abuse-blocker-state/get-abuse-blocker-state.query.ts new file mode 100644 index 0000000..ef974d4 --- /dev/null +++ b/src/modules/_plugin/queries/get-abuse-blocker-state/get-abuse-blocker-state.query.ts @@ -0,0 +1 @@ +export class GetAbuseBlockerStateQuery {} diff --git a/src/modules/_plugin/queries/get-abuse-blocker-state/index.ts b/src/modules/_plugin/queries/get-abuse-blocker-state/index.ts new file mode 100644 index 0000000..eb08e7b --- /dev/null +++ b/src/modules/_plugin/queries/get-abuse-blocker-state/index.ts @@ -0,0 +1,2 @@ +export * from './get-abuse-blocker-state.handler'; +export * from './get-abuse-blocker-state.query'; diff --git a/src/modules/_plugin/queries/get-abuse-blocker-stats/get-abuse-blocker-stats.handler.ts b/src/modules/_plugin/queries/get-abuse-blocker-stats/get-abuse-blocker-stats.handler.ts new file mode 100644 index 0000000..97433c0 --- /dev/null +++ b/src/modules/_plugin/queries/get-abuse-blocker-stats/get-abuse-blocker-stats.handler.ts @@ -0,0 +1,16 @@ +import { IQueryHandler, QueryHandler } from '@nestjs/cqrs'; + +import { PluginStateService } from '../../services/plugin-state.service'; +import { GetAbuseBlockerStatsQuery } from './get-abuse-blocker-stats.query'; + +@QueryHandler(GetAbuseBlockerStatsQuery) +export class GetAbuseBlockerStatsHandler implements IQueryHandler { + constructor(private readonly pluginState: PluginStateService) {} + + async execute() { + return { + available: this.pluginState.plugins.abuseBlocker, + ...this.pluginState.abuseBlocker.stats, + }; + } +} diff --git a/src/modules/_plugin/queries/get-abuse-blocker-stats/get-abuse-blocker-stats.query.ts b/src/modules/_plugin/queries/get-abuse-blocker-stats/get-abuse-blocker-stats.query.ts new file mode 100644 index 0000000..eced7e1 --- /dev/null +++ b/src/modules/_plugin/queries/get-abuse-blocker-stats/get-abuse-blocker-stats.query.ts @@ -0,0 +1 @@ +export class GetAbuseBlockerStatsQuery {} diff --git a/src/modules/_plugin/queries/get-abuse-blocker-stats/index.ts b/src/modules/_plugin/queries/get-abuse-blocker-stats/index.ts new file mode 100644 index 0000000..82a3f90 --- /dev/null +++ b/src/modules/_plugin/queries/get-abuse-blocker-stats/index.ts @@ -0,0 +1,2 @@ +export * from './get-abuse-blocker-stats.handler'; +export * from './get-abuse-blocker-stats.query'; diff --git a/src/modules/_plugin/queries/index.ts b/src/modules/_plugin/queries/index.ts index d7a12c3..14add3c 100644 --- a/src/modules/_plugin/queries/index.ts +++ b/src/modules/_plugin/queries/index.ts @@ -1,4 +1,11 @@ +import { GetAbuseBlockerStateHandler } from './get-abuse-blocker-state'; +import { GetAbuseBlockerStatsHandler } from './get-abuse-blocker-stats'; import { GetTorrentBlockerReportsCountHandler } from './get-torrent-blocker-reports-count'; import { GetTorrentBlockerStateHandler } from './get-torrent-blocker-state'; -export const QUERIES = [GetTorrentBlockerStateHandler, GetTorrentBlockerReportsCountHandler]; +export const QUERIES = [ + GetAbuseBlockerStateHandler, + GetAbuseBlockerStatsHandler, + GetTorrentBlockerStateHandler, + GetTorrentBlockerReportsCountHandler, +]; diff --git a/src/modules/_plugin/services/nft.service.ts b/src/modules/_plugin/services/nft.service.ts index 8bd4f34..532fe7c 100644 --- a/src/modules/_plugin/services/nft.service.ts +++ b/src/modules/_plugin/services/nft.service.ts @@ -17,6 +17,7 @@ export class NftService implements OnModuleDestroy, OnModuleInit { private readonly logger = new Logger(NftService.name); private nftManager: NftManager | null = null; private available = false; + private abuseRefreshQueue: Promise = Promise.resolve(); constructor( private readonly state: PluginStateService, @@ -33,6 +34,7 @@ export class NftService implements OnModuleDestroy, OnModuleInit { if (!capNetAdmin) return; this.state.setPlugins({ + abuseBlocker: false, connectionDrop: true, ingressFilter: false, torrentBlocker: false, @@ -46,6 +48,7 @@ export class NftService implements OnModuleDestroy, OnModuleInit { ingressAddrSets: [ NFT_TABLES_CONSTANTS.INGRESS_FILTER_IP_SET_NAME, NFT_TABLES_CONSTANTS.TORRENT_BLOCKER_SET_NAME, + NFT_TABLES_CONSTANTS.ABUSE_BLOCKER_SET_NAME, ], egressAddrSets: [NFT_TABLES_CONSTANTS.EGRESS_FILTER_IP_SET_NAME], egressPortSets: [NFT_TABLES_CONSTANTS.EGRESS_FILTER_PORT_SET_NAME], @@ -55,6 +58,7 @@ export class NftService implements OnModuleDestroy, OnModuleInit { this.available = true; this.state.setPlugins({ + abuseBlocker: true, connectionDrop: true, ingressFilter: true, torrentBlocker: true, @@ -121,6 +125,36 @@ export class NftService implements OnModuleDestroy, OnModuleInit { this.eventBus.publish(new DropConnectionsEvent([ip])); } + public async blockAbuseIp(ip: string, timeoutSeconds: number): Promise { + if (!this.nftManager) throw new Error('nftables is unavailable'); + await this.nftManager.addAddress({ + ip, + set: NFT_TABLES_CONSTANTS.ABUSE_BLOCKER_SET_NAME, + timeout: timeoutSeconds, + }); + this.eventBus.publish(new DropConnectionsEvent([ip])); + } + + public async refreshAbuseIp(ip: string, timeoutSeconds: number): Promise { + const operation = this.abuseRefreshQueue.then(async () => { + if (!this.nftManager) throw new Error('nftables is unavailable'); + + try { + await this.nftManager.removeAddresses({ + ips: [ip], + set: NFT_TABLES_CONSTANTS.ABUSE_BLOCKER_SET_NAME, + }); + } catch { + this.logger.debug(`[ABUSE-BLOCKER] ${ip} was not present before refresh.`); + } + + await this.blockAbuseIp(ip, timeoutSeconds); + }); + + this.abuseRefreshQueue = operation.catch(() => void 0); + return operation; + } + public async recreateTables(): Promise { if (!this.nftManager) return; await this.nftManager.createTable(); @@ -129,6 +163,7 @@ export class NftService implements OnModuleDestroy, OnModuleInit { private logAvailablePlugins(): void { const plugins = this.state.plugins; [ + { name: 'Abuse Blocker', enabled: plugins.abuseBlocker }, { name: 'Ingress Filter', enabled: plugins.ingressFilter }, { name: 'Egress Filter', enabled: plugins.egressFilter }, { name: 'Torrent Blocker', enabled: plugins.torrentBlocker }, @@ -194,6 +229,11 @@ export class NftService implements OnModuleDestroy, OnModuleInit { set: NFT_TABLES_CONSTANTS.INGRESS_FILTER_IP_SET_NAME, }); + await this.nftManager.removeAddresses({ + ips, + set: NFT_TABLES_CONSTANTS.ABUSE_BLOCKER_SET_NAME, + }); + return { isOk: true, response: new GenericResponseModel(true), diff --git a/src/modules/_plugin/services/plugin-state.service.ts b/src/modules/_plugin/services/plugin-state.service.ts index b5f90f7..1dc24d5 100644 --- a/src/modules/_plugin/services/plugin-state.service.ts +++ b/src/modules/_plugin/services/plugin-state.service.ts @@ -1,10 +1,16 @@ import { Injectable } from '@nestjs/common'; import { IPlugins } from '../interfaces'; -import { TorrentBlockerState, ConnectionDropState, PreStartState } from './states'; +import { + AbuseBlockerState, + TorrentBlockerState, + ConnectionDropState, + PreStartState, +} from './states'; @Injectable() export class PluginStateService { + public readonly abuseBlocker = new AbuseBlockerState(); public readonly torrentBlocker = new TorrentBlockerState(); public readonly connectionDrop = new ConnectionDropState(); public readonly preStart = new PreStartState(); @@ -13,6 +19,7 @@ export class PluginStateService { private lastConfigHash: string | null = null; private availablePlugins: IPlugins = { + abuseBlocker: false, connectionDrop: false, ingressFilter: false, torrentBlocker: false, @@ -44,6 +51,7 @@ export class PluginStateService { } resetState(): void { + this.abuseBlocker.reset(); this.torrentBlocker.reset(); this.connectionDrop.reset(); this.preStart.reset(); diff --git a/src/modules/_plugin/services/states/abuse-blocker.state.ts b/src/modules/_plugin/services/states/abuse-blocker.state.ts new file mode 100644 index 0000000..82e0f0d --- /dev/null +++ b/src/modules/_plugin/services/states/abuse-blocker.state.ts @@ -0,0 +1,444 @@ +import type { TNodePlugin } from '@remnawave/node-plugins'; + +import type { + AbuseBlockerCoverageMode, + AbuseBlockerPolicy, + AbuseBlockerReportModel, + AbuseBlockerRuleName, + AbuseBlockerSeverity, + XrayWebhookModel, +} from '@libs/contracts/models'; + +import { getNetworkKey, IpMatcher } from '../../utils/ip-address.utils'; + +type AbuseBlockerConfig = NonNullable; + +interface IDetectorKeyState { + destinations: Map; + fired: boolean; + lastFiredAt: number; + lastSeenAt: number; +} + +interface IUserState { + horizontal: Map; + sweep: Map; + scoreEvents: Array<{ score: number; timestamp: number }>; + lastSeenAt: number; +} + +export interface IAbuseBlockerObservation { + userId: string; + sourceIp: string; + destinationIp: string; + destinationPort: number; + timestamp: number; + xrayReport: XrayWebhookModel; +} + +export interface IAbuseBlockerDetection { + rule: AbuseBlockerRuleName; + key: string; + uniqueDestinations: number; + windowSeconds: number; + score: number; + subnet: string | null; +} + +export interface IAbuseBlockerAnalysis { + detections: IAbuseBlockerDetection[]; + scoreBefore: number; + scoreDelta: number; + scoreAfter: number; + severity: AbuseBlockerSeverity; + evidence: Array<{ destinationIp: string; destinationPort: number; lastSeenAt: Date }>; + shouldBlock: boolean; +} + +export class AbuseBlockerState { + private enabled = false; + private config: AbuseBlockerConfig | null = null; + private configFingerprint = ''; + private ignoredUsers = new Set(); + private ignoredSources = new IpMatcher([]); + private ignoredDestinations = new IpMatcher([]); + private users = new Map(); + private reports = new Map(); + private coverageMode: AbuseBlockerCoverageMode = 'partial'; + private skippedWebhookRules = 0; + private evictedUsers = 0; + private evictedKeys = 0; + private droppedReports = 0; + private lastError: string | null = null; + + get isEnabled(): boolean { + return this.enabled; + } + + get policy(): AbuseBlockerPolicy | null { + if (!this.config) return null; + const config = this.config; + return { + excludedPorts: config.excludedPorts, + scoreWindowSeconds: config.scoreWindowSeconds, + incidentCooldownSeconds: config.incidentCooldownSeconds, + suspiciousScore: config.suspiciousScore, + alertScore: config.alertScore, + blockScore: config.blockScore, + initialBlockSeconds: config.initialBlockSeconds, + repeatBlockSeconds: config.repeatBlockSeconds, + repeatWindowSeconds: config.repeatWindowSeconds, + evidenceLimit: config.evidenceLimit, + enhancedEvidenceLimit: config.enhancedEvidenceLimit, + maxTrackedUsers: config.maxTrackedUsers, + maxKeysPerUser: config.maxKeysPerUser, + reportBufferSize: config.reportBufferSize, + horizontalScan: config.horizontalScan, + destinationSweep: config.destinationSweep, + }; + } + + get fingerprint(): string { + return this.configFingerprint; + } + + get scoreWindowSeconds(): number { + return this.config?.scoreWindowSeconds ?? 0; + } + + configure(args: { + config: AbuseBlockerConfig; + configFingerprint: string; + ignoredUsers: string[]; + ignoredSources: string[]; + ignoredDestinations: string[]; + }): void { + this.enabled = true; + this.config = args.config; + this.configFingerprint = args.configFingerprint; + this.ignoredUsers = new Set(args.ignoredUsers); + this.ignoredSources = new IpMatcher(args.ignoredSources); + this.ignoredDestinations = new IpMatcher(args.ignoredDestinations); + } + + analyze(observation: IAbuseBlockerObservation): IAbuseBlockerAnalysis | null { + const config = this.config; + if (!this.enabled || !config) return null; + if (this.ignoredUsers.has(observation.userId)) return null; + if (this.ignoredSources.matches(observation.sourceIp)) return null; + if (this.ignoredDestinations.matches(observation.destinationIp)) return null; + if (config.excludedPorts.includes(observation.destinationPort)) return null; + + const user = this.getUserState(observation.userId, observation.timestamp); + this.pruneScore(user, observation.timestamp, config.scoreWindowSeconds); + const scoreBefore = user.scoreEvents.reduce((sum, event) => sum + event.score, 0); + const detections: IAbuseBlockerDetection[] = []; + + if (config.horizontalScan.enabled) { + const subnet = getNetworkKey( + observation.destinationIp, + config.horizontalScan.ipv4Prefix, + config.horizontalScan.ipv6Prefix, + ); + if (subnet) { + const key = `${observation.destinationPort}|${subnet}`; + const state = this.getDetectorState( + user.horizontal, + key, + user, + observation.timestamp, + ); + if ( + this.recordDestination( + state, + observation, + config.horizontalScan.windowSeconds, + config.horizontalScan.uniqueDestinations, + config.incidentCooldownSeconds, + ) + ) { + detections.push({ + rule: 'horizontal_scan', + key, + uniqueDestinations: state.destinations.size, + windowSeconds: config.horizontalScan.windowSeconds, + score: config.horizontalScan.score, + subnet, + }); + } + } + } + + if (config.destinationSweep.enabled) { + const key = String(observation.destinationPort); + const state = this.getDetectorState(user.sweep, key, user, observation.timestamp); + if ( + this.recordDestination( + state, + observation, + config.destinationSweep.windowSeconds, + config.destinationSweep.uniqueDestinations, + config.incidentCooldownSeconds, + ) + ) { + detections.push({ + rule: 'destination_sweep', + key, + uniqueDestinations: state.destinations.size, + windowSeconds: config.destinationSweep.windowSeconds, + score: config.destinationSweep.score, + subnet: null, + }); + } + } + + if (detections.length === 0) { + this.updateBufferedEvidence( + observation.userId, + user, + observation.destinationPort, + ); + return null; + } + + const scoreDelta = detections.reduce((sum, detection) => sum + detection.score, 0); + user.scoreEvents.push({ score: scoreDelta, timestamp: observation.timestamp }); + const scoreAfter = scoreBefore + scoreDelta; + if (scoreAfter < config.suspiciousScore) return null; + + const severity: AbuseBlockerSeverity = + scoreAfter >= config.blockScore + ? 'blocked' + : scoreAfter >= config.alertScore + ? 'alert' + : 'suspicious'; + const evidenceLimit = + scoreAfter >= config.alertScore ? config.enhancedEvidenceLimit : config.evidenceLimit; + + return { + detections, + scoreBefore, + scoreDelta, + scoreAfter, + severity, + evidence: this.collectEvidence(user, observation.destinationPort, evidenceLimit), + shouldBlock: scoreAfter >= config.blockScore, + }; + } + + addReport(report: AbuseBlockerReportModel): void { + const limit = this.config?.reportBufferSize ?? 1; + if (!this.reports.has(report.eventId) && this.reports.size >= limit) { + const oldest = this.reports.keys().next().value as string | undefined; + if (oldest) this.reports.delete(oldest); + this.droppedReports += 1; + } + this.reports.set(report.eventId, report); + } + + flushReports(): AbuseBlockerReportModel[] { + const reports = [...this.reports.values()]; + this.reports.clear(); + return reports; + } + + setCoverage(mode: AbuseBlockerCoverageMode, skippedWebhookRules: number): void { + this.coverageMode = mode; + this.skippedWebhookRules = skippedWebhookRules; + } + + setLastError(error: unknown): void { + this.lastError = error instanceof Error ? error.message : String(error); + } + + get stats() { + return { + enabled: this.enabled, + reportsCount: this.reports.size, + trackedUsers: this.users.size, + activeIncidents: [...this.users.values()].reduce( + (total, user) => + total + + [...user.horizontal.values(), ...user.sweep.values()].filter( + (state) => state.fired, + ).length, + 0, + ), + coverageMode: this.coverageMode, + skippedWebhookRules: this.skippedWebhookRules, + evictedUsers: this.evictedUsers, + evictedKeys: this.evictedKeys, + droppedReports: this.droppedReports, + lastError: this.lastError, + }; + } + + reset(): void { + this.enabled = false; + this.config = null; + this.configFingerprint = ''; + this.ignoredUsers.clear(); + this.ignoredSources = new IpMatcher([]); + this.ignoredDestinations = new IpMatcher([]); + this.users.clear(); + this.reports.clear(); + this.coverageMode = 'partial'; + this.skippedWebhookRules = 0; + this.evictedUsers = 0; + this.evictedKeys = 0; + this.droppedReports = 0; + this.lastError = null; + } + + private getUserState(userId: string, timestamp: number): IUserState { + const existing = this.users.get(userId); + if (existing) { + existing.lastSeenAt = timestamp; + this.users.delete(userId); + this.users.set(userId, existing); + return existing; + } + + const config = this.config!; + if (this.users.size >= config.maxTrackedUsers) { + const oldest = this.users.keys().next().value as string | undefined; + if (oldest) this.users.delete(oldest); + this.evictedUsers += 1; + } + + const state: IUserState = { + horizontal: new Map(), + sweep: new Map(), + scoreEvents: [], + lastSeenAt: timestamp, + }; + this.users.set(userId, state); + return state; + } + + private getDetectorState( + map: Map, + key: string, + user: IUserState, + timestamp: number, + ): IDetectorKeyState { + const existing = map.get(key); + if (existing) { + existing.lastSeenAt = timestamp; + map.delete(key); + map.set(key, existing); + return existing; + } + + const config = this.config!; + if (user.horizontal.size + user.sweep.size >= config.maxKeysPerUser) { + const candidates = [ + ...[...user.horizontal].map(([candidateKey, state]) => ({ + map: user.horizontal, + key: candidateKey, + lastSeenAt: state.lastSeenAt, + })), + ...[...user.sweep].map(([candidateKey, state]) => ({ + map: user.sweep, + key: candidateKey, + lastSeenAt: state.lastSeenAt, + })), + ]; + candidates.sort((a, b) => a.lastSeenAt - b.lastSeenAt); + const oldest = candidates[0]; + oldest?.map.delete(oldest.key); + this.evictedKeys += 1; + } + + const state: IDetectorKeyState = { + destinations: new Map(), + fired: false, + lastFiredAt: 0, + lastSeenAt: timestamp, + }; + map.set(key, state); + return state; + } + + private recordDestination( + state: IDetectorKeyState, + observation: IAbuseBlockerObservation, + windowSeconds: number, + threshold: number, + cooldownSeconds: number, + ): boolean { + const cutoff = observation.timestamp - windowSeconds * 1000; + for (const [destination, lastSeenAt] of state.destinations) { + if (lastSeenAt < cutoff) state.destinations.delete(destination); + } + + if ( + state.fired && + state.destinations.size < threshold && + observation.timestamp - state.lastFiredAt >= cooldownSeconds * 1000 + ) { + state.fired = false; + } + + state.destinations.set(observation.destinationIp, observation.timestamp); + state.lastSeenAt = observation.timestamp; + if (state.fired || state.destinations.size < threshold) return false; + if (observation.timestamp - state.lastFiredAt < cooldownSeconds * 1000) return false; + + state.fired = true; + state.lastFiredAt = observation.timestamp; + return true; + } + + private collectEvidence( + user: IUserState, + destinationPort: number, + limit: number, + ): Array<{ destinationIp: string; destinationPort: number; lastSeenAt: Date }> { + const destinations = new Map(); + const matchingStates = [ + ...[...user.horizontal].flatMap(([key, state]) => + key.startsWith(`${destinationPort}|`) ? [state] : [], + ), + ...(user.sweep.get(String(destinationPort)) + ? [user.sweep.get(String(destinationPort))!] + : []), + ]; + for (const state of matchingStates) { + for (const [destinationIp, timestamp] of state.destinations) { + const current = destinations.get(destinationIp) ?? 0; + if (timestamp > current) destinations.set(destinationIp, timestamp); + } + } + + return [...destinations] + .sort((a, b) => b[1] - a[1]) + .slice(0, limit) + .map(([destinationIp, timestamp]) => ({ + destinationIp, + destinationPort, + lastSeenAt: new Date(timestamp), + })); + } + + private pruneScore(user: IUserState, timestamp: number, windowSeconds: number): void { + const cutoff = timestamp - windowSeconds * 1000; + user.scoreEvents = user.scoreEvents.filter((event) => event.timestamp >= cutoff); + } + + private updateBufferedEvidence( + userId: string, + user: IUserState, + destinationPort: number, + ): void { + const limit = this.config?.enhancedEvidenceLimit; + if (!limit) return; + + for (const report of this.reports.values()) { + if (report.userId !== userId || report.destinationPort !== destinationPort) continue; + if (report.score.after < (this.config?.alertScore ?? Number.POSITIVE_INFINITY)) continue; + + report.evidence = this.collectEvidence(user, destinationPort, limit); + } + } +} diff --git a/src/modules/_plugin/services/states/index.ts b/src/modules/_plugin/services/states/index.ts index d121cc4..a70738f 100644 --- a/src/modules/_plugin/services/states/index.ts +++ b/src/modules/_plugin/services/states/index.ts @@ -1,3 +1,4 @@ +export { AbuseBlockerState } from './abuse-blocker.state'; export { ConnectionDropState } from './connection-drop.state'; export { TorrentBlockerState } from './torrent-blocker.state'; export { PreStartState } from './pre-start.state'; diff --git a/src/modules/_plugin/services/states/torrent-blocker.state.ts b/src/modules/_plugin/services/states/torrent-blocker.state.ts index b96e072..68ac3b4 100644 --- a/src/modules/_plugin/services/states/torrent-blocker.state.ts +++ b/src/modules/_plugin/services/states/torrent-blocker.state.ts @@ -9,6 +9,7 @@ export class TorrentBlockerState { private includeRuleTags = new Set(); private webhookUrl: string | null = null; private reports: TorrentBlockerReportModel[] = []; + private lastProcessedUsers = new Map(); get isEnabled(): boolean { return this.enabled; @@ -47,6 +48,13 @@ export class TorrentBlockerState { return this.ignoredUsers.has(userId); } + shouldProcess(userId: string, timestamp: number, deduplicationSeconds = 5): boolean { + const lastProcessedAt = this.lastProcessedUsers.get(userId) ?? 0; + if (timestamp - lastProcessedAt < deduplicationSeconds * 1000) return false; + this.lastProcessedUsers.set(userId, timestamp); + return true; + } + addReport(report: TorrentBlockerReportModel): void { this.reports.push(report); } @@ -67,6 +75,7 @@ export class TorrentBlockerState { this.ignoredIps.clear(); this.ignoredUsers.clear(); this.includeRuleTags.clear(); + this.lastProcessedUsers.clear(); } setIncludeRuleTags(tags: string[] | undefined): void { diff --git a/src/modules/_plugin/utils/ip-address.utils.ts b/src/modules/_plugin/utils/ip-address.utils.ts new file mode 100644 index 0000000..a5a4f74 --- /dev/null +++ b/src/modules/_plugin/utils/ip-address.utils.ts @@ -0,0 +1,151 @@ +import { isIP } from 'node:net'; + +export interface IParsedIpAddress { + family: 4 | 6; + bits: 32 | 128; + value: bigint; +} + +const parseIpv4 = (ip: string): bigint | null => { + const octets = ip.split('.').map(Number); + if (octets.length !== 4 || octets.some((octet) => !Number.isInteger(octet))) return null; + + return octets.reduce((value, octet) => (value << 8n) | BigInt(octet), 0n); +}; + +const expandIpv6Part = (part: string): string[] => { + if (!part) return []; + + const groups = part.split(':'); + const last = groups.at(-1); + if (!last || isIP(last) !== 4) return groups; + + const ipv4 = parseIpv4(last); + if (ipv4 === null) return groups; + + return [ + ...groups.slice(0, -1), + Number((ipv4 >> 16n) & 0xffffn).toString(16), + Number(ipv4 & 0xffffn).toString(16), + ]; +}; + +const parseIpv6 = (input: string): bigint | null => { + const ip = input.split('%')[0]; + const compressed = ip.split('::'); + if (compressed.length > 2) return null; + + const left = expandIpv6Part(compressed[0]); + const right = expandIpv6Part(compressed[1] ?? ''); + const missing = 8 - left.length - right.length; + if ((compressed.length === 1 && missing !== 0) || missing < 0) return null; + + const groups = [...left, ...Array.from({ length: missing }, () => '0'), ...right]; + if (groups.length !== 8) return null; + + let value = 0n; + for (const group of groups) { + if (!/^[0-9a-f]{1,4}$/i.test(group)) return null; + value = (value << 16n) | BigInt(`0x${group}`); + } + + return value; +}; + +export const parseIpAddress = (input: string): IParsedIpAddress | null => { + const normalized = input.replace(/^\[|\]$/g, '').split('%')[0]; + const family = isIP(normalized); + + if (family === 4) { + const value = parseIpv4(normalized); + return value === null ? null : { family: 4, bits: 32, value }; + } + + if (family === 6) { + const value = parseIpv6(normalized); + return value === null ? null : { family: 6, bits: 128, value }; + } + + return null; +}; + +export const getNetworkKey = ( + ip: string, + ipv4Prefix: number, + ipv6Prefix: number, +): string | null => { + const parsed = parseIpAddress(ip); + if (!parsed) return null; + + const prefix = parsed.family === 4 ? ipv4Prefix : ipv6Prefix; + const hostBits = BigInt(parsed.bits - prefix); + const network = hostBits === 0n ? parsed.value : (parsed.value >> hostBits) << hostBits; + + return `${parsed.family}:${network.toString(16)}/${prefix}`; +}; + +interface ICidrRange extends IParsedIpAddress { + prefix: number; + network: bigint; +} + +export class IpMatcher { + private readonly ranges: ICidrRange[]; + + constructor(values: string[]) { + this.ranges = values.flatMap((value) => { + const [ip, rawPrefix] = value.split('/'); + const parsed = parseIpAddress(ip); + if (!parsed) return []; + + const prefix = rawPrefix === undefined ? parsed.bits : Number(rawPrefix); + if (!Number.isInteger(prefix) || prefix < 0 || prefix > parsed.bits) return []; + + const hostBits = BigInt(parsed.bits - prefix); + const network = hostBits === 0n ? parsed.value : (parsed.value >> hostBits) << hostBits; + return [{ ...parsed, prefix, network }]; + }); + } + + matches(ip: string): boolean { + const parsed = parseIpAddress(ip); + if (!parsed) return false; + + return this.ranges.some((range) => { + if (range.family !== parsed.family) return false; + const hostBits = BigInt(range.bits - range.prefix); + const network = hostBits === 0n ? parsed.value : (parsed.value >> hostBits) << hostBits; + return network === range.network; + }); + } +} + +export interface IParsedNetworkEndpoint { + ip: string; + port: number | null; +} + +export const parseNetworkEndpoint = (input: string | null): IParsedNetworkEndpoint | null => { + if (!input) return null; + const value = input.replace(/^(?:tcp|udp):/i, ''); + + if (value.startsWith('[')) { + const closingBracket = value.indexOf(']'); + if (closingBracket < 0) return null; + const ip = value.slice(1, closingBracket); + if (!parseIpAddress(ip)) return null; + const rawPort = value.slice(closingBracket + 1).replace(/^:/, ''); + const port = rawPort ? Number(rawPort) : null; + return { ip, port: Number.isInteger(port) ? port : null }; + } + + if (parseIpAddress(value)) return { ip: value, port: null }; + + const separator = value.lastIndexOf(':'); + if (separator < 0) return null; + const ip = value.slice(0, separator); + const port = Number(value.slice(separator + 1)); + if (!parseIpAddress(ip) || !Number.isInteger(port)) return null; + + return { ip, port }; +}; diff --git a/src/modules/internal/internal.controller.ts b/src/modules/internal/internal.controller.ts index f6cc393..ba6f78d 100644 --- a/src/modules/internal/internal.controller.ts +++ b/src/modules/internal/internal.controller.ts @@ -3,8 +3,11 @@ import { EventBus } from '@nestjs/cqrs'; import { HttpExceptionFilter } from '@common/exception'; import { + XRAY_INTERNAL_ABUSE_WEBHOOK_PATH, XRAY_INTERNAL_API_CONTROLLER, XRAY_INTERNAL_API_PATH, + XRAY_INTERNAL_COMBINED_WEBHOOK_PATH, + XRAY_INTERNAL_TORRENT_WEBHOOK_PATH, XRAY_INTERNAL_WEBHOOK_PATH, } from '@libs/contracts/constants'; @@ -33,6 +36,24 @@ export class InternalController { @HttpCode(200) @Post(XRAY_INTERNAL_WEBHOOK_PATH) public handleWebhook(@Body() body: unknown): void { - void this.eventBus.publish(new XrayWebhookEvent(body)); + void this.eventBus.publish(new XrayWebhookEvent(body, 'torrent')); + } + + @HttpCode(200) + @Post(XRAY_INTERNAL_TORRENT_WEBHOOK_PATH) + public handleTorrentWebhook(@Body() body: unknown): void { + void this.eventBus.publish(new XrayWebhookEvent(body, 'torrent')); + } + + @HttpCode(200) + @Post(XRAY_INTERNAL_ABUSE_WEBHOOK_PATH) + public handleAbuseWebhook(@Body() body: unknown): void { + void this.eventBus.publish(new XrayWebhookEvent(body, 'abuse')); + } + + @HttpCode(200) + @Post(XRAY_INTERNAL_COMBINED_WEBHOOK_PATH) + public handleCombinedWebhook(@Body() body: unknown): void { + void this.eventBus.publish(new XrayWebhookEvent(body, 'combined')); } } diff --git a/src/modules/stats/models/get-system-stats.response.model.ts b/src/modules/stats/models/get-system-stats.response.model.ts index 1cbc088..277d9a4 100644 --- a/src/modules/stats/models/get-system-stats.response.model.ts +++ b/src/modules/stats/models/get-system-stats.response.model.ts @@ -14,6 +14,19 @@ interface IXrayStats { } interface IPluginStats { + abuseBlocker: { + available: boolean; + enabled: boolean; + reportsCount: number; + trackedUsers: number; + activeIncidents: number; + coverageMode: 'full' | 'partial'; + skippedWebhookRules: number; + evictedUsers: number; + evictedKeys: number; + droppedReports: number; + lastError: string | null; + }; torrentBlocker: { reportsCount: number; }; diff --git a/src/modules/stats/stats.service.ts b/src/modules/stats/stats.service.ts index 9c16ca7..29a13bf 100644 --- a/src/modules/stats/stats.service.ts +++ b/src/modules/stats/stats.service.ts @@ -1,13 +1,17 @@ import { Injectable, Logger } from '@nestjs/common'; import { QueryBus } from '@nestjs/cqrs'; -import { InjectXtls } from '@remnawave/xtls-sdk-nestjs'; import { XtlsApi } from '@remnawave/xtls-sdk'; +import { InjectXtls } from '@remnawave/xtls-sdk-nestjs'; import { ICommandResponse } from '@common/types/command-response.type'; import { getSystemStats } from '@common/utils/get-system-stats'; import { ERRORS } from '@libs/contracts/constants'; +import { GetAbuseBlockerStatsQuery } from '../_plugin/queries/get-abuse-blocker-stats'; +import { GetTorrentBlockerReportsCountQuery } from '../_plugin/queries/get-torrent-blocker-reports-count'; +import { GetInterfaceStatsQuery } from '../network-stats/queries/get-interface-stats/get-interface-stats.query'; +import { IGetUserOnlineStatusRequest } from './interfaces'; import { GetAllInboundsStatsResponseModel, GetAllOutboundsStatsResponseModel, @@ -20,9 +24,6 @@ import { GetUsersIpListResponseModel, GetUsersStatsResponseModel, } from './models'; -import { GetInterfaceStatsQuery } from '../network-stats/queries/get-interface-stats/get-interface-stats.query'; -import { GetTorrentBlockerReportsCountQuery } from '../_plugin/queries/get-torrent-blocker-reports-count'; -import { IGetUserOnlineStatusRequest } from './interfaces'; @Injectable() export class StatsService { @@ -75,12 +76,14 @@ export class StatsService { const reportsCount = await this.queryBus.execute( new GetTorrentBlockerReportsCountQuery(), ); + const abuseBlocker = await this.queryBus.execute(new GetAbuseBlockerStatsQuery()); return { isOk: true, response: new GetSystemStatsResponseModel( response.data, { + abuseBlocker, torrentBlocker: { reportsCount, }, diff --git a/src/modules/xray-core/xray.service.ts b/src/modules/xray-core/xray.service.ts index 3745b64..55c3b0b 100644 --- a/src/modules/xray-core/xray.service.ts +++ b/src/modules/xray-core/xray.service.ts @@ -20,6 +20,8 @@ import { IntegrationsService } from '@integration-modules/integrations.service'; import { ResetPluginsCommand } from '../_plugin/commands/reset-plugins/reset-plugins.command'; import { RunPreStartCommand } from '../_plugin/commands/run-pre-start/run-pre-start.command'; +import { SetAbuseBlockerCoverageCommand } from '../_plugin/commands/set-abuse-blocker-coverage'; +import { GetAbuseBlockerStateQuery } from '../_plugin/queries/get-abuse-blocker-state'; import { GetTorrentBlockerStateQuery } from '../_plugin/queries/get-torrent-blocker-state'; import { InternalService } from '../internal/internal.service'; import { GetInterfaceStatsQuery } from '../network-stats/queries/get-interface-stats/get-interface-stats.query'; @@ -183,11 +185,22 @@ export class XrayService implements OnApplicationBootstrap { new GetTorrentBlockerStateQuery(), ); - const fullConfig = generateApiConfig({ + const abuseBlockerState = await this.queryBus.execute(new GetAbuseBlockerStateQuery()); + + const generated = generateApiConfig({ config: body.xrayConfig, torrentBlockerState: isTorrentBlockerEnabled, + abuseBlockerState, internal: this.internal, }); + const fullConfig = generated.config; + + await this.commandBus.execute( + new SetAbuseBlockerCoverageCommand( + generated.abuseCoverage.mode, + generated.abuseCoverage.skippedWebhookRules, + ), + ); await this.internalService.extractUsersFromConfig(body.internals.hashes, fullConfig); From 2e47450d0d1e1d2ce376317fa24e4331b2a5e098 Mon Sep 17 00:00:00 2001 From: l0nelynx <113856580+l0nelynx@users.noreply.github.com> Date: Sat, 15 Aug 2026 10:51:49 +0300 Subject: [PATCH 3/3] test(plugin): cover abuse blocker detection and routing --- .../services/states/abuse-blocker.state.ts | 9 +- .../abuse-blocker/abuse-blocker.state.test.ts | 288 ++++++++++++++++++ .../abuse-blocker/generate-api-config.test.ts | 105 +++++++ tests/abuse-blocker/ip-address.utils.test.ts | 38 +++ tests/abuse-blocker/nft.service.test.ts | 75 +++++ tests/abuse-blocker/xray-webhook.test.ts | 62 ++++ 6 files changed, 571 insertions(+), 6 deletions(-) create mode 100644 tests/abuse-blocker/abuse-blocker.state.test.ts create mode 100644 tests/abuse-blocker/generate-api-config.test.ts create mode 100644 tests/abuse-blocker/ip-address.utils.test.ts create mode 100644 tests/abuse-blocker/nft.service.test.ts create mode 100644 tests/abuse-blocker/xray-webhook.test.ts diff --git a/src/modules/_plugin/services/states/abuse-blocker.state.ts b/src/modules/_plugin/services/states/abuse-blocker.state.ts index 82e0f0d..e6827d1 100644 --- a/src/modules/_plugin/services/states/abuse-blocker.state.ts +++ b/src/modules/_plugin/services/states/abuse-blocker.state.ts @@ -193,11 +193,7 @@ export class AbuseBlockerState { } if (detections.length === 0) { - this.updateBufferedEvidence( - observation.userId, - user, - observation.destinationPort, - ); + this.updateBufferedEvidence(observation.userId, user, observation.destinationPort); return null; } @@ -436,7 +432,8 @@ export class AbuseBlockerState { for (const report of this.reports.values()) { if (report.userId !== userId || report.destinationPort !== destinationPort) continue; - if (report.score.after < (this.config?.alertScore ?? Number.POSITIVE_INFINITY)) continue; + if (report.score.after < (this.config?.alertScore ?? Number.POSITIVE_INFINITY)) + continue; report.evidence = this.collectEvidence(user, destinationPort, limit); } diff --git a/tests/abuse-blocker/abuse-blocker.state.test.ts b/tests/abuse-blocker/abuse-blocker.state.test.ts new file mode 100644 index 0000000..23d18a5 --- /dev/null +++ b/tests/abuse-blocker/abuse-blocker.state.test.ts @@ -0,0 +1,288 @@ +import type { XrayWebhookModel } from '../../libs/contract/models'; + +import assert from 'node:assert/strict'; +import { describe, it } from 'node:test'; + +import { NodePluginSchema } from '@remnawave/node-plugins'; + +import { AbuseBlockerState } from '../../src/modules/_plugin/services/states/abuse-blocker.state'; + +const webhook: XrayWebhookModel = { + email: '42', + level: 0, + protocol: null, + network: 'tcp', + source: '198.51.100.10:12345', + destination: '192.0.2.1:22', + routeTarget: null, + originalTarget: null, + inboundTag: 'VLESS', + inboundName: null, + inboundLocal: null, + outboundTag: 'DIRECT', + ts: 0, +}; + +const createState = (overrides: Record = {}) => { + const config = NodePluginSchema.parse({ + abuseBlocker: { enabled: true, ...overrides }, + }).abuseBlocker!; + const state = new AbuseBlockerState(); + state.configure({ + config, + configFingerprint: 'test-config', + ignoredUsers: [], + ignoredSources: [], + ignoredDestinations: [], + }); + return state; +}; + +const observe = ( + state: AbuseBlockerState, + destinationIp: string, + index: number, + destinationPort = 22, +) => + state.analyze({ + userId: '42', + sourceIp: '198.51.100.10', + destinationIp, + destinationPort, + timestamp: 1_000_000 + index * 100, + xrayReport: webhook, + }); + +describe('AbuseBlockerState', () => { + it('fires a horizontal scan once at 20 unique destinations', () => { + const state = createState(); + for (let index = 1; index < 20; index += 1) { + assert.equal(observe(state, `192.0.2.${index}`, index), null); + } + + const result = observe(state, '192.0.2.20', 20); + assert.equal(result?.scoreAfter, 100); + assert.equal(result?.severity, 'alert'); + assert.equal(result?.detections[0].rule, 'horizontal_scan'); + assert.equal(observe(state, '192.0.2.21', 21), null); + }); + + it('counts unique destinations and ignores duplicates', () => { + const state = createState(); + for (let index = 0; index < 100; index += 1) { + assert.equal(observe(state, '192.0.2.1', index), null); + } + assert.equal(state.stats.activeIncidents, 0); + }); + + it('updates buffered alert evidence without adding score or another action', () => { + const state = createState(); + let alert = null; + for (let index = 1; index <= 20; index += 1) { + alert = observe(state, `192.0.2.${index}`, index) ?? alert; + } + assert.ok(alert); + + state.addReport({ + eventId: '00000000-0000-4000-8000-000000000001', + userId: '42', + destinationPort: 22, + score: { after: alert.scoreAfter }, + evidence: alert.evidence, + } as Parameters[0]); + + assert.equal(observe(state, '192.0.2.21', 21), null); + const [updated] = state.flushReports(); + assert.equal(updated.eventId, '00000000-0000-4000-8000-000000000001'); + assert.equal(updated.score.after, 100); + assert.equal(updated.evidence.length, 21); + assert.equal(updated.evidence[0].destinationIp, '192.0.2.21'); + }); + + it('combines detector scores and requests a block at 150', () => { + const state = createState(); + let lastResult = null; + for (let index = 0; index < 50; index += 1) { + const destination = index < 20 ? `192.0.2.${index + 1}` : `10.${index}.0.1`; + lastResult = observe(state, destination, index) ?? lastResult; + } + + assert.equal(lastResult?.scoreAfter, 150); + assert.equal(lastResult?.severity, 'blocked'); + assert.equal(lastResult?.shouldBlock, true); + assert.equal(lastResult?.detections[0].rule, 'destination_sweep'); + assert.equal(lastResult?.evidence.length, 50); + }); + + it('supports IPv6 /64 horizontal scans', () => { + const state = createState(); + let result = null; + for (let index = 1; index <= 20; index += 1) { + result = observe(state, `2001:db8:abcd:12::${index.toString(16)}`, index) ?? result; + } + + assert.equal(result?.detections[0].rule, 'horizontal_scan'); + assert.match(result?.detections[0].subnet ?? '', /^6:.*\/64$/); + }); + + it('includes the exact rolling-window boundary and expires older destinations', () => { + const state = createState({ + horizontalScan: { uniqueDestinations: 2, windowSeconds: 60 }, + destinationSweep: { enabled: false }, + }); + const analyzeAt = (destinationIp: string, timestamp: number) => + state.analyze({ + userId: '42', + sourceIp: '198.51.100.10', + destinationIp, + destinationPort: 22, + timestamp, + xrayReport: webhook, + }); + + assert.equal(analyzeAt('192.0.2.1', 1_000_000), null); + assert.equal(analyzeAt('192.0.2.2', 1_060_000)?.severity, 'alert'); + + const expired = createState({ + horizontalScan: { uniqueDestinations: 2, windowSeconds: 60 }, + destinationSweep: { enabled: false }, + }); + assert.equal( + expired.analyze({ + userId: '42', + sourceIp: '198.51.100.10', + destinationIp: '192.0.2.1', + destinationPort: 22, + timestamp: 1_000_000, + xrayReport: webhook, + }), + null, + ); + assert.equal( + expired.analyze({ + userId: '42', + sourceIp: '198.51.100.10', + destinationIp: '192.0.2.2', + destinationPort: 22, + timestamp: 1_060_001, + xrayReport: webhook, + }), + null, + ); + }); + + it('does not combine destinations across users or ports', () => { + const state = createState({ + horizontalScan: { uniqueDestinations: 2 }, + destinationSweep: { enabled: false }, + }); + const analyze = (userId: string, port: number, destinationIp: string, timestamp: number) => + state.analyze({ + userId, + sourceIp: '198.51.100.10', + destinationIp, + destinationPort: port, + timestamp, + xrayReport: { ...webhook, email: userId }, + }); + + assert.equal(analyze('42', 22, '192.0.2.1', 1_000_000), null); + assert.equal(analyze('43', 22, '192.0.2.2', 1_000_100), null); + assert.equal(analyze('42', 23, '192.0.2.2', 1_000_200), null); + }); + + it('re-arms a rule only after its window falls below threshold and cooldown elapses', () => { + const state = createState({ + horizontalScan: { uniqueDestinations: 2, windowSeconds: 60 }, + destinationSweep: { enabled: false }, + incidentCooldownSeconds: 300, + }); + const analyzeAt = (destinationIp: string, timestamp: number) => + state.analyze({ + userId: '42', + sourceIp: '198.51.100.10', + destinationIp, + destinationPort: 22, + timestamp, + xrayReport: webhook, + }); + + assert.equal(analyzeAt('192.0.2.1', 1_000_000), null); + assert.equal(analyzeAt('192.0.2.2', 1_000_100)?.scoreAfter, 100); + assert.equal(analyzeAt('192.0.2.3', 1_061_000), null); + assert.equal(analyzeAt('192.0.2.4', 1_301_000), null); + assert.equal(analyzeAt('192.0.2.5', 1_301_100)?.scoreAfter, 200); + }); + + it('expires score events outside the score window', () => { + const state = createState({ + horizontalScan: { enabled: false }, + destinationSweep: { uniqueDestinations: 2, score: 50 }, + scoreWindowSeconds: 3600, + }); + const analyzeAt = (port: number, destinationIp: string, timestamp: number) => + state.analyze({ + userId: '42', + sourceIp: '198.51.100.10', + destinationIp, + destinationPort: port, + timestamp, + xrayReport: webhook, + }); + + assert.equal(analyzeAt(22, '192.0.2.1', 1_000_000), null); + assert.equal(analyzeAt(22, '192.0.2.2', 1_000_100)?.scoreAfter, 50); + assert.equal(analyzeAt(23, '198.51.100.1', 4_600_101), null); + assert.equal(analyzeAt(23, '198.51.100.2', 4_600_200)?.scoreBefore, 0); + }); + + it('respects excluded ports and source ignore ranges', () => { + const config = NodePluginSchema.parse({ abuseBlocker: { enabled: true } }).abuseBlocker!; + const state = new AbuseBlockerState(); + state.configure({ + config, + configFingerprint: 'test-config', + ignoredUsers: [], + ignoredSources: ['198.51.100.0/24'], + ignoredDestinations: [], + }); + + assert.equal(observe(state, '192.0.2.1', 1), null); + assert.equal(observe(createState(), '192.0.2.1', 1, 443), null); + }); + + it('evicts the least recently used user at the configured limit', () => { + const state = createState({ maxTrackedUsers: 1 }); + observe(state, '192.0.2.1', 1); + state.analyze({ + userId: '43', + sourceIp: '198.51.100.11', + destinationIp: '192.0.2.2', + destinationPort: 22, + timestamp: 1_001_000, + xrayReport: { ...webhook, email: '43' }, + }); + + assert.equal(state.stats.trackedUsers, 1); + assert.equal(state.stats.evictedUsers, 1); + }); + + it('evicts detector keys and drops the oldest buffered report at configured limits', () => { + const state = createState({ maxKeysPerUser: 1, reportBufferSize: 1 }); + observe(state, '192.0.2.1', 1, 22); + observe(state, '192.0.2.2', 2, 23); + + const report = { + eventId: '00000000-0000-4000-8000-000000000001', + } as Parameters[0]; + state.addReport(report); + state.addReport({ ...report, eventId: '00000000-0000-4000-8000-000000000002' }); + + assert.ok(state.stats.evictedKeys > 0); + assert.equal(state.stats.droppedReports, 1); + assert.deepEqual( + state.flushReports().map((item) => item.eventId), + ['00000000-0000-4000-8000-000000000002'], + ); + }); +}); diff --git a/tests/abuse-blocker/generate-api-config.test.ts b/tests/abuse-blocker/generate-api-config.test.ts new file mode 100644 index 0000000..a4a7cba --- /dev/null +++ b/tests/abuse-blocker/generate-api-config.test.ts @@ -0,0 +1,105 @@ +import assert from 'node:assert/strict'; +import { describe, it } from 'node:test'; + +import { generateApiConfig } from '../../src/common/utils/generate-api-config'; + +const internal = { + socketPath: 'rw-internal.sock', + token: 'test-token', + xtlsApiSocketPath: 'rw-xray.sock', +}; + +const generate = (config: Record, torrentTags = new Set()) => + generateApiConfig({ + config, + abuseBlockerState: { enabled: true }, + torrentBlockerState: { enabled: torrentTags.size > 0, includeRuleTags: torrentTags }, + internal, + }); + +describe('generateApiConfig abuse routing', () => { + it('instruments explicit rules and an AsIs default route', () => { + const generated = generate({ + outbounds: [{ tag: 'DIRECT', protocol: 'freedom' }], + routing: { + rules: [{ ruleTag: 'PRIVATE', ip: ['geoip:private'], outboundTag: 'BLOCK' }], + }, + }); + const rules = (generated.config.routing as { rules: Record[] }).rules; + + assert.equal(generated.abuseCoverage.mode, 'full'); + assert.equal((rules[1].webhook as { deduplication: number }).deduplication, 0); + assert.equal(rules.at(-1)?.ruleTag, 'RW_ABUSE_DEFAULT'); + assert.equal(rules.at(-1)?.network, 'tcp'); + }); + + it('uses an IP catch-all for IPIfNonMatch', () => { + const generated = generate({ + outbounds: [{ tag: 'DIRECT', protocol: 'freedom' }], + routing: { domainStrategy: 'IPIfNonMatch', rules: [] }, + }); + const rules = (generated.config.routing as { rules: Record[] }).rules; + assert.deepEqual(rules.at(-1)?.ip, ['0.0.0.0/0', '::/0']); + }); + + it('preserves external webhooks and reports partial coverage', () => { + const external = { url: 'https://example.com/hook', deduplication: 10 }; + const generated = generate({ + outbounds: [{ tag: 'DIRECT', protocol: 'freedom' }], + routing: { + rules: [{ ruleTag: 'EXTERNAL', outboundTag: 'DIRECT', webhook: external }], + }, + }); + const rules = (generated.config.routing as { rules: Record[] }).rules; + + assert.deepEqual(rules[1].webhook, external); + assert.equal(generated.abuseCoverage.mode, 'partial'); + assert.equal(generated.abuseCoverage.skippedWebhookRules, 1); + }); + + it('does not replace an external webhook selected by torrentBlocker', () => { + const external = { url: 'https://example.com/hook', deduplication: 10 }; + const generated = generate( + { + outbounds: [{ tag: 'DIRECT', protocol: 'freedom' }], + routing: { + rules: [ + { + ruleTag: 'EXTERNAL', + outboundTag: 'DIRECT', + webhook: external, + }, + ], + }, + }, + new Set(['EXTERNAL']), + ); + const rules = (generated.config.routing as { rules: Record[] }).rules; + const externalRule = rules.find((rule) => rule.ruleTag === 'EXTERNAL'); + + assert.ok(externalRule); + assert.deepEqual(externalRule.webhook, external); + }); + + it('uses the combined endpoint when torrentBlocker observes the same rule', () => { + const generated = generate( + { + outbounds: [{ tag: 'DIRECT', protocol: 'freedom' }], + routing: { + rules: [{ ruleTag: 'WATCHED', outboundTag: 'DIRECT' }], + }, + }, + new Set(['WATCHED']), + ); + const rules = (generated.config.routing as { rules: Record[] }).rules; + const watched = rules.find((rule) => rule.ruleTag === 'WATCHED'); + + assert.ok(watched); + assert.match((watched.webhook as { url: string }).url, /\/internal\/webhook\/combined/); + assert.equal((watched.webhook as { deduplication: number }).deduplication, 0); + + const torrentRule = rules.find((rule) => rule.outboundTag === 'RW_TB_OUTBOUND_BLOCK'); + assert.ok(torrentRule); + assert.equal((torrentRule.webhook as { deduplication: number }).deduplication, 0); + }); +}); diff --git a/tests/abuse-blocker/ip-address.utils.test.ts b/tests/abuse-blocker/ip-address.utils.test.ts new file mode 100644 index 0000000..8fa7c79 --- /dev/null +++ b/tests/abuse-blocker/ip-address.utils.test.ts @@ -0,0 +1,38 @@ +import assert from 'node:assert/strict'; +import { describe, it } from 'node:test'; + +import { + getNetworkKey, + IpMatcher, + parseNetworkEndpoint, +} from '../../src/modules/_plugin/utils/ip-address.utils'; + +describe('IP utilities', () => { + it('groups IPv4 and IPv6 destinations by configured prefixes', () => { + assert.equal(getNetworkKey('192.0.2.42', 24, 64), '4:c0000200/24'); + assert.equal( + getNetworkKey('2001:db8:abcd:12::1', 24, 64), + '6:20010db8abcd00120000000000000000/64', + ); + }); + + it('matches exact addresses and CIDR ranges', () => { + const matcher = new IpMatcher(['192.0.2.0/24', '2001:db8::/32', '203.0.113.1']); + assert.equal(matcher.matches('192.0.2.99'), true); + assert.equal(matcher.matches('2001:db8:1::1'), true); + assert.equal(matcher.matches('203.0.113.1'), true); + assert.equal(matcher.matches('198.51.100.1'), false); + }); + + it('parses Xray IPv4 and bracketed IPv6 endpoints', () => { + assert.deepEqual(parseNetworkEndpoint('tcp:192.0.2.1:22'), { + ip: '192.0.2.1', + port: 22, + }); + assert.deepEqual(parseNetworkEndpoint('[2001:db8::1]:3389'), { + ip: '2001:db8::1', + port: 3389, + }); + assert.equal(parseNetworkEndpoint('example.com:22'), null); + }); +}); diff --git a/tests/abuse-blocker/nft.service.test.ts b/tests/abuse-blocker/nft.service.test.ts new file mode 100644 index 0000000..0ea7289 --- /dev/null +++ b/tests/abuse-blocker/nft.service.test.ts @@ -0,0 +1,75 @@ +import assert from 'node:assert/strict'; +import { describe, it } from 'node:test'; + +import { NFT_TABLES_CONSTANTS } from '../../src/modules/_plugin/constants/nfttables.contants'; +import { NftService } from '../../src/modules/_plugin/services/nft.service'; + +interface INftCall { + operation: 'add' | 'remove'; + ip: string; + set: string; + timeout?: number; +} + +const createService = () => { + const calls: INftCall[] = []; + const dropped: string[][] = []; + const manager = { + addAddress: async ({ ip, set, timeout }: { ip: string; set: string; timeout: number }) => { + calls.push({ operation: 'add', ip, set, timeout }); + }, + removeAddresses: async ({ ips, set }: { ips: string[]; set: string }) => { + calls.push({ operation: 'remove', ip: ips[0], set }); + }, + }; + const service = new NftService( + { plugins: {}, setPlugins: () => void 0 } as never, + { publish: (event: { ips: string[] }) => dropped.push(event.ips) } as never, + ); + Object.assign(service, { nftManager: manager }); + return { calls, dropped, service }; +}; + +describe('NftService abuse blocker', () => { + it('uses the dedicated timeout set for IPv4 and IPv6 and drops active connections', async () => { + const { calls, dropped, service } = createService(); + + await service.blockAbuseIp('198.51.100.10', 600); + await service.blockAbuseIp('2001:db8::10', 3600); + + assert.deepEqual(calls, [ + { + operation: 'add', + ip: '198.51.100.10', + set: NFT_TABLES_CONSTANTS.ABUSE_BLOCKER_SET_NAME, + timeout: 600, + }, + { + operation: 'add', + ip: '2001:db8::10', + set: NFT_TABLES_CONSTANTS.ABUSE_BLOCKER_SET_NAME, + timeout: 3600, + }, + ]); + assert.deepEqual(dropped, [['198.51.100.10'], ['2001:db8::10']]); + }); + + it('refreshes a block with a serialized remove then add', async () => { + const { calls, service } = createService(); + + await Promise.all([ + service.refreshAbuseIp('198.51.100.10', 3600), + service.refreshAbuseIp('198.51.100.11', 3600), + ]); + + assert.deepEqual( + calls.map((call) => `${call.operation}:${call.ip}`), + [ + 'remove:198.51.100.10', + 'add:198.51.100.10', + 'remove:198.51.100.11', + 'add:198.51.100.11', + ], + ); + }); +}); diff --git a/tests/abuse-blocker/xray-webhook.test.ts b/tests/abuse-blocker/xray-webhook.test.ts new file mode 100644 index 0000000..0590382 --- /dev/null +++ b/tests/abuse-blocker/xray-webhook.test.ts @@ -0,0 +1,62 @@ +import type { XrayWebhookModel } from '../../libs/contract/models'; + +import assert from 'node:assert/strict'; +import { describe, it } from 'node:test'; + +import { toAbuseBlockerObservation } from '../../src/modules/_plugin/events/xray-webhook/xray-webhook.handler'; + +const webhook: XrayWebhookModel = { + email: '42', + level: 0, + protocol: null, + network: 'tcp', + source: 'tcp:198.51.100.10:12345', + destination: '203.0.113.10:22', + routeTarget: '192.0.2.10:3389', + originalTarget: '10.0.0.10:445', + inboundTag: 'VLESS', + inboundName: null, + inboundLocal: null, + outboundTag: 'DIRECT', + ts: 123, +}; + +describe('abuse blocker Xray observations', () => { + it('prefers originalTarget and converts Xray seconds to milliseconds', () => { + const observation = toAbuseBlockerObservation(webhook); + + assert.equal(observation?.destinationIp, '10.0.0.10'); + assert.equal(observation?.destinationPort, 445); + assert.equal(observation?.timestamp, 123_000); + }); + + it('falls back through routeTarget to destination', () => { + assert.equal( + toAbuseBlockerObservation({ ...webhook, originalTarget: 'example.com:443' }) + ?.destinationIp, + '192.0.2.10', + ); + assert.equal( + toAbuseBlockerObservation({ + ...webhook, + originalTarget: null, + routeTarget: null, + })?.destinationIp, + '203.0.113.10', + ); + }); + + it('ignores UDP, non-numeric users, and domain-only destinations', () => { + assert.equal(toAbuseBlockerObservation({ ...webhook, network: 'udp' }), null); + assert.equal(toAbuseBlockerObservation({ ...webhook, email: 'user@example.com' }), null); + assert.equal( + toAbuseBlockerObservation({ + ...webhook, + originalTarget: 'example.com:22', + routeTarget: 'example.net:22', + destination: 'example.org:22', + }), + null, + ); + }); +});