Finish adding type hints to the HomeServer

changelog.d/9374.misc

@@ -0,0 +1 @@
+Add type hints to various parts of the code base.

synapse/federation/sender/__init__.py

@@ -477,7 +477,10 @@ def send_presence_to_destinations(
         self, states: Iterable[UserPresenceState], destinations: Iterable[str]
     ) -> None:
         """Send the given presence states to the given destinations.
-        destinations (list[str])
+
+        Args:
+            states: The user presence states to send.
+            destinations: The hostnames to send the updated states to.
         """
 
         if not states or not self.hs.config.use_presence:

synapse/handlers/deactivate_account.py

@@ -17,6 +17,7 @@
 from typing import TYPE_CHECKING, Optional
 
 from synapse.api.errors import SynapseError
+from synapse.handlers.device import DeviceHandler
 from synapse.metrics.background_process_metrics import run_as_background_process
 from synapse.types import Requester, UserID, create_requester
 
@@ -74,6 +75,8 @@ async def deactivate_account(
         Returns:
             True if identity server supports removing threepids, otherwise False.
         """
+        assert isinstance(self._device_handler, DeviceHandler)
+
         # FIXME: Theoretically there is a race here wherein user resets
         # password using threepid.
 

synapse/handlers/devicemessage.py

@@ -18,6 +18,7 @@
 
 from synapse.api.constants import EduTypes
 from synapse.api.errors import SynapseError
+from synapse.handlers.device import DeviceHandler
 from synapse.api.ratelimiting import Ratelimiter
 from synapse.logging.context import run_in_background
 from synapse.logging.opentracing import (
@@ -72,8 +73,10 @@ def __init__(self, hs: "HomeServer"):
         # sync. We do all device list resyncing on the master instance, so if
         # we're on a worker we hit the device resync replication API.
         if hs.config.worker.worker_app is None:
+            device_handler = hs.get_device_handler()
+            assert isinstance(device_handler, DeviceHandler)
             self._user_device_resync = (
-                hs.get_device_handler().device_list_updater.user_device_resync
+                device_handler.device_list_updater.user_device_resync
             )
         else:
             self._user_device_resync = (

synapse/handlers/e2e_keys.py

@@ -27,6 +27,7 @@
 from twisted.internet import defer
 
 from synapse.api.errors import CodeMessageException, Codes, NotFoundError, SynapseError
+from synapse.handlers.device import DeviceHandler
 from synapse.logging.context import make_deferred_yieldable, run_in_background
 from synapse.logging.opentracing import log_kv, set_tag, tag_args, trace
 from synapse.replication.http.devices import ReplicationUserDevicesResyncRestServlet
@@ -51,24 +52,26 @@ class E2eKeysHandler:
     def __init__(self, hs: "HomeServer"):
         self.store = hs.get_datastore()
         self.federation = hs.get_federation_client()
-        self.device_handler = hs.get_device_handler()
         self.is_mine = hs.is_mine
         self.clock = hs.get_clock()
 
-        self._edu_updater = SigningKeyEduUpdater(hs, self)
-
         federation_registry = hs.get_federation_registry()
 
-        self._is_master = hs.config.worker_app is None
-        if not self._is_master:
+        self.device_handler = None  # type: Optional[DeviceHandler]
+        if hs.config.worker_app is not None:
             self._user_device_resync_client = (
                 ReplicationUserDevicesResyncRestServlet.make_client(hs)
             )
         else:
+            device_handler = hs.get_device_handler()
+            assert isinstance(device_handler, DeviceHandler)
+            self.device_handler = device_handler
+
             # Only register this edu handler on master as it requires writing
             # device updates to the db
             #
             # FIXME: switch to m.signing_key_update when MSC1756 is merged into the spec
+            self._edu_updater = SigningKeyEduUpdater(hs, device_handler)
             federation_registry.register_edu_handler(
                 "org.matrix.signing_key_update",
                 self._edu_updater.incoming_signing_key_update,
@@ -194,7 +197,7 @@ async def query_devices(
 
         # Now fetch any devices that we don't have in our cache
         @trace
-        async def do_remote_query(destination):
+        async def do_remote_query(destination: str) -> None:
             """This is called when we are querying the device list of a user on
             a remote homeserver and their device list is not in the device list
             cache. If we share a room with this user and we're not querying for
@@ -226,16 +229,23 @@ async def do_remote_query(destination):
                 # probably be tracking their device lists. However, we haven't
                 # done an initial sync on the device list so we do it now.
                 try:
-                    if self._is_master:
-                        user_devices = await self.device_handler.device_list_updater.user_device_resync(
+                    if self.device_handler:
+                        user_device_resync = await self.device_handler.device_list_updater.user_device_resync(
                             user_id
                         )
                     else:
-                        user_devices = await self._user_device_resync_client(
+                        user_device_resync = await self._user_device_resync_client(
                             user_id=user_id
                         )
 
-                    user_devices = user_devices["devices"]
+                    if user_device_resync is None:
+                        failures[destination] = {
+                            "status": 503,
+                            "message": "Unable to resync devices",
+                        }
+                        continue
+
+                    user_devices = user_device_resync["devices"]
                     user_results = results.setdefault(user_id, {})
                     for device in user_devices:
                         user_results[device["device_id"]] = device["keys"]
@@ -439,7 +449,7 @@ async def claim_one_time_keys(
                     }
 
         @trace
-        async def claim_client_keys(destination):
+        async def claim_client_keys(destination: str) -> None:
             set_tag("destination", destination)
             device_keys = remote_queries[destination]
             try:
@@ -485,6 +495,7 @@ async def claim_client_keys(destination):
     async def upload_keys_for_user(
         self, user_id: str, device_id: str, keys: JsonDict
     ) -> JsonDict:
+        assert self.device_handler
 
         time_now = self.clock.time_msec()
 
@@ -610,6 +621,7 @@ async def upload_signing_keys_for_user(
             user_id: the user uploading the keys
             keys: the signing keys
         """
+        assert self.device_handler
 
         # if a master key is uploaded, then check it.  Otherwise, load the
         # stored master key, to check signatures on other keys
@@ -701,6 +713,8 @@ async def upload_signatures_for_device_keys(
         Raises:
             SynapseError: if the signatures dict is not valid.
         """
+        assert self.device_handler
+
         failures = {}
 
         # signatures to be stored.  Each item will be a SignatureListItem
@@ -1081,6 +1095,8 @@ async def _retrieve_cross_signing_keys_for_remote_user(
             A tuple of the retrieved key content, the key's ID and the matching VerifyKey.
             If the key cannot be retrieved, all values in the tuple will instead be None.
         """
+        assert self.device_handler
+
         try:
             remote_result = await self.federation.query_user_devices(
                 user.domain, user.to_string()
@@ -1281,11 +1297,11 @@ class SignatureListItem:
 class SigningKeyEduUpdater:
     """Handles incoming signing key updates from federation and updates the DB"""
 
-    def __init__(self, hs: "HomeServer", e2e_keys_handler: E2eKeysHandler):
+    def __init__(self, hs: "HomeServer", device_handler: DeviceHandler):
         self.store = hs.get_datastore()
         self.federation = hs.get_federation_client()
         self.clock = hs.get_clock()
-        self.e2e_keys_handler = e2e_keys_handler
+        self.device_handler = device_handler
 
         self._remote_edu_linearizer = Linearizer(name="remote_signing_key")
 
@@ -1341,8 +1357,7 @@ async def _handle_signing_key_updates(self, user_id: str) -> None:
             user_id: the user whose updates we are processing
         """
 
-        device_handler = self.e2e_keys_handler.device_handler
-        device_list_updater = device_handler.device_list_updater
+        device_list_updater = self.device_handler.device_list_updater
 
         with (await self._remote_edu_linearizer.queue(user_id)):
             pending_updates = self._pending_updates.pop(user_id, [])
@@ -1364,4 +1379,4 @@ async def _handle_signing_key_updates(self, user_id: str) -> None:
                 )
                 device_ids = device_ids + new_device_ids
 
-            await device_handler.notify_device_update(user_id, device_ids)
+            await self.device_handler.notify_device_update(user_id, device_ids)

synapse/handlers/federation.py

@@ -55,6 +55,7 @@
 from synapse.events.snapshot import EventContext
 from synapse.events.validator import EventValidator
 from synapse.handlers._base import BaseHandler
+from synapse.handlers.device import DeviceHandler
 from synapse.http.servlet import assert_params_in_dict
 from synapse.logging.context import (
     make_deferred_yieldable,
@@ -157,7 +158,9 @@ def __init__(self, hs: "HomeServer"):
                 ReplicationStoreRoomOnOutlierMembershipRestServlet.make_client(hs)
             )
         else:
-            self._device_list_updater = hs.get_device_handler().device_list_updater
+            device_handler = hs.get_device_handler()
+            assert isinstance(device_handler, DeviceHandler)
+            self._device_list_updater = device_handler.device_list_updater
             self._maybe_store_room_on_outlier_membership = (
                 self.store.maybe_store_room_on_outlier_membership
             )

synapse/handlers/groups_local.py

@@ -18,6 +18,7 @@
 from typing import TYPE_CHECKING, Dict, Iterable, List, Set
 
 from synapse.api.errors import HttpResponseException, RequestSendFailed, SynapseError
+from synapse.groups.groups_server import GroupsServerHandler
 from synapse.types import GroupID, JsonDict, get_domain_from_id
 
 if TYPE_CHECKING:
@@ -287,6 +288,7 @@ async def create_group(
         logger.info("Asking to create group with ID: %r", group_id)
 
         if self.is_mine_id(group_id):
+            assert isinstance(self.groups_server_handler, GroupsServerHandler)
             res = await self.groups_server_handler.create_group(
                 group_id, user_id, content
             )
@@ -314,6 +316,7 @@ async def join_group(
     ) -> JsonDict:
         """Request to join a group"""
         if self.is_mine_id(group_id):
+            assert isinstance(self.groups_server_handler, GroupsServerHandler)
             await self.groups_server_handler.join_group(group_id, user_id, content)
             local_attestation = None
             remote_attestation = None
@@ -360,6 +363,7 @@ async def accept_invite(
     ) -> JsonDict:
         """Accept an invite to a group"""
         if self.is_mine_id(group_id):
+            assert isinstance(self.groups_server_handler, GroupsServerHandler)
             await self.groups_server_handler.accept_invite(group_id, user_id, content)
             local_attestation = None
             remote_attestation = None
@@ -407,6 +411,7 @@ async def invite(
         """Invite a user to a group"""
         content = {"requester_user_id": requester_user_id, "config": config}
         if self.is_mine_id(group_id):
+            assert isinstance(self.groups_server_handler, GroupsServerHandler)
             res = await self.groups_server_handler.invite_to_group(
                 group_id, user_id, requester_user_id, content
             )
@@ -471,6 +476,7 @@ async def remove_user_from_group(
             # retry if the group server is currently down.
 
         if self.is_mine_id(group_id):
+            assert isinstance(self.groups_server_handler, GroupsServerHandler)
             res = await self.groups_server_handler.remove_user_from_group(
                 group_id, user_id, requester_user_id, content
             )

synapse/handlers/register.py

@@ -25,6 +25,7 @@
 from synapse.api.errors import AuthError, Codes, ConsentNotGivenError, SynapseError
 from synapse.appservice import ApplicationService
 from synapse.config.server import is_threepid_reserved
+from synapse.handlers.device import DeviceHandler
 from synapse.http.servlet import assert_params_in_dict
 from synapse.replication.http.login import RegisterDeviceReplicationServlet
 from synapse.replication.http.register import (
@@ -81,7 +82,9 @@ def __init__(self, hs: "HomeServer"):
                 ReplicationPostRegisterActionsServlet.make_client(hs)
             )
         else:
-            self.device_handler = hs.get_device_handler()
+            device_handler = hs.get_device_handler()
+            assert isinstance(device_handler, DeviceHandler)
+            self.device_handler = device_handler
             self._register_device_client = self.register_device_inner
             self.pusher_pool = hs.get_pusherpool()
 
@@ -437,10 +440,10 @@ async def _join_rooms(self, user_id: str) -> None:
 
                 if RoomAlias.is_valid(r):
                     (
-                        room_id,
+                        room,
                         remote_room_hosts,
                     ) = await room_member_handler.lookup_room_alias(room_alias)
-                    room_id = room_id.to_string()
+                    room_id = room.to_string()
                 else:
                     raise SynapseError(
                         400, "%s was not legal room ID or room alias" % (r,)

synapse/handlers/room.py

@@ -38,6 +38,7 @@
 from synapse.api.room_versions import KNOWN_ROOM_VERSIONS, RoomVersion
 from synapse.events import EventBase
 from synapse.events.utils import copy_power_levels_contents
+from synapse.handlers.room_member import RoomMemberMasterHandler
 from synapse.rest.admin._base import assert_user_is_admin
 from synapse.storage.state import StateFilter
 from synapse.types import (
@@ -1274,6 +1275,7 @@ async def shutdown_room(
                 migrated from the old room to the new.
             new_room_id: A string representing the room ID of the new room.
         """
+        assert isinstance(self.room_member_handler, RoomMemberMasterHandler)
 
         if not new_room_name:
             new_room_name = self.DEFAULT_ROOM_NAME

synapse/handlers/set_password.py

@@ -16,6 +16,7 @@
 from typing import TYPE_CHECKING, Optional
 
 from synapse.api.errors import Codes, StoreError, SynapseError
+from synapse.handlers.device import DeviceHandler
 from synapse.types import Requester
 
 from ._base import BaseHandler
@@ -41,6 +42,8 @@ async def set_password(
         logout_devices: bool,
         requester: Optional[Requester] = None,
     ) -> None:
+        assert isinstance(self._device_handler, DeviceHandler)
+
         if not self.hs.config.password_localdb_enabled:
             raise SynapseError(403, "Password change disabled", errcode=Codes.FORBIDDEN)
 

synapse/notifier.py

@@ -37,6 +37,7 @@
 from synapse.api.constants import EventTypes, HistoryVisibility, Membership
 from synapse.api.errors import AuthError
 from synapse.events import EventBase
+from synapse.federation.sender import FederationSender
 from synapse.handlers.presence import format_user_presence_state
 from synapse.logging.context import PreserveLoggingContext
 from synapse.logging.utils import log_function
@@ -220,7 +221,9 @@ def __init__(self, hs: "synapse.server.HomeServer"):
 
         self.federation_sender = None
         if hs.should_send_federation():
-            self.federation_sender = hs.get_federation_sender()
+            federation_sender = hs.get_federation_sender()
+            assert isinstance(federation_sender, FederationSender)
+            self.federation_sender = federation_sender
 
         self.state_handler = hs.get_state_handler()
 

synapse/replication/tcp/handler.py

@@ -34,6 +34,7 @@
 
 from twisted.internet.protocol import ReconnectingClientFactory
 
+from synapse.federation.send_queue import FederationRemoteSendQueue
 from synapse.metrics import LaterGauge
 from synapse.metrics.background_process_metrics import run_as_background_process
 from synapse.replication.tcp.client import DirectTcpReplicationClientFactory
@@ -213,7 +214,9 @@ def __init__(self, hs: "HomeServer"):
 
         self._federation_sender = None
         if self._is_master and not hs.config.send_federation:
-            self._federation_sender = hs.get_federation_sender()
+            federation_sender = hs.get_federation_sender()
+            assert isinstance(federation_sender, FederationRemoteSendQueue)
+            self._federation_sender = federation_sender
 
         self._server_notices_sender = None
         if self._is_master: