Stop forwarding non compliant user ids (#20338)

Closes #20116

> [!NOTE]
> This PR Follows #20115 which already implemented this for
`m.device_list_update`

> User IDs with localparts containing characters outside the range
U+0021 to U+007E, or with
> an empty localpart, are considered non-compliant. For current room
versions, servers must
> still accept events using such user IDs over federation; however they
SHOULD NOT forward
> such user IDs to clients when referenced outside the context of an
event. For example,
> device list updates from non-compliant user IDs would be dropped by
the receiving server.
>
> — [Spec
1.19](https://spec.matrix.org/v1.19/appendices/#historical-user-ids)

This PR prevents from forwarding non compliant user ids for the
following events:

- `m.signing_key_update` 8c3c35e95d
- `m.presence` 26a8dfe862
- `m.typing` 1110cf2d73
- `m.receipt` 9e2da8b0a6
- `m.direct_to_device` 9541232a30

changelog.d/20338.bugfix

@@ -0,0 +1 @@
+Stop forwarding non-compliant (grandfathered historical) remote user IDs to clients outside the context of an event, by dropping their incoming presence, typing, read receipt, to-device and signing key updates, and by leaving them out of `/sync` device list changes and newly joined users' presence, as per [the spec](https://spec.matrix.org/v1.19/appendices/#historical-user-ids).

synapse/handlers/device.py

@@ -70,7 +70,7 @@ from synapse.types import (
     UserID,
     get_domain_from_id,
     get_verify_key_from_cross_signing_key,
-    is_compliant_user_id_localpart,
+    is_compliant_user_id,
 )
 from synapse.util import stringutils
 from synapse.util.async_helpers import Linearizer
@@ -796,7 +796,20 @@ class DeviceHandler:
             if any(rid in joined_room_ids for rid in entries):
                 newly_left_users.discard(user_id)
 
-        return DeviceListUpdates(changed=users_that_have_changed, left=newly_left_users)
+        return DeviceListUpdates(
+            changed={
+                changed_user_id
+                for changed_user_id in users_that_have_changed
+                if self.hs.is_mine_id(changed_user_id)
+                or is_compliant_user_id(changed_user_id)
+            },
+            left={
+                left_user_id
+                for left_user_id in newly_left_users
+                if self.hs.is_mine_id(left_user_id)
+                or is_compliant_user_id(left_user_id)
+            },
+        )
 
     async def on_federation_query_user_devices(self, user_id: str) -> JsonDict:
         if not self.hs.is_mine(UserID.from_string(user_id)):
@@ -1539,11 +1552,11 @@ class DeviceListUpdater(DeviceListWorkerUpdater):
             )
             return
 
-        if not is_compliant_user_id_localpart(UserID.from_string(user_id).localpart):
+        if not is_compliant_user_id(user_id):
             # We SHOULD NOT forward non-compliant (grandfathered historical)
             # user IDs to clients outside the context of an event, and the spec
             # gives dropping their device list updates as the example. See
-            # https://spec.matrix.org/v1.14/appendices/#historical-user-ids
+            # https://spec.matrix.org/v1.19/appendices/#historical-user-ids
             logger.warning(
                 "Dropping device list update edu for non-compliant user ID %r from %r",
                 user_id,

synapse/handlers/devicemessage.py

@@ -42,7 +42,14 @@ from synapse.logging.opentracing import (
     log_kv,
     set_tag,
 )
-from synapse.types import JsonDict, Requester, StreamKeyType, UserID, get_domain_from_id
+from synapse.types import (
+    JsonDict,
+    Requester,
+    StreamKeyType,
+    UserID,
+    get_domain_from_id,
+    is_compliant_user_id,
+)
 from synapse.util import split_dict_to_fit_to_size
 from synapse.util.json import json_encoder
 from synapse.util.stringutils import random_string_insecure_fast
@@ -122,6 +129,14 @@ class DeviceMessageHandler:
                 sender_user_id,
             )
             return
+
+        if not is_compliant_user_id(sender_user_id):
+            logger.warning(
+                "Dropping device message from %r with non-compliant sender %r",
+                origin,
+                sender_user_id,
+            )
+            return
         message_type = content["type"]
         message_id = content["message_id"]
         for user_id, by_device in content["messages"].items():

synapse/handlers/e2e_keys.py

@@ -44,6 +44,7 @@ from synapse.types import (
     UserID,
     get_domain_from_id,
     get_verify_key_from_cross_signing_key,
+    is_compliant_user_id,
 )
 from synapse.util.async_helpers import Linearizer, concurrently_execute
 from synapse.util.cancellation import cancellable
@@ -1835,6 +1836,14 @@ class SigningKeyEduUpdater:
             logger.warning("Got signing key update edu for %r from %r", user_id, origin)
             return
 
+        if not is_compliant_user_id(user_id):
+            logger.warning(
+                "Dropping signing key update edu for non-compliant user ID %r from %r",
+                user_id,
+                origin,
+            )
+            return
+
         room_ids = await self.store.get_rooms_for_user(user_id)
         if not room_ids:
             # We don't share any rooms with this user. Ignore update, as we

synapse/handlers/presence.py

@@ -119,6 +119,7 @@ from synapse.types import (
     StreamKeyType,
     UserID,
     get_domain_from_id,
+    is_compliant_user_id,
 )
 from synapse.util.async_helpers import Linearizer
 from synapse.util.duration import Duration
@@ -1511,6 +1512,14 @@ class PresenceHandler(BasePresenceHandler):
                 )
                 continue
 
