implement polling reports in synapse (#259)

config/default.yaml

@@ -235,3 +235,8 @@ web:
   abuseReporting:
     # Whether to enable this feature.
     enabled: false
+
+# Whether or not to actively poll synapse for abuse reports, to be used
+# instead of intercepting client calls to synapse's abuse endpoint, when that
+# isn't possible/practical.
+pollReports: false

src/Mjolnir.ts

@@ -43,6 +43,7 @@ import { Healthz } from "./health/healthz";
 import { EventRedactionQueue, RedactUserInRoom } from "./queues/EventRedactionQueue";
 import { htmlEscape } from "./utils";
 import { ReportManager } from "./report/ReportManager";
+import { ReportPoller } from "./report/ReportPoller";
 import { WebAPIs } from "./webapis/WebAPIs";
 import { replaceRoomIdsWithPills } from "./utils";
 import RuleServer from "./models/RuleServer";
@@ -67,6 +68,11 @@ const ENABLED_PROTECTIONS_EVENT_TYPE = "org.matrix.mjolnir.enabled_protections";
 const PROTECTED_ROOMS_EVENT_TYPE = "org.matrix.mjolnir.protected_rooms";
 const WARN_UNPROTECTED_ROOM_EVENT_PREFIX = "org.matrix.mjolnir.unprotected_room_warning.for.";
 const CONSEQUENCE_EVENT_DATA = "org.matrix.mjolnir.consequence";
+/**
+ * Synapse will tell us where we last got to on polling reports, so we need
+ * to store that for pagination on further polls
+ */
+export const REPORT_POLL_EVENT_TYPE = "org.matrix.mjolnir.report_poll";
 
 export class Mjolnir {
     private displayName: string;
@@ -97,7 +103,10 @@ export class Mjolnir {
     private webapis: WebAPIs;
     private protectedRoomActivityTracker: ProtectedRoomActivityTracker;
     public taskQueue: ThrottlingQueue;
-
+    /*
+     * Config-enabled polling of reports in Synapse, so Mjolnir can react to reports
+     */
+    private reportPoller?: ReportPoller;
     /**
      * Adds a listener to the client that will automatically accept invitations.
      * @param {MatrixClient} client
@@ -256,12 +265,13 @@ export class Mjolnir {
         // Setup Web APIs
         console.log("Creating Web APIs");
         const reportManager = new ReportManager(this);
-        reportManager.on("report.new", this.handleReport);
+        reportManager.on("report.new", this.handleReport.bind(this));
         this.webapis = new WebAPIs(reportManager, this.ruleServer);
-
+        if (config.pollReports) {
+            this.reportPoller = new ReportPoller(this, reportManager);
+        }
         // Setup join/leave listener
         this.roomJoins = new RoomMemberManager(this.client);
-
         this.taskQueue = new ThrottlingQueue(this, config.backgroundDelayMS);
     }
 
@@ -302,6 +312,20 @@ export class Mjolnir {
             console.log("Starting web server");
             await this.webapis.start();
 
+            if (this.reportPoller) {
+                let reportPollSetting: { from: number } = { from: 0 };
+                try {
+                    reportPollSetting = await this.client.getAccountData(REPORT_POLL_EVENT_TYPE);
+                } catch (err) {
+                    if (err.body?.errcode !== "M_NOT_FOUND") {
+                        throw err;
+                    } else {
+                        this.logMessage(LogLevel.INFO, "Mjolnir@startup", "report poll setting does not exist yet");
+                    }
+                }
+                this.reportPoller.start(reportPollSetting.from);
+            }
+
             // Load the state.
             this.currentState = STATE_CHECKING_PERMISSIONS;
 
@@ -358,6 +382,7 @@ export class Mjolnir {
         LogService.info("Mjolnir", "Stopping Mjolnir...");
         this.client.stop();
         this.webapis.stop();
+        this.reportPoller?.stop();
     }
 
     public async logMessage(level: LogLevel, module: string, message: string | any, additionalRoomIds: string[] | string | null = null, isRecursive = false): Promise<any> {
@@ -1163,7 +1188,7 @@ export class Mjolnir {
         return await this.eventRedactionQueue.process(this, roomId);
     }
 
-    private async handleReport(roomId: string, reporterId: string, event: any, reason?: string) {
+    private async handleReport({ roomId, reporterId, event, reason }: { roomId: string, reporterId: string, event: any, reason?: string }) {
         for (const protection of this.enabledProtections) {
             await protection.handleReport(this, roomId, reporterId, event, reason);
         }

src/config.ts

@@ -53,6 +53,7 @@ interface IConfig {
      * of one background task and the start of the next one.
      */
     backgroundDelayMS: number;
+    pollReports: boolean;
     admin?: {
         enableMakeRoomAdminCommand?: boolean;
     }
@@ -122,6 +123,7 @@ const defaultConfig: IConfig = {
     automaticallyRedactForReasons: ["spam", "advertising"],
     protectAllJoinedRooms: false,
     backgroundDelayMS: 500,
+    pollReports: false,
     commands: {
         allowNoPrefix: false,
         additionalPrefixes: [],

src/report/ReportPoller.ts

@@ -0,0 +1,145 @@
+/*
+Copyright 2022 The Matrix.org Foundation C.I.C.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+import { Mjolnir, REPORT_POLL_EVENT_TYPE } from "../Mjolnir";
+import { ReportManager } from './ReportManager';
+import { LogLevel } from "matrix-bot-sdk";
+
+class InvalidStateError extends Error {}
+
+/**
+ * A class to poll synapse's report endpoint, so we can act on new reports
+ *
+ * @param mjolnir The running Mjolnir instance
+ * @param manager The report manager in to which we feed new reports
+ */
+export class ReportPoller {
+    /**
+     * https://matrix-org.github.io/synapse/latest/admin_api/event_reports.html
+     * "from" is an opaque token that is returned from the API to paginate reports
+     */
+    private from = 0;
+    /**
+     * The currently-pending report poll
+     */
+    private timeout: ReturnType<typeof setTimeout> | null = null;
+
+    constructor(
+        private mjolnir: Mjolnir,
+        private manager: ReportManager,
+    ) { }
+
+    private schedulePoll() {
+        if (this.timeout === null) {
+            /*
+             * Important that we use `setTimeout` here, not `setInterval`,
+             * because if there's networking problems and `getAbuseReports`
+             * hangs for longer thank the interval, it could cause a stampede
+             * of requests when networking problems resolve
+             */
+            this.timeout = setTimeout(
+                this.tryGetAbuseReports.bind(this),
+                30_000 // a minute in milliseconds
+            );
+        } else {
+            throw new InvalidStateError("poll already scheduled");
+        }
+    }
+
+    private async getAbuseReports() {
+        let response_: {
+            event_reports: { room_id: string, event_id: string, sender: string, reason: string }[],
+            next_token: number | undefined
+        } | undefined;
+        try {
+            response_ = await this.mjolnir.client.doRequest(
+                "GET",
+                "/_synapse/admin/v1/event_reports",
+                { from: this.from.toString() }
+            );
+        } catch (ex) {
+            await this.mjolnir.logMessage(LogLevel.ERROR, "getAbuseReports", `failed to poll events: ${ex}`);
+            return;
+        }
+
+        const response = response_!;
+        for (let report of response.event_reports) {
+            if (!(report.room_id in this.mjolnir.protectedRooms)) {
+                continue;
+            }
+
+            let event: any; // `any` because `handleServerAbuseReport` uses `any`
+            try {
+                event = (await this.mjolnir.client.doRequest(
+                    "GET",
+                    `/_synapse/admin/v1/rooms/${report.room_id}/context/${report.event_id}?limit=1`
+                )).event;
+            } catch (ex) {
+                this.mjolnir.logMessage(LogLevel.ERROR, "getAbuseReports", `failed to get context: ${ex}`);
+                continue;
+            }
+
+            await this.manager.handleServerAbuseReport({
+                roomId: report.room_id,
+                reporterId: report.sender,
+                event: event,
+                reason: report.reason,
+            });
+        }
+
+        /*
+         * This API endpoint returns an opaque `next_token` number that we
+         * need to give back to subsequent requests for pagination, so here we
+         * save it in account data
+         */
+        if (response.next_token !== undefined) {
+            this.from = response.next_token;
+            try {
+               await this.mjolnir.client.setAccountData(REPORT_POLL_EVENT_TYPE, { from: response.next_token });
+            } catch (ex) {
+                await this.mjolnir.logMessage(LogLevel.ERROR, "getAbuseReports", `failed to update progress: ${ex}`);
+            }
+        }
+    }
+
+    private async tryGetAbuseReports() {
+        this.timeout = null;
+
+        try {
+            await this.getAbuseReports()
+        } catch (ex) {
+            await this.mjolnir.logMessage(LogLevel.ERROR, "tryGetAbuseReports", `failed to get abuse reports: ${ex}`);
+        }
+
+        this.schedulePoll();
+    }
+    public start(startFrom: number) {
+        if (this.timeout === null) {
+            this.from = startFrom;
+            this.schedulePoll();
+        } else {
+            throw new InvalidStateError("cannot start an already started poll");
+        }
+    }
+    public stop() {
+        if (this.timeout !== null) {
+            clearTimeout(this.timeout);
+            this.timeout = null;
+        } else {
+            throw new InvalidStateError("cannot stop a poll that hasn't started");
+        }
+    }
+}

test/integration/reportPollingTest.ts

@@ -0,0 +1,54 @@
+import { strict as assert } from "assert";
+
+import config from "../../src/config";
+import { Mjolnir } from "../../src/Mjolnir";
+import { IProtection } from "../../src/protections/IProtection";
+import { PROTECTIONS } from "../../src/protections/protections";
+import { ProtectionSettingValidationError } from "../../src/protections/ProtectionSettings";
+import { NumberProtectionSetting, StringProtectionSetting, StringListProtectionSetting } from "../../src/protections/ProtectionSettings";
+import { newTestUser, noticeListener } from "./clientHelper";
+import { matrixClient, mjolnir } from "./mjolnirSetupUtils";
+
+describe("Test: Report polling", function() {
+    let client;
+    this.beforeEach(async function () {
+        client = await newTestUser({ name: { contains: "protection-settings" }});
+        await client.start();
+    })
+    this.afterEach(async function () {
+        await client.stop();
+    })
+    it("Mjolnir correctly retrieves a report from synapse", async function() {
+        this.timeout(40000);
+
+        const reportPromise = new Promise(async (resolve, reject) => {
+            await this.mjolnir.registerProtection(new class implements IProtection {
+                name = "jYvufI";
+                description = "A test protection";
+                settings = { };
+                handleEvent = async (mjolnir: Mjolnir, roomId: string, event: any) => { };
+                handleReport = (mjolnir: Mjolnir, roomId: string, reporterId: string, event: any, reason?: string) => {
+                    if (reason === "x5h1Je") {
+                        resolve(null);
+                    }
+                };
+            });
+        });
+        await this.mjolnir.enableProtection("jYvufI");
+
+        let protectedRoomId = await this.mjolnir.client.createRoom({ invite: [await client.getUserId()] });
+        await client.joinRoom(protectedRoomId);
+        await this.mjolnir.addProtectedRoom(protectedRoomId);
+
+        const eventId = await client.sendMessage(protectedRoomId, {msgtype: "m.text", body: "uwNd3q"});
+        await client.doRequest(
+            "POST",
+            `/_matrix/client/r0/rooms/${encodeURIComponent(protectedRoomId)}/report/${encodeURIComponent(eventId)}`, "", {
+                reason: "x5h1Je"
+            }
+        );
+
+        await reportPromise;
+    });
+});
+

tsconfig.json

@@ -22,6 +22,7 @@
         "./src/**/*",
         "./test/integration/manualLaunchScript.ts",
         "./test/integration/roomMembersTest.ts",
-        "./test/integration/banListTest.ts"
+        "./test/integration/banListTest.ts",
+        "./test/integration/reportPollingTest"
     ]
 }