Add a config option to drop emails if user-supplied content contains a given keyword.

changelog.d/535.misc

@@ -0,0 +1 @@
+Add a config option to drop emails if user-supplied content contains a given keyword.

sydent/config/email.py

@@ -76,6 +76,9 @@ def parse_config(self, cfg: ConfigParser) -> bool:
         third_party_invite_room_blocklist = cfg.get(
             "email", "email.third_party_invite_room_blocklist", fallback=""
         )
+        third_party_invite_keyword_blocklist = cfg.get(
+            "email", "email.third_party_invite_keyword_blocklist", fallback=""
+        )
         self.third_party_invite_homeserver_blocklist = {
             server
             for server in third_party_invite_homeserver_blocklist.split("\n")
@@ -86,6 +89,11 @@ def parse_config(self, cfg: ConfigParser) -> bool:
             for room_id in third_party_invite_room_blocklist.split("\n")
             if room_id  # filter out empty lines
         }
+        self.third_party_invite_keyword_blocklist = {
+            keyword
+            for keyword in third_party_invite_keyword_blocklist.split("\n")
+            if keyword
+        }
 
         self.email_sender_ratelimit_burst = cfg.getint(
             "email", "email.ratelimit_sender.burst", fallback=5

sydent/http/servlets/store_invite_servlet.py

@@ -94,6 +94,10 @@ def render_POST(self, request: Request) -> JsonDict:
             sender, "Limit exceeded for this sender"
         )
 
+        for keyword in self.sydent.config.email.third_party_invite_keyword_blocklist:
+            if keyword in [medium, address, roomId, sender]:
+                raise MatrixRestError(403, "M_UNAUTHORIZED", "Invite not allowed")
+
         globalAssocStore = GlobalAssociationStore(self.sydent)
         mxid = globalAssocStore.getMxid(medium, normalised_address)
         if mxid:

tests/test_invites.py

@@ -6,7 +6,7 @@
 from sydent.db.invite_tokens import JoinTokenStore
 from sydent.http.httpclient import FederationHttpClient
 from sydent.http.servlets.store_invite_servlet import StoreInviteServlet
-from tests.utils import make_sydent
+from tests.utils import make_request, make_sydent
 
 
 class ThreepidInvitesTestCase(unittest.TestCase):
@@ -19,6 +19,7 @@ def setUp(self):
                 # Used by test_invited_email_address_obfuscation
                 "email.third_party_invite_username_obfuscate_characters": "6",
                 "email.third_party_invite_domain_obfuscate_characters": "8",
+                "email.third_party_invite_keyword_blocklist": "evil\nbad",
             },
         }
         self.sydent = make_sydent(test_config=config)
@@ -102,6 +103,23 @@ def test_invited_email_address_obfuscation(self):
 
         self.assertEqual(redacted_address, "...@1...")
 
+    def test_third_party_invite_keyword_block_works(self):
+        invite_config = {
+            "medium": "email",
+            "address": "evil",
+            "room_id": "bad",
+            "sender": "@bad_dude:badness.org",
+        }
+        request, channel = make_request(
+            self.sydent.reactor,
+            "POST",
+            "/_matrix/identity/v2/store-invite",
+            invite_config,
+        )
+        store_invite_servlet = StoreInviteServlet(self.sydent)
+        request.render(store_invite_servlet)
+        self.assertEqual(channel.code, 403)
+
 
 class ThreepidInvitesNoDeleteTestCase(unittest.TestCase):
     """Test that invite tokens are not deleted when that is disabled."""