+            if not is_compliant_user_id(user_id):
+                logger.info(
+                    "Dropping presence update from %r for non-compliant user ID %r",
+                    origin,
+                    user_id,
+                )
+                continue
+
             presence_state = push.get("presence", None)
             if not presence_state:
                 logger.info(

synapse/handlers/receipts.py

@@ -32,6 +32,7 @@ from synapse.types import (
     StreamKeyType,
     UserID,
     get_domain_from_id,
+    is_compliant_user_id,
 )
 
 if TYPE_CHECKING:
@@ -116,6 +117,14 @@ class ReceiptsHandler:
                         )
                         continue
 
+                    if not is_compliant_user_id(user_id):
+                        logger.info(
+                            "Received receipt for non-compliant user ID %r from server %s, ignoring",
+                            user_id,
+                            origin,
+                        )
+                        continue
+
                     # Check if these receipts apply to a thread.
                     data = user_values.get("data", {})
                     thread_id = data.get("thread_id")

synapse/handlers/sync.py

@@ -81,6 +81,7 @@ from synapse.types import (
     StreamKeyType,
     StreamToken,
     UserID,
+    is_compliant_user_id,
 )
 from synapse.types.state import StateFilter
 from synapse.util.async_helpers import concurrently_execute
@@ -2787,6 +2788,11 @@ class SyncHandler:
             users = await self.store.get_users_in_room(room_id)
             extra_users_ids.update(users)
         extra_users_ids.discard(user.to_string())
