test(plugin): cover abuse blocker detection and routing

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);
         }

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<string, unknown> = {}) => {
+    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<typeof state.addReport>[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<typeof state.addReport>[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'],
+        );
+    });
+});

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<string, unknown>, torrentTags = new Set<string>()) =>
+    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<string, unknown>[] }).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<string, unknown>[] }).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<string, unknown>[] }).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<string, unknown>[] }).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<string, unknown>[] }).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);
+    });
+});

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);
+    });
+});

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',
+            ],
+        );
+    });
+});

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,
+        );
+    });
+});