Experimental federated user directory

changelog.d/19887.feature

@@ -0,0 +1 @@
+Pre-MSC implementation of a federated user directory.

synapse/config/experimental.py

@@ -625,3 +625,27 @@ def read_config(
 
         # MSC4491: Invite reasons in room creation
         self.msc4491_enabled: bool = experimental.get("msc4491_enabled", False)
+
+        # Pre-MSC implementation of federated user search.
+        self.bwi_federated_user_dir_enabled: bool = experimental.get(
+            "bwi_federated_user_dir_enabled", False
+        )
+
+        self.bwi_federated_user_dir_federation_search_timeout: int = (
+            self.parse_duration(
+                experimental.get(
+                    "bwi_federated_user_dir_federation_search_timeout", 2000
+                )
+            )
+        )
+
+        self.bwi_federated_user_dir_sync_interval_ms: int = self.parse_duration(
+            experimental.get("bwi_federated_user_dir_sync_interval", "4h")
+        )
+
+        if self.bwi_federated_user_dir_enabled:
+            if self.bwi_federated_user_dir_sync_interval_ms < 1:
+                raise ConfigError(
+                    "experimental_features.bwi_federated_user_dir_sync_interval must "
+                    "be positive"
+                )

synapse/federation/federation_client.py

@@ -36,6 +36,7 @@
     Optional,
     Sequence,
     TypeVar,
+    cast,
 )
 
 import attr
@@ -70,13 +71,21 @@
 from synapse.http.types import QueryParams
 from synapse.logging.opentracing import SynapseTags, log_kv, set_tag, tag_args, trace
 from synapse.metrics import SERVER_NAME_LABEL
-from synapse.types import JsonDict, StrCollection, UserID, get_domain_from_id
+from synapse.metrics.background_process_metrics import wrap_as_background_process
+from synapse.types import (
+    JsonDict,
+    RemoteUserDirectoryEntry,
+    StrCollection,
+    UserID,
+    get_domain_from_id,
+)
 from synapse.util.async_helpers import concurrently_execute
 from synapse.util.caches.expiringcache import ExpiringCache
 from synapse.util.duration import Duration
 from synapse.util.retryutils import NotRetryingDestination
 
 if TYPE_CHECKING:
+    from synapse.handlers.user_directory import UserDirectoryFederationHandler
     from synapse.server import HomeServer
 
 logger = logging.getLogger(__name__)
@@ -134,6 +143,9 @@ def __init__(self, hs: "HomeServer"):
         self._clock.looping_call(self._clear_tried_cache, Duration(minutes=1))
         self.state = hs.get_state_handler()
         self.transport_layer = hs.get_federation_transport_client()
+        self.user_directory_search_timeout = (
+            hs.config.experimental.bwi_federated_user_dir_federation_search_timeout
+        )
 
         self.server_name = hs.hostname
         self.signing_key = hs.signing_key
@@ -170,6 +182,19 @@ def __init__(self, hs: "HomeServer"):
             reset_expiry_on_get=False,
         )
 
+        # Periodically sync remote homeservers' user directories into our own,
+        # but only on the worker that runs background tasks.
+        if (
+            hs.config.experimental.bwi_federated_user_dir_enabled
+            and hs.config.worker.run_background_tasks
+        ):
+            self._clock.looping_call(
+                self._sync_federated_user_directory,
+                Duration(
+                    milliseconds=hs.config.experimental.bwi_federated_user_dir_sync_interval_ms
+                ),
+            )
+
     def _clear_tried_cache(self) -> None:
         """Clear pdu_destination_tried cache"""
         now = self._clock.time_msec()
@@ -1916,6 +1941,145 @@ def filter_user_id(user_id: str) -> bool:
 
         return filtered_statuses, filtered_failures
 