+        extra_users_ids = {
+            extra_user_id
+            for extra_user_id in extra_users_ids
+            if self._is_mine_id(extra_user_id) or is_compliant_user_id(extra_user_id)
+        }
 
         if extra_users_ids:
             states = await self.presence_handler.get_states(extra_users_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.

synapse/types/__init__.py

@@ -436,7 +436,7 @@ class UserID(DomainSpecificString):
     def is_valid_strict(cls, s: str) -> bool:
         """
         Parses the input string and attempts to ensure it is a valid and compliant user
-        ID according to https://spec.matrix.org/v1.17/appendices/#historical-user-ids.
+        ID according to https://spec.matrix.org/v1.19/appendices/#historical-user-ids.
 
         This should be used with care: there are existing non-compliant user IDs in the
         wild with empty or non-ASCII localparts, which will be rejected by this method.
@@ -575,7 +575,7 @@ def is_compliant_user_id_localpart(localpart: str) -> bool:
     """
     Validates that the given user ID localpart is within the "compliant" range,
     i.e. not empty and all characters are between U+0021 and U+007E inclusive.
-    See https://spec.matrix.org/v1.17/appendices/#historical-user-ids
+    See https://spec.matrix.org/v1.19/appendices/#historical-user-ids
 
     To check if a localpart is non-historical, use contains_invalid_mxid_characters instead.
 
@@ -593,6 +593,26 @@ def is_compliant_user_id_localpart(localpart: str) -> bool:
     return all(0x21 <= ord(c) <= 0x7E for c in localpart)
 
 
+def is_compliant_user_id(user_id: str) -> bool:
+    """
+    Validates that the localpart of the given user ID is within the "compliant"
+    range. See `is_compliant_user_id_localpart`.
+
+    Unlike parsing the user ID with `UserID.from_string`, this never raises on a
+    malformed user ID.
+
+    Args:
+        user_id: the user ID to be checked
+
+    Returns:
+        True if the localpart of the user ID is compliant, False otherwise
+    """
+    if not user_id.startswith(UserID.SIGIL):
+        return False
+    localpart, _, _ = user_id[1:].partition(":")
+    return is_compliant_user_id_localpart(localpart)
+
+
 UPPER_CASE_PATTERN = re.compile(b"[A-Z_]")
 
 # the following is a pattern which matches '=', and bytes which are not allowed in a mxid

tests/federation/test_federation_devices.py

@@ -23,6 +23,7 @@ from unittest.mock import AsyncMock, Mock
 
 from twisted.internet.testing import MemoryReactor
 
+from synapse.api.constants import EduTypes
 from synapse.handlers.device import DeviceListUpdater
 from synapse.server import HomeServer
 from synapse.types import JsonDict
@@ -103,7 +104,7 @@ class DeviceListResyncTestCase(unittest.HomeserverTestCase):
     def test_non_compliant_user_id_update_dropped(self) -> None:
         """Tests that device list updates from non-compliant (grandfathered
         historical) user IDs are dropped, per
-        https://spec.matrix.org/v1.14/appendices/#historical-user-ids
+        https://spec.matrix.org/v1.19/appendices/#historical-user-ids
         """
         remote_user_id = "@héllo:test_remote"
         remote_origin = "test_remote"
@@ -155,6 +156,44 @@ class DeviceListResyncTestCase(unittest.HomeserverTestCase):
         )
         self.assertNotIn(remote_user_id, need_resync)
 
