From 843d753cb5d9fe4460210f9e7b8840ff14718f8d Mon Sep 17 00:00:00 2001 From: Joshua Fenster Date: Wed, 27 May 2026 15:19:51 +0200 Subject: [PATCH 01/11] Implement federated user directory --- synapse/config/experimental.py | 20 ++ synapse/federation/federation_client.py | 215 +++++++++++++- synapse/federation/federation_server.py | 51 +++- synapse/federation/transport/client.py | 34 +++ .../federation/transport/server/__init__.py | 7 + .../federation/transport/server/federation.py | 39 ++- synapse/handlers/user_directory.py | 61 +++- synapse/server.py | 9 +- .../storage/databases/main/transactions.py | 14 + .../storage/databases/main/user_directory.py | 76 +++++ synapse/types/__init__.py | 14 + tests/federation/test_federation_client.py | 271 +++++++++++++++++- tests/federation/test_federation_server.py | 67 +++++ tests/handlers/test_user_directory.py | 117 +++++++- 14 files changed, 987 insertions(+), 8 deletions(-) diff --git a/synapse/config/experimental.py b/synapse/config/experimental.py index 2d74e9b8866..c826215b5b7 100644 --- a/synapse/config/experimental.py +++ b/synapse/config/experimental.py @@ -625,3 +625,23 @@ 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 = 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" + ) diff --git a/synapse/federation/federation_client.py b/synapse/federation/federation_client.py index 5b334543256..02bb4ac22d7 100644 --- a/synapse/federation/federation_client.py +++ b/synapse/federation/federation_client.py @@ -36,11 +36,14 @@ Optional, Sequence, TypeVar, + cast, ) import attr from prometheus_client import Counter +from twisted.internet import defer + from synapse.api.constants import Direction, EventContentFields, EventTypes, Membership from synapse.api.errors import ( CodeMessageException, @@ -68,15 +71,24 @@ from synapse.federation.transport.client import SendJoinResponse from synapse.http.client import is_unknown_endpoint from synapse.http.types import QueryParams +from synapse.logging.context import make_deferred_yieldable 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__) @@ -88,6 +100,10 @@ PDU_RETRY_TIME_MS = 1 * 60 * 1000 +# Localpart of the synthetic requester MXID used when querying remote homeservers +# during the periodic federated user directory sync. +FEDERATED_USER_DIR_SYNC_REQUESTER_LOCALPART = "_user_directory_sync" + T = TypeVar("T") @@ -134,6 +150,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 +189,25 @@ def __init__(self, hs: "HomeServer"): reset_expiry_on_get=False, ) + # Synthetic requester used for the federated user directory sync. The + # remote server requires the requester to belong to our server. + self._federated_user_dir_sync_requester = ( + f"@{FEDERATED_USER_DIR_SYNC_REQUESTER_LOCALPART}:{self.server_name}" + ) + + # 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 +1954,181 @@ def filter_user_id(user_id: str) -> bool: return filtered_statuses, filtered_failures + async def user_directory_search( + self, + requester: str, + 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, so no result limit is sent. + + Args: + requester: The user that initiated the request. + 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: + response = await self.transport_layer.user_directory_search( + requester, destination, timeout + ) + return response + except Exception as e: + # If something goes wrong, we still want to return what we have + logger.exception( + "Error searching user directory across federation[destination=%s] : %s", + destination, + e, + ) + return {"limited": False, "results": []} + + async def search_user_directory_across_federation( + self, + requester: str, + destinations: Collection[str], + limit: int = 10, + ) -> JsonDict: + """Fetch users from the directories of multiple federated servers. + + Args: + requester: The user that initiated the request. + destinations: The servers to query. + limit: Maximum number of results to return per server. + + Returns: + Combined results from all servers. + """ + + if not destinations: + return {"limited": False, "results": []} + + # Query each server individually and collect results + combined_results = [] + limited = False + + # Create a list of deferreds to query each server + query_tasks = [] + for destination in destinations: + if not self._is_mine_server_name(destination): + # Convert coroutine to Deferred + deferred = defer.ensureDeferred( + self.user_directory_search( + requester, + destination, + self.user_directory_search_timeout, + ) + ) + query_tasks.append(deferred) + + # Execute all queries in parallel + if query_tasks: + server_results = await make_deferred_yieldable( + defer.gatherResults( + query_tasks, + consumeErrors=True, + ) + ) + + # Process results from each server + for result in server_results: + if result.get("limited", False): + limited = True + combined_results.extend(result.get("results", [])) + + # Limit the total number of results + if len(combined_results) > limit: + combined_results = combined_results[:limit] + limited = True + + return {"limited": limited, "results": combined_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): + continue + + user_id = user.get("user_id") + if not isinstance(user_id, str): + continue + + display_name = user.get("display_name") + if not isinstance(display_name, str): + display_name = None + + avatar_url = user.get("avatar_url") + if not isinstance(avatar_url, str): + 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("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( + self._federated_user_dir_sync_requester, + 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.info( + "Federated user directory sync upserted %d remote users", + len(entries_by_user), + ) + async def federation_download_media( self, destination: str, diff --git a/synapse/federation/federation_server.py b/synapse/federation/federation_server.py index 2fc78c012e9..b86bb2e7251 100644 --- a/synapse/federation/federation_server.py +++ b/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,49 @@ 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, requester_id: str, 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: + requester_id: The user ID of the requester on the origin server. + 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 ``{"limited": False, "results": [...]}``. + """ + results = await self.store.get_users_in_user_dir() + + # Federation endpoint: only return users local to this homeserver. + filtered_results = [] + for user in results.get("results", []): + try: + if self.hs.is_mine_id(user["user_id"]): + filtered_results.append(user) + except SynapseError: + # Ignore malformed user IDs. + continue + + # The federation endpoint never truncates: it always returns the full + # local directory. + return {"limited": False, "results": filtered_results} + class FederationHandlerRegistry: """Allows classes to register themselves as handlers for a given EDU or diff --git a/synapse/federation/transport/client.py b/synapse/federation/transport/client.py index 5d5212ef96c..6fb676584b2 100644 --- a/synapse/federation/transport/client.py +++ b/synapse/federation/transport/client.py @@ -853,6 +853,40 @@ async def get_account_status( destination=destination, path=path, data={"user_ids": user_ids} ) + async def user_directory_search( + self, + requester: str, + 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: + requester: The user that initiated the request. + 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", + ) + content = {"requester": requester} + return await self.client.post_json( + destination, + path=path, + data=content, + # 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, diff --git a/synapse/federation/transport/server/__init__.py b/synapse/federation/transport/server/__init__.py index 0eff49cf73c..b7761fa0a6d 100644 --- a/synapse/federation/transport/server/__init__.py +++ b/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, diff --git a/synapse/federation/transport/server/federation.py b/synapse/federation/transport/server/federation.py index d783e6da518..78994c3071c 100644 --- a/synapse/federation/transport/server/federation.py +++ b/synapse/federation/transport/server/federation.py @@ -47,7 +47,7 @@ from synapse.http.site import SynapseRequest from synapse.media._base import DEFAULT_MAX_TIMEOUT_MS, MAXIMUM_ALLOWED_MAX_TIMEOUT_MS from synapse.media.thumbnailer import ThumbnailProvider -from synapse.types import JsonDict +from synapse.types import JsonDict, JsonMapping, get_domain_from_id from synapse.util import SYNAPSE_VERSION from synapse.util.ratelimitutils import FederationRateLimiter @@ -893,6 +893,42 @@ async def on_GET( self.media_repo.mark_recently_accessed(None, media_id) +class FederationUserDirectorySearchServlet(BaseFederationServerServlet): + """ + Implements a federation API endpoint for searching a server's user directory. + POST /_matrix/federation/v3/user_directory/search + Request: + { + "requester": "@user:example.com" + } + Response: + { + "limited": false, + "results": [ + { + "user_id": "@user:example.com", + "display_name": "Display Name", + "avatar_url": "mxc://example.com/avatar", + "m.user_directory.visibility": "local" + } + ] + } + """ + + PATH = "/user_directory/search" + PREFIX = FEDERATION_UNSTABLE_PREFIX + "/de.bwi.federated_user_dir" + RATELIMIT = True + + async def on_POST( + self, origin: str, content: JsonDict, query: dict[bytes, list[bytes]] + ) -> tuple[int, JsonMapping]: + requester = content.get("requester") + if requester is None or get_domain_from_id(requester) != origin: + raise SynapseError(400, "Missing or invalid requester", Codes.BAD_JSON) + + return await self.handler.on_user_directory_search_request(requester, origin) + + FEDERATION_SERVLET_CLASSES: tuple[type[BaseFederationServlet], ...] = ( FederationSendServlet, FederationEventServlet, @@ -925,4 +961,5 @@ async def on_GET( FederationV1SendKnockServlet, FederationMakeKnockServlet, FederationAccountStatusServlet, + FederationUserDirectorySearchServlet, ) diff --git a/synapse/handlers/user_directory.py b/synapse/handlers/user_directory.py index e5c4de03c5a..058086d23c7 100644 --- a/synapse/handlers/user_directory.py +++ b/synapse/handlers/user_directory.py @@ -21,7 +21,7 @@ import logging from http import HTTPStatus -from typing import TYPE_CHECKING, Optional +from typing import TYPE_CHECKING, Optional, Sequence from twisted.internet.interfaces import IDelayedCall @@ -39,7 +39,7 @@ from synapse.storage.databases.main.state_deltas import StateDelta from synapse.storage.databases.main.user_directory import SearchResult from synapse.storage.roommember import ProfileInfo -from synapse.types import UserID +from synapse.types import RemoteUserDirectoryEntry, UserID from synapse.util.duration import Duration from synapse.util.metrics import Measure from synapse.util.retryutils import NotRetryingDestination @@ -63,6 +63,15 @@ # every 15 seconds. INTERVAL_TO_ADD_MORE_SERVERS_TO_REFRESH_PROFILES = Duration(seconds=15) +# Sentinel "room" id used to mark remote users ingested via federated user +# directory sync as visible in local user directory searches. This is not a real +# room; it only reuses the `users_in_public_rooms` visibility mechanism. +FEDERATED_USER_DIR_CACHE_ROOM_LOCALPART = "bwi-fed-user-dir-cache" + + +def federated_user_dir_cache_room_id(server_name: str) -> str: + return f"!{FEDERATED_USER_DIR_CACHE_ROOM_LOCALPART}:{server_name}" + def calculate_time_of_next_retry(now_ts: int, retry_count: int) -> int: """ @@ -784,3 +793,51 @@ async def _unsafe_refresh_remote_profiles_for_remote_server( profile.get(ProfileFields.AVATAR_URL) ), ) + + +class UserDirectoryFederationHandler(UserDirectoryHandler): + """Extends the user directory handler with the ability to ingest remote + users discovered via federated user directory sync. + + The federation layer (which knows how to talk to other homeservers, how + often to sync and which destinations to contact) hands over already-parsed + `RemoteUserDirectoryEntry` objects. This handler is solely responsible for + persisting them into the same database tables used for locally discovered + users, so it has no knowledge of federation clients, transports or wire + formats. + """ + + def __init__(self, hs: "HomeServer"): + super().__init__(hs) + + # Sentinel "room" used to make ingested remote users visible in local + # user directory searches via the `users_in_public_rooms` mechanism. + self._federated_cache_room_id = federated_user_dir_cache_room_id( + self.server_name + ) + + async def upsert_remote_users( + self, users: Sequence[RemoteUserDirectoryEntry] + ) -> None: + """Persist remote users fetched from another homeserver's user directory. + + Remote users are stored in the same tables as locally discovered users + so that client searches only need to query the local database. Entries + for our own users are ignored. + """ + if not self.update_user_directory: + # Only the worker that owns the user directory should write to it. + return + + profiles = [ + (entry.user_id, entry.display_name, entry.avatar_url) + for entry in users + if not self.is_mine_id(entry.user_id) + ] + + if not profiles: + return + + await self.store.upsert_federated_remote_users( + self._federated_cache_room_id, profiles + ) diff --git a/synapse/server.py b/synapse/server.py index 8bf19f11b5d..69db9df4d71 100644 --- a/synapse/server.py +++ b/synapse/server.py @@ -136,7 +136,10 @@ from synapse.handlers.sync import SyncHandler from synapse.handlers.thread_subscriptions import ThreadSubscriptionsHandler from synapse.handlers.typing import FollowerTypingHandler, TypingWriterHandler -from synapse.handlers.user_directory import UserDirectoryHandler +from synapse.handlers.user_directory import ( + UserDirectoryFederationHandler, + UserDirectoryHandler, +) from synapse.handlers.worker_lock import WorkerLocksHandler from synapse.http.client import ( InsecureInterceptableContextFactory, @@ -1028,6 +1031,10 @@ def get_bulk_push_rule_evaluator(self) -> BulkPushRuleEvaluator: @cache_in_self def get_user_directory_handler(self) -> UserDirectoryHandler: + # Only load the federation-aware variant when the experimental feature + # is enabled; otherwise use the plain, federation-agnostic handler. + if self.config.experimental.bwi_federated_user_dir_enabled: + return UserDirectoryFederationHandler(self) return UserDirectoryHandler(self) @cache_in_self diff --git a/synapse/storage/databases/main/transactions.py b/synapse/storage/databases/main/transactions.py index a3d75002b99..80808eec8ec 100644 --- a/synapse/storage/databases/main/transactions.py +++ b/synapse/storage/databases/main/transactions.py @@ -534,6 +534,20 @@ def _get_catch_up_outstanding_destinations_txn( return destinations + async def get_known_destinations(self) -> list[str]: + """Return the names of all homeserver destinations we know about. + + These are servers we have previously recorded in the `destinations` + table (e.g. because we have federated with them). The local server is + not filtered out here; callers should do so if required. + """ + return await self.db_pool.simple_select_onecol( + table="destinations", + keyvalues={}, + retcol="destination", + desc="get_known_destinations", + ) + async def get_destinations_paginate( self, start: int, diff --git a/synapse/storage/databases/main/user_directory.py b/synapse/storage/databases/main/user_directory.py index 6c5abc71aea..f2fc74f14c1 100644 --- a/synapse/storage/databases/main/user_directory.py +++ b/synapse/storage/databases/main/user_directory.py @@ -657,6 +657,39 @@ async def update_profile_in_user_dir( [_UserDirProfile(user_id, display_name, avatar_url)], ) + async def upsert_federated_remote_users( + self, + cache_room_id: str, + users: Sequence[tuple[str, str | None, str | None]], + ) -> None: + """Upsert remote users discovered via federated user directory sync. + + Updates profile and search index entries and marks the users as visible + in user directory searches via the federated cache room. + """ + if not users: + return + + profiles = [ + _UserDirProfile(user_id, display_name, avatar_url) + for user_id, display_name, avatar_url in users + ] + + def _upsert_federated_remote_users_txn(txn: LoggingTransaction) -> None: + self._update_profiles_in_user_dir_txn(txn, profiles) + self.db_pool.simple_upsert_many_txn( + txn, + table="users_in_public_rooms", + key_names=["user_id", "room_id"], + key_values=[(profile.user_id, cache_room_id) for profile in profiles], + value_names=(), + value_values=(), + ) + + await self.db_pool.runInteraction( + "upsert_federated_remote_users", _upsert_federated_remote_users_txn + ) + def _update_profiles_in_user_dir_txn( self, txn: LoggingTransaction, @@ -1157,6 +1190,49 @@ async def search_user_dir( ], } + async def get_users_in_user_dir(self) -> SearchResult: + """Get every user stored in the user directory. + + Unlike :meth:`search_user_dir`, this does not match a search term: it + returns all profiles in the ``user_directory`` table. It is used by the + federation responder to hand a server's full local directory to a + remote homeserver. + + Returns: + A ``SearchResult`` of the form:: + + { + "limited": False, # always False; no term/limit is applied + "results": [ + { + "user_id": , + "display_name": , + "avatar_url": , + } + ] + } + """ + rows = cast( + list[tuple[str, str | None, str | None]], + await self.db_pool.simple_select_list( + table="user_directory", + keyvalues=None, + retcols=("user_id", "display_name", "avatar_url"), + desc="get_users_in_user_dir", + ), + ) + + results: list[UserProfile] = [ + { + "user_id": user_id, + "display_name": display_name, + "avatar_url": avatar_url, + } + for user_id, display_name, avatar_url in rows + ] + + return {"limited": False, "results": results} + def _filter_text_for_index(text: str) -> str: """Transforms text before it is inserted into the user directory index, or searched diff --git a/synapse/types/__init__.py b/synapse/types/__init__.py index a6fc806701d..4181dc57481 100644 --- a/synapse/types/__init__.py +++ b/synapse/types/__init__.py @@ -1642,6 +1642,20 @@ class UserProfile(TypedDict): avatar_url: str | None +@attr.s(auto_attribs=True, frozen=True, slots=True) +class RemoteUserDirectoryEntry: + """A remote user directory entry fetched from another homeserver. + + This is the boundary type used to hand parsed remote users from the + federation layer to the user directory, so the user directory does not need + to know anything about the federation wire format. + """ + + user_id: str + display_name: str | None = None + avatar_url: str | None = None + + @attr.s(auto_attribs=True, frozen=True, slots=True) class RetentionPolicy: min_lifetime: int | None = None diff --git a/tests/federation/test_federation_client.py b/tests/federation/test_federation_client.py index 656665cb424..f93ec34cbd4 100644 --- a/tests/federation/test_federation_client.py +++ b/tests/federation/test_federation_client.py @@ -20,18 +20,22 @@ # from unittest import mock +from unittest.mock import AsyncMock import twisted.web.client from twisted.internet import defer from twisted.internet.testing import MemoryReactor +from synapse.api.errors import HttpResponseException from synapse.api.room_versions import RoomVersions from synapse.events import EventBase from synapse.rest import admin -from synapse.rest.client import login, room +from synapse.rest.client import login, register, room, user_directory from synapse.server import HomeServer +from synapse.types import JsonDict from synapse.util.clock import Clock +from tests.storage.test_user_directory import GetUserDirectoryTables from tests.test_utils import FakeResponse, event_injection from tests.unittest import FederatingHomeserverTestCase @@ -41,8 +45,19 @@ class FederationClientTest(FederatingHomeserverTestCase): admin.register_servlets, room.register_servlets, login.register_servlets, + user_directory.register_servlets, + register.register_servlets, ] + def make_homeserver(self, reactor: MemoryReactor, clock: Clock) -> HomeServer: + """Create a homeserver with federation enabled and user directory enabled.""" + config = self.default_config() + config["user_directory"] = { + "enabled": True, + "search_all_users": True, + } + return self.setup_test_homeserver(config=config) + def prepare( self, reactor: MemoryReactor, clock: Clock, homeserver: HomeServer ) -> None: @@ -59,6 +74,8 @@ def prepare( self.creator = f"@creator:{self.OTHER_SERVER_NAME}" self.test_room_id = "!room_id" + self.federation_client = homeserver.get_federation_client() + self.transport_layer = self.federation_client.transport_layer def test_get_room_state(self) -> None: # mock up some events to use in the response. @@ -317,3 +334,255 @@ def test_backfill_invalid_signature_records_failed_pull_attempts( # This is 2 because it failed once from `self.OTHER_SERVER_NAME` and the # other from "yet.another.server" self.assertEqual(backfill_num_attempts, 2) + + def test_user_directory_search(self) -> None: + """Test that the federation client correctly handles user directory search requests.""" + # Mock the transport layer's user_directory_search method + mock_results = { + "limited": False, + "results": [ + { + "user_id": "@user:other.example.com", + "display_name": "Test User", + "avatar_url": "mxc://example.com/avatar", + } + ], + } + self.transport_layer.user_directory_search = AsyncMock( # type: ignore[method-assign] + return_value=mock_results + ) + + # Call the federation client method + result = self.get_success( + self.federation_client.user_directory_search( + "@requester:example.com", "other.example.com", 2000 + ) + ) + + # Check that the result is correct + self.assertEqual(result, mock_results) + + # Check that user_directory_search was called with the correct arguments + self.transport_layer.user_directory_search.assert_called_once_with( + "@requester:example.com", "other.example.com", 2000 + ) + + def test_user_directory_search_endpoint_not_found(self) -> None: + """Test that the federation client handles 404 responses correctly.""" + # Mock the transport layer to raise a 404 error + self.transport_layer.user_directory_search = AsyncMock( # type: ignore[method-assign] + side_effect=HttpResponseException( + 404, "Not Found", b'{"errcode": "M_NOT_FOUND"}' + ) + ) + + # Call the federation client method + result = self.get_success( + self.federation_client.user_directory_search( + "@requester:example.com", "other.example.com", 10 + ) + ) + + # Check that the result is an empty result set + self.assertEqual(result, {"limited": False, "results": []}) + + def test_search_user_directory_across_federation(self) -> None: + """Test that the federation client correctly handles searching across multiple servers.""" + + # Mock the user_directory_search method to return different results for different servers + async def mock_user_directory_search( + requester: str, destination: str, timeout: int + ) -> JsonDict: + if destination == "server1.example.com": + return { + "limited": False, + "results": [ + { + "user_id": "@user1:server1.example.com", + "display_name": "User 1", + "avatar_url": "mxc://example.com/avatar1", + } + ], + } + elif destination == "server2.example.com": + return { + "limited": False, + "results": [ + { + "user_id": "@user2:server2.example.com", + "display_name": "User 2", + "avatar_url": "mxc://example.com/avatar2", + } + ], + } + else: + return {"limited": False, "results": []} + + self.federation_client.user_directory_search = AsyncMock( # type: ignore[method-assign] + side_effect=mock_user_directory_search + ) + + # Call the federation client method + result = self.get_success( + self.federation_client.search_user_directory_across_federation( + "@requester:example.com", + ["server1.example.com", "server2.example.com"], + 10, + ) + ) + + # Check that the result contains results from both servers + self.assertEqual( + result, + { + "limited": False, + "results": [ + { + "user_id": "@user1:server1.example.com", + "display_name": "User 1", + "avatar_url": "mxc://example.com/avatar1", + }, + { + "user_id": "@user2:server2.example.com", + "display_name": "User 2", + "avatar_url": "mxc://example.com/avatar2", + }, + ], + }, + ) + + def test_search_user_directory_across_federation_with_limit(self) -> None: + """Test that the federation client correctly applies limits when searching across multiple servers.""" + + # Mock the user_directory_search method to return many results + async def mock_user_directory_search( + requester: str, destination: str, timeout: int + ) -> JsonDict: + return { + "limited": False, + "results": [ + { + "user_id": f"@user{i}:{destination}", + "display_name": f"User {i}", + "avatar_url": f"mxc://example.com/avatar{i}", + } + for i in range(10) + ], + } + + self.federation_client.user_directory_search = AsyncMock( # type: ignore[method-assign] + side_effect=mock_user_directory_search + ) + + # Call the federation client method with a limit of 5 + result = self.get_success( + self.federation_client.search_user_directory_across_federation( + "@requester:example.com", + ["server1.example.com", "server2.example.com"], + 5, + ) + ) + + # Check that the result is limited to 5 users + self.assertEqual(len(result["results"]), 5) + self.assertTrue(result["limited"]) + + def test_search_user_directory_across_federation_empty_destinations(self) -> None: + """Test that the federation client handles empty destination lists correctly.""" + # Call the federation client method with an empty destination list + result = self.get_success( + self.federation_client.search_user_directory_across_federation( + "@requester:example.com", [], 10 + ) + ) + + # Check that the result is an empty result set + self.assertEqual(result, {"limited": False, "results": []}) + + def test_search_user_directory_across_federation_server_error(self) -> None: + """Test that the federation client handles server errors correctly.""" + self.federation_client.transport_layer.user_directory_search = AsyncMock( # type: ignore[method-assign] + side_effect=HttpResponseException(500, "Internal Server Error", b"{}") + ) + + # Call the federation client method + result = self.get_success( + self.federation_client.search_user_directory_across_federation( + "@requester:example.com", + ["server1.example.com", "server2.example.com"], + 10, + ) + ) + + # Check that the result is an empty result set + self.assertEqual(result, {"limited": False, "results": []}) + + +class FederatedUserDirectorySyncTestCase(FederatingHomeserverTestCase): + """Tests the federation-side periodic user directory sync background job.""" + + servlets = [ + admin.register_servlets, + login.register_servlets, + register.register_servlets, + user_directory.register_servlets, + ] + + def make_homeserver(self, reactor: MemoryReactor, clock: Clock) -> HomeServer: + config = self.default_config() + # Enable writing to the user directory on this process. + config["update_user_directory_from_worker"] = None + config["user_directory"] = {"enabled": True, "search_all_users": True} + config["experimental_features"] = { + "bwi_federated_user_dir_enabled": True, + } + return self.setup_test_homeserver(config=config) + + def prepare( + self, reactor: MemoryReactor, clock: Clock, homeserver: HomeServer + ) -> None: + self.store = homeserver.get_datastores().main + self.federation_client = homeserver.get_federation_client() + self.user_dir_helper = GetUserDirectoryTables(self.store) + + def test_sync_uses_db_destinations_and_upserts_remote_users(self) -> None: + # Record a known destination in the DB. + self.get_success( + self.store.set_destination_retry_timings("remote.example.com", None, 0, 0) + ) + + self.federation_client.user_directory_search = AsyncMock( # type: ignore[method-assign] + return_value={ + "limited": False, + "results": [ + { + "user_id": "@bob:remote.example.com", + "display_name": "Bob Remote", + "avatar_url": None, + } + ], + } + ) + + self.get_success(self.federation_client._sync_federated_user_directory()) + + self.federation_client.user_directory_search.assert_called_once_with( + "@_user_directory_sync:test", + "remote.example.com", + self.hs.config.experimental.bwi_federated_user_dir_federation_search_timeout, + ) + + profiles = self.get_success( + self.user_dir_helper.get_profiles_in_user_directory() + ) + self.assertIn("@bob:remote.example.com", profiles) + + def test_sync_skips_own_server(self) -> None: + # Our own server name should never be queried, even if present in the DB. + self.get_success(self.store.set_destination_retry_timings("test", None, 0, 0)) + + self.federation_client.user_directory_search = AsyncMock() # type: ignore[method-assign] + + self.get_success(self.federation_client._sync_federated_user_directory()) + + self.federation_client.user_directory_search.assert_not_called() diff --git a/tests/federation/test_federation_server.py b/tests/federation/test_federation_server.py index 1bf0817bd24..2e999d34027 100644 --- a/tests/federation/test_federation_server.py +++ b/tests/federation/test_federation_server.py @@ -55,6 +55,19 @@ class FederationServerTests(unittest.FederatingHomeserverTestCase): login.register_servlets, ] + def default_config(self) -> JsonDict: + config = super().default_config() + config["user_directory"] = { + "enabled": True, + "search_all_users": True, + } + # The federation user directory search responder is only registered when + # the experimental feature is enabled. + config["experimental_features"] = { + "bwi_federated_user_dir_enabled": True, + } + return config + @parameterized.expand([(b"",), (b"foo",), (b'{"limit": Infinity}',)]) def test_bad_request(self, query_content: bytes) -> None: """ @@ -93,6 +106,60 @@ async def failing_handler(_origin: str, _content: JsonDict) -> None: ) self.assertEqual(500, channel.code, channel.result) + def test_federation_user_directory_search_servlet(self) -> None: + """Test that the federation user directory search servlet works correctly.""" + self.register_user("userlambda", "password") + + # Make a request to the servlet + channel = self.make_signed_federation_request( + "POST", + "/_matrix/federation/unstable/de.bwi.federated_user_dir/user_directory/search", + content={ + "requester": "@requester:other.example.com", + }, + ) + + # Check that the response is correct + self.assertEqual(channel.code, 200) + self.assertEqual(channel.json_body.get("limited", None), False) + results = channel.json_body.get("results", []) + self.assertEqual(len(results), 1) + self.assertEqual(results[0].get("user_id"), "@userlambda:test") + + def test_federation_user_directory_search_servlet_invalid_request(self) -> None: + """Test that the federation user directory search servlet rejects invalid requests.""" + self.register_user("user", "password") + + # Make a request with a missing requester + channel = self.make_signed_federation_request( + "POST", + "/_matrix/federation/unstable/de.bwi.federated_user_dir/user_directory/search", + content={}, + ) + + # Check that the response is an error + self.assertEqual(channel.code, 400) + self.assertEqual(channel.json_body["errcode"], "M_BAD_JSON") + + def test_federation_user_directory_search_servlet_no_results(self) -> None: + """An empty local directory yields no results.""" + # No local users are registered, so the directory is empty. + + # Make a request to the servlet + channel = self.make_signed_federation_request( + "POST", + "/_matrix/federation/unstable/de.bwi.federated_user_dir/user_directory/search", + content={ + "requester": "@requester:other.example.com", + }, + ) + + # Check that the response is correct + self.assertEqual(channel.code, 200) + self.assertEqual(channel.json_body.get("limited", None), False) + results = channel.json_body.get("results", []) + self.assertEqual(len(results), 0) + def _create_acl_event(content: JsonDict) -> EventBase: return make_test_event( diff --git a/tests/handlers/test_user_directory.py b/tests/handlers/test_user_directory.py index f50fa1f4a02..a2c014a614e 100644 --- a/tests/handlers/test_user_directory.py +++ b/tests/handlers/test_user_directory.py @@ -28,10 +28,17 @@ from synapse.api.errors import SynapseError from synapse.api.room_versions import RoomVersion, RoomVersions from synapse.appservice import ApplicationService +from synapse.handlers.user_directory import UserDirectoryFederationHandler from synapse.rest.client import login, register, room, user_directory from synapse.server import HomeServer from synapse.storage.roommember import ProfileInfo -from synapse.types import JsonDict, UserID, UserProfile, create_requester +from synapse.types import ( + JsonDict, + RemoteUserDirectoryEntry, + UserID, + UserProfile, + create_requester, +) from synapse.util.clock import Clock from tests import unittest @@ -1433,3 +1440,111 @@ async def get_remote_profile( display_name="Sir Bruce Bruceson", avatar_url="mxc://remote/789" ), ) + + +class FederatedUserDirectoryHandlerTestCase(unittest.HomeserverTestCase): + """Tests for ingesting remote users into the user directory. + + The handler is intentionally unaware of federation; it only persists + already-parsed `RemoteUserDirectoryEntry` objects. + """ + + servlets = [ + login.register_servlets, + register.register_servlets, + user_directory.register_servlets, + ] + + def make_homeserver(self, reactor: MemoryReactor, clock: Clock) -> HomeServer: + config = self.default_config() + # Enable updating the user directory on this (main) process. + config["update_user_directory_from_worker"] = None + # Enabling the feature makes get_user_directory_handler() return the + # federation-aware subclass. + config["experimental_features"] = { + "bwi_federated_user_dir_enabled": True, + } + return self.setup_test_homeserver(config=config) + + def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None: + self.store = hs.get_datastores().main + handler = hs.get_user_directory_handler() + # Enabling the feature makes the getter return the federation subclass. + assert isinstance(handler, UserDirectoryFederationHandler) + self.handler = handler + self.user_dir_helper = GetUserDirectoryTables(self.store) + + def test_upsert_remote_users_persists_profiles_and_visibility(self) -> None: + self.get_success( + self.handler.upsert_remote_users( + [ + RemoteUserDirectoryEntry( + user_id="@alice:remote.example.com", + display_name="Alice Remote", + avatar_url="mxc://remote.example.com/abc", + ) + ] + ) + ) + + profiles = self.get_success( + self.user_dir_helper.get_profiles_in_user_directory() + ) + self.assertEqual( + profiles.get("@alice:remote.example.com"), + ProfileInfo( + display_name="Alice Remote", + avatar_url="mxc://remote.example.com/abc", + ), + ) + + public_rooms = self.get_success( + self.user_dir_helper.get_users_in_public_rooms() + ) + self.assertIn( + ( + "@alice:remote.example.com", + self.handler._federated_cache_room_id, + ), + public_rooms, + ) + + def test_upsert_remote_users_ignores_local_users(self) -> None: + self.get_success( + self.handler.upsert_remote_users( + [ + RemoteUserDirectoryEntry( + user_id="@localuser:test", + display_name="Local User", + avatar_url=None, + ) + ] + ) + ) + + profiles = self.get_success( + self.user_dir_helper.get_profiles_in_user_directory() + ) + self.assertNotIn("@localuser:test", profiles) + + def test_search_users_returns_cached_remote_users(self) -> None: + self.get_success( + self.handler.upsert_remote_users( + [ + RemoteUserDirectoryEntry( + user_id="@carol:remote.example.com", + display_name="Carol Remote", + avatar_url=None, + ) + ] + ) + ) + + results = self.get_success( + self.handler.search_users("@searcher:test", "carol", 10) + ) + + self.assertIn( + "@carol:remote.example.com", + {user["user_id"] for user in results["results"]}, + ) From f90ee72c25837af3a2ce10349280c3510c006ffb Mon Sep 17 00:00:00 2001 From: "Kai A. Hiller" Date: Mon, 29 Jun 2026 12:17:34 +0200 Subject: [PATCH 02/11] Add newsfragment --- changelog.d/19887.feature | 1 + 1 file changed, 1 insertion(+) create mode 100644 changelog.d/19887.feature diff --git a/changelog.d/19887.feature b/changelog.d/19887.feature new file mode 100644 index 00000000000..577b5bcf469 --- /dev/null +++ b/changelog.d/19887.feature @@ -0,0 +1 @@ +Pre-MSC implementation of a federated user directory. \ No newline at end of file From 0d936e85b58bad483cc2319a7bbf817ab6697838 Mon Sep 17 00:00:00 2001 From: Joshua Fenster Date: Mon, 6 Jul 2026 14:11:13 +0200 Subject: [PATCH 03/11] adds missing prase_duration --- synapse/config/experimental.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/synapse/config/experimental.py b/synapse/config/experimental.py index c826215b5b7..bfb71e6ddfd 100644 --- a/synapse/config/experimental.py +++ b/synapse/config/experimental.py @@ -631,8 +631,8 @@ def read_config( "bwi_federated_user_dir_enabled", False ) - self.bwi_federated_user_dir_federation_search_timeout: int = experimental.get( - "bwi_federated_user_dir_federation_search_timeout", 2000 + 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( From 50f65b8fcb0bc4c332ded4b5baa0b49c78dfa689 Mon Sep 17 00:00:00 2001 From: Joshua Fenster Date: Mon, 6 Jul 2026 15:13:52 +0200 Subject: [PATCH 04/11] =?UTF-8?q?removes=C2=A0synthetic=20requester=20from?= =?UTF-8?q?=20server=20and=20client=20side=20(updates=20tests).?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- synapse/config/experimental.py | 8 +++++-- synapse/federation/federation_client.py | 18 +-------------- synapse/federation/federation_server.py | 3 +-- synapse/federation/transport/client.py | 5 +--- .../federation/transport/server/federation.py | 12 +++------- tests/federation/test_federation_client.py | 22 +++++------------- tests/federation/test_federation_server.py | 23 ++----------------- 7 files changed, 20 insertions(+), 71 deletions(-) diff --git a/synapse/config/experimental.py b/synapse/config/experimental.py index bfb71e6ddfd..2c838ba90a9 100644 --- a/synapse/config/experimental.py +++ b/synapse/config/experimental.py @@ -631,8 +631,12 @@ def read_config( "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_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( diff --git a/synapse/federation/federation_client.py b/synapse/federation/federation_client.py index 02bb4ac22d7..989d2da0558 100644 --- a/synapse/federation/federation_client.py +++ b/synapse/federation/federation_client.py @@ -100,10 +100,6 @@ PDU_RETRY_TIME_MS = 1 * 60 * 1000 -# Localpart of the synthetic requester MXID used when querying remote homeservers -# during the periodic federated user directory sync. -FEDERATED_USER_DIR_SYNC_REQUESTER_LOCALPART = "_user_directory_sync" - T = TypeVar("T") @@ -189,12 +185,6 @@ def __init__(self, hs: "HomeServer"): reset_expiry_on_get=False, ) - # Synthetic requester used for the federated user directory sync. The - # remote server requires the requester to belong to our server. - self._federated_user_dir_sync_requester = ( - f"@{FEDERATED_USER_DIR_SYNC_REQUESTER_LOCALPART}:{self.server_name}" - ) - # Periodically sync remote homeservers' user directories into our own, # but only on the worker that runs background tasks. if ( @@ -1956,7 +1946,6 @@ def filter_user_id(user_id: str) -> bool: async def user_directory_search( self, - requester: str, destination: str, timeout: int, ) -> JsonDict: @@ -1966,7 +1955,6 @@ async def user_directory_search( directory, so no result limit is sent. Args: - requester: The user that initiated the request. destination: The server to query. timeout: Timeout in milliseconds for the request. @@ -1975,7 +1963,7 @@ async def user_directory_search( """ try: response = await self.transport_layer.user_directory_search( - requester, destination, timeout + destination, timeout ) return response except Exception as e: @@ -1989,14 +1977,12 @@ async def user_directory_search( async def search_user_directory_across_federation( self, - requester: str, destinations: Collection[str], limit: int = 10, ) -> JsonDict: """Fetch users from the directories of multiple federated servers. Args: - requester: The user that initiated the request. destinations: The servers to query. limit: Maximum number of results to return per server. @@ -2018,7 +2004,6 @@ async def search_user_directory_across_federation( # Convert coroutine to Deferred deferred = defer.ensureDeferred( self.user_directory_search( - requester, destination, self.user_directory_search_timeout, ) @@ -2105,7 +2090,6 @@ async def _sync_federated_user_directory(self) -> None: continue response = await self.user_directory_search( - self._federated_user_dir_sync_requester, destination, self.user_directory_search_timeout, ) diff --git a/synapse/federation/federation_server.py b/synapse/federation/federation_server.py index b86bb2e7251..41938cc186d 100644 --- a/synapse/federation/federation_server.py +++ b/synapse/federation/federation_server.py @@ -1484,7 +1484,7 @@ 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, requester_id: str, origin: str + self, origin: str ) -> tuple[int, JsonMapping]: """Handle a user directory request from a remote server. @@ -1492,7 +1492,6 @@ async def on_user_directory_search_request( always syncs the full local directory rather than matching a term. Args: - requester_id: The user ID of the requester on the origin server. origin: The server that sent the request. Returns: diff --git a/synapse/federation/transport/client.py b/synapse/federation/transport/client.py index 6fb676584b2..ad796ecbf8a 100644 --- a/synapse/federation/transport/client.py +++ b/synapse/federation/transport/client.py @@ -855,7 +855,6 @@ async def get_account_status( async def user_directory_search( self, - requester: str, destination: str, timeout: int, ) -> JsonDict: @@ -866,7 +865,6 @@ async def user_directory_search( directory. Args: - requester: The user that initiated the request. destination: The server to query. timeout: timeout in milliseconds to get the response from destination. @@ -877,11 +875,10 @@ async def user_directory_search( FEDERATION_UNSTABLE_PREFIX, "/de.bwi.federated_user_dir" + "/user_directory/search", ) - content = {"requester": requester} return await self.client.post_json( destination, path=path, - data=content, + data={}, # ignore backoff for user search as we will set a small user_directory_search_timeout ignore_backoff=True, timeout=timeout, diff --git a/synapse/federation/transport/server/federation.py b/synapse/federation/transport/server/federation.py index 78994c3071c..a0dcd52950f 100644 --- a/synapse/federation/transport/server/federation.py +++ b/synapse/federation/transport/server/federation.py @@ -47,7 +47,7 @@ from synapse.http.site import SynapseRequest from synapse.media._base import DEFAULT_MAX_TIMEOUT_MS, MAXIMUM_ALLOWED_MAX_TIMEOUT_MS from synapse.media.thumbnailer import ThumbnailProvider -from synapse.types import JsonDict, JsonMapping, get_domain_from_id +from synapse.types import JsonDict, JsonMapping from synapse.util import SYNAPSE_VERSION from synapse.util.ratelimitutils import FederationRateLimiter @@ -898,9 +898,7 @@ class FederationUserDirectorySearchServlet(BaseFederationServerServlet): Implements a federation API endpoint for searching a server's user directory. POST /_matrix/federation/v3/user_directory/search Request: - { - "requester": "@user:example.com" - } + {} Response: { "limited": false, @@ -922,11 +920,7 @@ class FederationUserDirectorySearchServlet(BaseFederationServerServlet): async def on_POST( self, origin: str, content: JsonDict, query: dict[bytes, list[bytes]] ) -> tuple[int, JsonMapping]: - requester = content.get("requester") - if requester is None or get_domain_from_id(requester) != origin: - raise SynapseError(400, "Missing or invalid requester", Codes.BAD_JSON) - - return await self.handler.on_user_directory_search_request(requester, origin) + return await self.handler.on_user_directory_search_request(origin) FEDERATION_SERVLET_CLASSES: tuple[type[BaseFederationServlet], ...] = ( diff --git a/tests/federation/test_federation_client.py b/tests/federation/test_federation_client.py index f93ec34cbd4..cac3fa39b36 100644 --- a/tests/federation/test_federation_client.py +++ b/tests/federation/test_federation_client.py @@ -354,9 +354,7 @@ def test_user_directory_search(self) -> None: # Call the federation client method result = self.get_success( - self.federation_client.user_directory_search( - "@requester:example.com", "other.example.com", 2000 - ) + self.federation_client.user_directory_search("other.example.com", 2000) ) # Check that the result is correct @@ -364,7 +362,7 @@ def test_user_directory_search(self) -> None: # Check that user_directory_search was called with the correct arguments self.transport_layer.user_directory_search.assert_called_once_with( - "@requester:example.com", "other.example.com", 2000 + "other.example.com", 2000 ) def test_user_directory_search_endpoint_not_found(self) -> None: @@ -378,9 +376,7 @@ def test_user_directory_search_endpoint_not_found(self) -> None: # Call the federation client method result = self.get_success( - self.federation_client.user_directory_search( - "@requester:example.com", "other.example.com", 10 - ) + self.federation_client.user_directory_search("other.example.com", 10) ) # Check that the result is an empty result set @@ -391,7 +387,7 @@ def test_search_user_directory_across_federation(self) -> None: # Mock the user_directory_search method to return different results for different servers async def mock_user_directory_search( - requester: str, destination: str, timeout: int + destination: str, timeout: int ) -> JsonDict: if destination == "server1.example.com": return { @@ -425,7 +421,6 @@ async def mock_user_directory_search( # Call the federation client method result = self.get_success( self.federation_client.search_user_directory_across_federation( - "@requester:example.com", ["server1.example.com", "server2.example.com"], 10, ) @@ -456,7 +451,7 @@ def test_search_user_directory_across_federation_with_limit(self) -> None: # Mock the user_directory_search method to return many results async def mock_user_directory_search( - requester: str, destination: str, timeout: int + destination: str, timeout: int ) -> JsonDict: return { "limited": False, @@ -477,7 +472,6 @@ async def mock_user_directory_search( # Call the federation client method with a limit of 5 result = self.get_success( self.federation_client.search_user_directory_across_federation( - "@requester:example.com", ["server1.example.com", "server2.example.com"], 5, ) @@ -491,9 +485,7 @@ def test_search_user_directory_across_federation_empty_destinations(self) -> Non """Test that the federation client handles empty destination lists correctly.""" # Call the federation client method with an empty destination list result = self.get_success( - self.federation_client.search_user_directory_across_federation( - "@requester:example.com", [], 10 - ) + self.federation_client.search_user_directory_across_federation([], 10) ) # Check that the result is an empty result set @@ -508,7 +500,6 @@ def test_search_user_directory_across_federation_server_error(self) -> None: # Call the federation client method result = self.get_success( self.federation_client.search_user_directory_across_federation( - "@requester:example.com", ["server1.example.com", "server2.example.com"], 10, ) @@ -567,7 +558,6 @@ def test_sync_uses_db_destinations_and_upserts_remote_users(self) -> None: self.get_success(self.federation_client._sync_federated_user_directory()) self.federation_client.user_directory_search.assert_called_once_with( - "@_user_directory_sync:test", "remote.example.com", self.hs.config.experimental.bwi_federated_user_dir_federation_search_timeout, ) diff --git a/tests/federation/test_federation_server.py b/tests/federation/test_federation_server.py index 2e999d34027..97645811cd1 100644 --- a/tests/federation/test_federation_server.py +++ b/tests/federation/test_federation_server.py @@ -114,9 +114,7 @@ def test_federation_user_directory_search_servlet(self) -> None: channel = self.make_signed_federation_request( "POST", "/_matrix/federation/unstable/de.bwi.federated_user_dir/user_directory/search", - content={ - "requester": "@requester:other.example.com", - }, + content={}, ) # Check that the response is correct @@ -126,21 +124,6 @@ def test_federation_user_directory_search_servlet(self) -> None: self.assertEqual(len(results), 1) self.assertEqual(results[0].get("user_id"), "@userlambda:test") - def test_federation_user_directory_search_servlet_invalid_request(self) -> None: - """Test that the federation user directory search servlet rejects invalid requests.""" - self.register_user("user", "password") - - # Make a request with a missing requester - channel = self.make_signed_federation_request( - "POST", - "/_matrix/federation/unstable/de.bwi.federated_user_dir/user_directory/search", - content={}, - ) - - # Check that the response is an error - self.assertEqual(channel.code, 400) - self.assertEqual(channel.json_body["errcode"], "M_BAD_JSON") - def test_federation_user_directory_search_servlet_no_results(self) -> None: """An empty local directory yields no results.""" # No local users are registered, so the directory is empty. @@ -149,9 +132,7 @@ def test_federation_user_directory_search_servlet_no_results(self) -> None: channel = self.make_signed_federation_request( "POST", "/_matrix/federation/unstable/de.bwi.federated_user_dir/user_directory/search", - content={ - "requester": "@requester:other.example.com", - }, + content={}, ) # Check that the response is correct From 46a51686d503ecec167a63611a7aac07eb517e0d Mon Sep 17 00:00:00 2001 From: Joshua Fenster Date: Mon, 6 Jul 2026 15:23:06 +0200 Subject: [PATCH 05/11] removes m.user_directory.visibility from docs --- synapse/federation/transport/server/federation.py | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/synapse/federation/transport/server/federation.py b/synapse/federation/transport/server/federation.py index a0dcd52950f..135a7a74cba 100644 --- a/synapse/federation/transport/server/federation.py +++ b/synapse/federation/transport/server/federation.py @@ -906,8 +906,7 @@ class FederationUserDirectorySearchServlet(BaseFederationServerServlet): { "user_id": "@user:example.com", "display_name": "Display Name", - "avatar_url": "mxc://example.com/avatar", - "m.user_directory.visibility": "local" + "avatar_url": "mxc://example.com/avatar" } ] } From 4f69ad8975097289945a147991b961d548be69c1 Mon Sep 17 00:00:00 2001 From: Joshua Fenster Date: Mon, 6 Jul 2026 15:34:08 +0200 Subject: [PATCH 06/11] removes orphan search_user_directory_across_federation --- synapse/federation/federation_client.py | 60 ---------- tests/federation/test_federation_client.py | 127 --------------------- 2 files changed, 187 deletions(-) diff --git a/synapse/federation/federation_client.py b/synapse/federation/federation_client.py index 989d2da0558..15ef9f4a4b2 100644 --- a/synapse/federation/federation_client.py +++ b/synapse/federation/federation_client.py @@ -42,8 +42,6 @@ import attr from prometheus_client import Counter -from twisted.internet import defer - from synapse.api.constants import Direction, EventContentFields, EventTypes, Membership from synapse.api.errors import ( CodeMessageException, @@ -71,7 +69,6 @@ from synapse.federation.transport.client import SendJoinResponse from synapse.http.client import is_unknown_endpoint from synapse.http.types import QueryParams -from synapse.logging.context import make_deferred_yieldable from synapse.logging.opentracing import SynapseTags, log_kv, set_tag, tag_args, trace from synapse.metrics import SERVER_NAME_LABEL from synapse.metrics.background_process_metrics import wrap_as_background_process @@ -1975,63 +1972,6 @@ async def user_directory_search( ) return {"limited": False, "results": []} - async def search_user_directory_across_federation( - self, - destinations: Collection[str], - limit: int = 10, - ) -> JsonDict: - """Fetch users from the directories of multiple federated servers. - - Args: - destinations: The servers to query. - limit: Maximum number of results to return per server. - - Returns: - Combined results from all servers. - """ - - if not destinations: - return {"limited": False, "results": []} - - # Query each server individually and collect results - combined_results = [] - limited = False - - # Create a list of deferreds to query each server - query_tasks = [] - for destination in destinations: - if not self._is_mine_server_name(destination): - # Convert coroutine to Deferred - deferred = defer.ensureDeferred( - self.user_directory_search( - destination, - self.user_directory_search_timeout, - ) - ) - query_tasks.append(deferred) - - # Execute all queries in parallel - if query_tasks: - server_results = await make_deferred_yieldable( - defer.gatherResults( - query_tasks, - consumeErrors=True, - ) - ) - - # Process results from each server - for result in server_results: - if result.get("limited", False): - limited = True - combined_results.extend(result.get("results", [])) - - # Limit the total number of results - if len(combined_results) > limit: - combined_results = combined_results[:limit] - limited = True - - return {"limited": limited, "results": combined_results} - @staticmethod def _parse_remote_user_directory_results( response: JsonDict, diff --git a/tests/federation/test_federation_client.py b/tests/federation/test_federation_client.py index cac3fa39b36..e64c68a9916 100644 --- a/tests/federation/test_federation_client.py +++ b/tests/federation/test_federation_client.py @@ -32,7 +32,6 @@ from synapse.rest import admin from synapse.rest.client import login, register, room, user_directory from synapse.server import HomeServer -from synapse.types import JsonDict from synapse.util.clock import Clock from tests.storage.test_user_directory import GetUserDirectoryTables @@ -382,132 +381,6 @@ def test_user_directory_search_endpoint_not_found(self) -> None: # Check that the result is an empty result set self.assertEqual(result, {"limited": False, "results": []}) - def test_search_user_directory_across_federation(self) -> None: - """Test that the federation client correctly handles searching across multiple servers.""" - - # Mock the user_directory_search method to return different results for different servers - async def mock_user_directory_search( - destination: str, timeout: int - ) -> JsonDict: - if destination == "server1.example.com": - return { - "limited": False, - "results": [ - { - "user_id": "@user1:server1.example.com", - "display_name": "User 1", - "avatar_url": "mxc://example.com/avatar1", - } - ], - } - elif destination == "server2.example.com": - return { - "limited": False, - "results": [ - { - "user_id": "@user2:server2.example.com", - "display_name": "User 2", - "avatar_url": "mxc://example.com/avatar2", - } - ], - } - else: - return {"limited": False, "results": []} - - self.federation_client.user_directory_search = AsyncMock( # type: ignore[method-assign] - side_effect=mock_user_directory_search - ) - - # Call the federation client method - result = self.get_success( - self.federation_client.search_user_directory_across_federation( - ["server1.example.com", "server2.example.com"], - 10, - ) - ) - - # Check that the result contains results from both servers - self.assertEqual( - result, - { - "limited": False, - "results": [ - { - "user_id": "@user1:server1.example.com", - "display_name": "User 1", - "avatar_url": "mxc://example.com/avatar1", - }, - { - "user_id": "@user2:server2.example.com", - "display_name": "User 2", - "avatar_url": "mxc://example.com/avatar2", - }, - ], - }, - ) - - def test_search_user_directory_across_federation_with_limit(self) -> None: - """Test that the federation client correctly applies limits when searching across multiple servers.""" - - # Mock the user_directory_search method to return many results - async def mock_user_directory_search( - destination: str, timeout: int - ) -> JsonDict: - return { - "limited": False, - "results": [ - { - "user_id": f"@user{i}:{destination}", - "display_name": f"User {i}", - "avatar_url": f"mxc://example.com/avatar{i}", - } - for i in range(10) - ], - } - - self.federation_client.user_directory_search = AsyncMock( # type: ignore[method-assign] - side_effect=mock_user_directory_search - ) - - # Call the federation client method with a limit of 5 - result = self.get_success( - self.federation_client.search_user_directory_across_federation( - ["server1.example.com", "server2.example.com"], - 5, - ) - ) - - # Check that the result is limited to 5 users - self.assertEqual(len(result["results"]), 5) - self.assertTrue(result["limited"]) - - def test_search_user_directory_across_federation_empty_destinations(self) -> None: - """Test that the federation client handles empty destination lists correctly.""" - # Call the federation client method with an empty destination list - result = self.get_success( - self.federation_client.search_user_directory_across_federation([], 10) - ) - - # Check that the result is an empty result set - self.assertEqual(result, {"limited": False, "results": []}) - - def test_search_user_directory_across_federation_server_error(self) -> None: - """Test that the federation client handles server errors correctly.""" - self.federation_client.transport_layer.user_directory_search = AsyncMock( # type: ignore[method-assign] - side_effect=HttpResponseException(500, "Internal Server Error", b"{}") - ) - - # Call the federation client method - result = self.get_success( - self.federation_client.search_user_directory_across_federation( - ["server1.example.com", "server2.example.com"], - 10, - ) - ) - - # Check that the result is an empty result set - self.assertEqual(result, {"limited": False, "results": []}) - class FederatedUserDirectorySyncTestCase(FederatingHomeserverTestCase): """Tests the federation-side periodic user directory sync background job.""" From 36102dc1a402d18bc941e3b6897e4f3973ddc551 Mon Sep 17 00:00:00 2001 From: Joshua Fenster Date: Mon, 6 Jul 2026 15:41:03 +0200 Subject: [PATCH 07/11] Adds debug logging to _parse_remote_user_directory_results --- synapse/federation/federation_client.py | 25 +++++++++++++++++++++++-- 1 file changed, 23 insertions(+), 2 deletions(-) diff --git a/synapse/federation/federation_client.py b/synapse/federation/federation_client.py index 15ef9f4a4b2..67c1bf13db6 100644 --- a/synapse/federation/federation_client.py +++ b/synapse/federation/federation_client.py @@ -1984,18 +1984,39 @@ def _parse_remote_user_directory_results( 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( @@ -2019,7 +2040,7 @@ async def _sync_federated_user_directory(self) -> None: """ destinations = await self.store.get_known_destinations() if not destinations: - logger.debug("Federated user directory sync: no known destinations") + logger.debug("ending federated user directory sync: no known destinations") return # De-duplicate by user id across destinations. @@ -2048,7 +2069,7 @@ async def _sync_federated_user_directory(self) -> None: ) await handler.upsert_remote_users(list(entries_by_user.values())) - logger.info( + logger.debug( "Federated user directory sync upserted %d remote users", len(entries_by_user), ) From 4f51aa207e88c0989bbb84120b846f89775e5353 Mon Sep 17 00:00:00 2001 From: Joshua Fenster Date: Mon, 6 Jul 2026 16:17:05 +0200 Subject: [PATCH 08/11] Solves Ratelimiting and logging --- synapse/federation/federation_client.py | 21 +++++++++++++++------ 1 file changed, 15 insertions(+), 6 deletions(-) diff --git a/synapse/federation/federation_client.py b/synapse/federation/federation_client.py index 67c1bf13db6..c0d9eea9447 100644 --- a/synapse/federation/federation_client.py +++ b/synapse/federation/federation_client.py @@ -1959,18 +1959,27 @@ async def user_directory_search( The results containing a list of users from the remote directory. """ try: - response = await self.transport_layer.user_directory_search( + return await self.transport_layer.user_directory_search( destination, timeout ) - return response - except Exception as e: - # If something goes wrong, we still want to return what we have - logger.exception( - "Error searching user directory across federation[destination=%s] : %s", + 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 {"limited": False, "results": []} + except Exception: + # Unexpected error; log with a stack trace for debugging. + logger.exception( + "Unexpected error searching remote user directory [destination=%s]", + destination, + ) + return {"limited": False, "results": []} @staticmethod def _parse_remote_user_directory_results( From b341798a673d4cbc3161053f9a015c441544645a Mon Sep 17 00:00:00 2001 From: Joshua Fenster Date: Mon, 6 Jul 2026 16:34:42 +0200 Subject: [PATCH 09/11] Removes limited false response --- synapse/federation/federation_client.py | 4 ++-- synapse/federation/federation_server.py | 22 ++++++++----------- .../federation/transport/server/federation.py | 1 - tests/federation/test_federation_client.py | 4 +--- tests/federation/test_federation_server.py | 4 ++-- 5 files changed, 14 insertions(+), 21 deletions(-) diff --git a/synapse/federation/federation_client.py b/synapse/federation/federation_client.py index c0d9eea9447..2f8ade4aa0f 100644 --- a/synapse/federation/federation_client.py +++ b/synapse/federation/federation_client.py @@ -1972,14 +1972,14 @@ async def user_directory_search( destination, e, ) - return {"limited": False, "results": []} + 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 {"limited": False, "results": []} + return {"results": []} @staticmethod def _parse_remote_user_directory_results( diff --git a/synapse/federation/federation_server.py b/synapse/federation/federation_server.py index 41938cc186d..3bd93803fd9 100644 --- a/synapse/federation/federation_server.py +++ b/synapse/federation/federation_server.py @@ -1507,23 +1507,19 @@ async def _search_all_users(self) -> JsonDict: homeserver's own users (the table may also hold cached remote users). Returns: - A dict of the form ``{"limited": False, "results": [...]}``. + A dict of the form ``{"results": [...]}``. """ results = await self.store.get_users_in_user_dir() # Federation endpoint: only return users local to this homeserver. - filtered_results = [] - for user in results.get("results", []): - try: - if self.hs.is_mine_id(user["user_id"]): - filtered_results.append(user) - except SynapseError: - # Ignore malformed user IDs. - continue - - # The federation endpoint never truncates: it always returns the full - # local directory. - return {"limited": False, "results": filtered_results} + # is_mine_id is infallible and returns False for malformed user IDs. + filtered_results = [ + user + for user in results.get("results", []) + if self.hs.is_mine_id(user["user_id"]) + ] + + return {"results": filtered_results} class FederationHandlerRegistry: diff --git a/synapse/federation/transport/server/federation.py b/synapse/federation/transport/server/federation.py index 135a7a74cba..83e94d8087a 100644 --- a/synapse/federation/transport/server/federation.py +++ b/synapse/federation/transport/server/federation.py @@ -901,7 +901,6 @@ class FederationUserDirectorySearchServlet(BaseFederationServerServlet): {} Response: { - "limited": false, "results": [ { "user_id": "@user:example.com", diff --git a/tests/federation/test_federation_client.py b/tests/federation/test_federation_client.py index e64c68a9916..c5d7cb6773e 100644 --- a/tests/federation/test_federation_client.py +++ b/tests/federation/test_federation_client.py @@ -338,7 +338,6 @@ def test_user_directory_search(self) -> None: """Test that the federation client correctly handles user directory search requests.""" # Mock the transport layer's user_directory_search method mock_results = { - "limited": False, "results": [ { "user_id": "@user:other.example.com", @@ -379,7 +378,7 @@ def test_user_directory_search_endpoint_not_found(self) -> None: ) # Check that the result is an empty result set - self.assertEqual(result, {"limited": False, "results": []}) + self.assertEqual(result, {"results": []}) class FederatedUserDirectorySyncTestCase(FederatingHomeserverTestCase): @@ -417,7 +416,6 @@ def test_sync_uses_db_destinations_and_upserts_remote_users(self) -> None: self.federation_client.user_directory_search = AsyncMock( # type: ignore[method-assign] return_value={ - "limited": False, "results": [ { "user_id": "@bob:remote.example.com", diff --git a/tests/federation/test_federation_server.py b/tests/federation/test_federation_server.py index 97645811cd1..6e8c686b6d6 100644 --- a/tests/federation/test_federation_server.py +++ b/tests/federation/test_federation_server.py @@ -119,7 +119,7 @@ def test_federation_user_directory_search_servlet(self) -> None: # Check that the response is correct self.assertEqual(channel.code, 200) - self.assertEqual(channel.json_body.get("limited", None), False) + self.assertNotIn("limited", channel.json_body) results = channel.json_body.get("results", []) self.assertEqual(len(results), 1) self.assertEqual(results[0].get("user_id"), "@userlambda:test") @@ -137,7 +137,7 @@ def test_federation_user_directory_search_servlet_no_results(self) -> None: # Check that the response is correct self.assertEqual(channel.code, 200) - self.assertEqual(channel.json_body.get("limited", None), False) + self.assertNotIn("limited", channel.json_body) results = channel.json_body.get("results", []) self.assertEqual(len(results), 0) From 0dcf17bca19374c2976ba25c839673740a80c87c Mon Sep 17 00:00:00 2001 From: Joshua Fenster Date: Mon, 6 Jul 2026 16:59:54 +0200 Subject: [PATCH 10/11] Omit optional fields entirely when unset rather than sending --- synapse/federation/federation_server.py | 18 +++++++++++++----- .../storage/databases/main/user_directory.py | 6 +++--- 2 files changed, 16 insertions(+), 8 deletions(-) diff --git a/synapse/federation/federation_server.py b/synapse/federation/federation_server.py index 3bd93803fd9..79fb1c01a83 100644 --- a/synapse/federation/federation_server.py +++ b/synapse/federation/federation_server.py @@ -1513,11 +1513,19 @@ async def _search_all_users(self) -> JsonDict: # Federation endpoint: only return users local to this homeserver. # is_mine_id is infallible and returns False for malformed user IDs. - filtered_results = [ - user - for user in results.get("results", []) - if self.hs.is_mine_id(user["user_id"]) - ] + 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} diff --git a/synapse/storage/databases/main/user_directory.py b/synapse/storage/databases/main/user_directory.py index f2fc74f14c1..637958c300c 100644 --- a/synapse/storage/databases/main/user_directory.py +++ b/synapse/storage/databases/main/user_directory.py @@ -1193,13 +1193,13 @@ async def search_user_dir( async def get_users_in_user_dir(self) -> SearchResult: """Get every user stored in the user directory. - Unlike :meth:`search_user_dir`, this does not match a search term: it - returns all profiles in the ``user_directory`` table. It is used by the + Unlike `search_user_dir`, this does not match a search term: it + returns all profiles in the `user_directory` table. It is used by the federation responder to hand a server's full local directory to a remote homeserver. Returns: - A ``SearchResult`` of the form:: + A `SearchResult` of the form: { "limited": False, # always False; no term/limit is applied From 118be0cf5ee77a0992112cc47770e16f34902ffe Mon Sep 17 00:00:00 2001 From: Gordan Trevis Date: Mon, 6 Jul 2026 19:21:40 +0200 Subject: [PATCH 11/11] Apply suggestions from code review Co-authored-by: Andrew Morgan <1342360+anoadragon453@users.noreply.github.com> --- synapse/federation/federation_client.py | 5 +---- 1 file changed, 1 insertion(+), 4 deletions(-) diff --git a/synapse/federation/federation_client.py b/synapse/federation/federation_client.py index 2f8ade4aa0f..bb98e149d67 100644 --- a/synapse/federation/federation_client.py +++ b/synapse/federation/federation_client.py @@ -1946,10 +1946,7 @@ async def user_directory_search( 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, so no result limit is sent. + """Fetch the full user directory of a remote server. Args: destination: The server to query.