+    async def user_directory_search(
+        self,
+        destination: str,
+        timeout: int,
+    ) -> JsonDict:
+        """Fetch the full user directory of a remote server.
+
+        Args:
+            destination: The server to query.
+            timeout: Timeout in milliseconds for the request.
+
+        Returns:
+            The results containing a list of users from the remote directory.
+        """
+        try:
+            return await self.transport_layer.user_directory_search(
+                destination, timeout
+            )
+        except (RequestSendFailed, HttpResponseException) as e:
+            # A failing or unreachable destination shouldn't break the sync. The
+            # endpoint is rate-limited, and the transport layer surfaces a 429 as
+            # a RequestSendFailed after exhausting retries, so log without a
+            # stack trace.
+            logger.warning(
+                "Failed to search remote user directory [destination=%s]: %s",
+                destination,
+                e,
+            )
+            return {"results": []}
+        except Exception:
+            # Unexpected error; log with a stack trace for debugging.
+            logger.exception(
+                "Unexpected error searching remote user directory [destination=%s]",
+                destination,
+            )
+            return {"results": []}
+
+    @staticmethod
+    def _parse_remote_user_directory_results(
+        response: JsonDict,
+    ) -> list[RemoteUserDirectoryEntry]:
+        """Parse a remote user directory search response into typed entries.
+
+        Malformed entries are skipped. This keeps the federation wire format
+        contained within the federation layer.
+        """
+        entries: list[RemoteUserDirectoryEntry] = []
+        for user in response.get("results", []):
+            if not isinstance(user, dict):
+                logger.debug(
+                    "Skipping remote user directory entry that is not an object: %r",
+                    user,
+                )
+                continue
+
+            user_id = user.get("user_id")
+            if not isinstance(user_id, str):
+                logger.debug(
+                    "Skipping remote user directory entry with missing or "
+                    "non-string user_id: %r",
+                    user,
+                )
+                continue
+
+            display_name = user.get("display_name")
+            if not isinstance(display_name, str):
+                if display_name is not None:
+                    logger.debug(
+                        "Ignoring non-string display_name for remote user %s: %r",
+                        user_id,
+                        display_name,
+                    )
+                display_name = None
+
+            avatar_url = user.get("avatar_url")
+            if not isinstance(avatar_url, str):
+                if avatar_url is not None:
+                    logger.debug(
+                        "Ignoring non-string avatar_url for remote user %s: %r",
+                        user_id,
+                        avatar_url,
+                    )
+                avatar_url = None
+
+            entries.append(
+                RemoteUserDirectoryEntry(
+                    user_id=user_id,
+                    display_name=display_name,
+                    avatar_url=avatar_url,
+                )
+            )
+
+        return entries
+
+    @wrap_as_background_process("federated_user_directory_sync")
+    async def _sync_federated_user_directory(self) -> None:
+        """Periodically fetch users from known homeservers' user directories
+        and hand them to the local user directory for storage.
+
+        This is the federation-side background job: it decides which servers to
+        contact and how to query them. Persisting the results is delegated to
+        the user directory handler, which knows nothing about federation.
+        """
+        destinations = await self.store.get_known_destinations()
+        if not destinations:
+            logger.debug("ending federated user directory sync: no known destinations")
+            return
+
+        # De-duplicate by user id across destinations.
+        entries_by_user: dict[str, RemoteUserDirectoryEntry] = {}
+
+        for destination in destinations:
+            if self._is_mine_server_name(destination):
+                continue
+
+            response = await self.user_directory_search(
+                destination,
+                self.user_directory_search_timeout,
+            )
+
+            for entry in self._parse_remote_user_directory_results(response):
+                entries_by_user[entry.user_id] = entry
+
+        if not entries_by_user:
+            logger.debug("Federated user directory sync found no remote users")
+            return
+
+        # This job is only scheduled when the feature is enabled, in which case
+        # the homeserver exposes the federation-aware handler variant.
+        handler = cast(
+            "UserDirectoryFederationHandler", self.hs.get_user_directory_handler()
+        )
+        await handler.upsert_remote_users(list(entries_by_user.values()))
+
+        logger.debug(
+            "Federated user directory sync upserted %d remote users",
+            len(entries_by_user),
+        )
+
     async def federation_download_media(
         self,
         destination: str,

synapse/federation/federation_server.py

@@ -87,7 +87,13 @@
 from synapse.storage.databases.main.lock import Lock
 from synapse.storage.databases.main.roommember import extract_heroes_from_room_summary
 from synapse.storage.roommember import MemberSummary
-from synapse.types import JsonDict, StateMap, UserID, get_domain_from_id
+from synapse.types import (
+    JsonDict,
+    JsonMapping,
+    StateMap,
+    UserID,
+    get_domain_from_id,
+)
 from synapse.util import unwrapFirstError
 from synapse.util.async_helpers import Linearizer, concurrently_execute, gather_results
 from synapse.util.caches.response_cache import ResponseCache
@@ -1477,6 +1483,52 @@ async def check_server_matches_acl(self, server_name: str, room_id: str) -> None
         ):
             raise AuthError(code=403, msg="Server is banned from room")
 
+    async def on_user_directory_search_request(
+        self, origin: str
+    ) -> tuple[int, JsonMapping]:
+        """Handle a user directory request from a remote server.
+
+        Returns every searchable local user, since the federation endpoint
+        always syncs the full local directory rather than matching a term.
+
+        Args:
+            origin: The server that sent the request.
+
+        Returns:
+            A tuple of (response code, response json)
+        """
+        return 200, await self._search_all_users()
+
+    async def _search_all_users(self) -> JsonDict:
+        """Return all of this server's own users from the user directory.
+
+        Reads the directory straight from the database and filters to locally
+        owned users, since the federation endpoint must only expose this
+        homeserver's own users (the table may also hold cached remote users).
+
+        Returns:
+            A dict of the form ``{"results": [...]}``.
+        """
+        results = await self.store.get_users_in_user_dir()
+
+        # Federation endpoint: only return users local to this homeserver.
+        # is_mine_id is infallible and returns False for malformed user IDs.
+        filtered_results: list[JsonDict] = []
+        for user in results["results"]:
+            if not self.hs.is_mine_id(user["user_id"]):
+                continue
+
+            # Omit optional fields entirely when unset rather than sending
+            # null over the wire.
+            entry: JsonDict = {"user_id": user["user_id"]}
+            if user["display_name"] is not None:
+                entry["display_name"] = user["display_name"]
+            if user["avatar_url"] is not None:
+                entry["avatar_url"] = user["avatar_url"]
+            filtered_results.append(entry)
+
+        return {"results": filtered_results}
+
 
 class FederationHandlerRegistry:
     """Allows classes to register themselves as handlers for a given EDU or

synapse/federation/transport/client.py

@@ -853,6 +853,37 @@ async def get_account_status(
             destination=destination, path=path, data={"user_ids": user_ids}
         )
 
+    async def user_directory_search(
+        self,
+        destination: str,
+        timeout: int,
+    ) -> JsonDict:
+        """
+        Fetch users from the user directory of a remote server.
+
+        The federation endpoint always returns the remote server's full local
+        directory.
+
+        Args:
+            destination: The server to query.
+            timeout: timeout in milliseconds to get the response from destination.
+
+        Returns:
+            The search results.
+        """
+        path = _create_path(
+            FEDERATION_UNSTABLE_PREFIX,
+            "/de.bwi.federated_user_dir" + "/user_directory/search",
+        )
+        return await self.client.post_json(
+            destination,
+            path=path,
+            data={},
+            # ignore backoff for user search as we will set a small user_directory_search_timeout
+            ignore_backoff=True,
+            timeout=timeout,
+        )
+
     async def download_media_r0(
         self,
         destination: str,

synapse/federation/transport/server/__init__.py

@@ -35,6 +35,7 @@
     FederationMediaThumbnailServlet,
     FederationUnstableClientKeysClaimServlet,
     FederationUnstableGetExtremitiesServlet,
+    FederationUserDirectorySearchServlet,
 )
 from synapse.http.server import HttpServer, JsonResource
 from synapse.http.servlet import (
@@ -334,6 +335,12 @@ def register_servlets(
             ):
                 continue
 
+            if (
+                servletclass == FederationUserDirectorySearchServlet
+                and not hs.config.experimental.bwi_federated_user_dir_enabled
+            ):
+                continue
+
             servletclass(
                 hs=hs,
                 authenticator=authenticator,