Drop federation typing notifications from non-compliant user IDs

synapse/handlers/typing.py

@@ -39,6 +39,7 @@ from synapse.types import (
     StrCollection,
     StreamKeyType,
     UserID,
+    is_compliant_user_id,
 )
 from synapse.util.caches.stream_change_cache import StreamChangeCache
 from synapse.util.duration import Duration
@@ -422,6 +423,14 @@ class TypingWriterHandler(FollowerTypingHandler):
             )
             return
 
+        if not is_compliant_user_id(user_id):
+            logger.info(
+                "Dropping typing update from %r for non-compliant user ID %r",
+                origin,
+                user_id,
+            )
+            return
+
         # Let's check that the origin server is in the room before accepting the typing
         # event. We don't want to block waiting on a partial state so take an
         # approximation if needed.

tests/handlers/test_typing.py

@@ -349,6 +349,41 @@ class TypingNotificationsTestCase(unittest.HomeserverTestCase):
         self.assertEqual(events[0], [])
         self.assertEqual(events[1], 0)
 
+    def test_started_typing_remote_recv_non_compliant_user_id(self) -> None:
+        """A remote server sends a typing notification for a non-compliant
+        (grandfathered historical) user ID in a room we share. It is dropped.
+        """
+        non_compliant_user = UserID.from_string("@ónion:farm")
+        self.room_members = [U_APPLE, non_compliant_user]
+
+        self.assertEqual(self.event_source.get_current_key(), 0)
+
+        channel = self.make_request(
+            "PUT",
+            "/_matrix/federation/v1/send/1000000",
+            _make_edu_transaction_json(
+                EduTypes.TYPING,
+                content={
+                    "room_id": ROOM_ID,
+                    "user_id": non_compliant_user.to_string(),
+                    "typing": True,
+                },
+            ),
+            federation_auth_origin=b"farm",
+        )
+        self.assertEqual(channel.code, 200)
+
+        self.on_new_event.assert_not_called()
+
+        self.assertEqual(self.event_source.get_current_key(), 0)
+        events = self.get_success(
+            self.event_source.get_new_events(
+                user=U_APPLE, from_key=0, limit=0, room_ids=[ROOM_ID], is_guest=False
+            )
+        )
+        self.assertEqual(events[0], [])
+        self.assertEqual(events[1], 0)
+
     # Enable federation sending on the main process.
     @override_config({"federation_sender_instances": None})
     def test_stopped_typing(self) -> None: