changelog

Downgrade 'Closing scope' error to warning; prevent hammering sentry
2022-08-18 11:58:38 +01:00 · 2022-08-18 11:46:54 +01:00
98 changed files with 686 additions and 1073 deletions
@@ -1,84 +1,3 @@
-Synapse 1.66.0rc1 (2022-08-23)
-==============================
-
-This release removes the ability for homeservers to delegate email ownership
-verification and password reset confirmation to identity servers. This removal
-was originally planned for Synapse 1.64, but was later deferred until now.
-
-See the [upgrade notes](https://matrix-org.github.io/synapse/v1.66/upgrade.html#upgrading-to-v1660) for more details.
-
-Features
--------
-
- Improve validation of request bodies for the following client-server API endpoints: [`/account/password`](https://spec.matrix.org/v1.3/client-server-api/#post_matrixclientv3accountpassword), [`/account/password/email/requestToken`](https://spec.matrix.org/v1.3/client-server-api/#post_matrixclientv3accountpasswordemailrequesttoken), [`/account/deactivate`](https://spec.matrix.org/v1.3/client-server-api/#post_matrixclientv3accountdeactivate) and [`/account/3pid/email/requestToken`](https://spec.matrix.org/v1.3/client-server-api/#post_matrixclientv3account3pidemailrequesttoken). ([\#13188](https://github.com/matrix-org/synapse/issues/13188), [\#13563](https://github.com/matrix-org/synapse/issues/13563))
- Add forgotten status to [Room Details Admin API](https://matrix-org.github.io/synapse/latest/admin_api/rooms.html#room-details-api). ([\#13503](https://github.com/matrix-org/synapse/issues/13503))
- Add an experimental implementation for [MSC3852 (Expose user agents on `Device`)](https://github.com/matrix-org/matrix-spec-proposals/pull/3852). ([\#13549](https://github.com/matrix-org/synapse/issues/13549))
- Add `org.matrix.msc2716v4` experimental room version with updated content fields. Part of [MSC2716 (Importing history)](https://github.com/matrix-org/matrix-spec-proposals/pull/2716).  ([\#13551](https://github.com/matrix-org/synapse/issues/13551))
- Add support for compression to federation responses. ([\#13537](https://github.com/matrix-org/synapse/issues/13537))
- Improve performance of sending messages in rooms with thousands of local users. ([\#13522](https://github.com/matrix-org/synapse/issues/13522), [\#13547](https://github.com/matrix-org/synapse/issues/13547))
-
-
-Bugfixes
--------
-
- Faster room joins: make `/joined_members` block whilst the room is partial stated. ([\#13514](https://github.com/matrix-org/synapse/issues/13514))
- Fix a bug introduced in Synapse 1.21.0 where the [`/event_reports` Admin API](https://matrix-org.github.io/synapse/develop/admin_api/event_reports.html) could return a total count which was larger than the number of results you can actually query for. ([\#13525](https://github.com/matrix-org/synapse/issues/13525))
- Fix a bug introduced in Synapse 1.52.0 where sending server notices fails if `max_avatar_size` or `allowed_avatar_mimetypes` is set and not `system_mxid_avatar_url`. ([\#13566](https://github.com/matrix-org/synapse/issues/13566))
- Fix a bug where the `opentracing.force_tracing_for_users` config option would not apply to [`/sendToDevice`](https://spec.matrix.org/v1.3/client-server-api/#put_matrixclientv3sendtodeviceeventtypetxnid) and [`/keys/upload`](https://spec.matrix.org/v1.3/client-server-api/#post_matrixclientv3keysupload) requests. ([\#13574](https://github.com/matrix-org/synapse/issues/13574))
-
-
-Improved Documentation
----------------------
-
- Add `openssl` example for generating registration HMAC digest. ([\#13472](https://github.com/matrix-org/synapse/issues/13472))
- Tidy up Synapse's README. ([\#13491](https://github.com/matrix-org/synapse/issues/13491))
- Document that event purging related to the `redaction_retention_period` config option is executed only every 5 minutes. ([\#13492](https://github.com/matrix-org/synapse/issues/13492))
- Add a warning to retention documentation regarding the possibility of database corruption. ([\#13497](https://github.com/matrix-org/synapse/issues/13497))
- Document that the `DOCKER_BUILDKIT=1` flag is needed to build the docker image. ([\#13515](https://github.com/matrix-org/synapse/issues/13515))
- Add missing links in `user_consent` section of configuration manual. ([\#13536](https://github.com/matrix-org/synapse/issues/13536))
- Fix the doc and some warnings that were referring to the nonexistent `custom_templates_directory` setting (instead of `custom_template_directory`). ([\#13538](https://github.com/matrix-org/synapse/issues/13538))
-
-
-Deprecations and Removals
-------------------------
-
- Remove the ability for homeservers to delegate email ownership verification
-  and password reset confirmation to identity servers. See [upgrade notes](https://matrix-org.github.io/synapse/v1.66/upgrade.html#upgrading-to-v1660) for more details.
-
-Internal Changes
----------------
-
-### Faster room joins
-
- Update the rejected state of events during de-partial-stating. ([\#13459](https://github.com/matrix-org/synapse/issues/13459))
- Avoid blocking lazy-loading `/sync`s during partial joins due to remote memberships. Pull remote memberships from auth events instead of the room state. ([\#13477](https://github.com/matrix-org/synapse/issues/13477))
- Refuse to start when faster joins is enabled on a deployment with workers, since worker configurations are not currently supported. ([\#13531](https://github.com/matrix-org/synapse/issues/13531))
-
-### Metrics and tracing
-
- Allow use of both `@trace` and `@tag_args` stacked on the same function. ([\#13453](https://github.com/matrix-org/synapse/issues/13453))
- Instrument the federation/backfill part of `/messages` for understandable traces in Jaeger. ([\#13489](https://github.com/matrix-org/synapse/issues/13489))
- Instrument `FederationStateIdsServlet` (`/state_ids`) for understandable traces in Jaeger. ([\#13499](https://github.com/matrix-org/synapse/issues/13499), [\#13554](https://github.com/matrix-org/synapse/issues/13554))
- Track HTTP response times over 10 seconds from `/messages` (`synapse_room_message_list_rest_servlet_response_time_seconds`). ([\#13533](https://github.com/matrix-org/synapse/issues/13533))
- Add metrics to track how the rate limiter is affecting requests (sleep/reject). ([\#13534](https://github.com/matrix-org/synapse/issues/13534), [\#13541](https://github.com/matrix-org/synapse/issues/13541))
- Add metrics to time how long it takes us to do backfill processing (`synapse_federation_backfill_processing_before_time_seconds`, `synapse_federation_backfill_processing_after_time_seconds`). ([\#13535](https://github.com/matrix-org/synapse/issues/13535), [\#13584](https://github.com/matrix-org/synapse/issues/13584))
- Add metrics to track rate limiter queue timing (`synapse_rate_limit_queue_wait_time_seconds`). ([\#13544](https://github.com/matrix-org/synapse/issues/13544))
- Update metrics to track `/messages` response time by room size. ([\#13545](https://github.com/matrix-org/synapse/issues/13545))
-
-### Everything else
-
- Refactor methods in `synapse.api.auth.Auth` to use `Requester` objects everywhere instead of user IDs. ([\#13024](https://github.com/matrix-org/synapse/issues/13024))
- Clean-up tests for notifications. ([\#13471](https://github.com/matrix-org/synapse/issues/13471))
- Add some miscellaneous comments to document sync, especially around `compute_state_delta`. ([\#13474](https://github.com/matrix-org/synapse/issues/13474))
- Use literals in place of `HTTPStatus` constants in tests. ([\#13479](https://github.com/matrix-org/synapse/issues/13479), [\#13488](https://github.com/matrix-org/synapse/issues/13488))
- Add comments about how event push actions are rotated. ([\#13485](https://github.com/matrix-org/synapse/issues/13485))
- Modify HTML template content to better support mobile devices' screen sizes. ([\#13493](https://github.com/matrix-org/synapse/issues/13493))
- Add a linter script which will reject non-strict types in Pydantic models. ([\#13502](https://github.com/matrix-org/synapse/issues/13502))
- Reduce the number of tests using legacy TCP replication. ([\#13543](https://github.com/matrix-org/synapse/issues/13543))
- Allow specifying additional request fields when using the `HomeServerTestCase.login` helper method. ([\#13549](https://github.com/matrix-org/synapse/issues/13549))
- Make `HomeServerTestCase` load any configured homeserver modules automatically. ([\#13558](https://github.com/matrix-org/synapse/issues/13558))
-
-
 Synapse 1.65.0 (2022-08-16)
 ===========================

@@ -0,0 +1 @@
+Improve validation of request bodies for the following client-server API endpoints: [`/account/password`](https://spec.matrix.org/v1.3/client-server-api/#post_matrixclientv3accountpassword), [`/account/password/email/requestToken`](https://spec.matrix.org/v1.3/client-server-api/#post_matrixclientv3accountpasswordemailrequesttoken), [`/account/deactivate`](https://spec.matrix.org/v1.3/client-server-api/#post_matrixclientv3accountdeactivate) and [`/account/3pid/email/requestToken`](https://spec.matrix.org/v1.3/client-server-api/#post_matrixclientv3account3pidemailrequesttoken).
@@ -0,0 +1 @@
+Allow use of both `@trace` and `@tag_args` stacked on the same function (tracing).
@@ -0,0 +1 @@
+Faster joins: update the rejected state of events during de-partial-stating.
@@ -0,0 +1 @@
+Clean-up tests for notifications.
@@ -0,0 +1 @@
+Add `openssl` example for generating registration HMAC digest.
@@ -0,0 +1 @@
+Add some miscellaneous comments to document sync, especially around `compute_state_delta`.
@@ -0,0 +1 @@
+Use literals in place of `HTTPStatus` constants in tests.
@@ -0,0 +1 @@
+Add comments about how event push actions are rotated.
@@ -0,0 +1 @@
+Use literals in place of `HTTPStatus` constants in tests.
@@ -0,0 +1 @@
+Instrument the federation/backfill part of `/messages` for understandable traces in Jaeger.
@@ -0,0 +1 @@
+Tidy up Synapse's README.
@@ -0,0 +1 @@
+Document that event purging related to the `redaction_retention_period` config option is executed only every 5 minutes.
@@ -0,0 +1 @@
+Modify HTML template content to better support mobile devices' screen sizes.
@@ -0,0 +1,2 @@
+Add a warning to retention documentation regarding the possibility of database corruption.
+
@@ -0,0 +1 @@
+Instrument `FederationStateIdsServlet` (`/state_ids`) for understandable traces in Jaeger.
@@ -0,0 +1 @@
+Add a linter script which will reject non-strict types in Pydantic models.
@@ -0,0 +1 @@
+Add forgotten status to Room Details API.
@@ -0,0 +1 @@
+Faster room joins: make `/joined_members` block whilst the room is partial stated.
@@ -0,0 +1 @@
+Document that the `DOCKER_BUILDKIT=1` flag is needed to build the docker image.
@@ -0,0 +1 @@
+Improve performance of sending messages in rooms with thousands of local users.
@@ -0,0 +1 @@
+Fix a bug in the `/event_reports` Admin API which meant that the total count could be larger than the number of results you can actually query for.
@@ -0,0 +1 @@
+Faster room joins: Refuse to start when faster joins is enabled on a deployment with workers, since worker configurations are not currently supported.
@@ -0,0 +1 @@
+Track HTTP response times over 10 seconds from `/messages` (`synapse_room_message_list_rest_servlet_response_time_seconds`).
@@ -0,0 +1 @@
+Add metrics to track how the rate limiter is affecting requests (sleep/reject).
@@ -0,0 +1 @@
+Add metrics to time how long it takes us to do backfill processing (`synapse_federation_backfill_processing_before_time_seconds`, `synapse_federation_backfill_processing_after_time_seconds`).
@@ -0,0 +1 @@
+Add missing links in `user_consent` section of configuration manual.
@@ -0,0 +1 @@
+Fix the doc and some warnings that were referring to the nonexistent `custom_templates_directory` setting (instead of `custom_template_directory`).
@@ -0,0 +1 @@
+Add metrics to track rate limiter queue timing (`synapse_rate_limit_queue_wait_time_seconds`).
@@ -0,0 +1 @@
+Improve performance of sending messages in rooms with thousands of local users.
@@ -0,0 +1 @@
+Downgrade an error that had the capacity to spam Sentry instances to a warning.
@@ -1,9 +1,3 @@
-matrix-synapse-py3 (1.66.0~rc1) stable; urgency=medium
-
-  * New Synapse release 1.66.0rc1.
-
- -- Synapse Packaging team <packages@matrix.org>  Tue, 23 Aug 2022 09:48:55 +0100
-
 matrix-synapse-py3 (1.65.0) stable; urgency=medium

  * New Synapse release 1.65.0.
@@ -337,8 +337,6 @@ A response body like the following is returned:
 }
 ```

-_Changed in Synapse 1.66:_ Added the `forgotten` key to the response body.
-
 # Room Members API

 The Room Members admin API allows server admins to get a list of all members of a room.
@@ -753,7 +753,6 @@ A response body like the following is returned:
      "device_id": "QBUAZIFURK",
      "display_name": "android",
      "last_seen_ip": "1.2.3.4",
-      "last_seen_user_agent": "Mozilla/5.0 (X11; Linux x86_64; rv:103.0) Gecko/20100101 Firefox/103.0",
      "last_seen_ts": 1474491775024,
      "user_id": "<user_id>"
    },
@@ -761,7 +760,6 @@ A response body like the following is returned:
      "device_id": "AUIECTSRND",
      "display_name": "ios",
      "last_seen_ip": "1.2.3.5",
-      "last_seen_user_agent": "Mozilla/5.0 (X11; Linux x86_64; rv:103.0) Gecko/20100101 Firefox/103.0",
      "last_seen_ts": 1474491775025,
      "user_id": "<user_id>"
    }
@@ -788,8 +786,6 @@ The following fields are returned in the JSON response body:
    Absent if no name has been set.
  - `last_seen_ip` - The IP address where this device was last seen.
    (May be a few minutes out of date, for efficiency reasons).
-  - `last_seen_user_agent` - The user agent of the device when it was last seen.
-    (May be a few minutes out of date, for efficiency reasons).
  - `last_seen_ts` - The timestamp (in milliseconds since the unix epoch) when this
    devices was last seen. (May be a few minutes out of date, for efficiency reasons).
  - `user_id` - Owner of  device.
@@ -841,7 +837,6 @@ A response body like the following is returned:
  "device_id": "<device_id>",
  "display_name": "android",
  "last_seen_ip": "1.2.3.4",
-  "last_seen_user_agent": "Mozilla/5.0 (X11; Linux x86_64; rv:103.0) Gecko/20100101 Firefox/103.0",
  "last_seen_ts": 1474491775024,
  "user_id": "<user_id>"
 }
@@ -863,8 +858,6 @@ The following fields are returned in the JSON response body:
  Absent if no name has been set.
 - `last_seen_ip` - The IP address where this device was last seen.
  (May be a few minutes out of date, for efficiency reasons).
-  - `last_seen_user_agent` - The user agent of the device when it was last seen.
-  (May be a few minutes out of date, for efficiency reasons).
 - `last_seen_ts` - The timestamp (in milliseconds since the unix epoch) when this
  devices was last seen. (May be a few minutes out of date, for efficiency reasons).
 - `user_id` - Owner of  device.
@@ -89,25 +89,6 @@ process, for example:
    dpkg -i matrix-synapse-py3_1.3.0+stretch1_amd64.deb
    ```

-# Upgrading to v1.66.0
-
-## Delegation of email validation no longer supported
-
-As of this version, Synapse no longer allows the tasks of verifying email address
-ownership, and password reset confirmation, to be delegated to an identity server.
-This removal was previously planned for Synapse 1.64.0, but was
-[delayed](https://github.com/matrix-org/synapse/issues/13421) until now to give
-homeserver administrators more notice of the change.
-
-To continue to allow users to add email addresses to their homeserver accounts,
-and perform password resets, make sure that Synapse is configured with a working
-email server in the [`email` configuration
-section](https://matrix-org.github.io/synapse/latest/usage/configuration/config_documentation.html#email)
-(including, at a minimum, a `notif_from` setting.)
-
-Specifying an `email` setting under `account_threepid_delegates` will now cause
-an error at startup.
-
 # Upgrading to v1.64.0

 ## Deprecation of the ability to delegate e-mail verification to identity servers
@@ -444,7 +444,7 @@ Sub-options for each listener include:
   * `names`: a list of names of HTTP resources. See below for a list of valid resource names.

   * `compress`: set to true to enable gzip compression on HTTP bodies for this resource. This is currently only supported with the
-     `client`, `consent`, `metrics` and `federation` resources.
+     `client`, `consent` and `metrics` resources.

 * `additional_resources`: Only valid for an 'http' listener. A map of
   additional endpoints which should be loaded via dynamic modules.
@@ -2182,10 +2182,7 @@ their account.
 by the Matrix Identity Service API
 [specification](https://matrix.org/docs/spec/identity_service/latest).)

-*Deprecated in Synapse 1.64.0*: The `email` option is deprecated.
-
-*Removed in Synapse 1.66.0*: The `email` option has been removed.
-If present, Synapse will report a configuration error on startup.
+*Updated in Synapse 1.64.0*: The `email` option is deprecated.

 Example configuration:
 ```yaml
@@ -54,7 +54,7 @@ skip_gitignore = true

 [tool.poetry]
 name = "matrix-synapse"
-version = "1.66.0rc1"
+version = "1.65.0"
 description = "Homeserver for the Matrix decentralised comms protocol"
 authors = ["Matrix.org Team and Contributors <packages@matrix.org>"]
 license = "Apache-2.0"
@@ -37,7 +37,8 @@ from synapse.logging.opentracing import (
    start_active_span,
    trace,
 )
-from synapse.types import Requester, create_requester
+from synapse.storage.databases.main.registration import TokenLookupResult
+from synapse.types import Requester, UserID, create_requester

 if TYPE_CHECKING:
    from synapse.server import HomeServer
@@ -69,14 +70,14 @@ class Auth:
    async def check_user_in_room(
        self,
        room_id: str,
-        requester: Requester,
+        user_id: str,
        allow_departed_users: bool = False,
    ) -> Tuple[str, Optional[str]]:
        """Check if the user is in the room, or was at some point.
        Args:
            room_id: The room to check.

-            requester: The user making the request, according to the access token.
+            user_id: The user to check.

            current_state: Optional map of the current state of the room.
                If provided then that map is used to check whether they are a
@@ -93,7 +94,6 @@ class Auth:
            membership event ID of the user.
        """

-        user_id = requester.user.to_string()
        (
            membership,
            member_event_id,
@@ -182,69 +182,96 @@ class Auth:

            access_token = self.get_access_token_from_request(request)

-            # First check if it could be a request from an appservice
-            requester = await self._get_appservice_user(request)
-            if not requester:
-                # If not, it should be from a regular user
-                requester = await self.get_user_by_access_token(
-                    access_token, allow_expired=allow_expired
-                )
-
-                # Deny the request if the user account has expired.
-                # This check is only done for regular users, not appservice ones.
-                if not allow_expired:
-                    if await self._account_validity_handler.is_user_expired(
-                        requester.user.to_string()
-                    ):
-                        # Raise the error if either an account validity module has determined
-                        # the account has expired, or the legacy account validity
-                        # implementation is enabled and determined the account has expired
-                        raise AuthError(
-                            403,
-                            "User account has expired",
-                            errcode=Codes.EXPIRED_ACCOUNT,
-                        )
-
-            if ip_addr and (
-                not requester.app_service or self._track_appservice_user_ips
-            ):
-                # XXX(quenting): I'm 95% confident that we could skip setting the
-                # device_id to "dummy-device" for appservices, and that the only impact
-                # would be some rows which whould not deduplicate in the 'user_ips'
-                # table during the transition
-                recorded_device_id = (
-                    "dummy-device"
-                    if requester.device_id is None and requester.app_service is not None
-                    else requester.device_id
-                )
-                await self.store.insert_client_ip(
-                    user_id=requester.authenticated_entity,
-                    access_token=access_token,
-                    ip=ip_addr,
-                    user_agent=user_agent,
-                    device_id=recorded_device_id,
-                )
-
-                # Track also the puppeted user client IP if enabled and the user is puppeting
-                if (
-                    requester.user.to_string() != requester.authenticated_entity
-                    and self._track_puppeted_user_ips
-                ):
+            (
+                user_id,
+                device_id,
+                app_service,
+            ) = await self._get_appservice_user_id_and_device_id(request)
+            if user_id and app_service:
+                if ip_addr and self._track_appservice_user_ips:
                    await self.store.insert_client_ip(
-                        user_id=requester.user.to_string(),
+                        user_id=user_id,
                        access_token=access_token,
                        ip=ip_addr,
                        user_agent=user_agent,
-                        device_id=requester.device_id,
+                        device_id="dummy-device"
+                        if device_id is None
+                        else device_id,  # stubbed
                    )

-            if requester.is_guest and not allow_guest:
+                requester = create_requester(
+                    user_id, app_service=app_service, device_id=device_id
+                )
+
+                request.requester = user_id
+                return requester
+
+            user_info = await self.get_user_by_access_token(
+                access_token, allow_expired=allow_expired
+            )
+            token_id = user_info.token_id
+            is_guest = user_info.is_guest
+            shadow_banned = user_info.shadow_banned
+
+            # Deny the request if the user account has expired.
+            if not allow_expired:
+                if await self._account_validity_handler.is_user_expired(
+                    user_info.user_id
+                ):
+                    # Raise the error if either an account validity module has determined
+                    # the account has expired, or the legacy account validity
+                    # implementation is enabled and determined the account has expired
+                    raise AuthError(
+                        403,
+                        "User account has expired",
+                        errcode=Codes.EXPIRED_ACCOUNT,
+                    )
+
+            device_id = user_info.device_id
+
+            if access_token and ip_addr:
+                await self.store.insert_client_ip(
+                    user_id=user_info.token_owner,
+                    access_token=access_token,
+                    ip=ip_addr,
+                    user_agent=user_agent,
+                    device_id=device_id,
+                )
+                # Track also the puppeted user client IP if enabled and the user is puppeting
+                if (
+                    user_info.user_id != user_info.token_owner
+                    and self._track_puppeted_user_ips
+                ):
+                    await self.store.insert_client_ip(
+                        user_id=user_info.user_id,
+                        access_token=access_token,
+                        ip=ip_addr,
+                        user_agent=user_agent,
+                        device_id=device_id,
+                    )
+
+            if is_guest and not allow_guest:
                raise AuthError(
                    403,
                    "Guest access not allowed",
                    errcode=Codes.GUEST_ACCESS_FORBIDDEN,
                )

+            # Mark the token as used. This is used to invalidate old refresh
+            # tokens after some time.
+            if not user_info.token_used and token_id is not None:
+                await self.store.mark_access_token_as_used(token_id)
+
+            requester = create_requester(
+                user_info.user_id,
+                token_id,
+                is_guest,
+                shadow_banned,
+                device_id,
+                app_service=app_service,
+                authenticated_entity=user_info.token_owner,
+            )
+
            request.requester = requester
            return requester
        except KeyError:
@@ -281,7 +308,9 @@ class Auth:
                403, "Application service has not registered this user (%s)" % user_id
            )

-    async def _get_appservice_user(self, request: Request) -> Optional[Requester]:
+    async def _get_appservice_user_id_and_device_id(
+        self, request: Request
+    ) -> Tuple[Optional[str], Optional[str], Optional[ApplicationService]]:
        """
        Given a request, reads the request parameters to determine:
        - whether it's an application service that's making this request
@@ -296,13 +325,15 @@ class Auth:
             Must use `org.matrix.msc3202.device_id` in place of `device_id` for now.

        Returns:
-            the application service `Requester` of that request
+            3-tuple of
+            (user ID?, device ID?, application service?)

        Postconditions:
-        - The `app_service` field in the returned `Requester` is set
-        - The `user_id` field in the returned `Requester` is either the application
-          service sender or the controlled user set by the `user_id` URI parameter
-        - The returned application service is permitted to control the returned user ID.
+        - If an application service is returned, so is a user ID
+        - A user ID is never returned without an application service
+        - A device ID is never returned without a user ID or an application service
+        - The returned application service, if present, is permitted to control the
+          returned user ID.
        - The returned device ID, if present, has been checked to be a valid device ID
          for the returned user ID.
        """
@@ -312,12 +343,12 @@ class Auth:
            self.get_access_token_from_request(request)
        )
        if app_service is None:
-            return None
+            return None, None, None

        if app_service.ip_range_whitelist:
            ip_address = IPAddress(request.getClientAddress().host)
            if ip_address not in app_service.ip_range_whitelist:
-                return None
+                return None, None, None

        # This will always be set by the time Twisted calls us.
        assert request.args is not None
@@ -351,15 +382,13 @@ class Auth:
                    Codes.EXCLUSIVE,
                )

-        return create_requester(
-            effective_user_id, app_service=app_service, device_id=effective_device_id
-        )
+        return effective_user_id, effective_device_id, app_service

    async def get_user_by_access_token(
        self,
        token: str,
        allow_expired: bool = False,
-    ) -> Requester:
+    ) -> TokenLookupResult:
        """Validate access token and get user_id from it

        Args:
@@ -376,9 +405,9 @@ class Auth:

        # First look in the database to see if the access token is present
        # as an opaque token.
-        user_info = await self.store.get_user_by_access_token(token)
-        if user_info:
-            valid_until_ms = user_info.valid_until_ms
+        r = await self.store.get_user_by_access_token(token)
+        if r:
+            valid_until_ms = r.valid_until_ms
            if (
                not allow_expired
                and valid_until_ms is not None
@@ -390,20 +419,7 @@ class Auth:
                    msg="Access token has expired", soft_logout=True
                )

-            # Mark the token as used. This is used to invalidate old refresh
-            # tokens after some time.
-            await self.store.mark_access_token_as_used(user_info.token_id)
-
-            requester = create_requester(
-                user_id=user_info.user_id,
-                access_token_id=user_info.token_id,
-                is_guest=user_info.is_guest,
-                shadow_banned=user_info.shadow_banned,
-                device_id=user_info.device_id,
-                authenticated_entity=user_info.token_owner,
-            )
-
-            return requester
+            return r

        # If the token isn't found in the database, then it could still be a
        # macaroon for a guest, so we check that here.
@@ -429,12 +445,11 @@ class Auth:
                    "Guest access token used for regular user"
                )

-            return create_requester(
+            return TokenLookupResult(
                user_id=user_id,
                is_guest=True,
                # all guests get the same device id
                device_id=GUEST_DEVICE_ID,
-                authenticated_entity=user_id,
            )
        except (
            pymacaroons.exceptions.MacaroonException,
@@ -457,33 +472,32 @@ class Auth:
        request.requester = create_requester(service.sender, app_service=service)
        return service

-    async def is_server_admin(self, requester: Requester) -> bool:
+    async def is_server_admin(self, user: UserID) -> bool:
        """Check if the given user is a local server admin.

        Args:
-            requester: The user making the request, according to the access token.
+            user: user to check

        Returns:
            True if the user is an admin
        """
-        return await self.store.is_server_admin(requester.user)
+        return await self.store.is_server_admin(user)

-    async def check_can_change_room_list(
-        self, room_id: str, requester: Requester
-    ) -> bool:
+    async def check_can_change_room_list(self, room_id: str, user: UserID) -> bool:
        """Determine whether the user is allowed to edit the room's entry in the
        published room list.

        Args:
-            room_id: The room to check.
-            requester: The user making the request, according to the access token.
+            room_id
+            user
        """

-        is_admin = await self.is_server_admin(requester)
+        is_admin = await self.is_server_admin(user)
        if is_admin:
            return True

-        await self.check_user_in_room(room_id, requester)
+        user_id = user.to_string()
+        await self.check_user_in_room(room_id, user_id)

        # We currently require the user is a "moderator" in the room. We do this
        # by checking if they would (theoretically) be able to change the
@@ -502,9 +516,7 @@ class Auth:
        send_level = event_auth.get_send_level(
            EventTypes.CanonicalAlias, "", power_level_event
        )
-        user_level = event_auth.get_user_power_level(
-            requester.user.to_string(), auth_events
-        )
+        user_level = event_auth.get_user_power_level(user_id, auth_events)

        return user_level >= send_level

@@ -562,16 +574,16 @@ class Auth:

    @trace
    async def check_user_in_room_or_world_readable(
-        self, room_id: str, requester: Requester, allow_departed_users: bool = False
+        self, room_id: str, user_id: str, allow_departed_users: bool = False
    ) -> Tuple[str, Optional[str]]:
        """Checks that the user is or was in the room or the room is world
        readable. If it isn't then an exception is raised.

        Args:
-            room_id: The room to check.
-            requester: The user making the request, according to the access token.
-            allow_departed_users: If True, accept users that were previously
-                members but have now departed.
+            room_id: room to check
+            user_id: user to check
+            allow_departed_users: if True, accept users that were previously
+                members but have now departed

        Returns:
            Resolves to the current membership of the user in the room and the
@@ -586,7 +598,7 @@ class Auth:
            #  * The user is a guest user, and has joined the room
            # else it will throw.
            return await self.check_user_in_room(
-                room_id, requester, allow_departed_users=allow_departed_users
+                room_id, user_id, allow_departed_users=allow_departed_users
            )
        except AuthError:
            visibility = await self._storage_controllers.state.get_current_state_event(
@@ -601,6 +613,6 @@ class Auth:
            raise UnstableSpecAuthError(
                403,
                "User %s not in room %s, and room previews are disabled"
-                % (requester.user, room_id),
+                % (user_id, room_id),
                errcode=Codes.NOT_JOINED,
            )
@@ -216,11 +216,11 @@ class EventContentFields:
    MSC2716_HISTORICAL: Final = "org.matrix.msc2716.historical"
    # For "insertion" events to indicate what the next batch ID should be in
    # order to connect to it
-    MSC2716_NEXT_BATCH_ID: Final = "next_batch_id"
+    MSC2716_NEXT_BATCH_ID: Final = "org.matrix.msc2716.next_batch_id"
    # Used on "batch" events to indicate which insertion event it connects to
-    MSC2716_BATCH_ID: Final = "batch_id"
+    MSC2716_BATCH_ID: Final = "org.matrix.msc2716.batch_id"
    # For "marker" events
-    MSC2716_INSERTION_EVENT_REFERENCE: Final = "insertion_event_reference"
+    MSC2716_MARKER_INSERTION: Final = "org.matrix.msc2716.marker.insertion"

    # The authorising user for joining a restricted room.
    AUTHORISING_USER: Final = "join_authorised_via_users_server"
@@ -269,6 +269,24 @@ class RoomVersions:
        msc3787_knock_restricted_join_rule=False,
        msc3667_int_only_power_levels=False,
    )
+    MSC2716v3 = RoomVersion(
+        "org.matrix.msc2716v3",
+        RoomDisposition.UNSTABLE,
+        EventFormatVersions.V3,
+        StateResolutionVersions.V2,
+        enforce_key_validity=True,
+        special_case_aliases_auth=False,
+        strict_canonicaljson=True,
+        limit_notifications_power_levels=True,
+        msc2176_redaction_rules=False,
+        msc3083_join_rules=False,
+        msc3375_redaction_rules=False,
+        msc2403_knocking=True,
+        msc2716_historical=True,
+        msc2716_redactions=True,
+        msc3787_knock_restricted_join_rule=False,
+        msc3667_int_only_power_levels=False,
+    )
    MSC3787 = RoomVersion(
        "org.matrix.msc3787",
        RoomDisposition.UNSTABLE,
@@ -305,24 +323,6 @@ class RoomVersions:
        msc3787_knock_restricted_join_rule=True,
        msc3667_int_only_power_levels=True,
    )
-    MSC2716v4 = RoomVersion(
-        "org.matrix.msc2716v4",
-        RoomDisposition.UNSTABLE,
-        EventFormatVersions.V3,
-        StateResolutionVersions.V2,
-        enforce_key_validity=True,
-        special_case_aliases_auth=False,
-        strict_canonicaljson=True,
-        limit_notifications_power_levels=True,
-        msc2176_redaction_rules=False,
-        msc3083_join_rules=False,
-        msc3375_redaction_rules=False,
-        msc2403_knocking=True,
-        msc2716_historical=True,
-        msc2716_redactions=True,
-        msc3787_knock_restricted_join_rule=False,
-        msc3667_int_only_power_levels=False,
-    )


 KNOWN_ROOM_VERSIONS: Dict[str, RoomVersion] = {
@@ -338,9 +338,9 @@ KNOWN_ROOM_VERSIONS: Dict[str, RoomVersion] = {
        RoomVersions.V7,
        RoomVersions.V8,
        RoomVersions.V9,
+        RoomVersions.MSC2716v3,
        RoomVersions.MSC3787,
        RoomVersions.V10,
-        RoomVersions.MSC2716v4,
    )
 }

@@ -44,6 +44,7 @@ from synapse.app._base import (
    register_start,
 )
 from synapse.config._base import ConfigError, format_config_error
+from synapse.config.emailconfig import ThreepidBehaviour
 from synapse.config.homeserver import HomeServerConfig
 from synapse.config.server import ListenerConfig
 from synapse.federation.transport.server import TransportLayerServer
@@ -201,7 +202,7 @@ class SynapseHomeServer(HomeServer):
                }
            )

-            if self.config.email.can_verify_email:
+            if self.config.email.threepid_behaviour_email == ThreepidBehaviour.LOCAL:
                from synapse.rest.synapse.client.password_reset import (
                    PasswordResetSubmitTokenResource,
                )
@@ -219,10 +220,7 @@ class SynapseHomeServer(HomeServer):
            resources.update({"/_matrix/consent": consent_resource})

        if name == "federation":
-            federation_resource: Resource = TransportLayerServer(self)
-            if compress:
-                federation_resource = gz_wrap(federation_resource)
-            resources.update({FEDERATION_PREFIX: federation_resource})
+            resources.update({FEDERATION_PREFIX: TransportLayerServer(self)})

        if name == "openid":
            resources.update(
@@ -18,6 +18,7 @@
 import email.utils
 import logging
 import os
+from enum import Enum
 from typing import Any

 import attr
@@ -135,22 +136,40 @@ class EmailConfig(Config):

        self.email_enable_notifs = email_config.get("enable_notifs", False)

+        self.threepid_behaviour_email = (
+            # Have Synapse handle the email sending if account_threepid_delegates.email
+            # is not defined
+            # msisdn is currently always remote while Synapse does not support any method of
+            # sending SMS messages
+            ThreepidBehaviour.REMOTE
+            if self.root.registration.account_threepid_delegate_email
+            else ThreepidBehaviour.LOCAL
+        )
+
        if config.get("trust_identity_server_for_password_resets"):
            raise ConfigError(
-                'The config option "trust_identity_server_for_password_resets" '
-                "is no longer supported. Please remove it from the config file."
+                'The config option "trust_identity_server_for_password_resets" has been removed.'
+                "Please consult the configuration manual at docs/usage/configuration/config_documentation.md for "
+                "details and update your config file."
            )

-        # If we have email config settings, assume that we can verify ownership of
-        # email addresses.
-        self.can_verify_email = email_config != {}
+        self.local_threepid_handling_disabled_due_to_email_config = False
+        if (
+            self.threepid_behaviour_email == ThreepidBehaviour.LOCAL
+            and email_config == {}
+        ):
+            # We cannot warn the user this has happened here
+            # Instead do so when a user attempts to reset their password
+            self.local_threepid_handling_disabled_due_to_email_config = True
+
+            self.threepid_behaviour_email = ThreepidBehaviour.OFF

        # Get lifetime of a validation token in milliseconds
        self.email_validation_token_lifetime = self.parse_duration(
            email_config.get("validation_token_lifetime", "1h")
        )

-        if self.can_verify_email:
+        if self.threepid_behaviour_email == ThreepidBehaviour.LOCAL:
            missing = []
            if not self.email_notif_from:
                missing.append("email.notif_from")
@@ -341,3 +360,18 @@ class EmailConfig(Config):
                    "Config option email.invite_client_location must be a http or https URL",
                    path=("email", "invite_client_location"),
                )
+
+
+class ThreepidBehaviour(Enum):
+    """
+    Enum to define the behaviour of Synapse with regards to when it contacts an identity
+    server for 3pid registration and password resets
+
+    REMOTE = use an external server to send tokens
+    LOCAL = send tokens ourselves
+    OFF = disable registration via 3pid and password resets
+    """
+
+    REMOTE = "remote"
+    LOCAL = "local"
+    OFF = "off"
@@ -90,6 +90,3 @@ class ExperimentalConfig(Config):

        # MSC3848: Introduce errcodes for specific event sending failures
        self.msc3848_enabled: bool = experimental.get("msc3848_enabled", False)
-
-        # MSC3852: Expose last seen user agent field on /_matrix/client/v3/devices.
-        self.msc3852_enabled: bool = experimental.get("msc3852_enabled", False)
@@ -13,6 +13,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 import argparse
+import logging
 from typing import Any, Optional

 from synapse.api.constants import RoomCreationPreset
@@ -20,11 +21,15 @@ from synapse.config._base import Config, ConfigError
 from synapse.types import JsonDict, RoomAlias, UserID
 from synapse.util.stringutils import random_string_with_symbols, strtobool

-NO_EMAIL_DELEGATE_ERROR = """\
-Delegation of email verification to an identity server is no longer supported. To
+logger = logging.getLogger(__name__)
+
+LEGACY_EMAIL_DELEGATE_WARNING = """\
+Delegation of email verification to an identity server is now deprecated. To
 continue to allow users to add email addresses to their accounts, and use them for
 password resets, configure Synapse with an SMTP server via the `email` setting, and
 remove `account_threepid_delegates.email`.
+
+This will be an error in a future version.
 """


@@ -59,7 +64,9 @@ class RegistrationConfig(Config):

        account_threepid_delegates = config.get("account_threepid_delegates") or {}
        if "email" in account_threepid_delegates:
-            raise ConfigError(NO_EMAIL_DELEGATE_ERROR)
+            logger.warning(LEGACY_EMAIL_DELEGATE_WARNING)
+
+        self.account_threepid_delegate_email = account_threepid_delegates.get("email")
        self.account_threepid_delegate_msisdn = account_threepid_delegates.get("msisdn")
        self.default_identity_server = config.get("default_identity_server")
        self.allow_guest_access = config.get("allow_guest_access", False)
@@ -161,7 +161,7 @@ def prune_event_dict(room_version: RoomVersion, event_dict: JsonDict) -> JsonDic
    elif room_version.msc2716_redactions and event_type == EventTypes.MSC2716_BATCH:
        add_fields(EventContentFields.MSC2716_BATCH_ID)
    elif room_version.msc2716_redactions and event_type == EventTypes.MSC2716_MARKER:
-        add_fields(EventContentFields.MSC2716_INSERTION_EVENT_REFERENCE)
+        add_fields(EventContentFields.MSC2716_MARKER_INSERTION)

    allowed_fields = {k: v for k, v in event_dict.items() if k in allowed_keys}

@@ -280,7 +280,7 @@ class AuthHandler:
        that it isn't stolen by re-authenticating them.

        Args:
-            requester: The user making the request, according to the access token.
+            requester: The user, as given by the access token

            request: The request sent by the client.

@@ -1435,25 +1435,20 @@ class AuthHandler:
            access_token: access token to be deleted

        """
-        token = await self.store.get_user_by_access_token(access_token)
-        if not token:
-            # At this point, the token should already have been fetched once by
-            # the caller, so this should not happen, unless of a race condition
-            # between two delete requests
-            raise SynapseError(HTTPStatus.UNAUTHORIZED, "Unrecognised access token")
+        user_info = await self.auth.get_user_by_access_token(access_token)
        await self.store.delete_access_token(access_token)

        # see if any modules want to know about this
        await self.password_auth_provider.on_logged_out(
-            user_id=token.user_id,
-            device_id=token.device_id,
+            user_id=user_info.user_id,
+            device_id=user_info.device_id,
            access_token=access_token,
        )

        # delete pushers associated with this access token
-        if token.token_id is not None:
+        if user_info.token_id is not None:
            await self.hs.get_pusherpool().remove_pushers_by_access_token(
-                token.user_id, (token.token_id,)
+                user_info.user_id, (user_info.token_id,)
            )

    async def delete_access_tokens_for_user(
@@ -74,7 +74,6 @@ class DeviceWorkerHandler:
        self._state_storage = hs.get_storage_controllers().state
        self._auth_handler = hs.get_auth_handler()
        self.server_name = hs.hostname
-        self._msc3852_enabled = hs.config.experimental.msc3852_enabled

    @trace
    async def get_devices_by_user(self, user_id: str) -> List[JsonDict]:
@@ -748,13 +747,7 @@ def _update_device_from_client_ips(
    device: JsonDict, client_ips: Mapping[Tuple[str, str], Mapping[str, Any]]
 ) -> None:
    ip = client_ips.get((device["user_id"], device["device_id"]), {})
-    device.update(
-        {
-            "last_seen_user_agent": ip.get("user_agent"),
-            "last_seen_ts": ip.get("last_seen"),
-            "last_seen_ip": ip.get("ip"),
-        }
-    )
+    device.update({"last_seen_ts": ip.get("last_seen"), "last_seen_ip": ip.get("ip")})


 class DeviceListUpdater:
@@ -30,7 +30,7 @@ from synapse.api.errors import (
 from synapse.appservice import ApplicationService
 from synapse.module_api import NOT_SPAM
 from synapse.storage.databases.main.directory import RoomAliasMapping
-from synapse.types import JsonDict, Requester, RoomAlias, get_domain_from_id
+from synapse.types import JsonDict, Requester, RoomAlias, UserID, get_domain_from_id

 if TYPE_CHECKING:
    from synapse.server import HomeServer
@@ -133,7 +133,7 @@ class DirectoryHandler:
        else:
            # Server admins are not subject to the same constraints as normal
            # users when creating an alias (e.g. being in the room).
-            is_admin = await self.auth.is_server_admin(requester)
+            is_admin = await self.auth.is_server_admin(requester.user)

            if (self.require_membership and check_membership) and not is_admin:
                rooms_for_user = await self.store.get_rooms_for_user(user_id)
@@ -197,7 +197,7 @@ class DirectoryHandler:
        user_id = requester.user.to_string()

        try:
-            can_delete = await self._user_can_delete_alias(room_alias, requester)
+            can_delete = await self._user_can_delete_alias(room_alias, user_id)
        except StoreError as e:
            if e.code == 404:
                raise NotFoundError("Unknown room alias")
@@ -400,9 +400,7 @@ class DirectoryHandler:
        # either no interested services, or no service with an exclusive lock
        return True

-    async def _user_can_delete_alias(
-        self, alias: RoomAlias, requester: Requester
-    ) -> bool:
+    async def _user_can_delete_alias(self, alias: RoomAlias, user_id: str) -> bool:
        """Determine whether a user can delete an alias.

        One of the following must be true:
@@ -415,7 +413,7 @@ class DirectoryHandler:
        """
        creator = await self.store.get_room_alias_creator(alias.to_string())

-        if creator == requester.user.to_string():
+        if creator == user_id:
            return True

        # Resolve the alias to the corresponding room.
@@ -424,7 +422,9 @@ class DirectoryHandler:
        if not room_id:
            return False

-        return await self.auth.check_can_change_room_list(room_id, requester)
+        return await self.auth.check_can_change_room_list(
+            room_id, UserID.from_string(user_id)
+        )

    async def edit_published_room_list(
        self, requester: Requester, room_id: str, visibility: str
@@ -463,7 +463,7 @@ class DirectoryHandler:
            raise SynapseError(400, "Unknown room")

        can_change_room_list = await self.auth.check_can_change_room_list(
-            room_id, requester
+            room_id, requester.user
        )
        if not can_change_room_list:
            raise AuthError(
@@ -528,8 +528,10 @@ class DirectoryHandler:
        Get a list of the aliases that currently point to this room on this server
        """
        # allow access to server admins and current members of the room
-        is_admin = await self.auth.is_server_admin(requester)
+        is_admin = await self.auth.is_server_admin(requester.user)
        if not is_admin:
-            await self.auth.check_user_in_room_or_world_readable(room_id, requester)
+            await self.auth.check_user_in_room_or_world_readable(
+                room_id, requester.user.to_string()
+            )

        return await self.store.get_aliases_for_room(room_id)
@@ -86,14 +86,9 @@ backfill_processing_before_timer = Histogram(
    "sec",
    [],
    buckets=(
-        0.1,
-        0.5,
        1.0,
-        2.5,
        5.0,
-        7.5,
        10.0,
-        15.0,
        20.0,
        30.0,
        40.0,
@@ -487,7 +482,7 @@ class FederationHandler:

        processing_end_time = self.clock.time_msec()
        backfill_processing_before_timer.observe(
-            (processing_end_time - processing_start_time) / 1000
+            (processing_start_time - processing_end_time) / 1000
        )

        success = await try_backfill(likely_domains)
@@ -104,25 +104,15 @@ backfill_processing_after_timer = Histogram(
    "sec",
    [],
    buckets=(
-        0.1,
-        0.25,
-        0.5,
        1.0,
-        2.5,
        5.0,
-        7.5,
        10.0,
-        15.0,
        20.0,
-        25.0,
        30.0,
        40.0,
-        50.0,
        60.0,
        80.0,
-        100.0,
        120.0,
-        150.0,
        180.0,
        "+Inf",
    ),
@@ -1394,7 +1384,7 @@ class FederationEventHandler:
        logger.debug("_handle_marker_event: received %s", marker_event)

        insertion_event_id = marker_event.content.get(
-            EventContentFields.MSC2716_INSERTION_EVENT_REFERENCE
+            EventContentFields.MSC2716_MARKER_INSERTION
        )

        if insertion_event_id is None:
@@ -26,6 +26,7 @@ from synapse.api.errors import (
    SynapseError,
 )
 from synapse.api.ratelimiting import Ratelimiter
+from synapse.config.emailconfig import ThreepidBehaviour
 from synapse.http import RequestTimedOutError
 from synapse.http.client import SimpleHttpClient
 from synapse.http.site import SynapseRequest
@@ -415,6 +416,48 @@ class IdentityHandler:

        return session_id

+    async def request_email_token(
+        self,
+        id_server: str,
+        email: str,
+        client_secret: str,
+        send_attempt: int,
+        next_link: Optional[str] = None,
+    ) -> JsonDict:
+        """
+        Request an external server send an email on our behalf for the purposes of threepid
+        validation.
+
+        Args:
+            id_server: The identity server to proxy to
+            email: The email to send the message to
+            client_secret: The unique client_secret sends by the user
+            send_attempt: Which attempt this is
+            next_link: A link to redirect the user to once they submit the token
+
+        Returns:
+            The json response body from the server
+        """
+        params = {
+            "email": email,
+            "client_secret": client_secret,
+            "send_attempt": send_attempt,
+        }
+        if next_link:
+            params["next_link"] = next_link
+
+        try:
+            data = await self.http_client.post_json_get_json(
+                id_server + "/_matrix/identity/api/v1/validate/email/requestToken",
+                params,
+            )
+            return data
+        except HttpResponseException as e:
+            logger.info("Proxied requestToken failed: %r", e)
+            raise e.to_synapse_error()
+        except RequestTimedOutError:
+            raise SynapseError(500, "Timed out contacting identity server")
+
    async def requestMsisdnToken(
        self,
        id_server: str,
@@ -488,7 +531,18 @@ class IdentityHandler:
        validation_session = None

        # Try to validate as email
-        if self.hs.config.email.can_verify_email:
+        if self.hs.config.email.threepid_behaviour_email == ThreepidBehaviour.REMOTE:
+            # Remote emails will only be used if a valid identity server is provided.
+            assert (
+                self.hs.config.registration.account_threepid_delegate_email is not None
+            )
+
+            # Ask our delegated email identity server
+            validation_session = await self.threepid_from_creds(
+                self.hs.config.registration.account_threepid_delegate_email,
+                threepid_creds,
+            )
+        elif self.hs.config.email.threepid_behaviour_email == ThreepidBehaviour.LOCAL:
            # Get a validated session matching these details
            validation_session = await self.store.get_threepid_validation_session(
                "email", client_secret, sid=sid, validated=True
@@ -309,18 +309,18 @@ class InitialSyncHandler:
        if blocked:
            raise SynapseError(403, "This room has been blocked on this server")

+        user_id = requester.user.to_string()
+
        (
            membership,
            member_event_id,
        ) = await self.auth.check_user_in_room_or_world_readable(
            room_id,
-            requester,
+            user_id,
            allow_departed_users=True,
        )
        is_peeking = member_event_id is None

-        user_id = requester.user.to_string()
-
        if membership == Membership.JOIN:
            result = await self._room_initial_sync_joined(
                user_id, room_id, pagin_config, membership, is_peeking
@@ -104,7 +104,7 @@ class MessageHandler:

    async def get_room_data(
        self,
-        requester: Requester,
+        user_id: str,
        room_id: str,
        event_type: str,
        state_key: str,
@@ -112,7 +112,7 @@ class MessageHandler:
        """Get data from a room.

        Args:
-            requester: The user who did the request.
+            user_id
            room_id
            event_type
            state_key
@@ -125,7 +125,7 @@ class MessageHandler:
            membership,
            membership_event_id,
        ) = await self.auth.check_user_in_room_or_world_readable(
-            room_id, requester, allow_departed_users=True
+            room_id, user_id, allow_departed_users=True
        )

        if membership == Membership.JOIN:
@@ -161,10 +161,11 @@ class MessageHandler:

    async def get_state_events(
        self,
-        requester: Requester,
+        user_id: str,
        room_id: str,
        state_filter: Optional[StateFilter] = None,
        at_token: Optional[StreamToken] = None,
+        is_guest: bool = False,
    ) -> List[dict]:
        """Retrieve all state events for a given room. If the user is
        joined to the room then return the current state. If the user has
@@ -173,13 +174,14 @@ class MessageHandler:
        visible.

        Args:
-            requester: The user requesting state events.
+            user_id: The user requesting state events.
            room_id: The room ID to get all state events from.
            state_filter: The state filter used to fetch state from the database.
            at_token: the stream token of the at which we are requesting
                the stats. If the user is not allowed to view the state as of that
                stream token, we raise a 403 SynapseError. If None, returns the current
                state based on the current_state_events table.
+            is_guest: whether this user is a guest
        Returns:
            A list of dicts representing state events. [{}, {}, {}]
        Raises:
@@ -189,7 +191,6 @@ class MessageHandler:
            members of this room.
        """
        state_filter = state_filter or StateFilter.all()
-        user_id = requester.user.to_string()

        if at_token:
            last_event_id = (
@@ -222,7 +223,7 @@ class MessageHandler:
                membership,
                membership_event_id,
            ) = await self.auth.check_user_in_room_or_world_readable(
-                room_id, requester, allow_departed_users=True
+                room_id, user_id, allow_departed_users=True
            )

            if membership == Membership.JOIN:
@@ -316,11 +317,12 @@ class MessageHandler:
        Returns:
            A dict of user_id to profile info
        """
+        user_id = requester.user.to_string()
        if not requester.app_service:
            # We check AS auth after fetching the room membership, as it
            # requires us to pull out all joined members anyway.
            membership, _ = await self.auth.check_user_in_room_or_world_readable(
-                room_id, requester, allow_departed_users=True
+                room_id, user_id, allow_departed_users=True
            )
            if membership != Membership.JOIN:
                raise SynapseError(
@@ -338,10 +340,7 @@ class MessageHandler:
        # If this is an AS, double check that they are allowed to see the members.
        # This can either be because the AS user is in the room or because there
        # is a user in the room that the AS is "interested in"
-        if (
-            requester.app_service
-            and requester.user.to_string() not in users_with_profile
-        ):
+        if requester.app_service and user_id not in users_with_profile:
            for uid in users_with_profile:
                if requester.app_service.is_interested_in_user(uid):
                    break
@@ -464,7 +464,7 @@ class PaginationHandler:
                membership,
                member_event_id,
            ) = await self.auth.check_user_in_room_or_world_readable(
-                room_id, requester, allow_departed_users=True
+                room_id, user_id, allow_departed_users=True
            )

            if pagin_config.direction == "b":
@@ -29,13 +29,7 @@ from synapse.api.constants import (
    JoinRules,
    LoginType,
 )
-from synapse.api.errors import (
-    AuthError,
-    Codes,
-    ConsentNotGivenError,
-    InvalidClientTokenError,
-    SynapseError,
-)
+from synapse.api.errors import AuthError, Codes, ConsentNotGivenError, SynapseError
 from synapse.appservice import ApplicationService
 from synapse.config.server import is_threepid_reserved
 from synapse.http.servlet import assert_params_in_dict
@@ -186,7 +180,10 @@ class RegistrationHandler:
                )
            if guest_access_token:
                user_data = await self.auth.get_user_by_access_token(guest_access_token)
-                if not user_data.is_guest or user_data.user.localpart != localpart:
+                if (
+                    not user_data.is_guest
+                    or UserID.from_string(user_data.user_id).localpart != localpart
+                ):
                    raise AuthError(
                        403,
                        "Cannot register taken user ID without valid guest "
@@ -621,7 +618,7 @@ class RegistrationHandler:
        user_id = user.to_string()
        service = self.store.get_app_service_by_token(as_token)
        if not service:
-            raise InvalidClientTokenError()
+            raise AuthError(403, "Invalid application service token.")
        if not service.is_interested_in_user(user_id):
            raise SynapseError(
                400,
@@ -103,7 +103,7 @@ class RelationsHandler:

        # TODO Properly handle a user leaving a room.
        (_, member_event_id) = await self._auth.check_user_in_room_or_world_readable(
-            room_id, requester, allow_departed_users=True
+            room_id, user_id, allow_departed_users=True
        )

        # This gets the original event and checks that a) the event exists and
@@ -721,7 +721,7 @@ class RoomCreationHandler:
            # allow the server notices mxid to create rooms
            is_requester_admin = True
        else:
-            is_requester_admin = await self.auth.is_server_admin(requester)
+            is_requester_admin = await self.auth.is_server_admin(requester.user)

        # Let the third party rules modify the room creation config if needed, or abort
        # the room creation entirely with an exception.
@@ -1279,7 +1279,7 @@ class RoomContextHandler:
        """
        user = requester.user
        if use_admin_priviledge:
-            await assert_user_is_admin(self.auth, requester)
+            await assert_user_is_admin(self.auth, requester.user)

        before_limit = math.floor(limit / 2.0)
        after_limit = limit - before_limit
@@ -179,7 +179,7 @@ class RoomMemberHandler(metaclass=abc.ABCMeta):
        """Try and join a room that this server is not in

        Args:
-            requester: The user making the request, according to the access token.
+            requester
            remote_room_hosts: List of servers that can be used to join via.
            room_id: Room that we are trying to join
            user: User who is trying to join
@@ -689,7 +689,7 @@ class RoomMemberHandler(metaclass=abc.ABCMeta):
                errcode=Codes.BAD_JSON,
            )

-        if "avatar_url" in content and content.get("avatar_url") is not None:
+        if "avatar_url" in content:
            if not await self.profile_handler.check_avatar_size_and_mime_type(
                content["avatar_url"],
            ):
@@ -744,7 +744,7 @@ class RoomMemberHandler(metaclass=abc.ABCMeta):
                is_requester_admin = True

            else:
-                is_requester_admin = await self.auth.is_server_admin(requester)
+                is_requester_admin = await self.auth.is_server_admin(requester.user)

            if not is_requester_admin:
                if self.config.server.block_non_admin_invites:
@@ -868,7 +868,7 @@ class RoomMemberHandler(metaclass=abc.ABCMeta):
                bypass_spam_checker = True

            else:
-                bypass_spam_checker = await self.auth.is_server_admin(requester)
+                bypass_spam_checker = await self.auth.is_server_admin(requester.user)

            inviter = await self._get_inviter(target.to_string(), room_id)
            if (
@@ -1410,7 +1410,7 @@ class RoomMemberHandler(metaclass=abc.ABCMeta):
            ShadowBanError if the requester has been shadow-banned.
        """
        if self.config.server.block_non_admin_invites:
-            is_requester_admin = await self.auth.is_server_admin(requester)
+            is_requester_admin = await self.auth.is_server_admin(requester.user)
            if not is_requester_admin:
                raise SynapseError(
                    403, "Invites have been disabled on this server", Codes.FORBIDDEN
@@ -1693,7 +1693,7 @@ class RoomMemberMasterHandler(RoomMemberHandler):
            check_complexity
            and self.hs.config.server.limit_remote_rooms.admins_can_join
        ):
-            check_complexity = not await self.store.is_server_admin(user)
+            check_complexity = not await self.auth.is_server_admin(user)

        if check_complexity:
            # Fetch the room complexity
@@ -16,11 +16,9 @@ import logging
 from typing import (
    TYPE_CHECKING,
    Any,
-    Collection,
    Dict,
    FrozenSet,
    List,
-    Mapping,
    Optional,
    Sequence,
    Set,
@@ -519,17 +517,10 @@ class SyncHandler:
                # ensure that we always include current state in the timeline
                current_state_ids: FrozenSet[str] = frozenset()
                if any(e.is_state() for e in recents):
-                    # FIXME(faster_joins): We use the partial state here as
-                    # we don't want to block `/sync` on finishing a lazy join.
-                    # Which should be fine once
-                    # https://github.com/matrix-org/synapse/issues/12989 is resolved,
-                    # since we shouldn't reach here anymore?
-                    # Note that we use the current state as a whitelist for filtering
-                    # `recents`, so partial state is only a problem when a membership
-                    # event turns up in `recents` but has not made it into the current
-                    # state.
                    current_state_ids_map = (
-                        await self.store.get_partial_current_state_ids(room_id)
+                        await self._state_storage_controller.get_current_state_ids(
+                            room_id
+                        )
                    )
                    current_state_ids = frozenset(current_state_ids_map.values())

@@ -598,13 +589,7 @@ class SyncHandler:
                if any(e.is_state() for e in loaded_recents):
                    # FIXME(faster_joins): We use the partial state here as
                    # we don't want to block `/sync` on finishing a lazy join.
-                    # Which should be fine once
-                    # https://github.com/matrix-org/synapse/issues/12989 is resolved,
-                    # since we shouldn't reach here anymore?
-                    # Note that we use the current state as a whitelist for filtering
-                    # `loaded_recents`, so partial state is only a problem when a
-                    # membership event turns up in `loaded_recents` but has not made it
-                    # into the current state.
+                    # Is this the correct way of doing it?
                    current_state_ids_map = (
                        await self.store.get_partial_current_state_ids(room_id)
                    )
@@ -652,10 +637,7 @@ class SyncHandler:
        )

    async def get_state_after_event(
-        self,
-        event_id: str,
-        state_filter: Optional[StateFilter] = None,
-        await_full_state: bool = True,
+        self, event_id: str, state_filter: Optional[StateFilter] = None
    ) -> StateMap[str]:
        """
        Get the room state after the given event
@@ -663,14 +645,9 @@ class SyncHandler:
        Args:
            event_id: event of interest
            state_filter: The state filter used to fetch state from the database.
-            await_full_state: if `True`, will block if we do not yet have complete state
-                at the event and `state_filter` is not satisfied by partial state.
-                Defaults to `True`.
        """
        state_ids = await self._state_storage_controller.get_state_ids_for_event(
-            event_id,
-            state_filter=state_filter or StateFilter.all(),
-            await_full_state=await_full_state,
+            event_id, state_filter=state_filter or StateFilter.all()
        )

        # using get_metadata_for_events here (instead of get_event) sidesteps an issue
@@ -693,7 +670,6 @@ class SyncHandler:
        room_id: str,
        stream_position: StreamToken,
        state_filter: Optional[StateFilter] = None,
-        await_full_state: bool = True,
    ) -> StateMap[str]:
        """Get the room state at a particular stream position

@@ -701,9 +677,6 @@ class SyncHandler:
            room_id: room for which to get state
            stream_position: point at which to get state
            state_filter: The state filter used to fetch state from the database.
-            await_full_state: if `True`, will block if we do not yet have complete state
-                at the last event in the room before `stream_position` and
-                `state_filter` is not satisfied by partial state. Defaults to `True`.
        """
        # FIXME: This gets the state at the latest event before the stream ordering,
        # which might not be the same as the "current state" of the room at the time
@@ -715,9 +688,7 @@ class SyncHandler:

        if last_event_id:
            state = await self.get_state_after_event(
-                last_event_id,
-                state_filter=state_filter or StateFilter.all(),
-                await_full_state=await_full_state,
+                last_event_id, state_filter=state_filter or StateFilter.all()
            )

        else:
@@ -920,15 +891,7 @@ class SyncHandler:
        with Measure(self.clock, "compute_state_delta"):
            # The memberships needed for events in the timeline.
            # Only calculated when `lazy_load_members` is on.
-            members_to_fetch: Optional[Set[str]] = None
-
-            # A dictionary mapping user IDs to the first event in the timeline sent by
-            # them. Only calculated when `lazy_load_members` is on.
-            first_event_by_sender_map: Optional[Dict[str, EventBase]] = None
-
-            # The contribution to the room state from state events in the timeline.
-            # Only contains the last event for any given state key.
-            timeline_state: StateMap[str]
+            members_to_fetch = None

            lazy_load_members = sync_config.filter_collection.lazy_load_members()
            include_redundant_members = (
@@ -939,23 +902,10 @@ class SyncHandler:
                # We only request state for the members needed to display the
                # timeline:

-                timeline_state = {}
-
-                members_to_fetch = set()
-                first_event_by_sender_map = {}
-                for event in batch.events:
-                    # Build the map from user IDs to the first timeline event they sent.
-                    if event.sender not in first_event_by_sender_map:
-                        first_event_by_sender_map[event.sender] = event
-
-                    # We need the event's sender, unless their membership was in a
-                    # previous timeline event.
-                    if (EventTypes.Member, event.sender) not in timeline_state:
-                        members_to_fetch.add(event.sender)
-                    # FIXME: we also care about invite targets etc.
-
-                    if event.is_state():
-                        timeline_state[(event.type, event.state_key)] = event.event_id
+                members_to_fetch = {
+                    event.sender  # FIXME: we also care about invite targets etc.
+                    for event in batch.events
+                }

                if full_state:
                    # always make sure we LL ourselves so we know we're in the room
@@ -965,21 +915,16 @@ class SyncHandler:
                    members_to_fetch.add(sync_config.user.to_string())

                state_filter = StateFilter.from_lazy_load_member_list(members_to_fetch)
-
-                # We are happy to use partial state to compute the `/sync` response.
-                # Since partial state may not include the lazy-loaded memberships we
-                # require, we fix up the state response afterwards with memberships from
-                # auth events.
-                await_full_state = False
            else:
-                timeline_state = {
-                    (event.type, event.state_key): event.event_id
-                    for event in batch.events
-                    if event.is_state()
-                }
-
                state_filter = StateFilter.all()
-                await_full_state = True
+
+            # The contribution to the room state from state events in the timeline.
+            # Only contains the last event for any given state key.
+            timeline_state = {
+                (event.type, event.state_key): event.event_id
+                for event in batch.events
+                if event.is_state()
+            }

            # Now calculate the state to return in the sync response for the room.
            # This is more or less the change in state between the end of the previous
@@ -991,26 +936,19 @@ class SyncHandler:
                if batch:
                    state_at_timeline_end = (
                        await self._state_storage_controller.get_state_ids_for_event(
-                            batch.events[-1].event_id,
-                            state_filter=state_filter,
-                            await_full_state=await_full_state,
+                            batch.events[-1].event_id, state_filter=state_filter
                        )
                    )

                    state_at_timeline_start = (
                        await self._state_storage_controller.get_state_ids_for_event(
-                            batch.events[0].event_id,
-                            state_filter=state_filter,
-                            await_full_state=await_full_state,
+                            batch.events[0].event_id, state_filter=state_filter
                        )
                    )

                else:
                    state_at_timeline_end = await self.get_state_at(
-                        room_id,
-                        stream_position=now_token,
-                        state_filter=state_filter,
-                        await_full_state=await_full_state,
+                        room_id, stream_position=now_token, state_filter=state_filter
                    )

                    state_at_timeline_start = state_at_timeline_end
@@ -1026,19 +964,14 @@ class SyncHandler:
                if batch:
                    state_at_timeline_start = (
                        await self._state_storage_controller.get_state_ids_for_event(
-                            batch.events[0].event_id,
-                            state_filter=state_filter,
-                            await_full_state=await_full_state,
+                            batch.events[0].event_id, state_filter=state_filter
                        )
                    )
                else:
                    # We can get here if the user has ignored the senders of all
                    # the recent events.
                    state_at_timeline_start = await self.get_state_at(
-                        room_id,
-                        stream_position=now_token,
-                        state_filter=state_filter,
-                        await_full_state=await_full_state,
+                        room_id, stream_position=now_token, state_filter=state_filter
                    )

                # for now, we disable LL for gappy syncs - see
@@ -1060,28 +993,20 @@ class SyncHandler:
                # is indeed the case.
                assert since_token is not None
                state_at_previous_sync = await self.get_state_at(
-                    room_id,
-                    stream_position=since_token,
-                    state_filter=state_filter,
-                    await_full_state=await_full_state,
+                    room_id, stream_position=since_token, state_filter=state_filter
                )

                if batch:
                    state_at_timeline_end = (
                        await self._state_storage_controller.get_state_ids_for_event(
-                            batch.events[-1].event_id,
-                            state_filter=state_filter,
-                            await_full_state=await_full_state,
+                            batch.events[-1].event_id, state_filter=state_filter
                        )
                    )
                else:
                    # We can get here if the user has ignored the senders of all
                    # the recent events.
                    state_at_timeline_end = await self.get_state_at(
-                        room_id,
-                        stream_position=now_token,
-                        state_filter=state_filter,
-                        await_full_state=await_full_state,
+                        room_id, stream_position=now_token, state_filter=state_filter
                    )

                state_ids = _calculate_state(
@@ -1111,23 +1036,8 @@ class SyncHandler:
                                (EventTypes.Member, member)
                                for member in members_to_fetch
                            ),
-                            await_full_state=False,
                        )

-            # If we only have partial state for the room, `state_ids` may be missing the
-            # memberships we wanted. We attempt to find some by digging through the auth
-            # events of timeline events.
-            if lazy_load_members and await self.store.is_partial_state_room(room_id):
-                assert members_to_fetch is not None
-                assert first_event_by_sender_map is not None
-
-                additional_state_ids = (
-                    await self._find_missing_partial_state_memberships(
-                        room_id, members_to_fetch, first_event_by_sender_map, state_ids
-                    )
-                )
-                state_ids = {**state_ids, **additional_state_ids}
-
            # At this point, if `lazy_load_members` is enabled, `state_ids` includes
            # the memberships of all event senders in the timeline. This is because we
            # may not have sent the memberships in a previous sync.
@@ -1176,99 +1086,6 @@ class SyncHandler:
            if e.type != EventTypes.Aliases  # until MSC2261 or alternative solution
        }

-    async def _find_missing_partial_state_memberships(
-        self,
-        room_id: str,
-        members_to_fetch: Collection[str],
-        events_with_membership_auth: Mapping[str, EventBase],
-        found_state_ids: StateMap[str],
-    ) -> StateMap[str]:
-        """Finds missing memberships from a set of auth events and returns them as a
-        state map.
-
-        Args:
-            room_id: The partial state room to find the remaining memberships for.
-            members_to_fetch: The memberships to find.
-            events_with_membership_auth: A mapping from user IDs to events whose auth
-                events are known to contain their membership.
-            found_state_ids: A dict from (type, state_key) -> state_event_id, containing
-                memberships that have been previously found. Entries in
-                `members_to_fetch` that have a membership in `found_state_ids` are
-                ignored.
-
-        Returns:
-            A dict from ("m.room.member", state_key) -> state_event_id, containing the
-            memberships missing from `found_state_ids`.
-
-        Raises:
-            KeyError: if `events_with_membership_auth` does not have an entry for a
-                missing membership. Memberships in `found_state_ids` do not need an
-                entry in `events_with_membership_auth`.
-        """
-        additional_state_ids: MutableStateMap[str] = {}
-
-        # Tracks the missing members for logging purposes.
-        missing_members = set()
-
-        # Identify memberships missing from `found_state_ids` and pick out the auth
-        # events in which to look for them.
-        auth_event_ids: Set[str] = set()
-        for member in members_to_fetch:
-            if (EventTypes.Member, member) in found_state_ids:
-                continue
-
-            missing_members.add(member)
-            event_with_membership_auth = events_with_membership_auth[member]
-            auth_event_ids.update(event_with_membership_auth.auth_event_ids())
-
-        auth_events = await self.store.get_events(auth_event_ids)
-
-        # Run through the missing memberships once more, picking out the memberships
-        # from the pile of auth events we have just fetched.
-        for member in members_to_fetch:
-            if (EventTypes.Member, member) in found_state_ids:
-                continue
-
-            event_with_membership_auth = events_with_membership_auth[member]
-
-            # Dig through the auth events to find the desired membership.
-            for auth_event_id in event_with_membership_auth.auth_event_ids():
-                # We only store events once we have all their auth events,
-                # so the auth event must be in the pile we have just
-                # fetched.
-                auth_event = auth_events[auth_event_id]
-
-                if (
-                    auth_event.type == EventTypes.Member
-                    and auth_event.state_key == member
-                ):
-                    missing_members.remove(member)
-                    additional_state_ids[
-                        (EventTypes.Member, member)
-                    ] = auth_event.event_id
-                    break
-
-        if missing_members:
-            # There really shouldn't be any missing memberships now. Either:
-            #  * we couldn't find an auth event, which shouldn't happen because we do
-            #    not persist events with persisting their auth events first, or
-            #  * the set of auth events did not contain a membership we wanted, which
-            #    means our caller didn't compute the events in `members_to_fetch`
-            #    correctly, or we somehow accepted an event whose auth events were
-            #    dodgy.
-            logger.error(
-                "Failed to find memberships for %s in partial state room "
-                "%s in the auth events of %s.",
-                missing_members,
-                room_id,
-                [
-                    events_with_membership_auth[member].event_id
-                    for member in missing_members
-                ],
-            )
-
-        return additional_state_ids
-
    async def unread_notifs_for_room_id(
        self, room_id: str, sync_config: SyncConfig
    ) -> NotifCounts:
@@ -1913,11 +1730,7 @@ class SyncHandler:
                continue

            if room_id in sync_result_builder.joined_room_ids or has_join:
-                old_state_ids = await self.get_state_at(
-                    room_id,
-                    since_token,
-                    state_filter=StateFilter.from_types([(EventTypes.Member, user_id)]),
-                )
+                old_state_ids = await self.get_state_at(room_id, since_token)
                old_mem_ev_id = old_state_ids.get((EventTypes.Member, user_id), None)
                old_mem_ev = None
                if old_mem_ev_id:
@@ -1943,13 +1756,7 @@ class SyncHandler:
                    newly_left_rooms.append(room_id)
                else:
                    if not old_state_ids:
-                        old_state_ids = await self.get_state_at(
-                            room_id,
-                            since_token,
-                            state_filter=StateFilter.from_types(
-                                [(EventTypes.Member, user_id)]
-                            ),
-                        )
+                        old_state_ids = await self.get_state_at(room_id, since_token)
                        old_mem_ev_id = old_state_ids.get(
                            (EventTypes.Member, user_id), None
                        )
@@ -253,11 +253,12 @@ class TypingWriterHandler(FollowerTypingHandler):
        self, target_user: UserID, requester: Requester, room_id: str, timeout: int
    ) -> None:
        target_user_id = target_user.to_string()
+        auth_user_id = requester.user.to_string()

        if not self.is_mine_id(target_user_id):
            raise SynapseError(400, "User is not hosted on this homeserver")

-        if target_user != requester.user:
+        if target_user_id != auth_user_id:
            raise AuthError(400, "Cannot set another user's typing state")

        if requester.shadow_banned:
@@ -265,7 +266,7 @@ class TypingWriterHandler(FollowerTypingHandler):
            await self.clock.sleep(random.randint(1, 10))
            raise ShadowBanError()

-        await self.auth.check_user_in_room(room_id, requester)
+        await self.auth.check_user_in_room(room_id, target_user_id)

        logger.debug("%s has started typing in %s", target_user_id, room_id)

@@ -288,11 +289,12 @@ class TypingWriterHandler(FollowerTypingHandler):
        self, target_user: UserID, requester: Requester, room_id: str
    ) -> None:
        target_user_id = target_user.to_string()
+        auth_user_id = requester.user.to_string()

        if not self.is_mine_id(target_user_id):
            raise SynapseError(400, "User is not hosted on this homeserver")

-        if target_user != requester.user:
+        if target_user_id != auth_user_id:
            raise AuthError(400, "Cannot set another user's typing state")

        if requester.shadow_banned:
@@ -300,7 +302,7 @@ class TypingWriterHandler(FollowerTypingHandler):
            await self.clock.sleep(random.randint(1, 10))
            raise ShadowBanError()

-        await self.auth.check_user_in_room(room_id, requester)
+        await self.auth.check_user_in_room(room_id, target_user_id)

        logger.debug("%s has stopped typing in %s", target_user_id, room_id)

@@ -19,6 +19,7 @@ from twisted.web.client import PartialDownloadError

 from synapse.api.constants import LoginType
 from synapse.api.errors import Codes, LoginError, SynapseError
+from synapse.config.emailconfig import ThreepidBehaviour
 from synapse.util import json_decoder

 if TYPE_CHECKING:
@@ -152,7 +153,7 @@ class _BaseThreepidAuthChecker:

        logger.info("Getting validated threepid. threepidcreds: %r", (threepid_creds,))

-        # msisdns are currently always verified via the IS
+        # msisdns are currently always ThreepidBehaviour.REMOTE
        if medium == "msisdn":
            if not self.hs.config.registration.account_threepid_delegate_msisdn:
                raise SynapseError(
@@ -163,7 +164,18 @@ class _BaseThreepidAuthChecker:
                threepid_creds,
            )
        elif medium == "email":
-            if self.hs.config.email.can_verify_email:
+            if (
+                self.hs.config.email.threepid_behaviour_email
+                == ThreepidBehaviour.REMOTE
+            ):
+                assert self.hs.config.registration.account_threepid_delegate_email
+                threepid = await identity_handler.threepid_from_creds(
+                    self.hs.config.registration.account_threepid_delegate_email,
+                    threepid_creds,
+                )
+            elif (
+                self.hs.config.email.threepid_behaviour_email == ThreepidBehaviour.LOCAL
+            ):
                threepid = None
                row = await self.store.get_threepid_validation_session(
                    medium,
@@ -215,7 +227,10 @@ class EmailIdentityAuthChecker(UserInteractiveAuthChecker, _BaseThreepidAuthChec
        _BaseThreepidAuthChecker.__init__(self, hs)

    def is_enabled(self) -> bool:
-        return self.hs.config.email.can_verify_email
+        return self.hs.config.email.threepid_behaviour_email in (
+            ThreepidBehaviour.REMOTE,
+            ThreepidBehaviour.LOCAL,
+        )

    async def check_auth(self, authdict: dict, clientip: str) -> Any:
        return await self._check_threepid("email", authdict)
@@ -226,7 +226,7 @@ class SynapseRequest(Request):

            # If this is a request where the target user doesn't match the user who
            # authenticated (e.g. and admin is puppetting a user) then we return both.
-            if requester != authenticated_entity:
+            if self._requester.user.to_string() != authenticated_entity:
                return requester, authenticated_entity

            return requester, None
@@ -156,7 +156,7 @@ class _LogContextScope(Scope):
    def close(self) -> None:
        active_scope = self.manager.active
        if active_scope is not self:
-            logger.error(
+            logger.warning(
                "Closing scope %s which is not the currently-active one %s",
                self,
                active_scope,
@@ -19,7 +19,7 @@ from typing import Iterable, Pattern
 from synapse.api.auth import Auth
 from synapse.api.errors import AuthError
 from synapse.http.site import SynapseRequest
-from synapse.types import Requester
+from synapse.types import UserID


 def admin_patterns(path_regex: str, version: str = "v1") -> Iterable[Pattern]:
@@ -48,19 +48,19 @@ async def assert_requester_is_admin(auth: Auth, request: SynapseRequest) -> None
        AuthError if the requester is not a server admin
    """
    requester = await auth.get_user_by_req(request)
-    await assert_user_is_admin(auth, requester)
+    await assert_user_is_admin(auth, requester.user)


-async def assert_user_is_admin(auth: Auth, requester: Requester) -> None:
+async def assert_user_is_admin(auth: Auth, user_id: UserID) -> None:
    """Verify that the given user is an admin user

    Args:
        auth: Auth singleton
-        requester: The user making the request, according to the access token.
+        user_id: user to check

    Raises:
        AuthError if the user is not a server admin
    """
-    is_admin = await auth.is_server_admin(requester)
+    is_admin = await auth.is_server_admin(user_id)
    if not is_admin:
        raise AuthError(HTTPStatus.FORBIDDEN, "You are not a server admin")
@@ -54,7 +54,7 @@ class QuarantineMediaInRoom(RestServlet):
        self, request: SynapseRequest, room_id: str
    ) -> Tuple[int, JsonDict]:
        requester = await self.auth.get_user_by_req(request)
-        await assert_user_is_admin(self.auth, requester)
+        await assert_user_is_admin(self.auth, requester.user)

        logging.info("Quarantining room: %s", room_id)

@@ -81,7 +81,7 @@ class QuarantineMediaByUser(RestServlet):
        self, request: SynapseRequest, user_id: str
    ) -> Tuple[int, JsonDict]:
        requester = await self.auth.get_user_by_req(request)
-        await assert_user_is_admin(self.auth, requester)
+        await assert_user_is_admin(self.auth, requester.user)

        logging.info("Quarantining media by user: %s", user_id)

@@ -110,7 +110,7 @@ class QuarantineMediaByID(RestServlet):
        self, request: SynapseRequest, server_name: str, media_id: str
    ) -> Tuple[int, JsonDict]:
        requester = await self.auth.get_user_by_req(request)
-        await assert_user_is_admin(self.auth, requester)
+        await assert_user_is_admin(self.auth, requester.user)

        logging.info("Quarantining media by ID: %s/%s", server_name, media_id)

@@ -75,7 +75,7 @@ class RoomRestV2Servlet(RestServlet):
    ) -> Tuple[int, JsonDict]:

        requester = await self._auth.get_user_by_req(request)
-        await assert_user_is_admin(self._auth, requester)
+        await assert_user_is_admin(self._auth, requester.user)

        content = parse_json_object_from_request(request)

@@ -327,7 +327,7 @@ class RoomRestServlet(RestServlet):
        pagination_handler: "PaginationHandler",
    ) -> Tuple[int, JsonDict]:
        requester = await auth.get_user_by_req(request)
-        await assert_user_is_admin(auth, requester)
+        await assert_user_is_admin(auth, requester.user)

        content = parse_json_object_from_request(request)

@@ -461,7 +461,7 @@ class JoinRoomAliasServlet(ResolveRoomIdMixin, RestServlet):
        assert request.args is not None

        requester = await self.auth.get_user_by_req(request)
-        await assert_user_is_admin(self.auth, requester)
+        await assert_user_is_admin(self.auth, requester.user)

        content = parse_json_object_from_request(request)

@@ -551,7 +551,7 @@ class MakeRoomAdminRestServlet(ResolveRoomIdMixin, RestServlet):
        self, request: SynapseRequest, room_identifier: str
    ) -> Tuple[int, JsonDict]:
        requester = await self.auth.get_user_by_req(request)
-        await assert_user_is_admin(self.auth, requester)
+        await assert_user_is_admin(self.auth, requester.user)
        content = parse_json_object_from_request(request, allow_empty_body=True)

        room_id, _ = await self.resolve_room_id(room_identifier)
@@ -742,7 +742,7 @@ class RoomEventContextServlet(RestServlet):
        self, request: SynapseRequest, room_id: str, event_id: str
    ) -> Tuple[int, JsonDict]:
        requester = await self.auth.get_user_by_req(request, allow_guest=False)
-        await assert_user_is_admin(self.auth, requester)
+        await assert_user_is_admin(self.auth, requester.user)

        limit = parse_integer(request, "limit", default=10)

@@ -834,7 +834,7 @@ class BlockRoomRestServlet(RestServlet):
        self, request: SynapseRequest, room_id: str
    ) -> Tuple[int, JsonDict]:
        requester = await self._auth.get_user_by_req(request)
-        await assert_user_is_admin(self._auth, requester)
+        await assert_user_is_admin(self._auth, requester.user)

        content = parse_json_object_from_request(request)

@@ -183,7 +183,7 @@ class UserRestServletV2(RestServlet):
        self, request: SynapseRequest, user_id: str
    ) -> Tuple[int, JsonDict]:
        requester = await self.auth.get_user_by_req(request)
-        await assert_user_is_admin(self.auth, requester)
+        await assert_user_is_admin(self.auth, requester.user)

        target_user = UserID.from_string(user_id)
        body = parse_json_object_from_request(request)
@@ -575,9 +575,10 @@ class WhoisRestServlet(RestServlet):
    ) -> Tuple[int, JsonDict]:
        target_user = UserID.from_string(user_id)
        requester = await self.auth.get_user_by_req(request)
+        auth_user = requester.user

-        if target_user != requester.user:
-            await assert_user_is_admin(self.auth, requester)
+        if target_user != auth_user:
+            await assert_user_is_admin(self.auth, auth_user)

        if not self.is_mine(target_user):
            raise SynapseError(HTTPStatus.BAD_REQUEST, "Can only whois a local user")
@@ -600,7 +601,7 @@ class DeactivateAccountRestServlet(RestServlet):
        self, request: SynapseRequest, target_user_id: str
    ) -> Tuple[int, JsonDict]:
        requester = await self.auth.get_user_by_req(request)
-        await assert_user_is_admin(self.auth, requester)
+        await assert_user_is_admin(self.auth, requester.user)

        if not self.is_mine(UserID.from_string(target_user_id)):
            raise SynapseError(
@@ -692,7 +693,7 @@ class ResetPasswordRestServlet(RestServlet):
        This needs user to have administrator access in Synapse.
        """
        requester = await self.auth.get_user_by_req(request)
-        await assert_user_is_admin(self.auth, requester)
+        await assert_user_is_admin(self.auth, requester.user)

        UserID.from_string(target_user_id)

@@ -806,7 +807,7 @@ class UserAdminServlet(RestServlet):
        self, request: SynapseRequest, user_id: str
    ) -> Tuple[int, JsonDict]:
        requester = await self.auth.get_user_by_req(request)
-        await assert_user_is_admin(self.auth, requester)
+        await assert_user_is_admin(self.auth, requester.user)
        auth_user = requester.user

        target_user = UserID.from_string(user_id)
@@ -920,7 +921,7 @@ class UserTokenRestServlet(RestServlet):
        self, request: SynapseRequest, user_id: str
    ) -> Tuple[int, JsonDict]:
        requester = await self.auth.get_user_by_req(request)
-        await assert_user_is_admin(self.auth, requester)
+        await assert_user_is_admin(self.auth, requester.user)
        auth_user = requester.user

        if not self.is_mine_id(user_id):
@@ -29,6 +29,7 @@ from synapse.api.errors import (
    SynapseError,
    ThreepidValidationError,
 )
+from synapse.config.emailconfig import ThreepidBehaviour
 from synapse.handlers.ui_auth import UIAuthSessionDataConstants
 from synapse.http.server import HttpServer, finish_request, respond_with_html
 from synapse.http.servlet import (
@@ -67,7 +68,7 @@ class EmailPasswordRequestTokenRestServlet(RestServlet):
        self.config = hs.config
        self.identity_handler = hs.get_identity_handler()

-        if self.config.email.can_verify_email:
+        if self.config.email.threepid_behaviour_email == ThreepidBehaviour.LOCAL:
            self.mailer = Mailer(
                hs=self.hs,
                app_name=self.config.email.email_app_name,
@@ -76,10 +77,11 @@ class EmailPasswordRequestTokenRestServlet(RestServlet):
            )

    async def on_POST(self, request: SynapseRequest) -> Tuple[int, JsonDict]:
-        if not self.config.email.can_verify_email:
-            logger.warning(
-                "User password resets have been disabled due to lack of email config"
-            )
+        if self.config.email.threepid_behaviour_email == ThreepidBehaviour.OFF:
+            if self.config.email.local_threepid_handling_disabled_due_to_email_config:
+                logger.warning(
+                    "User password resets have been disabled due to lack of email config"
+                )
            raise SynapseError(
                400, "Email-based password resets have been disabled on this server"
            )
@@ -115,20 +117,35 @@ class EmailPasswordRequestTokenRestServlet(RestServlet):

            raise SynapseError(400, "Email not found", Codes.THREEPID_NOT_FOUND)

-        # Send password reset emails from Synapse
-        sid = await self.identity_handler.send_threepid_validation(
-            body.email,
-            body.client_secret,
-            body.send_attempt,
-            self.mailer.send_password_reset_mail,
-            body.next_link,
-        )
+        if self.config.email.threepid_behaviour_email == ThreepidBehaviour.REMOTE:
+            assert self.hs.config.registration.account_threepid_delegate_email
+
+            # Have the configured identity server handle the request
+            ret = await self.identity_handler.request_email_token(
+                self.hs.config.registration.account_threepid_delegate_email,
+                body.email,
+                body.client_secret,
+                body.send_attempt,
+                body.next_link,
+            )
+        else:
+            # Send password reset emails from Synapse
+            sid = await self.identity_handler.send_threepid_validation(
+                body.email,
+                body.client_secret,
+                body.send_attempt,
+                self.mailer.send_password_reset_mail,
+                body.next_link,
+            )
+
+            # Wrap the session id in a JSON object
+            ret = {"sid": sid}
+
        threepid_send_requests.labels(type="email", reason="password_reset").observe(
            body.send_attempt
        )

-        # Wrap the session id in a JSON object
-        return 200, {"sid": sid}
+        return 200, ret


 class PasswordRestServlet(RestServlet):
@@ -179,7 +196,7 @@ class PasswordRestServlet(RestServlet):
                params, session_id = await self.auth_handler.validate_user_via_ui_auth(
                    requester,
                    request,
-                    body.dict(exclude_unset=True),
+                    body.dict(),
                    "modify your account password",
                )
            except InteractiveAuthIncompleteError as e:
@@ -202,7 +219,7 @@ class PasswordRestServlet(RestServlet):
                result, params, session_id = await self.auth_handler.check_ui_auth(
                    [[LoginType.EMAIL_IDENTITY]],
                    request,
-                    body.dict(exclude_unset=True),
+                    body.dict(),
                    "modify your account password",
                )
            except InteractiveAuthIncompleteError as e:
@@ -299,7 +316,7 @@ class DeactivateAccountRestServlet(RestServlet):
        await self.auth_handler.validate_user_via_ui_auth(
            requester,
            request,
-            body.dict(exclude_unset=True),
+            body.dict(),
            "deactivate your account",
        )
        result = await self._deactivate_account_handler.deactivate_account(
@@ -323,7 +340,7 @@ class EmailThreepidRequestTokenRestServlet(RestServlet):
        self.identity_handler = hs.get_identity_handler()
        self.store = self.hs.get_datastores().main

-        if self.config.email.can_verify_email:
+        if self.config.email.threepid_behaviour_email == ThreepidBehaviour.LOCAL:
            self.mailer = Mailer(
                hs=self.hs,
                app_name=self.config.email.email_app_name,
@@ -332,10 +349,11 @@ class EmailThreepidRequestTokenRestServlet(RestServlet):
            )

    async def on_POST(self, request: SynapseRequest) -> Tuple[int, JsonDict]:
-        if not self.config.email.can_verify_email:
-            logger.warning(
-                "Adding emails have been disabled due to lack of an email config"
-            )
+        if self.config.email.threepid_behaviour_email == ThreepidBehaviour.OFF:
+            if self.config.email.local_threepid_handling_disabled_due_to_email_config:
+                logger.warning(
+                    "Adding emails have been disabled due to lack of an email config"
+                )
            raise SynapseError(
                400,
                "Adding an email to your account is disabled on this server",
@@ -373,21 +391,35 @@ class EmailThreepidRequestTokenRestServlet(RestServlet):

            raise SynapseError(400, "Email is already in use", Codes.THREEPID_IN_USE)

-        # Send threepid validation emails from Synapse
-        sid = await self.identity_handler.send_threepid_validation(
-            body.email,
-            body.client_secret,
-            body.send_attempt,
-            self.mailer.send_add_threepid_mail,
-            body.next_link,
-        )
+        if self.config.email.threepid_behaviour_email == ThreepidBehaviour.REMOTE:
+            assert self.hs.config.registration.account_threepid_delegate_email
+
+            # Have the configured identity server handle the request
+            ret = await self.identity_handler.request_email_token(
+                self.hs.config.registration.account_threepid_delegate_email,
+                body.email,
+                body.client_secret,
+                body.send_attempt,
+                body.next_link,
+            )
+        else:
+            # Send threepid validation emails from Synapse
+            sid = await self.identity_handler.send_threepid_validation(
+                body.email,
+                body.client_secret,
+                body.send_attempt,
+                self.mailer.send_add_threepid_mail,
+                body.next_link,
+            )
+
+            # Wrap the session id in a JSON object
+            ret = {"sid": sid}

        threepid_send_requests.labels(type="email", reason="add_threepid").observe(
            body.send_attempt
        )

-        # Wrap the session id in a JSON object
-        return 200, {"sid": sid}
+        return 200, ret


 class MsisdnThreepidRequestTokenRestServlet(RestServlet):
@@ -480,19 +512,25 @@ class AddThreepidEmailSubmitTokenServlet(RestServlet):
        self.config = hs.config
        self.clock = hs.get_clock()
        self.store = hs.get_datastores().main
-        if self.config.email.can_verify_email:
+        if self.config.email.threepid_behaviour_email == ThreepidBehaviour.LOCAL:
            self._failure_email_template = (
                self.config.email.email_add_threepid_template_failure_html
            )

    async def on_GET(self, request: Request) -> None:
-        if not self.config.email.can_verify_email:
-            logger.warning(
-                "Adding emails have been disabled due to lack of an email config"
-            )
+        if self.config.email.threepid_behaviour_email == ThreepidBehaviour.OFF:
+            if self.config.email.local_threepid_handling_disabled_due_to_email_config:
+                logger.warning(
+                    "Adding emails have been disabled due to lack of an email config"
+                )
            raise SynapseError(
                400, "Adding an email to your account is disabled on this server"
            )
+        elif self.config.email.threepid_behaviour_email == ThreepidBehaviour.REMOTE:
+            raise SynapseError(
+                400,
+                "This homeserver is not validating threepids.",
+            )

        sid = parse_string(request, "sid", required=True)
        token = parse_string(request, "token", required=True)
@@ -42,26 +42,12 @@ class DevicesRestServlet(RestServlet):
        self.hs = hs
        self.auth = hs.get_auth()
        self.device_handler = hs.get_device_handler()
-        self._msc3852_enabled = hs.config.experimental.msc3852_enabled

    async def on_GET(self, request: SynapseRequest) -> Tuple[int, JsonDict]:
        requester = await self.auth.get_user_by_req(request, allow_guest=True)
        devices = await self.device_handler.get_devices_by_user(
            requester.user.to_string()
        )
-
-        # If MSC3852 is disabled, then the "last_seen_user_agent" field will be
-        # removed from each device. If it is enabled, then the field name will
-        # be replaced by the unstable identifier.
-        #
-        # When MSC3852 is accepted, this block of code can just be removed to
-        # expose "last_seen_user_agent" to clients.
-        for device in devices:
-            last_seen_user_agent = device["last_seen_user_agent"]
-            del device["last_seen_user_agent"]
-            if self._msc3852_enabled:
-                device["org.matrix.msc3852.last_seen_user_agent"] = last_seen_user_agent
-
        return 200, {"devices": devices}


@@ -122,7 +108,6 @@ class DeviceRestServlet(RestServlet):
        self.auth = hs.get_auth()
        self.device_handler = hs.get_device_handler()
        self.auth_handler = hs.get_auth_handler()
-        self._msc3852_enabled = hs.config.experimental.msc3852_enabled

    async def on_GET(
        self, request: SynapseRequest, device_id: str
@@ -133,18 +118,6 @@ class DeviceRestServlet(RestServlet):
        )
        if device is None:
            raise NotFoundError("No device found")
-
-        # If MSC3852 is disabled, then the "last_seen_user_agent" field will be
-        # removed from each device. If it is enabled, then the field name will
-        # be replaced by the unstable identifier.
-        #
-        # When MSC3852 is accepted, this block of code can just be removed to
-        # expose "last_seen_user_agent" to clients.
-        last_seen_user_agent = device["last_seen_user_agent"]
-        del device["last_seen_user_agent"]
-        if self._msc3852_enabled:
-            device["org.matrix.msc3852.last_seen_user_agent"] = last_seen_user_agent
-
        return 200, device

    @interactive_auth_handler
@@ -26,7 +26,7 @@ from synapse.http.servlet import (
    parse_string,
 )
 from synapse.http.site import SynapseRequest
-from synapse.logging.opentracing import log_kv, set_tag
+from synapse.logging.opentracing import log_kv, set_tag, trace_with_opname
 from synapse.types import JsonDict, StreamToken

 from ._base import client_patterns, interactive_auth_handler
@@ -71,6 +71,7 @@ class KeyUploadServlet(RestServlet):
        self.e2e_keys_handler = hs.get_e2e_keys_handler()
        self.device_handler = hs.get_device_handler()

+    @trace_with_opname("upload_keys")
    async def on_POST(
        self, request: SynapseRequest, device_id: Optional[str]
    ) -> Tuple[int, JsonDict]:
@@ -66,7 +66,7 @@ class ProfileDisplaynameRestServlet(RestServlet):
    ) -> Tuple[int, JsonDict]:
        requester = await self.auth.get_user_by_req(request, allow_guest=True)
        user = UserID.from_string(user_id)
-        is_admin = await self.auth.is_server_admin(requester)
+        is_admin = await self.auth.is_server_admin(requester.user)

        content = parse_json_object_from_request(request)

@@ -123,7 +123,7 @@ class ProfileAvatarURLRestServlet(RestServlet):
    ) -> Tuple[int, JsonDict]:
        requester = await self.auth.get_user_by_req(request)
        user = UserID.from_string(user_id)
-        is_admin = await self.auth.is_server_admin(requester)
+        is_admin = await self.auth.is_server_admin(requester.user)

        content = parse_json_object_from_request(request)
        try:
@@ -31,6 +31,7 @@ from synapse.api.errors import (
 )
 from synapse.api.ratelimiting import Ratelimiter
 from synapse.config import ConfigError
+from synapse.config.emailconfig import ThreepidBehaviour
 from synapse.config.homeserver import HomeServerConfig
 from synapse.config.ratelimiting import FederationRatelimitSettings
 from synapse.config.server import is_threepid_reserved
@@ -73,7 +74,7 @@ class EmailRegisterRequestTokenRestServlet(RestServlet):
        self.identity_handler = hs.get_identity_handler()
        self.config = hs.config

-        if self.hs.config.email.can_verify_email:
+        if self.hs.config.email.threepid_behaviour_email == ThreepidBehaviour.LOCAL:
            self.mailer = Mailer(
                hs=self.hs,
                app_name=self.config.email.email_app_name,
@@ -82,10 +83,13 @@ class EmailRegisterRequestTokenRestServlet(RestServlet):
            )

    async def on_POST(self, request: SynapseRequest) -> Tuple[int, JsonDict]:
-        if not self.hs.config.email.can_verify_email:
-            logger.warning(
-                "Email registration has been disabled due to lack of email config"
-            )
+        if self.hs.config.email.threepid_behaviour_email == ThreepidBehaviour.OFF:
+            if (
+                self.hs.config.email.local_threepid_handling_disabled_due_to_email_config
+            ):
+                logger.warning(
+                    "Email registration has been disabled due to lack of email config"
+                )
            raise SynapseError(
                400, "Email-based registration has been disabled on this server"
            )
@@ -134,21 +138,35 @@ class EmailRegisterRequestTokenRestServlet(RestServlet):

            raise SynapseError(400, "Email is already in use", Codes.THREEPID_IN_USE)

-        # Send registration emails from Synapse
-        sid = await self.identity_handler.send_threepid_validation(
-            email,
-            client_secret,
-            send_attempt,
-            self.mailer.send_registration_mail,
-            next_link,
-        )
+        if self.config.email.threepid_behaviour_email == ThreepidBehaviour.REMOTE:
+            assert self.hs.config.registration.account_threepid_delegate_email
+
+            # Have the configured identity server handle the request
+            ret = await self.identity_handler.request_email_token(
+                self.hs.config.registration.account_threepid_delegate_email,
+                email,
+                client_secret,
+                send_attempt,
+                next_link,
+            )
+        else:
+            # Send registration emails from Synapse,
+            # wrapping the session id in a JSON object.
+            ret = {
+                "sid": await self.identity_handler.send_threepid_validation(
+                    email,
+                    client_secret,
+                    send_attempt,
+                    self.mailer.send_registration_mail,
+                    next_link,
+                )
+            }

        threepid_send_requests.labels(type="email", reason="register").observe(
            send_attempt
        )

-        # Wrap the session id in a JSON object
-        return 200, {"sid": sid}
+        return 200, ret


 class MsisdnRegisterRequestTokenRestServlet(RestServlet):
@@ -242,7 +260,7 @@ class RegistrationSubmitTokenServlet(RestServlet):
        self.clock = hs.get_clock()
        self.store = hs.get_datastores().main

-        if self.config.email.can_verify_email:
+        if self.config.email.threepid_behaviour_email == ThreepidBehaviour.LOCAL:
            self._failure_email_template = (
                self.config.email.email_registration_template_failure_html
            )
@@ -252,10 +270,11 @@ class RegistrationSubmitTokenServlet(RestServlet):
            raise SynapseError(
                400, "This medium is currently not supported for registration"
            )
-        if not self.config.email.can_verify_email:
-            logger.warning(
-                "User registration via email has been disabled due to lack of email config"
-            )
+        if self.config.email.threepid_behaviour_email == ThreepidBehaviour.OFF:
+            if self.config.email.local_threepid_handling_disabled_due_to_email_config:
+                logger.warning(
+                    "User registration via email has been disabled due to lack of email config"
+                )
            raise SynapseError(
                400, "Email-based registration is disabled on this server"
            )
@@ -465,6 +484,9 @@ class RegisterRestServlet(RestServlet):
                    "Appservice token must be provided when using a type of m.login.application_service",
                )

+            # Verify the AS
+            self.auth.get_appservice_by_req(request)
+
            # Set the desired user according to the AS API (which uses the
            # 'user' key not 'username'). Since this is a new addition, we'll
            # fallback to 'username' if they gave one.
@@ -16,7 +16,6 @@
 """ This module contains REST servlets to do with rooms: /rooms/<paths> """
 import logging
 import re
-from enum import Enum
 from typing import TYPE_CHECKING, Awaitable, Dict, List, Optional, Tuple
 from urllib import parse as urlparse

@@ -49,7 +48,6 @@ from synapse.http.servlet import (
    parse_strings_from_args,
 )
 from synapse.http.site import SynapseRequest
-from synapse.logging.context import make_deferred_yieldable, run_in_background
 from synapse.logging.opentracing import set_tag
 from synapse.rest.client._base import client_patterns
 from synapse.rest.client.transactions import HttpTransactionCache
@@ -64,33 +62,6 @@ if TYPE_CHECKING:

 logger = logging.getLogger(__name__)

-
-class _RoomSize(Enum):
-    """
-    Enum to differentiate sizes of rooms. This is a pretty good approximation
-    about how hard it will be to get events in the room. We could also look at
-    room "complexity".
-    """
-
-    # This doesn't necessarily mean the room is a DM, just that there is a DM
-    # amount of people there.
-    DM_SIZE = "direct_message_size"
-    SMALL = "small"
-    SUBSTANTIAL = "substantial"
-    LARGE = "large"
-
-    @staticmethod
-    def from_member_count(member_count: int) -> "_RoomSize":
-        if member_count <= 2:
-            return _RoomSize.DM_SIZE
-        elif member_count < 100:
-            return _RoomSize.SMALL
-        elif member_count < 1000:
-            return _RoomSize.SUBSTANTIAL
-        else:
-            return _RoomSize.LARGE
-
-
 # This is an extra metric on top of `synapse_http_server_response_time_seconds`
 # which times the same sort of thing but this one allows us to see values
 # greater than 10s. We use a separate dedicated histogram with its own buckets
@@ -99,11 +70,7 @@ class _RoomSize(Enum):
 messsages_response_timer = Histogram(
    "synapse_room_message_list_rest_servlet_response_time_seconds",
    "sec",
-    # We have a label for room size so we can try to see a more realistic
-    # picture of /messages response time for bigger rooms. We don't want the
-    # tiny rooms that can always respond fast skewing our results when we're trying
-    # to optimize the bigger cases.
-    ["room_size"],
+    [],
    buckets=(
        0.005,
        0.01,
@@ -116,13 +83,9 @@ messsages_response_timer = Histogram(
        2.5,
        5.0,
        10.0,
-        20.0,
        30.0,
        60.0,
-        80.0,
-        100.0,
        120.0,
-        150.0,
        180.0,
        "+Inf",
    ),
@@ -233,7 +196,7 @@ class RoomStateEventRestServlet(TransactionRestServlet):

        msg_handler = self.message_handler
        data = await msg_handler.get_room_data(
-            requester=requester,
+            user_id=requester.user.to_string(),
            room_id=room_id,
            event_type=event_type,
            state_key=state_key,
@@ -578,7 +541,7 @@ class RoomMemberListRestServlet(RestServlet):

        events = await handler.get_state_events(
            room_id=room_id,
-            requester=requester,
+            user_id=requester.user.to_string(),
            at_token=at_token,
            state_filter=StateFilter.from_types([(EventTypes.Member, None)]),
        )
@@ -624,26 +587,14 @@ class RoomMessageListRestServlet(RestServlet):
    def __init__(self, hs: "HomeServer"):
        super().__init__()
        self._hs = hs
-        self.clock = hs.get_clock()
        self.pagination_handler = hs.get_pagination_handler()
        self.auth = hs.get_auth()
        self.store = hs.get_datastores().main

+    @messsages_response_timer.time()
    async def on_GET(
        self, request: SynapseRequest, room_id: str
    ) -> Tuple[int, JsonDict]:
-        processing_start_time = self.clock.time_msec()
-        # Fire off and hope that we get a result by the end.
-        #
-        # We're using the mypy type ignore comment because the `@cached`
-        # decorator on `get_number_joined_users_in_room` doesn't play well with
-        # the type system. Maybe in the future, it can use some ParamSpec
-        # wizardry to fix it up.
-        room_member_count_deferred = run_in_background(  # type: ignore[call-arg]
-            self.store.get_number_joined_users_in_room,
-            room_id,  # type: ignore[arg-type]
-        )
-
        requester = await self.auth.get_user_by_req(request, allow_guest=True)
        pagination_config = await PaginationConfig.from_request(
            self.store, request, default_limit=10
@@ -674,12 +625,6 @@ class RoomMessageListRestServlet(RestServlet):
            event_filter=event_filter,
        )

-        processing_end_time = self.clock.time_msec()
-        room_member_count = await make_deferred_yieldable(room_member_count_deferred)
-        messsages_response_timer.labels(
-            room_size=_RoomSize.from_member_count(room_member_count)
-        ).observe((processing_end_time - processing_start_time) / 1000)
-
        return 200, msgs


@@ -700,7 +645,8 @@ class RoomStateRestServlet(RestServlet):
        # Get all the current state for this room
        events = await self.message_handler.get_state_events(
            room_id=room_id,
-            requester=requester,
+            user_id=requester.user.to_string(),
+            is_guest=requester.is_guest,
        )
        return 200, events

@@ -758,7 +704,7 @@ class RoomEventServlet(RestServlet):
            == "true"
        )
        if include_unredacted_content and not await self.auth.is_server_admin(
-            requester
+            requester.user
        ):
            power_level_event = (
                await self._storage_controllers.state.get_current_state_event(
@@ -1263,7 +1209,9 @@ class TimestampLookupRestServlet(RestServlet):
        self, request: SynapseRequest, room_id: str
    ) -> Tuple[int, JsonDict]:
        requester = await self._auth.get_user_by_req(request)
-        await self._auth.check_user_in_room_or_world_readable(room_id, requester)
+        await self._auth.check_user_in_room_or_world_readable(
+            room_id, requester.user.to_string()
+        )

        timestamp = parse_integer(request, "ts", required=True)
        direction = parse_string(request, "dir", default="f", allowed_values=["f", "b"])
@@ -19,7 +19,7 @@ from synapse.http import servlet
 from synapse.http.server import HttpServer
 from synapse.http.servlet import assert_params_in_dict, parse_json_object_from_request
 from synapse.http.site import SynapseRequest
-from synapse.logging.opentracing import set_tag
+from synapse.logging.opentracing import set_tag, trace_with_opname
 from synapse.rest.client.transactions import HttpTransactionCache
 from synapse.types import JsonDict

@@ -43,6 +43,7 @@ class SendToDeviceRestServlet(servlet.RestServlet):
        self.txns = HttpTransactionCache(hs)
        self.device_message_handler = hs.get_device_message_handler()

+    @trace_with_opname("sendToDevice")
    def on_PUT(
        self, request: SynapseRequest, message_type: str, txn_id: str
    ) -> Awaitable[Tuple[int, JsonDict]]:
@@ -17,6 +17,7 @@ from typing import TYPE_CHECKING, Tuple
 from twisted.web.server import Request

 from synapse.api.errors import ThreepidValidationError
+from synapse.config.emailconfig import ThreepidBehaviour
 from synapse.http.server import DirectServeHtmlResource
 from synapse.http.servlet import parse_string
 from synapse.util.stringutils import assert_valid_client_secret
@@ -45,6 +46,9 @@ class PasswordResetSubmitTokenResource(DirectServeHtmlResource):
        self.clock = hs.get_clock()
        self.store = hs.get_datastores().main

+        self._local_threepid_handling_disabled_due_to_email_config = (
+            hs.config.email.local_threepid_handling_disabled_due_to_email_config
+        )
        self._confirmation_email_template = (
            hs.config.email.email_password_reset_template_confirmation_html
        )
@@ -55,8 +59,8 @@ class PasswordResetSubmitTokenResource(DirectServeHtmlResource):
            hs.config.email.email_password_reset_template_failure_html
        )

-        # This resource should only be mounted if email validation is enabled
-        assert hs.config.email.can_verify_email
+        # This resource should not be mounted if threepid behaviour is not LOCAL
+        assert hs.config.email.threepid_behaviour_email == ThreepidBehaviour.LOCAL

    async def _async_render_GET(self, request: Request) -> Tuple[int, bytes]:
        sid = parse_string(request, "sid", required=True)
@@ -244,7 +244,7 @@ class ServerNoticesManager:
        assert self.server_notices_mxid is not None

        notice_user_data_in_room = await self._message_handler.get_room_data(
-            create_requester(self.server_notices_mxid),
+            self.server_notices_mxid,
            room_id,
            EventTypes.Member,
            self.server_notices_mxid,
@@ -234,7 +234,6 @@ class StateStorageController:
        self,
        event_ids: Collection[str],
        state_filter: Optional[StateFilter] = None,
-        await_full_state: bool = True,
    ) -> Dict[str, StateMap[str]]:
        """
        Get the state dicts corresponding to a list of events, containing the event_ids
@@ -243,9 +242,6 @@ class StateStorageController:
        Args:
            event_ids: events whose state should be returned
            state_filter: The state filter used to fetch state from the database.
-            await_full_state: if `True`, will block if we do not yet have complete state
-                at these events and `state_filter` is not satisfied by partial state.
-                Defaults to `True`.

        Returns:
            A dict from event_id -> (type, state_key) -> event_id
@@ -254,12 +250,8 @@ class StateStorageController:
            RuntimeError if we don't have a state group for one or more of the events
                (ie they are outliers or unknown)
        """
-        if (
-            await_full_state
-            and state_filter
-            and not state_filter.must_await_full_state(self._is_mine_id)
-        ):
-            # Full state is not required if the state filter is restrictive enough.
+        await_full_state = True
+        if state_filter and not state_filter.must_await_full_state(self._is_mine_id):
            await_full_state = False

        event_to_groups = await self.get_state_group_for_events(
@@ -302,10 +294,7 @@ class StateStorageController:

    @trace
    async def get_state_ids_for_event(
-        self,
-        event_id: str,
-        state_filter: Optional[StateFilter] = None,
-        await_full_state: bool = True,
+        self, event_id: str, state_filter: Optional[StateFilter] = None
    ) -> StateMap[str]:
        """
        Get the state dict corresponding to a particular event
@@ -313,9 +302,6 @@ class StateStorageController:
        Args:
            event_id: event whose state should be returned
            state_filter: The state filter used to fetch state from the database.
-            await_full_state: if `True`, will block if we do not yet have complete state
-                at the event and `state_filter` is not satisfied by partial state.
-                Defaults to `True`.

        Returns:
            A dict from (type, state_key) -> state_event_id
@@ -325,9 +311,7 @@ class StateStorageController:
                outlier or is unknown)
        """
        state_map = await self.get_state_ids_for_events(
-            [event_id],
-            state_filter or StateFilter.all(),
-            await_full_state=await_full_state,
+            [event_id], state_filter or StateFilter.all()
        )
        return state_map[event_id]

@@ -69,9 +69,9 @@ class TokenLookupResult:
    """

    user_id: str
-    token_id: int
    is_guest: bool = False
    shadow_banned: bool = False
+    token_id: Optional[int] = None
    device_id: Optional[str] = None
    valid_until_ms: Optional[int] = None
    token_owner: str = attr.ib()
@@ -30,7 +30,7 @@ from synapse.logging.context import (
    run_in_background,
 )
 from synapse.logging.opentracing import start_active_span
-from synapse.metrics import Histogram, LaterGauge
+from synapse.metrics import Histogram
 from synapse.util import Clock

 if typing.TYPE_CHECKING:
@@ -74,27 +74,6 @@ class FederationRateLimiter:
            str, "_PerHostRatelimiter"
        ] = collections.defaultdict(new_limiter)

-        # We track the number of affected hosts per time-period so we can
-        # differentiate one really noisy homeserver from a general
-        # ratelimit tuning problem across the federation.
-        LaterGauge(
-            "synapse_rate_limit_sleep_affected_hosts",
-            "Number of hosts that had requests put to sleep",
-            [],
-            lambda: sum(
-                ratelimiter.should_sleep() for ratelimiter in self.ratelimiters.values()
-            ),
-        )
-        LaterGauge(
-            "synapse_rate_limit_reject_affected_hosts",
-            "Number of hosts that had requests rejected",
-            [],
-            lambda: sum(
-                ratelimiter.should_reject()
-                for ratelimiter in self.ratelimiters.values()
-            ),
-        )
-
    def ratelimit(self, host: str) -> "_GeneratorContextManager[defer.Deferred[None]]":
        """Used to ratelimit an incoming request from a given host

@@ -154,33 +133,12 @@ class _PerHostRatelimiter:
        self.host = host

        request_id = object()
-        # Ideally we'd use `Deferred.fromCoroutine()` here, to save on redundant
-        # type-checking, but we'd need Twisted >= 21.2.
-        ret = defer.ensureDeferred(self._on_enter_with_tracing(request_id))
+        ret = self._on_enter(request_id)
        try:
            yield ret
        finally:
            self._on_exit(request_id)

-    def should_reject(self) -> bool:
-        """
-        Whether to reject the request if we already have too many queued up
-        (either sleeping or in the ready queue).
-        """
-        queue_size = len(self.ready_request_queue) + len(self.sleeping_requests)
-        return queue_size > self.reject_limit
-
-    def should_sleep(self) -> bool:
-        """
-        Whether to sleep the request if we already have too many requests coming
-        through within the window.
-        """
-        return len(self.request_times) > self.sleep_limit
-
-    async def _on_enter_with_tracing(self, request_id: object) -> None:
-        with start_active_span("ratelimit wait"), queue_wait_timer.time():
-            await self._on_enter(request_id)
-
    def _on_enter(self, request_id: object) -> "defer.Deferred[None]":
        time_now = self.clock.time_msec()

@@ -191,7 +149,8 @@ class _PerHostRatelimiter:

        # reject the request if we already have too many queued up (either
        # sleeping or in the ready queue).
-        if self.should_reject():
+        queue_size = len(self.ready_request_queue) + len(self.sleeping_requests)
+        if queue_size > self.reject_limit:
            logger.debug("Ratelimiter(%s): rejecting request", self.host)
            rate_limit_reject_counter.inc()
            raise LimitExceededError(
@@ -221,7 +180,7 @@ class _PerHostRatelimiter:
            len(self.request_times),
        )

-        if self.should_sleep():
+        if len(self.request_times) > self.sleep_limit:
            logger.debug(
                "Ratelimiter(%s) [%s]: sleeping request for %f sec",
                self.host,
@@ -263,8 +222,17 @@ class _PerHostRatelimiter:
            # Ensure that we've properly cleaned up.
            self.sleeping_requests.discard(request_id)
            self.ready_request_queue.pop(request_id, None)
+            wait_span_scope.__exit__(None, None, None)
+            wait_timer_cm.__exit__(None, None, None)
            return r

+        # Tracing
+        wait_span_scope = start_active_span("ratelimit wait")
+        wait_span_scope.__enter__()
+        # Metrics
+        wait_timer_cm = queue_wait_timer.time()
+        wait_timer_cm.__enter__()
+
        ret_defer.addCallbacks(on_start, on_err)
        ret_defer.addBoth(on_both)
        return make_deferred_yieldable(ret_defer)
@@ -284,13 +284,10 @@ class AuthTestCase(unittest.HomeserverTestCase):
            TokenLookupResult(
                user_id="@baldrick:matrix.org",
                device_id="device",
-                token_id=5,
                token_owner="@admin:matrix.org",
-                token_used=True,
            )
        )
        self.store.insert_client_ip = simple_async_mock(None)
-        self.store.mark_access_token_as_used = simple_async_mock(None)
        request = Mock(args={})
        request.getClientAddress.return_value.host = "127.0.0.1"
        request.args[b"access_token"] = [self.test_token]
@@ -304,13 +301,10 @@ class AuthTestCase(unittest.HomeserverTestCase):
            TokenLookupResult(
                user_id="@baldrick:matrix.org",
                device_id="device",
-                token_id=5,
                token_owner="@admin:matrix.org",
-                token_used=True,
            )
        )
        self.store.insert_client_ip = simple_async_mock(None)
-        self.store.mark_access_token_as_used = simple_async_mock(None)
        request = Mock(args={})
        request.getClientAddress.return_value.host = "127.0.0.1"
        request.args[b"access_token"] = [self.test_token]
@@ -353,7 +347,7 @@ class AuthTestCase(unittest.HomeserverTestCase):
        serialized = macaroon.serialize()

        user_info = self.get_success(self.auth.get_user_by_access_token(serialized))
-        self.assertEqual(user_id, user_info.user.to_string())
+        self.assertEqual(user_id, user_info.user_id)
        self.assertTrue(user_info.is_guest)
        self.store.get_user_by_id.assert_called_with(user_id)

@@ -141,6 +141,10 @@ class PresenceRouterTestCase(FederatingHomeserverTestCase):
        hs = self.setup_test_homeserver(
            federation_transport_client=fed_transport_client,
        )
+        # Load the modules into the homeserver
+        module_api = hs.get_module_api()
+        for module, config in hs.config.modules.loaded_modules:
+            module(config=config, api=module_api)

        load_legacy_presence_router(hs)

@@ -322,18 +322,3 @@ class DeactivateAccountTestCase(HomeserverTestCase):
                )
            ),
        )
-
-    def test_deactivate_account_needs_auth(self) -> None:
-        """
-        Tests that making a request to /deactivate with an empty body
-        succeeds in starting the user-interactive auth flow.
-        """
-        req = self.make_request(
-            "POST",
-            "account/deactivate",
-            {},
-            access_token=self.token,
-        )
-
-        self.assertEqual(req.code, 401, req)
-        self.assertEqual(req.json_body["flows"], [{"stages": ["m.login.password"]}])
@@ -21,6 +21,7 @@ from unittest.mock import Mock
 import synapse
 from synapse.api.constants import LoginType
 from synapse.api.errors import Codes
+from synapse.handlers.auth import load_legacy_password_auth_providers
 from synapse.module_api import ModuleApi
 from synapse.rest.client import account, devices, login, logout, register
 from synapse.types import JsonDict, UserID
@@ -166,6 +167,16 @@ class PasswordAuthProviderTests(unittest.HomeserverTestCase):
        mock_password_provider.reset_mock()
        super().setUp()

+    def make_homeserver(self, reactor, clock):
+        hs = self.setup_test_homeserver()
+        # Load the modules into the homeserver
+        module_api = hs.get_module_api()
+        for module, config in hs.config.modules.loaded_modules:
+            module(config=config, api=module_api)
+        load_legacy_password_auth_providers(hs)
+
+        return hs
+
    @override_config(legacy_providers_config(LegacyPasswordOnlyAuthProvider))
    def test_password_only_auth_progiver_login_legacy(self):
        self.password_only_auth_provider_login_test_body()
@@ -22,6 +22,7 @@ from synapse.api.errors import (
    ResourceLimitError,
    SynapseError,
 )
+from synapse.events.spamcheck import load_legacy_spam_checkers
 from synapse.spam_checker_api import RegistrationBehaviour
 from synapse.types import RoomAlias, RoomID, UserID, create_requester

@@ -143,6 +144,12 @@ class RegistrationTestCase(unittest.HomeserverTestCase):
            config=hs_config, federation_client=self.mock_federation_client
        )

+        load_legacy_spam_checkers(hs)
+
+        module_api = hs.get_module_api()
+        for module, config in hs.config.modules.loaded_modules:
+            module(config=config, api=module_api)
+
        return hs

    def prepare(self, reactor, clock, hs):
@@ -14,7 +14,7 @@ from synapse.server import HomeServer
 from synapse.types import UserID, create_requester
 from synapse.util import Clock

-from tests.replication._base import BaseMultiWorkerStreamTestCase
+from tests.replication._base import RedisMultiWorkerStreamTestCase
 from tests.server import make_request
 from tests.test_utils import make_awaitable
 from tests.unittest import FederatingHomeserverTestCase, override_config
@@ -216,7 +216,7 @@ class TestJoinsLimitedByPerRoomRateLimiter(FederatingHomeserverTestCase):
    #   - trying to remote-join again.


-class TestReplicatedJoinsLimitedByPerRoomRateLimiter(BaseMultiWorkerStreamTestCase):
+class TestReplicatedJoinsLimitedByPerRoomRateLimiter(RedisMultiWorkerStreamTestCase):
    servlets = [
        synapse.rest.admin.register_servlets,
        synapse.rest.client.login.register_servlets,
@@ -25,7 +25,7 @@ from synapse.api.constants import EduTypes
 from synapse.api.errors import AuthError
 from synapse.federation.transport.server import TransportLayerServer
 from synapse.server import HomeServer
-from synapse.types import JsonDict, Requester, UserID, create_requester
+from synapse.types import JsonDict, UserID, create_requester
 from synapse.util import Clock

 from tests import unittest
@@ -117,10 +117,8 @@ class TypingNotificationsTestCase(unittest.HomeserverTestCase):

        self.room_members = []

-        async def check_user_in_room(room_id: str, requester: Requester) -> None:
-            if requester.user.to_string() not in [
-                u.to_string() for u in self.room_members
-            ]:
+        async def check_user_in_room(room_id: str, user_id: str) -> None:
+            if user_id not in [u.to_string() for u in self.room_members]:
                raise AuthError(401, "User is not in the room")
            return None

@@ -30,6 +30,7 @@ from tests.replication._base import BaseMultiWorkerStreamTestCase
 from tests.test_utils import simple_async_mock
 from tests.test_utils.event_injection import inject_member_event
 from tests.unittest import HomeserverTestCase, override_config
+from tests.utils import USE_POSTGRES_FOR_TESTS


 class ModuleApiTestCase(HomeserverTestCase):
@@ -737,6 +738,11 @@ class ModuleApiTestCase(HomeserverTestCase):
 class ModuleApiWorkerTestCase(BaseMultiWorkerStreamTestCase):
    """For testing ModuleApi functionality in a multi-worker setup"""

+    # Testing stream ID replication from the main to worker processes requires postgres
+    # (due to needing `MultiWriterIdGenerator`).
+    if not USE_POSTGRES_FOR_TESTS:
+        skip = "Requires Postgres"
+
    servlets = [
        admin.register_servlets,
        login.register_servlets,
@@ -746,6 +752,7 @@ class ModuleApiWorkerTestCase(BaseMultiWorkerStreamTestCase):

    def default_config(self):
        conf = super().default_config()
+        conf["redis"] = {"enabled": "true"}
        conf["stream_writers"] = {"presence": ["presence_writer"]}
        conf["instance_map"] = {
            "presence_writer": {"host": "testserv", "port": 1001},
@@ -24,11 +24,11 @@ from synapse.http.site import SynapseRequest, SynapseSite
 from synapse.replication.http import ReplicationRestResource
 from synapse.replication.tcp.client import ReplicationDataHandler
 from synapse.replication.tcp.handler import ReplicationCommandHandler
-from synapse.replication.tcp.protocol import (
-    ClientReplicationStreamProtocol,
+from synapse.replication.tcp.protocol import ClientReplicationStreamProtocol
+from synapse.replication.tcp.resource import (
+    ReplicationStreamProtocolFactory,
    ServerReplicationStreamProtocol,
 )
-from synapse.replication.tcp.resource import ReplicationStreamProtocolFactory
 from synapse.server import HomeServer

 from tests import unittest
@@ -220,34 +220,15 @@ class BaseStreamTestCase(unittest.HomeserverTestCase):
 class BaseMultiWorkerStreamTestCase(unittest.HomeserverTestCase):
    """Base class for tests running multiple workers.

-    Enables Redis, providing a fake Redis server.
-
    Automatically handle HTTP replication requests from workers to master,
    unlike `BaseStreamTestCase`.
    """

-    if not hiredis:
-        skip = "Requires hiredis"
-
-    if not USE_POSTGRES_FOR_TESTS:
-        # Redis replication only takes place on Postgres
-        skip = "Requires Postgres"
-
-    def default_config(self) -> Dict[str, Any]:
-        """
-        Overrides the default config to enable Redis.
-        Even if the test only uses make_worker_hs, the main process needs Redis
-        enabled otherwise it won't create a Fake Redis server to listen on the
-        Redis port and accept fake TCP connections.
-        """
-        base = super().default_config()
-        base["redis"] = {"enabled": True}
-        return base
-
    def setUp(self):
        super().setUp()

        # build a replication server
+        self.server_factory = ReplicationStreamProtocolFactory(self.hs)
        self.streamer = self.hs.get_replication_streamer()

        # Fake in memory Redis server that servers can connect to.
@@ -266,14 +247,15 @@ class BaseMultiWorkerStreamTestCase(unittest.HomeserverTestCase):
        # handling inbound HTTP requests to that instance.
        self._hs_to_site = {self.hs: self.site}

-        # Handle attempts to connect to fake redis server.
-        self.reactor.add_tcp_client_callback(
-            "localhost",
-            6379,
-            self.connect_any_redis_attempts,
-        )
+        if self.hs.config.redis.redis_enabled:
+            # Handle attempts to connect to fake redis server.
+            self.reactor.add_tcp_client_callback(
+                "localhost",
+                6379,
+                self.connect_any_redis_attempts,
+            )

-        self.hs.get_replication_command_handler().start_replication(self.hs)
+            self.hs.get_replication_command_handler().start_replication(self.hs)

        # When we see a connection attempt to the master replication listener we
        # automatically set up the connection. This is so that tests don't
@@ -357,6 +339,27 @@ class BaseMultiWorkerStreamTestCase(unittest.HomeserverTestCase):
        store = worker_hs.get_datastores().main
        store.db_pool._db_pool = self.database_pool._db_pool

+        # Set up TCP replication between master and the new worker if we don't
+        # have Redis support enabled.
+        if not worker_hs.config.redis.redis_enabled:
+            repl_handler = ReplicationCommandHandler(worker_hs)
+            client = ClientReplicationStreamProtocol(
+                worker_hs,
+                "client",
+                "test",
+                self.clock,
+                repl_handler,
+            )
+            server = self.server_factory.buildProtocol(
+                IPv4Address("TCP", "127.0.0.1", 0)
+            )
+
+            client_transport = FakeTransport(server, self.reactor)
+            client.makeConnection(client_transport)
+
+            server_transport = FakeTransport(client, self.reactor)
+            server.makeConnection(server_transport)
+
        # Set up a resource for the worker
        resource = ReplicationRestResource(worker_hs)

@@ -375,7 +378,8 @@ class BaseMultiWorkerStreamTestCase(unittest.HomeserverTestCase):
            reactor=self.reactor,
        )

-        worker_hs.get_replication_command_handler().start_replication(worker_hs)
+        if worker_hs.config.redis.redis_enabled:
+            worker_hs.get_replication_command_handler().start_replication(worker_hs)

        return worker_hs

@@ -578,3 +582,27 @@ class FakeRedisPubSubProtocol(Protocol):

    def connectionLost(self, reason):
        self._server.remove_subscriber(self)
+
+
+class RedisMultiWorkerStreamTestCase(BaseMultiWorkerStreamTestCase):
+    """
+    A test case that enables Redis, providing a fake Redis server.
+    """
+
+    if not hiredis:
+        skip = "Requires hiredis"
+
+    if not USE_POSTGRES_FOR_TESTS:
+        # Redis replication only takes place on Postgres
+        skip = "Requires Postgres"
+
+    def default_config(self) -> Dict[str, Any]:
+        """
+        Overrides the default config to enable Redis.
+        Even if the test only uses make_worker_hs, the main process needs Redis
+        enabled otherwise it won't create a Fake Redis server to listen on the
+        Redis port and accept fake TCP connections.
+        """
+        base = super().default_config()
+        base["redis"] = {"enabled": True}
+        return base
@@ -12,10 +12,10 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.

-from tests.replication._base import BaseMultiWorkerStreamTestCase
+from tests.replication._base import RedisMultiWorkerStreamTestCase


-class ChannelsTestCase(BaseMultiWorkerStreamTestCase):
+class ChannelsTestCase(RedisMultiWorkerStreamTestCase):
    def test_subscribed_to_enough_redis_channels(self) -> None:
        # The default main process is subscribed to the USER_IP channel.
        self.assertCountEqual(
@@ -20,6 +20,7 @@ from synapse.storage.util.id_generators import MultiWriterIdGenerator

 from tests.replication._base import BaseMultiWorkerStreamTestCase
 from tests.server import make_request
+from tests.utils import USE_POSTGRES_FOR_TESTS

 logger = logging.getLogger(__name__)

@@ -27,6 +28,11 @@ logger = logging.getLogger(__name__)
 class EventPersisterShardTestCase(BaseMultiWorkerStreamTestCase):
    """Checks event persisting sharding works"""

+    # Event persister sharding requires postgres (due to needing
+    # `MultiWriterIdGenerator`).
+    if not USE_POSTGRES_FOR_TESTS:
+        skip = "Requires Postgres"
+
    servlets = [
        admin.register_servlets_for_client_rest_resource,
        room.register_servlets,
@@ -44,6 +50,7 @@ class EventPersisterShardTestCase(BaseMultiWorkerStreamTestCase):

    def default_config(self):
        conf = super().default_config()
+        conf["redis"] = {"enabled": "true"}
        conf["stream_writers"] = {"events": ["worker1", "worker2"]}
        conf["instance_map"] = {
            "worker1": {"host": "testserv", "port": 1001},
@@ -159,62 +159,6 @@ class ServerNoticeTestCase(unittest.HomeserverTestCase):
        self.assertEqual(Codes.UNKNOWN, channel.json_body["errcode"])
        self.assertEqual("'msgtype' not in content", channel.json_body["error"])

-    @override_config(
-        {
-            "server_notices": {
-                "system_mxid_localpart": "notices",
-                "system_mxid_avatar_url": "somthingwrong",
-            },
-            "max_avatar_size": "10M",
-        }
-    )
-    def test_invalid_avatar_url(self) -> None:
-        """If avatar url in homeserver.yaml is invalid and
-        "check avatar size and mime type" is set, an error is returned.
-        TODO: Should be checked when reading the configuration."""
-        channel = self.make_request(
-            "POST",
-            self.url,
-            access_token=self.admin_user_tok,
-            content={
-                "user_id": self.other_user,
-                "content": {"msgtype": "m.text", "body": "test msg"},
-            },
-        )
-
-        self.assertEqual(500, channel.code, msg=channel.json_body)
-        self.assertEqual(Codes.UNKNOWN, channel.json_body["errcode"])
-
-    @override_config(
-        {
-            "server_notices": {
-                "system_mxid_localpart": "notices",
-                "system_mxid_display_name": "test display name",
-                "system_mxid_avatar_url": None,
-            },
-            "max_avatar_size": "10M",
-        }
-    )
-    def test_displayname_is_set_avatar_is_none(self) -> None:
-        """
-        Tests that sending a server notices is successfully,
-        if a display_name is set, avatar_url is `None` and
-        "check avatar size and mime type" is set.
-        """
-        channel = self.make_request(
-            "POST",
-            self.url,
-            access_token=self.admin_user_tok,
-            content={
-                "user_id": self.other_user,
-                "content": {"msgtype": "m.text", "body": "test msg"},
-            },
-        )
-        self.assertEqual(200, channel.code, msg=channel.json_body)
-
-        # user has one invite
-        self._check_invite_and_join_status(self.other_user, 1, 0)
-
    def test_server_notice_disabled(self) -> None:
        """Tests that server returns error if server notice is disabled"""
        channel = self.make_request(
@@ -1,4 +1,4 @@
-# Copyright 2018-2022 The Matrix.org Foundation C.I.C.
+# Copyright 2018-2021 The Matrix.org Foundation C.I.C.
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -904,96 +904,6 @@ class UsersListTestCase(unittest.HomeserverTestCase):
            )


-class UserDevicesTestCase(unittest.HomeserverTestCase):
-    """
-    Tests user device management-related Admin APIs.
-    """
-
-    servlets = [
-        synapse.rest.admin.register_servlets,
-        login.register_servlets,
-        sync.register_servlets,
-    ]
-
-    def prepare(
-        self, reactor: MemoryReactor, clock: Clock, homeserver: HomeServer
-    ) -> None:
-        # Set up an Admin user to query the Admin API with.
-        self.admin_user_id = self.register_user("admin", "pass", admin=True)
-        self.admin_user_token = self.login("admin", "pass")
-
-        # Set up a test user to query the devices of.
-        self.other_user_device_id = "TESTDEVICEID"
-        self.other_user_device_display_name = "My Test Device"
-        self.other_user_client_ip = "1.2.3.4"
-        self.other_user_user_agent = "EquestriaTechnology/123.0"
-
-        self.other_user_id = self.register_user("user", "pass", displayname="User1")
-        self.other_user_token = self.login(
-            "user",
-            "pass",
-            device_id=self.other_user_device_id,
-            additional_request_fields={
-                "initial_device_display_name": self.other_user_device_display_name,
-            },
-        )
-
-        # Have the "other user" make a request so that the "last_seen_*" fields are
-        # populated in the tests below.
-        channel = self.make_request(
-            "GET",
-            "/_matrix/client/v3/sync",
-            access_token=self.other_user_token,
-            client_ip=self.other_user_client_ip,
-            custom_headers=[
-                ("User-Agent", self.other_user_user_agent),
-            ],
-        )
-        self.assertEqual(200, channel.code, msg=channel.json_body)
-
-    def test_list_user_devices(self) -> None:
-        """
-        Tests that a user's devices and attributes are listed correctly via the Admin API.
-        """
-        # Request all devices of "other user"
-        channel = self.make_request(
-            "GET",
-            f"/_synapse/admin/v2/users/{self.other_user_id}/devices",
-            access_token=self.admin_user_token,
-        )
-        self.assertEqual(200, channel.code, msg=channel.json_body)
-
-        # Double-check we got the single device expected
-        user_devices = channel.json_body["devices"]
-        self.assertEqual(len(user_devices), 1)
-        self.assertEqual(channel.json_body["total"], 1)
-
-        # Check that all the attributes of the device reported are as expected.
-        self._validate_attributes_of_device_response(user_devices[0])
-
-        # Request just a single device for "other user" by its ID
-        channel = self.make_request(
-            "GET",
-            f"/_synapse/admin/v2/users/{self.other_user_id}/devices/"
-            f"{self.other_user_device_id}",
-            access_token=self.admin_user_token,
-        )
-        self.assertEqual(200, channel.code, msg=channel.json_body)
-
-        # Check that all the attributes of the device reported are as expected.
-        self._validate_attributes_of_device_response(channel.json_body)
-
-    def _validate_attributes_of_device_response(self, response: JsonDict) -> None:
-        # Check that all device expected attributes are present
-        self.assertEqual(response["user_id"], self.other_user_id)
-        self.assertEqual(response["device_id"], self.other_user_device_id)
-        self.assertEqual(response["display_name"], self.other_user_device_display_name)
-        self.assertEqual(response["last_seen_ip"], self.other_user_client_ip)
-        self.assertEqual(response["last_seen_user_agent"], self.other_user_user_agent)
-        self.assertIsInstance(response["last_seen_ts"], int)
-        self.assertGreater(response["last_seen_ts"], 0)
-
-
 class DeactivateAccountTestCase(unittest.HomeserverTestCase):

    servlets = [
@@ -586,9 +586,9 @@ class RegisterRestServletTestCase(unittest.HomeserverTestCase):
                "require_at_registration": True,
            },
            "account_threepid_delegates": {
+                "email": "https://id_server",
                "msisdn": "https://id_server",
            },
-            "email": {"notif_from": "Synapse <synapse@example.com>"},
        }
    )
    def test_advertised_flows_captcha_and_terms_and_3pids(self) -> None:
@@ -20,7 +20,7 @@ from synapse.api.constants import EventTypes
 from synapse.rest import admin
 from synapse.rest.client import login, room
 from synapse.server import HomeServer
-from synapse.types import JsonDict, create_requester
+from synapse.types import JsonDict
 from synapse.util import Clock
 from synapse.visibility import filter_events_for_client

@@ -188,7 +188,7 @@ class RetentionTestCase(unittest.HomeserverTestCase):
        message_handler = self.hs.get_message_handler()
        create_event = self.get_success(
            message_handler.get_room_data(
-                create_requester(self.user_id), room_id, EventTypes.Create, state_key=""
+                self.user_id, room_id, EventTypes.Create, state_key=""
            )
        )

@@ -26,7 +26,7 @@ from synapse.rest.client import (
    room_upgrade_rest_servlet,
 )
 from synapse.server import HomeServer
-from synapse.types import UserID, create_requester
+from synapse.types import UserID
 from synapse.util import Clock

 from tests import unittest
@@ -275,7 +275,7 @@ class ProfileTestCase(_ShadowBannedBase):
        message_handler = self.hs.get_message_handler()
        event = self.get_success(
            message_handler.get_room_data(
-                create_requester(self.banned_user_id),
+                self.banned_user_id,
                room_id,
                "m.room.member",
                self.banned_user_id,
@@ -310,7 +310,7 @@ class ProfileTestCase(_ShadowBannedBase):
        message_handler = self.hs.get_message_handler()
        event = self.get_success(
            message_handler.get_room_data(
-                create_requester(self.banned_user_id),
+                self.banned_user_id,
                room_id,
                "m.room.member",
                self.banned_user_id,
@@ -61,10 +61,6 @@ from twisted.web.resource import IResource
 from twisted.web.server import Request, Site

 from synapse.config.database import DatabaseConnectionConfig
-from synapse.events.presence_router import load_legacy_presence_router
-from synapse.events.spamcheck import load_legacy_spam_checkers
-from synapse.events.third_party_rules import load_legacy_third_party_event_rules
-from synapse.handlers.auth import load_legacy_password_auth_providers
 from synapse.http.site import SynapseRequest
 from synapse.logging.context import ContextResourceUsage
 from synapse.server import HomeServer
@@ -917,14 +913,4 @@ def setup_test_homeserver(
    # Make the threadpool and database transactions synchronous for testing.
    _make_test_homeserver_synchronous(hs)

-    # Load any configured modules into the homeserver
-    module_api = hs.get_module_api()
-    for module, config in hs.config.modules.loaded_modules:
-        module(config=config, api=module_api)
-
-    load_legacy_spam_checkers(hs)
-    load_legacy_third_party_event_rules(hs)
-    load_legacy_presence_router(hs)
-    load_legacy_password_auth_providers(hs)
-
    return hs
@@ -11,19 +11,16 @@
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 # See the License for the specific language governing permissions and
 # limitations under the License.
-from unittest.mock import Mock

-from twisted.test.proto_helpers import MemoryReactor
+from unittest.mock import Mock

 from synapse.api.constants import EventTypes, LimitBlockingTypes, ServerNoticeMsgType
 from synapse.api.errors import ResourceLimitError
 from synapse.rest import admin
 from synapse.rest.client import login, room, sync
-from synapse.server import HomeServer
 from synapse.server_notices.resource_limits_server_notices import (
    ResourceLimitsServerNotices,
 )
-from synapse.util import Clock

 from tests import unittest
 from tests.test_utils import make_awaitable
@@ -55,7 +52,7 @@ class TestResourceLimitsServerNotices(unittest.HomeserverTestCase):

        return config

-    def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None:
+    def prepare(self, reactor, clock, hs):
        self.server_notices_sender = self.hs.get_server_notices_sender()

        # relying on [1] is far from ideal, but the only case where
@@ -254,7 +251,7 @@ class TestResourceLimitsServerNoticesWithRealRooms(unittest.HomeserverTestCase):
        c["admin_contact"] = "mailto:user@test.com"
        return c

-    def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None:
+    def prepare(self, reactor, clock, hs):
        self.store = self.hs.get_datastores().main
        self.server_notices_sender = self.hs.get_server_notices_sender()
        self.server_notices_manager = self.hs.get_server_notices_manager()
@@ -677,29 +677,14 @@ class HomeserverTestCase(TestCase):
        username: str,
        password: str,
        device_id: Optional[str] = None,
-        additional_request_fields: Optional[Dict[str, str]] = None,
        custom_headers: Optional[Iterable[CustomHeaderType]] = None,
    ) -> str:
        """
        Log in a user, and get an access token. Requires the Login API be registered.
-
-        Args:
-            username: The localpart to assign to the new user.
-            password: The password to assign to the new user.
-            device_id: An optional device ID to assign to the new device created during
-                login.
-            additional_request_fields: A dictionary containing any additional /login
-                request fields and their values.
-            custom_headers: Custom HTTP headers and values to add to the /login request.
-
-        Returns:
-            The newly registered user's Matrix ID.
        """
        body = {"type": "m.login.password", "user": username, "password": password}
        if device_id:
            body["device_id"] = device_id
-        if additional_request_fields:
-            body.update(additional_request_fields)

        channel = self.make_request(
            "POST",
Author	SHA1	Message	Date
Andrew Morgan	9108872355	changelog	2022-08-18 11:58:38 +01:00
Andrew Morgan	40ce117f1f	Downgrade 'Closing scope' error to warning; prevent hammering sentry	2022-08-18 11:46:54 +01:00
				`@@ -0,0 +1 @@`
				Improve validation of request bodies for the following client-server API endpoints: [`/account/password`](https://spec.matrix.org/v1.3/client-server-api/#post_matrixclientv3accountpassword), [`/account/password/email/requestToken`](https://spec.matrix.org/v1.3/client-server-api/#post_matrixclientv3accountpasswordemailrequesttoken), [`/account/deactivate`](https://spec.matrix.org/v1.3/client-server-api/#post_matrixclientv3accountdeactivate) and [`/account/3pid/email/requestToken`](https://spec.matrix.org/v1.3/client-server-api/#post_matrixclientv3account3pidemailrequesttoken).
				`@@ -0,0 +1 @@`
				Allow use of both `@trace` and `@tag_args` stacked on the same function (tracing).
				`@@ -0,0 +1 @@`
				`Faster joins: update the rejected state of events during de-partial-stating.`
				`@@ -0,0 +1 @@`
				Add `openssl` example for generating registration HMAC digest.
				`@@ -0,0 +1 @@`
				Add some miscellaneous comments to document sync, especially around `compute_state_delta`.
				`@@ -0,0 +1 @@`
				Use literals in place of `HTTPStatus` constants in tests.
				`@@ -0,0 +1 @@`
				`Add comments about how event push actions are rotated.`
				`@@ -0,0 +1 @@`
				Instrument the federation/backfill part of `/messages` for understandable traces in Jaeger.
				`@@ -0,0 +1 @@`
				Document that event purging related to the `redaction_retention_period` config option is executed only every 5 minutes.