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 diff --git a/synapse/config/experimental.py b/synapse/config/experimental.py index 2d74e9b8866..2c838ba90a9 100644 --- a/synapse/config/experimental.py +++ b/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" + ) diff --git a/synapse/federation/federation_client.py b/synapse/federation/federation_client.py index 5b334543256..bb98e149d67 100644 --- a/synapse/federation/federation_client.py +++ b/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, diff --git a/synapse/federation/federation_server.py b/synapse/federation/federation_server.py index 2fc78c012e9..79fb1c01a83 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,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 diff --git a/synapse/federation/transport/client.py b/synapse/federation/transport/client.py index 5d5212ef96c..ad796ecbf8a 100644 --- a/synapse/federation/transport/client.py +++ b/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, 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..83e94d8087a 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 from synapse.util import SYNAPSE_VERSION from synapse.util.ratelimitutils import FederationRateLimiter @@ -893,6 +893,34 @@ 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: + {} + Response: + { + "results": [ + { + "user_id": "@user:example.com", + "display_name": "Display Name", + "avatar_url": "mxc://example.com/avatar" + } + ] + } + """ + + 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]: + return await self.handler.on_user_directory_search_request(origin) + + FEDERATION_SERVLET_CLASSES: tuple[type[BaseFederationServlet], ...] = ( FederationSendServlet, FederationEventServlet, @@ -925,4 +953,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..637958c300c 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 `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..c5d7cb6773e 100644 --- a/tests/federation/test_federation_client.py +++ b/tests/federation/test_federation_client.py @@ -20,18 +20,21 @@ # 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.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 +44,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 +73,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 +333,117 @@ 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 = { + "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("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( + "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("other.example.com", 10) + ) + + # Check that the result is an empty result set + self.assertEqual(result, {"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={ + "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( + "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..6e8c686b6d6 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,41 @@ 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={}, + ) + + # Check that the response is correct + self.assertEqual(channel.code, 200) + 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") + + 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={}, + ) + + # Check that the response is correct + self.assertEqual(channel.code, 200) + self.assertNotIn("limited", channel.json_body) + 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"]}, + )