+    def test_non_compliant_user_id_signing_key_update_dropped(self) -> None:
+        """Tests that signing key updates from non-compliant (grandfathered
+        historical) user IDs are dropped, while those from compliant user IDs
+        sharing the same room are stored.
+        """
+        compliant_user_id = "@john:test_remote"
+        non_compliant_user_id = "@héllo:test_remote"
+        remote_origin = "test_remote"
+        remote_master_key = "85T7JXPFBAySB/jwby4S3lBPTqY3+Zg53nYuGmu1ggY"
+
+        # Share room with both users.
+        self.store.get_rooms_for_user = AsyncMock(return_value=["!someroom:test"])
+
+        federation_registry = self.hs.get_federation_registry()
+        for user_id in (compliant_user_id, non_compliant_user_id):
+            self.get_success(
+                federation_registry.on_edu(
+                    EduTypes.SIGNING_KEY_UPDATE,
+                    remote_origin,
+                    {
+                        "user_id": user_id,
+                        "master_key": {
+                            "user_id": user_id,
+                            "usage": ["master"],
+                            "keys": {"ed25519:" + remote_master_key: remote_master_key},
+                        },
+                    },
+                )
+            )
+
+        keys = self.get_success(
+            self.store.get_e2e_cross_signing_keys_bulk(
+                user_ids=[compliant_user_id, non_compliant_user_id]
+            ),
+        )
+        self.assertIsNotNone(keys[compliant_user_id])
+        self.assertIsNone(keys[non_compliant_user_id])
+
     def test_cross_signing_keys_retry(self) -> None:
         """Tests that resyncing a device list correctly processes cross-signing keys from
         the remote server.

tests/handlers/test_presence.py

@@ -591,6 +591,35 @@ class PresenceUpdateTestCase(unittest.HomeserverTestCase):
             # Assert the user is now unavailable.
             self.assertEqual(state.state, PresenceState.UNAVAILABLE)
 
+    def test_non_compliant_user_id_federation_presence_dropped(self) -> None:
+        """A remote server pushes presence for a compliant and a non-compliant
+        (grandfathered historical) user ID. Only the compliant one is applied.
+        """
+        compliant_user_id = UserID.from_string("@john:remote")
+        non_compliant_user_id = UserID.from_string("@héllo:remote")
+
+        presence_handler = self.hs.get_presence_handler()
+        assert isinstance(presence_handler, PresenceHandler)
+        self.get_success(
+            presence_handler.incoming_presence(
+                "remote",
+                {
+                    "push": [
+                        {
+                            "user_id": user_id.to_string(),
+                            "presence": PresenceState.ONLINE,
+                        }
+                        for user_id in (compliant_user_id, non_compliant_user_id)
+                    ]
+                },
+            )
+        )
+
+        state = self.get_success(presence_handler.get_state(compliant_user_id))
+        self.assertEqual(state.state, PresenceState.ONLINE)
+        state = self.get_success(presence_handler.get_state(non_compliant_user_id))
+        self.assertEqual(state.state, PresenceState.OFFLINE)
+
 
 class PresenceTimeoutTestCase(unittest.TestCase):
     """Tests different timers and that the timer does not change `status_msg` of user."""

tests/handlers/test_receipts.py

@@ -23,7 +23,9 @@ from copy import deepcopy
 
 from twisted.internet.testing import MemoryReactor
 
-from synapse.api.constants import EduTypes, ReceiptTypes
+from synapse.api.constants import EduTypes, Membership, ReceiptTypes
+from synapse.rest import admin
+from synapse.rest.client import login, room
 from synapse.server import HomeServer
 from synapse.types import JsonDict
 from synapse.util.clock import Clock
@@ -340,3 +342,71 @@ class ReceiptsTestCase(unittest.HomeserverTestCase):
             events, "@me:server.org"
         )
         self.assertEqual(filtered_events, expected_output)
+
+
+class FederationReceiptsTestCase(unittest.FederatingHomeserverTestCase):
+    servlets = [
+        admin.register_servlets,
+        login.register_servlets,
+        room.register_servlets,
+    ]
+
+    def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None:
+        super().prepare(reactor, clock, hs)
+        hs.get_receipts_handler()
+
+    def test_non_compliant_user_id_receipt_dropped(self) -> None:
+        """A remote server sends read receipts for a compliant and a
+        non-compliant (grandfathered historical) user ID, both joined to a room
+        we share. Only the compliant one is stored.
+        """
+        local_user_id = self.register_user("alice", "pass")
+        local_user_token = self.login("alice", "pass")
+        room_id = self.helper.create_room_as(local_user_id, tok=local_user_token)
+        event_id = self.helper.send(room_id, tok=local_user_token)["event_id"]
+
+        compliant_user_id = f"@john:{self.OTHER_SERVER_NAME}"
+        non_compliant_user_id = f"@héllo:{self.OTHER_SERVER_NAME}"
+        for user_id in (compliant_user_id, non_compliant_user_id):
+            self.inject_room_member(room_id, user_id, Membership.JOIN)
+
+        channel = self.make_signed_federation_request(
+            "PUT",
+            "/_matrix/federation/v1/send/txn",
+            {
+                "edus": [
+                    {
+                        "edu_type": EduTypes.RECEIPT,
+                        "content": {
+                            room_id: {
+                                ReceiptTypes.READ: {
+                                    user_id: {
+                                        "event_ids": [event_id],
+                                        "data": {"ts": 1436451550453},
+                                    }
+                                    for user_id in (
+                                        compliant_user_id,
+                                        non_compliant_user_id,
+                                    )
+                                }
+                            }
+                        },
+                    }
+                ]
+            },
+        )
+        self.assertEqual(channel.code, 200, channel.result)
+
+        store = self.hs.get_datastores().main
+        self.assertEqual(
+            self.get_success(
+                store.get_receipts_for_user(compliant_user_id, [ReceiptTypes.READ])
+            ),
+            {room_id: event_id},
+        )
+        self.assertEqual(
+            self.get_success(
+                store.get_receipts_for_user(non_compliant_user_id, [ReceiptTypes.READ])
+            ),
+            {},
+        )

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:

tests/rest/client/test_sendtodevice.py

@@ -543,6 +543,48 @@ class SendToDeviceTestCase(HomeserverTestCase):
             ],
         )
 
+    def test_remote_non_compliant_sender(self) -> None:
+        """
+        Tests that a to-device message from a non-compliant (grandfathered
+        historical) sender is dropped, while one from a compliant sender on the
+        same server is delivered.
+        """
+        user2 = self.register_user("u2", "pass")
+        user2_tok = self.login("u2", "pass", "d2")
+
+        federation_registry = self.hs.get_federation_registry()
+
+        for sender, content in (
+            ("@héllo:example.org", {"foo": "bar"}),
+            ("@user:example.org", {"hiss": "meow"}),
+        ):
+            self.get_success(
+                federation_registry.on_edu(
+                    EduTypes.DIRECT_TO_DEVICE,
+                    "example.org",
+                    {
+                        "sender": sender,
+                        "type": "org.example.test",
+                        "messages": {user2: {"d2": content}},
+                        "message_id": "1",
+                    },
+                )
+            )
+
+        channel = self.make_request("GET", "/sync", access_token=user2_tok)
+        self.assertEqual(channel.code, 200, channel.result)
+        messages = channel.json_body.get("to_device", {}).get("events", [])
+        self.assertEqual(
+            messages,
+            [
+                {
+                    "content": {"hiss": "meow"},
+                    "sender": "@user:example.org",
+                    "type": "org.example.test",
+                }
+            ],
+        )
+
     def test_limited_sync(self) -> None:
         """If a limited sync for to-devices happens the next /sync should respond immediately."""
 

tests/rest/client/test_sync.py

@@ -30,6 +30,7 @@ from synapse.api.constants import (
     EventContentFields,
     EventTypes,
     JoinRules,
+    Membership,
     ReceiptTypes,
     RelationTypes,
 )
@@ -47,6 +48,7 @@ from tests.rest.client.test_rooms import make_request_with_cancellation_test
 from tests.server import FakeChannel, TimedOutException
 from tests.test_utils.event_injection import (
     inject_event,
+    inject_member_event,
     persist_message_and_state_event_in_one_batch,
 )
 
@@ -838,6 +840,49 @@ class SyncStateAfterTestCase(unittest.HomeserverTestCase):
         self.assertNotIn("org.matrix.msc4222.state_after", room_keys)
 
 
+class PresenceSyncTestCase(unittest.HomeserverTestCase):
+    """
+    Tests regarding the `presence` section of sync.
+    """
+
+    servlets = [
+        synapse.rest.admin.register_servlets,
+        login.register_servlets,
+        room.register_servlets,
+        sync.register_servlets,
+    ]
+
+    def test_non_compliant_remote_user_joining_not_in_presence(self) -> None:
+        """A compliant and a non-compliant (grandfathered historical) remote user
+        join a room alice is in. Alice's incremental sync only includes the
+        presence of the compliant one.
+        """
+        alice_user_id = self.register_user("alice", "correcthorse")
+        alice_access_token = self.login(alice_user_id, "correcthorse")
+        room_id = self.helper.create_room_as(alice_user_id, tok=alice_access_token)
+
+        channel = self.make_request("GET", "/sync", access_token=alice_access_token)
+        self.assertEqual(channel.code, 200, channel.json_body)
+        next_batch = channel.json_body["next_batch"]
+
+        compliant_user_id = "@john:remote"
+        non_compliant_user_id = "@héllo:remote"
+        for user_id in (compliant_user_id, non_compliant_user_id):
+            self.get_success(
+                inject_member_event(self.hs, room_id, user_id, Membership.JOIN)
+            )
+
+        channel = self.make_request(
+            "GET", f"/sync?since={next_batch}", access_token=alice_access_token
+        )
+        self.assertEqual(channel.code, 200, channel.json_body)
+        presence_senders = [
+            event["sender"] for event in channel.json_body["presence"]["events"]
+        ]
+        self.assertIn(compliant_user_id, presence_senders)
+        self.assertNotIn(non_compliant_user_id, presence_senders)
+
+
 class DeviceListSyncTestCase(unittest.HomeserverTestCase):
     """
     Tests regarding device list (`device_lists`) changes.
@@ -1023,6 +1068,41 @@ class DeviceListSyncTestCase(unittest.HomeserverTestCase):
             alice_user_id, device_list_changes, incremental_sync_channel.json_body
         )
 
+    def test_non_compliant_remote_user_not_in_device_list_changes(self) -> None:
+        """A compliant and a non-compliant (grandfathered historical) remote user
+        join, then leave, a room alice is in. Alice's incremental syncs only list
+        the compliant one in `device_lists.changed` and `device_lists.left`.
+        """
+        alice_user_id = self.register_user("alice", "correcthorse")
+        alice_access_token = self.login(alice_user_id, "correcthorse")
+        room_id = self.helper.create_room_as(alice_user_id, tok=alice_access_token)
+
+        channel = self.make_request("GET", "/sync", access_token=alice_access_token)
+        self.assertEqual(channel.code, 200, channel.json_body)
+        next_batch = channel.json_body["next_batch"]
+
+        compliant_user_id = "@john:remote"
+        non_compliant_user_id = "@héllo:remote"
+
+        for membership, device_lists_field in (
+            (Membership.JOIN, "changed"),
+            (Membership.LEAVE, "left"),
+        ):
+            for user_id in (compliant_user_id, non_compliant_user_id):
+                self.get_success(
+                    inject_member_event(self.hs, room_id, user_id, membership)
+                )
+
+            channel = self.make_request(
+                "GET", f"/sync?since={next_batch}", access_token=alice_access_token
+            )
+            self.assertEqual(channel.code, 200, channel.json_body)
+            next_batch = channel.json_body["next_batch"]
+
+            device_lists = channel.json_body["device_lists"][device_lists_field]
+            self.assertIn(compliant_user_id, device_lists)
+            self.assertNotIn(non_compliant_user_id, device_lists)
+
 
 class DeviceOneTimeKeysSyncTestCase(unittest.HomeserverTestCase):
     """

tests/test_types.py

@@ -37,6 +37,7 @@ from synapse.types import (
     UserID,
     get_domain_from_id,
     get_localpart_from_id,
+    is_compliant_user_id,
     map_username_to_mxid_localpart,
 )
 
@@ -142,6 +143,21 @@ class MapUsernameTestCase(unittest.TestCase):
         self.assertEqual(map_username_to_mxid_localpart("têst".encode()), "t=c3=aast")
 
 
+class IsCompliantUserIdTestCase(unittest.TestCase):
+    def test_compliant(self) -> None:
+        self.assertTrue(is_compliant_user_id("@test_1234:test"))
+        self.assertTrue(is_compliant_user_id("@TEST!~:test"))
+
+    def test_non_compliant(self) -> None:
+        self.assertFalse(is_compliant_user_id("@têst:test"))
+        self.assertFalse(is_compliant_user_id("@te st:test"))
+        self.assertFalse(is_compliant_user_id("@:test"))
+
+    def test_malformed_does_not_raise(self) -> None:
+        self.assertFalse(is_compliant_user_id(""))
+        self.assertFalse(is_compliant_user_id("@"))
+
+
 @parameterized_class(
     ("token_type",),
     [