Stabilise MAS integration (#18759)

This can be reviewed commit by commit

There are a few improvements over the experimental support:

- authorisation of Synapse <-> MAS requests is simplified, with a single
shared secret, removing the need for provisioning a client on the MAS
side
- the tests actually spawn a real server, allowing us to test the rust
introspection layer
- we now check that the device advertised in introspection actually
exist, making it so that when a user logs out, the tokens are
immediately invalidated, even if the cache doesn't expire
- it doesn't rely on discovery anymore, rather on a static endpoint
base. This means users don't have to override the introspection endpoint
to avoid internet roundtrips
- it doesn't depend on `authlib` anymore, as we simplified a lot the
calls done from Synapse to MAS

We still have to update the MAS documentation about the Synapse setup,
but that can be done later.

---------

Co-authored-by: reivilibre <oliverw@element.io>

changelog.d/18759.feature

@@ -0,0 +1 @@
+Stable support for delegating authentication to [Matrix Authentication Service](https://github.com/element-hq/matrix-authentication-service/).

docs/upgrade.md

@@ -164,7 +164,29 @@ The Grafana dashboard JSON in `contrib/grafana/synapse.json` has been updated to
 this change but you will need to manually update your own existing Grafana dashboards
 using these metrics.
 
+## Stable integration with Matrix Authentication Service
 
+Support for [Matrix Authentication Service (MAS)](https://github.com/element-hq/matrix-authentication-service) is now stable, with a simplified configuration.
+This stable integration requires MAS 0.20.0 or later.
+
+The existing `experimental_features.msc3861` configuration option is now deprecated and will be removed in Synapse v1.137.0.
+
+Synapse deployments already using MAS should now use the new configuration options:
+
+```yaml
+matrix_authentication_service:
+  # Enable the MAS integration
+  enabled: true
+  # The base URL where Synapse will contact MAS
+  endpoint: http://localhost:8080
+  # The shared secret used to authenticate MAS requests, must be the same as `matrix.secret` in the MAS configuration
+  # See https://element-hq.github.io/matrix-authentication-service/reference/configuration.html#matrix
+  secret: "asecurerandomsecretstring"
+```
+
+They must remove the `experimental_features.msc3861` configuration option from their configuration.
+
+They can also remove the client previously used by Synapse [in the MAS configuration](https://element-hq.github.io/matrix-authentication-service/reference/configuration.html#clients) as it is no longer in use.
 
 # Upgrading to v1.135.0
 
@@ -186,10 +208,10 @@ native ICU library on your system is no longer required.
 ## Documented endpoint which can be delegated to a federation worker
 
 The endpoint `^/_matrix/federation/v1/version$` can be delegated to a federation
-worker. This is not new behaviour, but had not been documented yet. The 
-[list of delegatable endpoints](workers.md#synapseappgeneric_worker) has 
+worker. This is not new behaviour, but had not been documented yet. The
+[list of delegatable endpoints](workers.md#synapseappgeneric_worker) has
 been updated to include it. Make sure to check your reverse proxy rules if you
-are using workers. 
+are using workers.
 
 # Upgrading to v1.126.0
 

docs/usage/configuration/config_documentation.md

@@ -643,6 +643,28 @@ no_proxy_hosts:
 - 172.30.0.0/16
 ```
 ---
+### `matrix_authentication_service`
+
+*(object)* The `matrix_authentication_service` setting configures integration with [Matrix Authentication Service (MAS)](https://github.com/element-hq/matrix-authentication-service).
+
+This setting has the following sub-options:
+
+* `enabled` (boolean): Whether or not to enable the MAS integration. If this is set to `false`, Synapse will use its legacy internal authentication API. Defaults to `false`.
+
+* `endpoint` (string): The URL where Synapse can reach MAS. This *must* have the `discovery` and `oauth` resources mounted. Defaults to `"http://localhost:8080"`.
+
+* `secret` (string|null): A shared secret that will be used to authenticate requests from and to MAS.
+
+* `secret_path` (string|null): Alternative to `secret`, reading the shared secret from a file. The file should be a plain text file, containing only the secret. Synapse reads the secret from the given file once at startup.
+
+Example configuration:
+```yaml
+matrix_authentication_service:
+  enabled: true
+  secret: someverysecuresecret
+  endpoint: http://localhost:8080
+```
+---
 ### `dummy_events_threshold`
 
 *(integer)* Forward extremities can build up in a room due to networking delays between homeservers. Once this happens in a large room, calculation of the state of that room can become quite expensive. To mitigate this, once the number of forward extremities reaches a given threshold, Synapse will send an `org.matrix.dummy_event` event, which will reduce the forward extremities in the room.

schema/synapse-config.schema.yaml

@@ -656,6 +656,43 @@ properties:
       - - master.hostname.example.com
         - 10.1.0.0/16
         - 172.30.0.0/16
+  matrix_authentication_service:
+    type: object
+    description: >-
+      The `matrix_authentication_service` setting configures integration with
+      [Matrix Authentication Service (MAS)](https://github.com/element-hq/matrix-authentication-service).
+    properties:
+      enabled:
+        type: boolean
+        description: >-
+          Whether or not to enable the MAS integration. If this is set to
+          `false`, Synapse will use its legacy internal authentication API.
+        default: false
+
+      endpoint:
+        type: string
+        format: uri
+        description: >-
+          The URL where Synapse can reach MAS. This *must* have the `discovery`
+          and `oauth` resources mounted.
+        default: http://localhost:8080
+
+      secret:
+        type: ["string", "null"]
+        description: >-
+          A shared secret that will be used to authenticate requests from and to MAS.
+
+      secret_path:
+        type: ["string", "null"]
+        description: >-
+          Alternative to `secret`, reading the shared secret from a file.
+          The file should be a plain text file, containing only the secret.
+          Synapse reads the secret from the given file once at startup.
+
+    examples:
+      - enabled: true
+        secret: someverysecuresecret
+        endpoint: http://localhost:8080
   dummy_events_threshold:
     type: integer
     description: >-

synapse/_pydantic_compat.py

@@ -34,9 +34,11 @@ HAS_PYDANTIC_V2: bool = Version(pydantic_version).major == 2
 
 if TYPE_CHECKING or HAS_PYDANTIC_V2:
     from pydantic.v1 import (
+        AnyHttpUrl,
         BaseModel,
         Extra,
         Field,
+        FilePath,
         MissingError,
         PydanticValueError,
         StrictBool,
@@ -55,9 +57,11 @@ if TYPE_CHECKING or HAS_PYDANTIC_V2:
     from pydantic.v1.typing import get_args
 else:
     from pydantic import (
+        AnyHttpUrl,
         BaseModel,
         Extra,
         Field,
+        FilePath,
         MissingError,
         PydanticValueError,
         StrictBool,
@@ -77,6 +81,7 @@ else:
 
 __all__ = (
     "HAS_PYDANTIC_V2",
+    "AnyHttpUrl",
     "BaseModel",
     "constr",
     "conbytes",
@@ -85,6 +90,7 @@ __all__ = (
     "ErrorWrapper",
     "Extra",
     "Field",
+    "FilePath",
     "get_args",
     "MissingError",
     "parse_obj_as",

synapse/api/auth/__init__.py

@@ -20,10 +20,13 @@
 #
 from typing import TYPE_CHECKING, Optional, Protocol, Tuple
 
+from prometheus_client import Histogram
+
 from twisted.web.server import Request
 
 from synapse.appservice import ApplicationService
 from synapse.http.site import SynapseRequest
+from synapse.metrics import SERVER_NAME_LABEL
 from synapse.types import Requester
 
 if TYPE_CHECKING:
@@ -33,6 +36,13 @@ if TYPE_CHECKING:
 GUEST_DEVICE_ID = "guest_device"
 
 
+introspection_response_timer = Histogram(
+    "synapse_api_auth_delegated_introspection_response",
+    "Time taken to get a response for an introspection request",
+    labelnames=["code", SERVER_NAME_LABEL],
+)
+
+
 class Auth(Protocol):
     """The interface that an auth provider must implement."""
 

synapse/api/auth/mas.py

@@ -0,0 +1,432 @@
+#
+# This file is licensed under the Affero General Public License (AGPL) version 3.
+#
+# Copyright (C) 2025 New Vector, Ltd
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as
+# published by the Free Software Foundation, either version 3 of the
+# License, or (at your option) any later version.
+#
+# See the GNU Affero General Public License for more details:
+# <https://www.gnu.org/licenses/agpl-3.0.html>.
+#
+#
+import logging
+from typing import TYPE_CHECKING, Optional
+from urllib.parse import urlencode
+
+from synapse._pydantic_compat import (
+    BaseModel,
+    Extra,
+    StrictBool,
+    StrictInt,
+    StrictStr,
+    ValidationError,
+)
+from synapse.api.auth.base import BaseAuth
+from synapse.api.errors import (
+    AuthError,
+    HttpResponseException,
+    InvalidClientTokenError,
+    SynapseError,
+    UnrecognizedRequestError,
+)
+from synapse.http.site import SynapseRequest
+from synapse.logging.context import PreserveLoggingContext
+from synapse.logging.opentracing import (
+    active_span,
+    force_tracing,
+    inject_request_headers,
+    start_active_span,
+)
+from synapse.metrics import SERVER_NAME_LABEL
+from synapse.synapse_rust.http_client import HttpClient
+from synapse.types import JsonDict, Requester, UserID, create_requester
+from synapse.util import json_decoder
+from synapse.util.caches.cached_call import RetryOnExceptionCachedCall
+from synapse.util.caches.response_cache import ResponseCache, ResponseCacheContext
+
+from . import introspection_response_timer
+
+if TYPE_CHECKING:
+    from synapse.rest.admin.experimental_features import ExperimentalFeature
+    from synapse.server import HomeServer
+
+logger = logging.getLogger(__name__)
+
+# Scope as defined by MSC2967
+# https://github.com/matrix-org/matrix-spec-proposals/pull/2967
+SCOPE_MATRIX_API = "urn:matrix:org.matrix.msc2967.client:api:*"
+SCOPE_MATRIX_DEVICE_PREFIX = "urn:matrix:org.matrix.msc2967.client:device:"
+
+
+class ServerMetadata(BaseModel):
+    class Config:
+        extra = Extra.allow
+
+    issuer: StrictStr
+    account_management_uri: StrictStr
+
+
+class IntrospectionResponse(BaseModel):
+    retrieved_at_ms: StrictInt
+    active: StrictBool
+    scope: Optional[StrictStr]
+    username: Optional[StrictStr]
+    sub: Optional[StrictStr]
+    device_id: Optional[StrictStr]
+    expires_in: Optional[StrictInt]
+
+    class Config:
+        extra = Extra.allow
+
+    def get_scope_set(self) -> set[str]:
+        if not self.scope:
+            return set()
+
+        return {token for token in self.scope.split(" ") if token}
+
+    def is_active(self, now_ms: int) -> bool:
+        if not self.active:
+            return False
+
+        # Compatibility tokens don't expire and don't have an 'expires_in' field
+        if self.expires_in is None:
+            return True
+
+        absolute_expiry_ms = self.expires_in * 1000 + self.retrieved_at_ms
+        return now_ms < absolute_expiry_ms
+
+
+class MasDelegatedAuth(BaseAuth):
+    def __init__(self, hs: "HomeServer"):
+        super().__init__(hs)
+
+        self.server_name = hs.hostname
+        self._clock = hs.get_clock()
+        self._config = hs.config.mas
+
+        self._http_client = hs.get_proxied_http_client()
+        self._rust_http_client = HttpClient(
+            reactor=hs.get_reactor(),
+            user_agent=self._http_client.user_agent.decode("utf8"),
+        )
+        self._server_metadata = RetryOnExceptionCachedCall[ServerMetadata](
+            self._load_metadata
+        )
+        self._force_tracing_for_users = hs.config.tracing.force_tracing_for_users
+
+        # # Token Introspection Cache
+        # This remembers what users/devices are represented by which access tokens,
+        # in order to reduce overall system load:
+        # - on Synapse (as requests are relatively expensive)
+        # - on the network
+        # - on MAS
+        #
+        # Since there is no invalidation mechanism currently,
+        # the entries expire after 2 minutes.
+        # This does mean tokens can be treated as valid by Synapse
+        # for longer than reality.
+        #
+        # Ideally, tokens should logically be invalidated in the following circumstances:
+        # - If a session logout happens.
+        #   In this case, MAS will delete the device within Synapse
+        #   anyway and this is good enough as an invalidation.
+        # - If the client refreshes their token in MAS.
+        #   In this case, the device still exists and it's not the end of the world for
+        #   the old access token to continue working for a short time.
+        self._introspection_cache: ResponseCache[str] = ResponseCache(
+            clock=self._clock,
+            name="mas_token_introspection",
+            server_name=self.server_name,
+            timeout_ms=120_000,
+            # don't log because the keys are access tokens
+            enable_logging=False,
+        )
+
+    @property
+    def _metadata_url(self) -> str:
+        return f"{self._config.endpoint.rstrip('/')}/.well-known/openid-configuration"
+
+    @property
+    def _introspection_endpoint(self) -> str:
+        return f"{self._config.endpoint.rstrip('/')}/oauth2/introspect"
+
+    async def _load_metadata(self) -> ServerMetadata:
+        response = await self._http_client.get_json(self._metadata_url)
+        metadata = ServerMetadata(**response)
+        return metadata
+
+    async def issuer(self) -> str:
+        metadata = await self._server_metadata.get()
+        return metadata.issuer
+
+    async def account_management_url(self) -> str:
+        metadata = await self._server_metadata.get()
+        return metadata.account_management_uri
+
+    async def auth_metadata(self) -> JsonDict:
+        metadata = await self._server_metadata.get()
+        return metadata.dict()
+
+    def is_request_using_the_shared_secret(self, request: SynapseRequest) -> bool:
+        """
+        Check if the request is using the shared secret.
+
+        Args:
+            request: The request to check.
+
+        Returns:
+            True if the request is using the shared secret, False otherwise.
+        """
+        access_token = self.get_access_token_from_request(request)
+        shared_secret = self._config.secret()
+        if not shared_secret:
+            return False
+
+        return access_token == shared_secret
+
+    async def _introspect_token(
+        self, token: str, cache_context: ResponseCacheContext[str]
+    ) -> IntrospectionResponse:
+        """
+        Send a token to the introspection endpoint and returns the introspection response
+
+        Parameters:
+            token: The token to introspect
+
+        Raises:
+            HttpResponseException: If the introspection endpoint returns a non-2xx response
+            ValueError: If the introspection endpoint returns an invalid JSON response
+            JSONDecodeError: If the introspection endpoint returns a non-JSON response
+            Exception: If the HTTP request fails
+
+        Returns:
+            The introspection response
+        """
+
+        # By default, we shouldn't cache the result unless we know it's valid
+        cache_context.should_cache = False
+        raw_headers: dict[str, str] = {
+            "Content-Type": "application/x-www-form-urlencoded",
+            "Accept": "application/json",
+            "Authorization": f"Bearer {self._config.secret()}",
+            # Tell MAS that we support reading the device ID as an explicit
+            # value, not encoded in the scope. This is supported by MAS 0.15+
+            "X-MAS-Supports-Device-Id": "1",
+        }
+
+        args = {"token": token, "token_type_hint": "access_token"}
+        body = urlencode(args, True)
+
+        # Do the actual request
+
+        logger.debug("Fetching token from MAS")
+        start_time = self._clock.time()
+        try:
+            with start_active_span("mas-introspect-token"):
+                inject_request_headers(raw_headers)
+                with PreserveLoggingContext():
+                    resp_body = await self._rust_http_client.post(
+                        url=self._introspection_endpoint,
+                        response_limit=1 * 1024 * 1024,
+                        headers=raw_headers,
+                        request_body=body,
+                    )
+        except HttpResponseException as e:
+            end_time = self._clock.time()
+            introspection_response_timer.labels(
+                code=e.code, **{SERVER_NAME_LABEL: self.server_name}
+            ).observe(end_time - start_time)
+            raise
+        except Exception:
+            end_time = self._clock.time()
+            introspection_response_timer.labels(
+                code="ERR", **{SERVER_NAME_LABEL: self.server_name}
+            ).observe(end_time - start_time)
+            raise
+
+        logger.debug("Fetched token from MAS")
+
+        end_time = self._clock.time()
+        introspection_response_timer.labels(
+            code=200, **{SERVER_NAME_LABEL: self.server_name}
+        ).observe(end_time - start_time)
+
+        raw_response = json_decoder.decode(resp_body.decode("utf-8"))
+        try:
+            response = IntrospectionResponse(
+                retrieved_at_ms=self._clock.time_msec(),
+                **raw_response,
+            )
+        except ValidationError as e:
+            raise ValueError(
+                "The introspection endpoint returned an invalid JSON response"
+            ) from e
+
+        # We had a valid response, so we can cache it
+        cache_context.should_cache = True
+        return response
+
+    async def is_server_admin(self, requester: Requester) -> bool:
+        return "urn:synapse:admin:*" in requester.scope
+
+    async def get_user_by_req(
+        self,
+        request: SynapseRequest,
+        allow_guest: bool = False,
+        allow_expired: bool = False,
+        allow_locked: bool = False,
+    ) -> Requester:
+        parent_span = active_span()
+        with start_active_span("get_user_by_req"):
+            access_token = self.get_access_token_from_request(request)
+
+            requester = await self.get_appservice_user(request, access_token)
+            if not requester:
+                requester = await self.get_user_by_access_token(
+                    token=access_token,
+                    allow_expired=allow_expired,
+                )
+
+            await self._record_request(request, requester)
+
+            request.requester = requester
+
+            if parent_span:
+                if requester.authenticated_entity in self._force_tracing_for_users:
+                    # request tracing is enabled for this user, so we need to force it
+                    # tracing on for the parent span (which will be the servlet span).
+                    #
+                    # It's too late for the get_user_by_req span to inherit the setting,
+                    # so we also force it on for that.
+                    force_tracing()
+                    force_tracing(parent_span)
+                parent_span.set_tag(
+                    "authenticated_entity", requester.authenticated_entity
+                )
+                parent_span.set_tag("user_id", requester.user.to_string())
+                if requester.device_id is not None:
+                    parent_span.set_tag("device_id", requester.device_id)
+                if requester.app_service is not None:
+                    parent_span.set_tag("appservice_id", requester.app_service.id)
+            return requester
+
+    async def get_user_by_access_token(
+        self,
+        token: str,
+        allow_expired: bool = False,
+    ) -> Requester:
+        try:
+            introspection_result = await self._introspection_cache.wrap(
+                token, self._introspect_token, token, cache_context=True
+            )
+        except Exception:
+            logger.exception("Failed to introspect token")
+            raise SynapseError(503, "Unable to introspect the access token")
+
+        logger.debug("Introspection result: %r", introspection_result)
+        if not introspection_result.is_active(self._clock.time_msec()):
+            raise InvalidClientTokenError("Token is not active")
+
+        # Let's look at the scope
+        scope = introspection_result.get_scope_set()
+
+        # Determine type of user based on presence of particular scopes
+        if SCOPE_MATRIX_API not in scope:
+            raise InvalidClientTokenError(
+                "Token doesn't grant access to the Matrix C-S API"
+            )
+
+        if introspection_result.username is None:
+            raise AuthError(
+                500,
+                "Invalid username claim in the introspection result",
+            )
+
+        user_id = UserID(
+            localpart=introspection_result.username,
+            domain=self.server_name,
+        )
+
+        # Try to find a user from the username claim
+        user_info = await self.store.get_user_by_id(user_id=user_id.to_string())
+        if user_info is None:
+            raise AuthError(
+                500,
+                "User not found",
+            )
+
+        # MAS will give us the device ID as an explicit value for *compatibility* sessions
+        # If present, we get it from here, if not we get it in the scope for next-gen sessions
+        device_id = introspection_result.device_id
+        if device_id is None:
+            # Find device_ids in scope
+            # We only allow a single device_id in the scope, so we find them all in the
+            # scope list, and raise if there are more than one. The OIDC server should be
+            # the one enforcing valid scopes, so we raise a 500 if we find an invalid scope.
+            device_ids = [
+                tok[len(SCOPE_MATRIX_DEVICE_PREFIX) :]
+                for tok in scope
+                if tok.startswith(SCOPE_MATRIX_DEVICE_PREFIX)
+            ]
+
+            if len(device_ids) > 1:
+                raise AuthError(
+                    500,
+                    "Multiple device IDs in scope",
+                )
+
+            device_id = device_ids[0] if device_ids else None
+
+        if device_id is not None:
+            # Sanity check the device_id
+            if len(device_id) > 255 or len(device_id) < 1:
+                raise AuthError(
+                    500,
+                    "Invalid device ID in introspection result",
+                )
+
+            # Make sure the device exists. This helps with introspection cache
+            # invalidation: if we log out, the device gets deleted by MAS
+            device = await self.store.get_device(
+                user_id=user_id.to_string(),
+                device_id=device_id,
+            )
+            if device is None:
+                # Invalidate the introspection cache, the device was deleted
+                self._introspection_cache.unset(token)
+                raise InvalidClientTokenError("Token is not active")
+
+        return create_requester(
+            user_id=user_id,
+            device_id=device_id,
+            scope=scope,
+        )
+
+    async def get_user_by_req_experimental_feature(
+        self,
+        request: SynapseRequest,
+        feature: "ExperimentalFeature",
+        allow_guest: bool = False,
+        allow_expired: bool = False,
+        allow_locked: bool = False,
+    ) -> Requester:
+        try:
+            requester = await self.get_user_by_req(
+                request,
+                allow_guest=allow_guest,
+                allow_expired=allow_expired,
+                allow_locked=allow_locked,
+            )
+            if await self.store.is_feature_enabled(requester.user.to_string(), feature):
+                return requester
+
+            raise UnrecognizedRequestError(code=404)
+        except (AuthError, InvalidClientTokenError):
+            if feature.is_globally_enabled(self.hs.config):
+                # If its globally enabled then return the auth error
+                raise
+
+            raise UnrecognizedRequestError(code=404)

synapse/api/auth/msc3861_delegated.py

@@ -28,7 +28,6 @@ from authlib.oauth2.auth import encode_client_secret_basic, encode_client_secret
 from authlib.oauth2.rfc7523 import ClientSecretJWT, PrivateKeyJWT, private_key_jwt_sign
 from authlib.oauth2.rfc7662 import IntrospectionToken
 from authlib.oidc.discovery import OpenIDProviderMetadata, get_well_known_url
-from prometheus_client import Histogram
 
 from synapse.api.auth.base import BaseAuth
 from synapse.api.errors import (
@@ -54,19 +53,14 @@ from synapse.util import json_decoder
 from synapse.util.caches.cached_call import RetryOnExceptionCachedCall
 from synapse.util.caches.response_cache import ResponseCache, ResponseCacheContext
 
+from . import introspection_response_timer
+
 if TYPE_CHECKING:
     from synapse.rest.admin.experimental_features import ExperimentalFeature
     from synapse.server import HomeServer
 
 logger = logging.getLogger(__name__)
 
-introspection_response_timer = Histogram(
-    "synapse_api_auth_delegated_introspection_response",
-    "Time taken to get a response for an introspection request",
-    labelnames=["code", SERVER_NAME_LABEL],
-)
-
-
 # Scope as defined by MSC2967
 # https://github.com/matrix-org/matrix-spec-proposals/pull/2967
 SCOPE_MATRIX_API = "urn:matrix:org.matrix.msc2967.client:api:*"

synapse/config/_base.pyi

@@ -36,6 +36,7 @@ from synapse.config import (  # noqa: F401
     jwt,
     key,
     logger,
+    mas,
     metrics,
     modules,
     oembed,
@@ -124,6 +125,7 @@ class RootConfig:
     background_updates: background_updates.BackgroundUpdateConfig
     auto_accept_invites: auto_accept_invites.AutoAcceptInvitesConfig
     user_types: user_types.UserTypesConfig
+    mas: mas.MasConfig
 
     config_classes: List[Type["Config"]] = ...
     config_files: List[str]

synapse/config/auth.py

@@ -36,13 +36,14 @@ class AuthConfig(Config):
         if password_config is None:
             password_config = {}
 
-        # The default value of password_config.enabled is True, unless msc3861 is enabled.
-        msc3861_enabled = (
-            (config.get("experimental_features") or {})
-            .get("msc3861", {})
-            .get("enabled", False)
-        )
-        passwords_enabled = password_config.get("enabled", not msc3861_enabled)
+        auth_delegated = (config.get("experimental_features") or {}).get(
+            "msc3861", {}
+        ).get("enabled", False) or (
+            config.get("matrix_authentication_service") or {}
+        ).get("enabled", False)
+
+        # The default value of password_config.enabled is True, unless auth is delegated
+        passwords_enabled = password_config.get("enabled", not auth_delegated)
 
         # 'only_for_reauth' allows users who have previously set a password to use it,
         # even though passwords would otherwise be disabled.

synapse/config/homeserver.py

@@ -36,6 +36,7 @@ from .federation import FederationConfig
 from .jwt import JWTConfig
 from .key import KeyConfig
 from .logger import LoggingConfig
+from .mas import MasConfig
 from .metrics import MetricsConfig
 from .modules import ModulesConfig
 from .oembed import OembedConfig
@@ -109,4 +110,6 @@ class HomeServerConfig(RootConfig):
         BackgroundUpdateConfig,
         AutoAcceptInvitesConfig,
         UserTypesConfig,
+        # This must be last, as it checks for conflicts with other config options.
+        MasConfig,
     ]

synapse/config/mas.py

@@ -0,0 +1,192 @@
+#
+# This file is licensed under the Affero General Public License (AGPL) version 3.
+#
+# Copyright (C) 2025 New Vector, Ltd
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as
+# published by the Free Software Foundation, either version 3 of the
+# License, or (at your option) any later version.
+#
+# See the GNU Affero General Public License for more details:
+# <https://www.gnu.org/licenses/agpl-3.0.html>.
+#
+#
+
+from typing import Any, Optional
+
+from synapse._pydantic_compat import (
+    AnyHttpUrl,
+    Field,
+    FilePath,
+    StrictBool,
+    StrictStr,
+    ValidationError,
+    validator,
+)
+from synapse.config.experimental import read_secret_from_file_once
+from synapse.types import JsonDict
+from synapse.util.pydantic_models import ParseModel
+
+from ._base import Config, ConfigError, RootConfig
+
+
+class MasConfigModel(ParseModel):
+    enabled: StrictBool = False
+    endpoint: AnyHttpUrl = Field(default="http://localhost:8080")
+    secret: Optional[StrictStr] = Field(default=None)
+    secret_path: Optional[FilePath] = Field(default=None)
+
+    @validator("secret")
+    def validate_secret_is_set_if_enabled(cls, v: Any, values: dict) -> Any:
+        if values.get("enabled", False) and not values.get("secret_path") and not v:
+            raise ValueError(
+                "You must set a `secret` or `secret_path` when enabling Matrix Authentication Service integration."
+            )
+
+        return v
+
+    @validator("secret_path")
+    def validate_secret_path_is_set_if_enabled(cls, v: Any, values: dict) -> Any:
+        if values.get("secret"):
+            raise ValueError(
+                "`secret` and `secret_path` cannot be set at the same time."
+            )
+
+        return v
+
+
+class MasConfig(Config):
+    section = "mas"
+
+    def read_config(
+        self, config: JsonDict, allow_secrets_in_config: bool, **kwargs: Any
+    ) -> None:
+        mas_config = config.get("matrix_authentication_service", {})
+        if mas_config is None:
+            mas_config = {}
+
+        try:
+            parsed = MasConfigModel(**mas_config)
+        except ValidationError as e:
+            raise ConfigError(
+                "Could not validate Matrix Authentication Service configuration",
+                path=("matrix_authentication_service",),
+            ) from e
+
+        if parsed.secret and not allow_secrets_in_config:
+            raise ConfigError(
+                "Config options that expect an in-line secret as value are disabled",
+                ("matrix_authentication_service", "secret"),
+            )
+
+        self.enabled = parsed.enabled
+        self.endpoint = parsed.endpoint
+        self._secret = parsed.secret
+        self._secret_path = parsed.secret_path
+
+        self.check_config_conflicts(self.root)
+
+    def check_config_conflicts(
+        self,
+        root: RootConfig,
+    ) -> None:
+        """Checks for any configuration conflicts with other parts of Synapse.
+
+        Raises:
+            ConfigError: If there are any configuration conflicts.
+        """
+
+        if not self.enabled:
+            return
+
+        if root.experimental.msc3861.enabled:
+            raise ConfigError(
+                "Experimental MSC3861 was replaced by Matrix Authentication Service."
+                "Please disable MSC3861 or disable Matrix Authentication Service.",
+                ("experimental", "msc3861"),
+            )
+
+        if (
+            root.auth.password_enabled_for_reauth
+            or root.auth.password_enabled_for_login
+        ):
+            raise ConfigError(
+                "Password auth cannot be enabled when OAuth delegation is enabled",
+                ("password_config", "enabled"),
+            )
+
+        if root.registration.enable_registration:
+            raise ConfigError(
+                "Registration cannot be enabled when OAuth delegation is enabled",
+                ("enable_registration",),
+            )
+
+        # We only need to test the user consent version, as if it must be set if the user_consent section was present in the config
+        if root.consent.user_consent_version is not None:
+            raise ConfigError(
+                "User consent cannot be enabled when OAuth delegation is enabled",
+                ("user_consent",),
+            )
+
+        if (
+            root.oidc.oidc_enabled
+            or root.saml2.saml2_enabled
+            or root.cas.cas_enabled
+            or root.jwt.jwt_enabled
+        ):
+            raise ConfigError("SSO cannot be enabled when OAuth delegation is enabled")
+
+        if bool(root.authproviders.password_providers):
+            raise ConfigError(
+                "Password auth providers cannot be enabled when OAuth delegation is enabled"
+            )
+
+        if root.captcha.enable_registration_captcha:
+            raise ConfigError(
+                "CAPTCHA cannot be enabled when OAuth delegation is enabled",
+                ("captcha", "enable_registration_captcha"),
+            )
+
+        if root.auth.login_via_existing_enabled:
+            raise ConfigError(
+                "Login via existing session cannot be enabled when OAuth delegation is enabled",
+                ("login_via_existing_session", "enabled"),
+            )
+
+        if root.registration.refresh_token_lifetime:
+            raise ConfigError(
+                "refresh_token_lifetime cannot be set when OAuth delegation is enabled",
+                ("refresh_token_lifetime",),
+            )
+
+        if root.registration.nonrefreshable_access_token_lifetime:
+            raise ConfigError(
+                "nonrefreshable_access_token_lifetime cannot be set when OAuth delegation is enabled",
+                ("nonrefreshable_access_token_lifetime",),
+            )
+
+        if root.registration.session_lifetime:
+            raise ConfigError(
+                "session_lifetime cannot be set when OAuth delegation is enabled",
+                ("session_lifetime",),
+            )
+
+        if root.registration.enable_3pid_changes:
+            raise ConfigError(
+                "enable_3pid_changes cannot be enabled when OAuth delegation is enabled",
+                ("enable_3pid_changes",),
+            )
+
+    def secret(self) -> str:
+        if self._secret is not None:
+            return self._secret
+        elif self._secret_path is not None:
+            return read_secret_from_file_once(
+                str(self._secret_path),
+                ("matrix_authentication_service", "secret_path"),
+            )
+        else:
+            raise RuntimeError(
+                "Neither `secret` nor `secret_path` are set, this is a bug.",
+            )

synapse/config/registration.py

@@ -148,15 +148,14 @@ class RegistrationConfig(Config):
         self.enable_set_displayname = config.get("enable_set_displayname", True)
         self.enable_set_avatar_url = config.get("enable_set_avatar_url", True)
 
+        auth_delegated = (config.get("experimental_features") or {}).get(
+            "msc3861", {}
+        ).get("enabled", False) or (
+            config.get("matrix_authentication_service") or {}
+        ).get("enabled", False)
+
         # The default value of enable_3pid_changes is True, unless msc3861 is enabled.
-        msc3861_enabled = (
-            (config.get("experimental_features") or {})
-            .get("msc3861", {})
-            .get("enabled", False)
-        )
-        self.enable_3pid_changes = config.get(
-            "enable_3pid_changes", not msc3861_enabled
-        )
+        self.enable_3pid_changes = config.get("enable_3pid_changes", not auth_delegated)
 
         self.disable_msisdn_registration = config.get(
             "disable_msisdn_registration", False

synapse/handlers/auth.py

@@ -282,7 +282,9 @@ class AuthHandler:
         # response.
         self._extra_attributes: Dict[str, SsoLoginExtraAttributes] = {}
 
-        self.msc3861_oauth_delegation_enabled = hs.config.experimental.msc3861.enabled
+        self._auth_delegation_enabled = (
+            hs.config.mas.enabled or hs.config.experimental.msc3861.enabled
+        )
 
     async def validate_user_via_ui_auth(
         self,
@@ -333,7 +335,7 @@ class AuthHandler:
             LimitExceededError if the ratelimiter's failed request count for this
                 user is too high to proceed
         """
-        if self.msc3861_oauth_delegation_enabled:
+        if self._auth_delegation_enabled:
             raise SynapseError(
                 HTTPStatus.INTERNAL_SERVER_ERROR, "UIA shouldn't be used with MSC3861"
             )

synapse/module_api/__init__.py

@@ -342,7 +342,9 @@ class ModuleApi:
         self._device_handler = hs.get_device_handler()
         self.custom_template_dir = hs.config.server.custom_template_directory
         self._callbacks = hs.get_module_api_callbacks()
-        self.msc3861_oauth_delegation_enabled = hs.config.experimental.msc3861.enabled
+        self._auth_delegation_enabled = (
+            hs.config.mas.enabled or hs.config.experimental.msc3861.enabled
+        )
         self._event_serializer = hs.get_event_client_serializer()
 
         try:
@@ -549,7 +551,7 @@ class ModuleApi:
 
         Added in Synapse v1.46.0.
         """
-        if self.msc3861_oauth_delegation_enabled:
+        if self._auth_delegation_enabled:
             raise ConfigError(
                 "Cannot use password auth provider callbacks when OAuth delegation is enabled"
             )

synapse/rest/admin/__init__.py

@@ -272,11 +272,15 @@ def register_servlets(hs: "HomeServer", http_server: HttpServer) -> None:
     # Admin servlets below may not work on workers.
     if hs.config.worker.worker_app is not None:
         # Some admin servlets can be mounted on workers when MSC3861 is enabled.
+        # Note that this is only for MSC3861 mode, as modern MAS using the
+        # matrix_authentication_service integration uses the dedicated MAS API.
         if hs.config.experimental.msc3861.enabled:
             register_servlets_for_msc3861_delegation(hs, http_server)
 
         return
 
+    auth_delegated = hs.config.mas.enabled or hs.config.experimental.msc3861.enabled
+
     register_servlets_for_client_rest_resource(hs, http_server)
     BlockRoomRestServlet(hs).register(http_server)
     ListRoomRestServlet(hs).register(http_server)
@@ -287,10 +291,10 @@ def register_servlets(hs: "HomeServer", http_server: HttpServer) -> None:
     DeleteRoomStatusByRoomIdRestServlet(hs).register(http_server)
     JoinRoomAliasServlet(hs).register(http_server)
     VersionServlet(hs).register(http_server)
-    if not hs.config.experimental.msc3861.enabled:
+    if not auth_delegated:
         UserAdminServlet(hs).register(http_server)
     UserMembershipRestServlet(hs).register(http_server)
-    if not hs.config.experimental.msc3861.enabled:
+    if not auth_delegated:
         UserTokenRestServlet(hs).register(http_server)
     UserRestServletV2(hs).register(http_server)
     UsersRestServletV2(hs).register(http_server)
@@ -307,7 +311,7 @@ def register_servlets(hs: "HomeServer", http_server: HttpServer) -> None:
     RoomEventContextServlet(hs).register(http_server)
     RateLimitRestServlet(hs).register(http_server)
     UsernameAvailableRestServlet(hs).register(http_server)
-    if not hs.config.experimental.msc3861.enabled:
+    if not auth_delegated:
         ListRegistrationTokensRestServlet(hs).register(http_server)
         NewRegistrationTokenRestServlet(hs).register(http_server)
         RegistrationTokenRestServlet(hs).register(http_server)
@@ -341,16 +345,18 @@ def register_servlets_for_client_rest_resource(
     hs: "HomeServer", http_server: HttpServer
 ) -> None:
     """Register only the servlets which need to be exposed on /_matrix/client/xxx"""
+    auth_delegated = hs.config.mas.enabled or hs.config.experimental.msc3861.enabled
+
     WhoisRestServlet(hs).register(http_server)
     PurgeHistoryStatusRestServlet(hs).register(http_server)
     PurgeHistoryRestServlet(hs).register(http_server)
     # The following resources can only be run on the main process.
     if hs.config.worker.worker_app is None:
         DeactivateAccountRestServlet(hs).register(http_server)
-        if not hs.config.experimental.msc3861.enabled:
+        if not auth_delegated:
             ResetPasswordRestServlet(hs).register(http_server)
     SearchUsersRestServlet(hs).register(http_server)
-    if not hs.config.experimental.msc3861.enabled:
+    if not auth_delegated:
         UserRegisterServlet(hs).register(http_server)
         AccountValidityRenewServlet(hs).register(http_server)
 

synapse/rest/admin/users.py

@@ -109,7 +109,9 @@ class UsersRestServletV2(RestServlet):
         self.auth = hs.get_auth()
         self.admin_handler = hs.get_admin_handler()
         self._msc3866_enabled = hs.config.experimental.msc3866.enabled
-        self._msc3861_enabled = hs.config.experimental.msc3861.enabled
+        self._auth_delegation_enabled = (
+            hs.config.mas.enabled or hs.config.experimental.msc3861.enabled
+        )
 
     async def on_GET(self, request: SynapseRequest) -> Tuple[int, JsonDict]:
         await assert_requester_is_admin(self.auth, request)
@@ -121,10 +123,10 @@ class UsersRestServletV2(RestServlet):
         name = parse_string(request, "name", encoding="utf-8")
 
         guests = parse_boolean(request, "guests", default=True)
-        if self._msc3861_enabled and guests:
+        if self._auth_delegation_enabled and guests:
             raise SynapseError(
                 HTTPStatus.BAD_REQUEST,
-                "The guests parameter is not supported when MSC3861 is enabled.",
+                "The guests parameter is not supported when delegating to MAS.",
                 errcode=Codes.INVALID_PARAM,
             )
 

synapse/rest/client/account.py

@@ -613,7 +613,7 @@ class ThreepidRestServlet(RestServlet):
     # ThreePidBindRestServelet.PostBody with an `alias_generator` to handle
     # `threePidCreds` versus `three_pid_creds`.
     async def on_POST(self, request: SynapseRequest) -> Tuple[int, JsonDict]:
-        if self.hs.config.experimental.msc3861.enabled:
+        if self.hs.config.mas.enabled or self.hs.config.experimental.msc3861.enabled:
             raise NotFoundError(errcode=Codes.UNRECOGNIZED)
 
         if not self.hs.config.registration.enable_3pid_changes:
@@ -905,18 +905,19 @@ class AccountStatusRestServlet(RestServlet):
 
 
 def register_servlets(hs: "HomeServer", http_server: HttpServer) -> None:
+    auth_delegated = hs.config.mas.enabled or hs.config.experimental.msc3861.enabled
+
     ThreepidRestServlet(hs).register(http_server)
     WhoamiRestServlet(hs).register(http_server)
 
-    if not hs.config.experimental.msc3861.enabled:
+    if not auth_delegated:
         DeactivateAccountRestServlet(hs).register(http_server)
 
-    # These servlets are only registered on the main process
     if hs.config.worker.worker_app is None:
         ThreepidBindRestServlet(hs).register(http_server)
         ThreepidUnbindRestServlet(hs).register(http_server)
 
-        if not hs.config.experimental.msc3861.enabled:
+        if not auth_delegated:
             EmailPasswordRequestTokenRestServlet(hs).register(http_server)
             PasswordRestServlet(hs).register(http_server)
             EmailThreepidRequestTokenRestServlet(hs).register(http_server)
@@ -926,5 +927,5 @@ def register_servlets(hs: "HomeServer", http_server: HttpServer) -> None:
             ThreepidAddRestServlet(hs).register(http_server)
             ThreepidDeleteRestServlet(hs).register(http_server)
 
-        if hs.config.experimental.msc3720_enabled:
-            AccountStatusRestServlet(hs).register(http_server)
+    if hs.config.experimental.msc3720_enabled:
+        AccountStatusRestServlet(hs).register(http_server)

synapse/rest/client/auth.py

@@ -20,10 +20,11 @@
 #
 
 import logging
-from typing import TYPE_CHECKING, cast
+from typing import TYPE_CHECKING
 
 from twisted.web.server import Request
 
+from synapse.api.auth.mas import MasDelegatedAuth
 from synapse.api.constants import LoginType
 from synapse.api.errors import LoginError, SynapseError
 from synapse.api.urls import CLIENT_API_PREFIX
@@ -66,22 +67,30 @@ class AuthRestServlet(RestServlet):
         if not session:
             raise SynapseError(400, "No session supplied")
 
-        if (
-            self.hs.config.experimental.msc3861.enabled
-            and stagetype == "org.matrix.cross_signing_reset"
-        ):
-            # If MSC3861 is enabled, we can assume self._auth is an instance of MSC3861DelegatedAuth
-            # We import lazily here because of the authlib requirement
-            from synapse.api.auth.msc3861_delegated import MSC3861DelegatedAuth
+        if stagetype == "org.matrix.cross_signing_reset":
+            if self.hs.config.mas.enabled:
+                assert isinstance(self.auth, MasDelegatedAuth)
 
-            auth = cast(MSC3861DelegatedAuth, self.auth)
-
-            url = await auth.account_management_url()
-            if url is not None:
+                url = await self.auth.account_management_url()
                 url = f"{url}?action=org.matrix.cross_signing_reset"
-            else:
-                url = await auth.issuer()
-            respond_with_redirect(request, str.encode(url))
+                return respond_with_redirect(
+                    request,
+                    url.encode(),
+                )
+
+            elif self.hs.config.experimental.msc3861.enabled:
+                # If MSC3861 is enabled, we can assume self._auth is an instance of MSC3861DelegatedAuth
+                # We import lazily here because of the authlib requirement
+                from synapse.api.auth.msc3861_delegated import MSC3861DelegatedAuth
+
+                assert isinstance(self.auth, MSC3861DelegatedAuth)
+
+                base = await self.auth.account_management_url()
+                if base is not None:
+                    url = f"{base}?action=org.matrix.cross_signing_reset"
+                else:
+                    url = await self.auth.issuer()
+                return respond_with_redirect(request, url.encode())
 
         if stagetype == LoginType.RECAPTCHA:
             html = self.recaptcha_template.render(

synapse/rest/client/auth_metadata.py

@@ -15,6 +15,7 @@ import logging
 import typing
 from typing import Tuple, cast
 
+from synapse.api.auth.mas import MasDelegatedAuth
 from synapse.api.errors import Codes, SynapseError
 from synapse.http.server import HttpServer
 from synapse.http.servlet import RestServlet
@@ -48,13 +49,18 @@ class AuthIssuerServlet(RestServlet):
         self._auth = hs.get_auth()
 
     async def on_GET(self, request: SynapseRequest) -> Tuple[int, JsonDict]:
-        if self._config.experimental.msc3861.enabled:
+        if self._config.mas.enabled:
+            assert isinstance(self._auth, MasDelegatedAuth)
+            return 200, {"issuer": await self._auth.issuer()}
+
+        elif self._config.experimental.msc3861.enabled:
             # If MSC3861 is enabled, we can assume self._auth is an instance of MSC3861DelegatedAuth
             # We import lazily here because of the authlib requirement
             from synapse.api.auth.msc3861_delegated import MSC3861DelegatedAuth
 
-            auth = cast(MSC3861DelegatedAuth, self._auth)
-            return 200, {"issuer": await auth.issuer()}
+            assert isinstance(self._auth, MSC3861DelegatedAuth)
+            return 200, {"issuer": await self._auth.issuer()}
+
         else:
             # Wouldn't expect this to be reached: the servelet shouldn't have been
             # registered. Still, fail gracefully if we are registered for some reason.
@@ -82,13 +88,18 @@ class AuthMetadataServlet(RestServlet):
         self._auth = hs.get_auth()
 
     async def on_GET(self, request: SynapseRequest) -> Tuple[int, JsonDict]:
-        if self._config.experimental.msc3861.enabled:
+        if self._config.mas.enabled:
+            assert isinstance(self._auth, MasDelegatedAuth)
+            return 200, await self._auth.auth_metadata()
+
+        elif self._config.experimental.msc3861.enabled:
             # If MSC3861 is enabled, we can assume self._auth is an instance of MSC3861DelegatedAuth
             # We import lazily here because of the authlib requirement
             from synapse.api.auth.msc3861_delegated import MSC3861DelegatedAuth
 
             auth = cast(MSC3861DelegatedAuth, self._auth)
             return 200, await auth.auth_metadata()
+
         else:
             # Wouldn't expect this to be reached: the servlet shouldn't have been
             # registered. Still, fail gracefully if we are registered for some reason.
@@ -100,7 +111,6 @@ class AuthMetadataServlet(RestServlet):
 
 
 def register_servlets(hs: "HomeServer", http_server: HttpServer) -> None:
-    # We use the MSC3861 values as they are used by multiple MSCs
-    if hs.config.experimental.msc3861.enabled:
+    if hs.config.mas.enabled or hs.config.experimental.msc3861.enabled:
         AuthIssuerServlet(hs).register(http_server)
         AuthMetadataServlet(hs).register(http_server)

synapse/rest/client/devices.py

@@ -144,7 +144,9 @@ class DeviceRestServlet(RestServlet):
         self.device_handler = handler
         self.auth_handler = hs.get_auth_handler()
         self._msc3852_enabled = hs.config.experimental.msc3852_enabled
-        self._msc3861_oauth_delegation_enabled = hs.config.experimental.msc3861.enabled
+        self._auth_delegation_enabled = (
+            hs.config.mas.enabled or hs.config.experimental.msc3861.enabled
+        )
 
     async def on_GET(
         self, request: SynapseRequest, device_id: str
@@ -196,7 +198,7 @@ class DeviceRestServlet(RestServlet):
             pass
 
         else:
-            if self._msc3861_oauth_delegation_enabled:
+            if self._auth_delegation_enabled:
                 raise UnrecognizedRequestError(code=404)
 
             await self.auth_handler.validate_user_via_ui_auth(
@@ -573,7 +575,8 @@ class DehydratedDeviceV2Servlet(RestServlet):
 
 
 def register_servlets(hs: "HomeServer", http_server: HttpServer) -> None:
-    if not hs.config.experimental.msc3861.enabled:
+    auth_delegated = hs.config.mas.enabled or hs.config.experimental.msc3861.enabled
+    if not auth_delegated:
         DeleteDevicesRestServlet(hs).register(http_server)
     DevicesRestServlet(hs).register(http_server)
     DeviceRestServlet(hs).register(http_server)

synapse/rest/client/keys.py

@@ -23,8 +23,9 @@
 import logging
 import re
 from collections import Counter
-from typing import TYPE_CHECKING, Any, Dict, Optional, Tuple, cast
+from typing import TYPE_CHECKING, Any, Dict, Optional, Tuple
 
+from synapse.api.auth.mas import MasDelegatedAuth
 from synapse.api.errors import (
     InteractiveAuthIncompleteError,
     InvalidAPICallError,
@@ -404,19 +405,11 @@ class SigningKeyUploadServlet(RestServlet):
         if is_cross_signing_setup:
             # With MSC3861, UIA is not possible. Instead, the auth service has to
             # explicitly mark the master key as replaceable.
-            if self.hs.config.experimental.msc3861.enabled:
+            if self.hs.config.mas.enabled:
                 if not master_key_updatable_without_uia:
-                    # If MSC3861 is enabled, we can assume self.auth is an instance of MSC3861DelegatedAuth
-                    # We import lazily here because of the authlib requirement
-                    from synapse.api.auth.msc3861_delegated import MSC3861DelegatedAuth
-
-                    auth = cast(MSC3861DelegatedAuth, self.auth)
-
-                    uri = await auth.account_management_url()
-                    if uri is not None:
-                        url = f"{uri}?action=org.matrix.cross_signing_reset"
-                    else:
-                        url = await auth.issuer()
+                    assert isinstance(self.auth, MasDelegatedAuth)
+                    url = await self.auth.account_management_url()
+                    url = f"{url}?action=org.matrix.cross_signing_reset"
 
                     # We use a dummy session ID as this isn't really a UIA flow, but we
                     # reuse the same API shape for better client compatibility.
@@ -437,6 +430,41 @@ class SigningKeyUploadServlet(RestServlet):
                             "then try again.",
                         },
                     )
+
+            elif self.hs.config.experimental.msc3861.enabled:
+                if not master_key_updatable_without_uia:
+                    # If MSC3861 is enabled, we can assume self.auth is an instance of MSC3861DelegatedAuth
+                    # We import lazily here because of the authlib requirement
+                    from synapse.api.auth.msc3861_delegated import MSC3861DelegatedAuth
+
+                    assert isinstance(self.auth, MSC3861DelegatedAuth)
+
+                    uri = await self.auth.account_management_url()
+                    if uri is not None:
+                        url = f"{uri}?action=org.matrix.cross_signing_reset"
+                    else:
+                        url = await self.auth.issuer()
+
+                    # We use a dummy session ID as this isn't really a UIA flow, but we
+                    # reuse the same API shape for better client compatibility.
+                    raise InteractiveAuthIncompleteError(
+                        "dummy",
+                        {
+                            "session": "dummy",
+                            "flows": [
+                                {"stages": ["org.matrix.cross_signing_reset"]},
+                            ],
+                            "params": {
+                                "org.matrix.cross_signing_reset": {
+                                    "url": url,
+                                },
+                            },
+                            "msg": "To reset your end-to-end encryption cross-signing "
+                            f"identity, you first need to approve it at {url} and "
+                            "then try again.",
+                        },
+                    )
+
             else:
                 # Without MSC3861, we require UIA.
                 await self.auth_handler.validate_user_via_ui_auth(

synapse/rest/client/login.py

@@ -715,7 +715,7 @@ class CasTicketServlet(RestServlet):
 
 
 def register_servlets(hs: "HomeServer", http_server: HttpServer) -> None:
-    if hs.config.experimental.msc3861.enabled:
+    if hs.config.mas.enabled or hs.config.experimental.msc3861.enabled:
         return
 
     LoginRestServlet(hs).register(http_server)

synapse/rest/client/logout.py

@@ -86,7 +86,7 @@ class LogoutAllRestServlet(RestServlet):
 
 
 def register_servlets(hs: "HomeServer", http_server: HttpServer) -> None:
-    if hs.config.experimental.msc3861.enabled:
+    if hs.config.mas.enabled or hs.config.experimental.msc3861.enabled:
         return
 
     LogoutRestServlet(hs).register(http_server)

synapse/rest/client/register.py

@@ -1044,7 +1044,7 @@ def _calculate_registration_flows(
 
 
 def register_servlets(hs: "HomeServer", http_server: HttpServer) -> None:
-    if hs.config.experimental.msc3861.enabled:
+    if hs.config.mas.enabled or hs.config.experimental.msc3861.enabled:
         RegisterAppServiceOnlyRestServlet(hs).register(http_server)
         return
 

synapse/rest/synapse/client/__init__.py

@@ -56,8 +56,9 @@ def build_synapse_client_resource_tree(hs: "HomeServer") -> Mapping[str, Resourc
         "/_synapse/client/unsubscribe": UnsubscribeResource(hs),
     }
 
-    # Expose the JWKS endpoint if OAuth2 delegation is enabled
-    if hs.config.experimental.msc3861.enabled:
+    if hs.config.mas.enabled:
+        resources["/_synapse/mas"] = MasResource(hs)
+    elif hs.config.experimental.msc3861.enabled:
         from synapse.rest.synapse.client.jwks import JwksResource
 
         resources["/_synapse/jwks"] = JwksResource(hs)

synapse/rest/synapse/mas/_base.py

@@ -16,6 +16,7 @@
 
 from typing import TYPE_CHECKING, cast
 
+from synapse.api.auth.mas import MasDelegatedAuth
 from synapse.api.errors import SynapseError
 from synapse.http.server import DirectServeJsonResource
 
@@ -27,14 +28,21 @@ if TYPE_CHECKING:
 
 class MasBaseResource(DirectServeJsonResource):
     def __init__(self, hs: "HomeServer"):
-        # Importing this module requires authlib, which is an optional
-        # dependency but required if msc3861 is enabled
-        from synapse.api.auth.msc3861_delegated import MSC3861DelegatedAuth
+        auth = hs.get_auth()
+        if hs.config.mas.enabled:
+            assert isinstance(auth, MasDelegatedAuth)
+
+            self._is_request_from_mas = auth.is_request_using_the_shared_secret
+        else:
+            # Importing this module requires authlib, which is an optional
+            # dependency but required if msc3861 is enabled
+            from synapse.api.auth.msc3861_delegated import MSC3861DelegatedAuth
+
+            assert isinstance(auth, MSC3861DelegatedAuth)
+
+            self._is_request_from_mas = auth.is_request_using_the_admin_token
 
         DirectServeJsonResource.__init__(self, extract_context=True)
-        auth = hs.get_auth()
-        assert isinstance(auth, MSC3861DelegatedAuth)
-        self.msc3861_auth = auth
         self.store = cast("GenericWorkerStore", hs.get_datastores().main)
         self.hostname = hs.hostname
 
@@ -43,5 +51,5 @@ class MasBaseResource(DirectServeJsonResource):
 
         Throws a 403 if the request is not coming from MAS.
         """
-        if not self.msc3861_auth.is_request_using_the_admin_token(request):
+        if not self._is_request_from_mas(request):
             raise SynapseError(403, "This endpoint must only be called by MAS")

synapse/rest/well_known.py

@@ -18,11 +18,12 @@
 #
 #
 import logging
-from typing import TYPE_CHECKING, Optional, Tuple, cast
+from typing import TYPE_CHECKING, Optional, Tuple
 
 from twisted.web.resource import Resource
 from twisted.web.server import Request
 
+from synapse.api.auth.mas import MasDelegatedAuth
 from synapse.api.errors import NotFoundError
 from synapse.http.server import DirectServeJsonResource
 from synapse.http.site import SynapseRequest
@@ -52,18 +53,25 @@ class WellKnownBuilder:
                 "base_url": self._config.registration.default_identity_server
             }
 
-        # We use the MSC3861 values as they are used by multiple MSCs
-        if self._config.experimental.msc3861.enabled:
+        if self._config.mas.enabled:
+            assert isinstance(self._auth, MasDelegatedAuth)
+
+            result["org.matrix.msc2965.authentication"] = {
+                "issuer": await self._auth.issuer(),
+                "account": await self._auth.account_management_url(),
+            }
+
+        elif self._config.experimental.msc3861.enabled:
             # If MSC3861 is enabled, we can assume self._auth is an instance of MSC3861DelegatedAuth
             # We import lazily here because of the authlib requirement
             from synapse.api.auth.msc3861_delegated import MSC3861DelegatedAuth
 
-            auth = cast(MSC3861DelegatedAuth, self._auth)
+            assert isinstance(self._auth, MSC3861DelegatedAuth)
 
             result["org.matrix.msc2965.authentication"] = {
-                "issuer": await auth.issuer(),
+                "issuer": await self._auth.issuer(),
             }
-            account_management_url = await auth.account_management_url()
+            account_management_url = await self._auth.account_management_url()
             if account_management_url is not None:
                 result["org.matrix.msc2965.authentication"]["account"] = (
                     account_management_url

synapse/server.py

@@ -40,6 +40,7 @@ from twisted.web.resource import Resource
 
 from synapse.api.auth import Auth
 from synapse.api.auth.internal import InternalAuth
+from synapse.api.auth.mas import MasDelegatedAuth
 from synapse.api.auth_blocking import AuthBlocking
 from synapse.api.filtering import Filtering
 from synapse.api.ratelimiting import Ratelimiter, RequestRatelimiter
@@ -451,6 +452,8 @@ class HomeServer(metaclass=abc.ABCMeta):
 
     @cache_in_self
     def get_auth(self) -> Auth:
+        if self.config.mas.enabled:
+            return MasDelegatedAuth(self)
         if self.config.experimental.msc3861.enabled:
             from synapse.api.auth.msc3861_delegated import MSC3861DelegatedAuth
 

synapse/synapse_rust/http_client.pyi

@@ -10,17 +10,19 @@
 # See the GNU Affero General Public License for more details:
 # <https://www.gnu.org/licenses/agpl-3.0.html>.
 
-from typing import Awaitable, Mapping
+from typing import Mapping
+
+from twisted.internet.defer import Deferred
 
 from synapse.types import ISynapseReactor
 
 class HttpClient:
     def __init__(self, reactor: ISynapseReactor, user_agent: str) -> None: ...
-    def get(self, url: str, response_limit: int) -> Awaitable[bytes]: ...
+    def get(self, url: str, response_limit: int) -> Deferred[bytes]: ...
     def post(
         self,
         url: str,
         response_limit: int,
         headers: Mapping[str, str],
         request_body: str,
-    ) -> Awaitable[bytes]: ...
+    ) -> Deferred[bytes]: ...

tests/config/test_oauth_delegation.py

@@ -20,6 +20,7 @@
 #
 
 import os
+import tempfile
 from unittest.mock import Mock
 
 from synapse.config import ConfigError
@@ -275,3 +276,168 @@ class MSC3861OAuthDelegation(TestCase):
         self.config_dict["enable_3pid_changes"] = True
         with self.assertRaises(ConfigError):
             self.parse_config()
+
+
+class MasAuthDelegation(TestCase):
+    """Test that the Homeserver fails to initialize if the config is invalid."""
+
+    def setUp(self) -> None:
+        self.config_dict: JsonDict = {
+            **default_config("test"),
+            "public_baseurl": BASE_URL,
+            "enable_registration": False,
+            "matrix_authentication_service": {
+                "enabled": True,
+                "endpoint": "http://localhost:1324/",
+                "secret": "verysecret",
+            },
+        }
+
+    def parse_config(self) -> HomeServerConfig:
+        config = HomeServerConfig()
+        config.parse_config_dict(self.config_dict, "", "")
+        return config
+
+    def test_endpoint_has_to_be_a_url(self) -> None:
+        self.config_dict["matrix_authentication_service"]["endpoint"] = "not a url"
+        with self.assertRaises(ConfigError):
+            self.parse_config()
+
+    def test_secret_and_secret_path_are_mutually_exclusive(self) -> None:
+        with tempfile.NamedTemporaryFile() as f:
+            self.config_dict["matrix_authentication_service"]["secret"] = "verysecret"
+            self.config_dict["matrix_authentication_service"]["secret_path"] = f.name
+            with self.assertRaises(ConfigError):
+                self.parse_config()
+
+    def test_secret_path_loads_secret(self) -> None:
+        with tempfile.NamedTemporaryFile(buffering=0) as f:
+            f.write(b"53C237")
+            del self.config_dict["matrix_authentication_service"]["secret"]
+            self.config_dict["matrix_authentication_service"]["secret_path"] = f.name
+            config = self.parse_config()
+            self.assertEqual(config.mas.secret(), "53C237")
+
+    def test_secret_path_must_exist(self) -> None:
+        del self.config_dict["matrix_authentication_service"]["secret"]
+        self.config_dict["matrix_authentication_service"]["secret_path"] = (
+            "/not/a/valid/file"
+        )
+        with self.assertRaises(ConfigError):
+            self.parse_config()
+
+    def test_registration_cannot_be_enabled(self) -> None:
+        self.config_dict["enable_registration"] = True
+        with self.assertRaises(ConfigError):
+            self.parse_config()
+
+    def test_user_consent_cannot_be_enabled(self) -> None:
+        tmpdir = self.mktemp()
+        os.mkdir(tmpdir)
+        self.config_dict["user_consent"] = {
+            "require_at_registration": True,
+            "version": "1",
+            "template_dir": tmpdir,
+            "server_notice_content": {
+                "msgtype": "m.text",
+                "body": "foo",
+            },
+        }
+        with self.assertRaises(ConfigError):
+            self.parse_config()
+
+    def test_password_config_cannot_be_enabled(self) -> None:
+        self.config_dict["password_config"] = {"enabled": True}
+        with self.assertRaises(ConfigError):
+            self.parse_config()
+
+    @skip_unless(HAS_AUTHLIB, "requires authlib")
+    def test_oidc_sso_cannot_be_enabled(self) -> None:
+        self.config_dict["oidc_providers"] = [
+            {
+                "idp_id": "microsoft",
+                "idp_name": "Microsoft",
+                "issuer": "https://login.microsoftonline.com/<tenant id>/v2.0",
+                "client_id": "<client id>",
+                "client_secret": "<client secret>",
+                "scopes": ["openid", "profile"],
+                "authorization_endpoint": "https://login.microsoftonline.com/<tenant id>/oauth2/v2.0/authorize",
+                "token_endpoint": "https://login.microsoftonline.com/<tenant id>/oauth2/v2.0/token",
+                "userinfo_endpoint": "https://graph.microsoft.com/oidc/userinfo",
+            }
+        ]
+
+        with self.assertRaises(ConfigError):
+            self.parse_config()
+
+    def test_cas_sso_cannot_be_enabled(self) -> None:
+        self.config_dict["cas_config"] = {
+            "enabled": True,
+            "server_url": "https://cas-server.com",
+            "displayname_attribute": "name",
+            "required_attributes": {"userGroup": "staff", "department": "None"},
+        }
+
+        with self.assertRaises(ConfigError):
+            self.parse_config()
+
+    def test_auth_providers_cannot_be_enabled(self) -> None:
+        self.config_dict["modules"] = [
+            {
+                "module": f"{__name__}.{CustomAuthModule.__qualname__}",
+                "config": {},
+            }
+        ]
+
+        # This requires actually setting up an HS, as the module will be run on setup,
+        # which should raise as the module tries to register an auth provider
+        config = self.parse_config()
+        reactor, clock = get_clock()
+        with self.assertRaises(ConfigError):
+            setup_test_homeserver(
+                self.addCleanup, reactor=reactor, clock=clock, config=config
+            )
+
+    @skip_unless(HAS_AUTHLIB, "requires authlib")
+    def test_jwt_auth_cannot_be_enabled(self) -> None:
+        self.config_dict["jwt_config"] = {
+            "enabled": True,
+            "secret": "my-secret-token",
+            "algorithm": "HS256",
+        }
+
+        with self.assertRaises(ConfigError):
+            self.parse_config()
+
+    def test_login_via_existing_session_cannot_be_enabled(self) -> None:
+        self.config_dict["login_via_existing_session"] = {"enabled": True}
+        with self.assertRaises(ConfigError):
+            self.parse_config()
+
+    def test_captcha_cannot_be_enabled(self) -> None:
+        self.config_dict.update(
+            enable_registration_captcha=True,
+            recaptcha_public_key="test",
+            recaptcha_private_key="test",
+        )
+        with self.assertRaises(ConfigError):
+            self.parse_config()
+
+    def test_refreshable_tokens_cannot_be_enabled(self) -> None:
+        self.config_dict.update(
+            refresh_token_lifetime="24h",
+            refreshable_access_token_lifetime="10m",
+            nonrefreshable_access_token_lifetime="24h",
+        )
+        with self.assertRaises(ConfigError):
+            self.parse_config()
+
+    def test_session_lifetime_cannot_be_set(self) -> None:
+        self.config_dict["session_lifetime"] = "24h"
+        with self.assertRaises(ConfigError):
+            self.parse_config()
+
+    def test_enable_3pid_changes_cannot_be_enabled(self) -> None:
+        self.config_dict["enable_3pid_changes"] = True
+        with self.assertRaises(ConfigError):
+            self.parse_config()

tests/handlers/test_oauth_delegation.py

@@ -20,12 +20,16 @@
 #
 
 import json
+import threading
+import time
 from http import HTTPStatus
+from http.server import BaseHTTPRequestHandler, HTTPServer
 from io import BytesIO
-from typing import Any, Dict, Union
+from typing import Any, Coroutine, Dict, Generator, Optional, TypeVar, Union
 from unittest.mock import ANY, AsyncMock, Mock
 from urllib.parse import parse_qs
 
+from parameterized import parameterized_class
 from signedjson.key import (
     encode_verify_key_base64,
     generate_signing_key,
@@ -33,8 +37,10 @@ from signedjson.key import (
 )
 from signedjson.sign import sign_json
 
+from twisted.internet.defer import Deferred, ensureDeferred
 from twisted.internet.testing import MemoryReactor
 
+from synapse.api.auth.mas import MasDelegatedAuth
 from synapse.api.errors import (
     AuthError,
     Codes,
@@ -48,7 +54,7 @@ from synapse.http.site import SynapseRequest
 from synapse.rest import admin
 from synapse.rest.client import account, devices, keys, login, logout, register
 from synapse.server import HomeServer
-from synapse.types import JsonDict, UserID
+from synapse.types import JsonDict, UserID, create_requester
 from synapse.util import Clock
 
 from tests.server import FakeChannel
@@ -109,12 +115,7 @@ async def get_json(url: str) -> JsonDict:
 class MSC3861OAuthDelegation(HomeserverTestCase):
     servlets = [
         account.register_servlets,
-        devices.register_servlets,
         keys.register_servlets,
-        register.register_servlets,
-        login.register_servlets,
-        logout.register_servlets,
-        admin.register_servlets,
     ]
 
     def default_config(self) -> Dict[str, Any]:
@@ -635,6 +636,535 @@ class MSC3861OAuthDelegation(HomeserverTestCase):
 
         self.assertEqual(channel.code, HTTPStatus.UNAUTHORIZED, channel.json_body)
 
+    def test_admin_token(self) -> None:
+        """The handler should return a requester with admin rights when admin_token is used."""
+        self._set_introspection_returnvalue({"active": False})
+
+        request = Mock(args={})
+        request.args[b"access_token"] = [b"admin_token_value"]
+        request.requestHeaders.getRawHeaders = mock_getRawHeaders()
+        requester = self.get_success(self.auth.get_user_by_req(request))
+        self.assertEqual(
+            requester.user.to_string(),
+            OIDC_ADMIN_USERID,
+        )
+        self.assertEqual(requester.is_guest, False)
+        self.assertEqual(requester.device_id, None)
+        self.assertEqual(
+            get_awaitable_result(self.auth.is_server_admin(requester)), True
+        )
+
+        # There should be no call to the introspection endpoint
+        self._rust_client.post.assert_not_called()
+
+    @override_config({"mau_stats_only": True})
+    def test_request_tracking(self) -> None:
+        """Using an access token should update the client_ips and MAU tables."""
+        # To start, there are no MAU users.
+        store = self.hs.get_datastores().main
+        mau = self.get_success(store.get_monthly_active_count())
+        self.assertEqual(mau, 0)
+
+        known_token = "token-token-GOOD-:)"
+
+        async def mock_http_client_request(
+            url: str, request_body: str, **kwargs: Any
+        ) -> bytes:
+            """Mocked auth provider response."""
+            token = parse_qs(request_body)["token"][0]
+            if token == known_token:
+                return json.dumps(
+                    {
+                        "active": True,
+                        "scope": MATRIX_USER_SCOPE,
+                        "sub": SUBJECT,
+                        "username": USERNAME,
+                    },
+                ).encode("utf-8")
+
+            return json.dumps({"active": False}).encode("utf-8")
+
+        self._rust_client.post = mock_http_client_request
+
+        EXAMPLE_IPV4_ADDR = "123.123.123.123"
+        EXAMPLE_USER_AGENT = "httprettygood"
+
+        # First test a known access token
+        channel = FakeChannel(self.site, self.reactor)
+        # type-ignore: FakeChannel is a mock of an HTTPChannel, not a proper HTTPChannel
+        req = SynapseRequest(channel, self.site, self.hs.hostname)  # type: ignore[arg-type]
+        req.client.host = EXAMPLE_IPV4_ADDR
+        req.requestHeaders.addRawHeader("Authorization", f"Bearer {known_token}")
+        req.requestHeaders.addRawHeader("User-Agent", EXAMPLE_USER_AGENT)
+        req.content = BytesIO(b"")
+        req.requestReceived(
+            b"GET",
+            b"/_matrix/client/v3/account/whoami",
+            b"1.1",
+        )
+        channel.await_result()
+        self.assertEqual(channel.code, HTTPStatus.OK, channel.json_body)
+        self.assertEqual(channel.json_body["user_id"], USER_ID, channel.json_body)
+
+        # Expect to see one MAU entry, from the first request
+        mau = self.get_success(store.get_monthly_active_count())
+        self.assertEqual(mau, 1)
+
+        conn_infos = self.get_success(
+            store.get_user_ip_and_agents(UserID.from_string(USER_ID))
+        )
+        self.assertEqual(len(conn_infos), 1, conn_infos)
+        conn_info = conn_infos[0]
+        self.assertEqual(conn_info["access_token"], known_token)
+        self.assertEqual(conn_info["ip"], EXAMPLE_IPV4_ADDR)
+        self.assertEqual(conn_info["user_agent"], EXAMPLE_USER_AGENT)
+
+        # Now test MAS making a request using the special __oidc_admin token
+        MAS_IPV4_ADDR = "127.0.0.1"
+        MAS_USER_AGENT = "masmasmas"
+
+        channel = FakeChannel(self.site, self.reactor)
+        req = SynapseRequest(channel, self.site, self.hs.hostname)  # type: ignore[arg-type]
+        req.client.host = MAS_IPV4_ADDR
+        req.requestHeaders.addRawHeader(
+            "Authorization", f"Bearer {self.auth._admin_token()}"
+        )
+        req.requestHeaders.addRawHeader("User-Agent", MAS_USER_AGENT)
+        req.content = BytesIO(b"")
+        req.requestReceived(
+            b"GET",
+            b"/_matrix/client/v3/account/whoami",
+            b"1.1",
+        )
+        channel.await_result()
+        self.assertEqual(channel.code, HTTPStatus.OK, channel.json_body)
+        self.assertEqual(
+            channel.json_body["user_id"], OIDC_ADMIN_USERID, channel.json_body
+        )
+
+        # Still expect to see one MAU entry, from the first request
+        mau = self.get_success(store.get_monthly_active_count())
+        self.assertEqual(mau, 1)
+
+        conn_infos = self.get_success(
+            store.get_user_ip_and_agents(UserID.from_string(OIDC_ADMIN_USERID))
+        )
+        self.assertEqual(conn_infos, [])
+
+
+class FakeMasHandler(BaseHTTPRequestHandler):
+    server: "FakeMasServer"
+
+    def do_POST(self) -> None:
+        self.server.calls += 1
+
+        if self.path != "/oauth2/introspect":
+            self.send_response(404)
+            self.end_headers()
+            self.wfile.close()
+            return
+
+        auth = self.headers.get("Authorization")
+        if auth is None or auth != f"Bearer {self.server.secret}":
+            self.send_response(401)
+            self.end_headers()
+            self.wfile.close()
+            return
+
+        content_length = self.headers.get("Content-Length")
+        if content_length is None:
+            self.send_response(400)
+            self.end_headers()
+            self.wfile.close()
+            return
+
+        raw_body = self.rfile.read(int(content_length))
+        body = parse_qs(raw_body)
+        param = body.get(b"token")
+        if param is None:
+            self.send_response(400)
+            self.end_headers()
+            self.wfile.close()
+            return
+
+        self.server.last_token_seen = param[0].decode("utf-8")
+
+        self.send_response(200)
+        self.send_header("Content-Type", "application/json")
+        self.end_headers()
+        self.wfile.write(json.dumps(self.server.introspection_response).encode("utf-8"))
+
+    def log_message(self, format: str, *args: Any) -> None:
+        # Don't log anything; by default, the server logs to stderr
+        pass
+
+
+class FakeMasServer(HTTPServer):
+    """A fake MAS server for testing.
+
+    This opens a real HTTP server on a random port, on a separate thread.
+    """
+
+    introspection_response: JsonDict = {}
+    """Determines what the response to the introspection endpoint will be."""
+
+    secret: str = "verysecret"
+    """The shared secret used to authenticate the introspection endpoint."""
+
+    last_token_seen: Optional[str] = None
+    """What is the last access token seen by the introspection endpoint."""
+
+    calls: int = 0
+    """How many times has the introspection endpoint been called."""
+
+    _thread: threading.Thread
+
+    def __init__(self) -> None:
+        super().__init__(("127.0.0.1", 0), FakeMasHandler)
+
+        self._thread = threading.Thread(
+            target=self.serve_forever,
+            name="FakeMasServer",
+            kwargs={"poll_interval": 0.01},
+            daemon=True,
+        )
+        self._thread.start()
+
+    def shutdown(self) -> None:
+        super().shutdown()
+        self._thread.join()
+
+    @property
+    def endpoint(self) -> str:
+        return f"http://127.0.0.1:{self.server_port}/"
+
+
+T = TypeVar("T")
+
+
+class MasAuthDelegation(HomeserverTestCase):
+    server: FakeMasServer
+
+    def till_deferred_has_result(
+        self,
+        awaitable: Union[
+            "Coroutine[Deferred[Any], Any, T]",
+            "Generator[Deferred[Any], Any, T]",
+            "Deferred[T]",
+        ],
+    ) -> "Deferred[T]":
+        """Wait until a deferred has a result.
+
+        This is useful because the Rust HTTP client will resolve the deferred
+        using reactor.callFromThread, which are only run when we call
+        reactor.advance.
+        """
+        deferred = ensureDeferred(awaitable)
+        tries = 0
+        while not deferred.called:
+            time.sleep(0.1)
+            self.reactor.advance(0)
+            tries += 1
+            if tries > 100:
+                raise Exception("Timed out waiting for deferred to resolve")
+
+        return deferred
+
+    def default_config(self) -> Dict[str, Any]:
+        config = super().default_config()
+        config["public_baseurl"] = BASE_URL
+        config["disable_registration"] = True
+        config["matrix_authentication_service"] = {
+            "enabled": True,
+            "endpoint": self.server.endpoint,
+            "secret": self.server.secret,
+        }
+        return config
+
+    def make_homeserver(self, reactor: MemoryReactor, clock: Clock) -> HomeServer:
+        self.server = FakeMasServer()
+        hs = self.setup_test_homeserver()
+        # This triggers the server startup hooks, which starts the Tokio thread pool
+        reactor.run()
+        self._auth = checked_cast(MasDelegatedAuth, hs.get_auth())
+        return hs
+
+    def prepare(
+        self, reactor: MemoryReactor, clock: Clock, homeserver: HomeServer
+    ) -> None:
+        # Provision the user and the device we use in the tests.
+        store = homeserver.get_datastores().main
+        self.get_success(store.register_user(USER_ID))
+        self.get_success(
+            store.store_device(USER_ID, DEVICE, initial_device_display_name=None)
+        )
+
+    def tearDown(self) -> None:
+        self.server.shutdown()
+        # MemoryReactor doesn't trigger the shutdown phases, and we want the
+        # Tokio thread pool to be stopped
+        # XXX: This logic should probably get moved somewhere else
+        shutdown_triggers = self.reactor.triggers.get("shutdown", {})
+        for phase in ["before", "during", "after"]:
+            triggers = shutdown_triggers.get(phase, [])
+            for callbable, args, kwargs in triggers:
+                callbable(*args, **kwargs)
+
+    def test_simple_introspection(self) -> None:
+        self.server.introspection_response = {
+            "active": True,
+            "sub": SUBJECT,
+            "scope": " ".join(
+                [
+                    MATRIX_USER_SCOPE,
+                    f"{MATRIX_DEVICE_SCOPE_PREFIX}{DEVICE}",
+                ]
+            ),
+            "username": USERNAME,
+            "expires_in": 60,
+        }
+
+        requester = self.get_success(
+            self.till_deferred_has_result(
+                self._auth.get_user_by_access_token("some_token")
+            )
+        )
+
+        self.assertEquals(requester.user.to_string(), USER_ID)
+        self.assertEquals(requester.device_id, DEVICE)
+        self.assertFalse(self.get_success(self._auth.is_server_admin(requester)))
+
+        self.assertEquals(
+            self.server.last_token_seen,
+            "some_token",
+        )
+
+    def test_unexpiring_token(self) -> None:
+        self.server.introspection_response = {
+            "active": True,
+            "sub": SUBJECT,
+            "scope": " ".join(
+                [
+                    MATRIX_USER_SCOPE,
+                    f"{MATRIX_DEVICE_SCOPE_PREFIX}{DEVICE}",
+                ]
+            ),
+            "username": USERNAME,
+        }
+
+        requester = self.get_success(
+            self.till_deferred_has_result(
+                self._auth.get_user_by_access_token("some_token")
+            )
+        )
+
+        self.assertEquals(requester.user.to_string(), USER_ID)
+        self.assertEquals(requester.device_id, DEVICE)
+        self.assertFalse(self.get_success(self._auth.is_server_admin(requester)))
+
+        self.assertEquals(
+            self.server.last_token_seen,
+            "some_token",
+        )
+
+    def test_inexistent_device(self) -> None:
+        self.server.introspection_response = {
+            "active": True,
+            "sub": SUBJECT,
+            "scope": " ".join(
+                [
+                    MATRIX_USER_SCOPE,
+                    f"{MATRIX_DEVICE_SCOPE_PREFIX}ABCDEF",
+                ]
+            ),
+            "username": USERNAME,
+            "expires_in": 60,
+        }
+
+        failure = self.get_failure(
+            self.till_deferred_has_result(
+                self._auth.get_user_by_access_token("some_token")
+            ),
+            InvalidClientTokenError,
+        )
+        self.assertEqual(failure.value.code, 401)
+
+    def test_inexistent_user(self) -> None:
+        self.server.introspection_response = {
+            "active": True,
+            "sub": SUBJECT,
+            "scope": " ".join([MATRIX_USER_SCOPE]),
+            "username": "inexistent_user",
+            "expires_in": 60,
+        }
+
+        failure = self.get_failure(
+            self.till_deferred_has_result(
+                self._auth.get_user_by_access_token("some_token")
+            ),
+            AuthError,
+        )
+        # This is a 500, it should never happen really
+        self.assertEqual(failure.value.code, 500)
+
+    def test_missing_scope(self) -> None:
+        self.server.introspection_response = {
+            "active": True,
+            "sub": SUBJECT,
+            "scope": "openid",
+            "username": USERNAME,
+            "expires_in": 60,
+        }
+
+        failure = self.get_failure(
+            self.till_deferred_has_result(
+                self._auth.get_user_by_access_token("some_token")
+            ),
+            InvalidClientTokenError,
+        )
+        self.assertEqual(failure.value.code, 401)
+
+    def test_invalid_response(self) -> None:
+        self.server.introspection_response = {}
+
+        failure = self.get_failure(
+            self.till_deferred_has_result(
+                self._auth.get_user_by_access_token("some_token")
+            ),
+            SynapseError,
+        )
+        self.assertEqual(failure.value.code, 503)
+
+    def test_device_id_in_body(self) -> None:
+        self.server.introspection_response = {
+            "active": True,
+            "sub": SUBJECT,
+            "scope": MATRIX_USER_SCOPE,
+            "username": USERNAME,
+            "expires_in": 60,
+            "device_id": DEVICE,
+        }
+
+        requester = self.get_success(
+            self.till_deferred_has_result(
+                self._auth.get_user_by_access_token("some_token")
+            )
+        )
+
+        self.assertEqual(requester.device_id, DEVICE)
+
+    def test_admin_scope(self) -> None:
+        self.server.introspection_response = {
+            "active": True,
+            "sub": SUBJECT,
+            "scope": " ".join([SYNAPSE_ADMIN_SCOPE, MATRIX_USER_SCOPE]),
+            "username": USERNAME,
+            "expires_in": 60,
+        }
+
+        requester = self.get_success(
+            self.till_deferred_has_result(
+                self._auth.get_user_by_access_token("some_token")
+            )
+        )
+
+        self.assertEqual(requester.user.to_string(), USER_ID)
+        self.assertTrue(self.get_success(self._auth.is_server_admin(requester)))
+
+    def test_cached_expired_introspection(self) -> None:
+        """The handler should raise an error if the introspection response gives
+        an expiry time, the introspection response is cached and then the entry is
+        re-requested after it has expired."""
+
+        self.server.introspection_response = {
+            "active": True,
+            "sub": SUBJECT,
+            "scope": " ".join(
+                [
+                    MATRIX_USER_SCOPE,
+                    f"{MATRIX_DEVICE_SCOPE_PREFIX}{DEVICE}",
+                ]
+            ),
+            "username": USERNAME,
+            "expires_in": 60,
+        }
+
+        self.assertEqual(self.server.calls, 0)
+
+        request = Mock(args={})
+        request.args[b"access_token"] = [b"some_token"]
+        request.requestHeaders.getRawHeaders = mock_getRawHeaders()
+
+        # The first CS-API request causes a successful introspection
+        self.get_success(
+            self.till_deferred_has_result(self._auth.get_user_by_req(request))
+        )
+        self.assertEqual(self.server.calls, 1)
+
+        # Sleep for 60 seconds so the token expires.
+        self.reactor.advance(60.0)
+
+        # Now the CS-API request fails because the token expired
+        self.assertFailure(
+            self.till_deferred_has_result(self._auth.get_user_by_req(request)),
+            InvalidClientTokenError,
+        )
+        # Ensure another introspection request was not sent
+        self.assertEqual(self.server.calls, 1)
+
+
+@parameterized_class(
+    ("config",),
+    [
+        (
+            {
+                "matrix_authentication_service": {
+                    "enabled": True,
+                    "endpoint": "http://localhost:1234/",
+                    "secret": "secret",
+                },
+            },
+        ),
+    ]
+    # Run the tests with experimental delegation only if authlib is available
+    + [
+        (
+            {
+                "experimental_features": {
+                    "msc3861": {
+                        "enabled": True,
+                        "issuer": ISSUER,
+                        "client_id": CLIENT_ID,
+                        "client_auth_method": "client_secret_post",
+                        "client_secret": CLIENT_SECRET,
+                        "admin_token": "admin_token_value",
+                    }
+                }
+            },
+        ),
+    ]
+    * HAS_AUTHLIB,
+)
+class DisabledEndpointsTestCase(HomeserverTestCase):
+    servlets = [
+        account.register_servlets,
+        devices.register_servlets,
+        keys.register_servlets,
+        register.register_servlets,
+        login.register_servlets,
+        logout.register_servlets,
+        admin.register_servlets,
+    ]
+
+    config: Dict[str, Any]
+
+    def default_config(self) -> Dict[str, Any]:
+        config = super().default_config()
+        config["public_baseurl"] = BASE_URL
+        config["disable_registration"] = True
+        config.update(self.config)
+        return config
+
     def expect_unauthorized(
         self, method: str, path: str, content: Union[bytes, str, JsonDict] = ""
     ) -> None:
@@ -774,13 +1304,11 @@ class MSC3861OAuthDelegation(HomeserverTestCase):
 
         # Because we still support those endpoints with ASes, it checks the
         # access token before returning 404
-        self._set_introspection_returnvalue(
-            {
-                "active": True,
-                "sub": SUBJECT,
-                "scope": " ".join([MATRIX_USER_SCOPE, MATRIX_DEVICE_SCOPE]),
-                "username": USERNAME,
-            },
+        self.hs.get_auth().get_user_by_req = AsyncMock(  # type: ignore[method-assign]
+            return_value=create_requester(
+                user_id=USER_ID,
+                device_id=DEVICE,
+            )
         )
 
         self.expect_unrecognized("POST", "/_matrix/client/v3/delete_devices", auth=True)
@@ -810,118 +1338,3 @@ class MSC3861OAuthDelegation(HomeserverTestCase):
         self.expect_unrecognized("GET", "/_synapse/admin/v1/users/foo/admin")
         self.expect_unrecognized("PUT", "/_synapse/admin/v1/users/foo/admin")
         self.expect_unrecognized("POST", "/_synapse/admin/v1/account_validity/validity")
-
-    def test_admin_token(self) -> None:
-        """The handler should return a requester with admin rights when admin_token is used."""
-        self._set_introspection_returnvalue({"active": False})
-
-        request = Mock(args={})
-        request.args[b"access_token"] = [b"admin_token_value"]
-        request.requestHeaders.getRawHeaders = mock_getRawHeaders()
-        requester = self.get_success(self.auth.get_user_by_req(request))
-        self.assertEqual(
-            requester.user.to_string(),
-            OIDC_ADMIN_USERID,
-        )
-        self.assertEqual(requester.is_guest, False)
-        self.assertEqual(requester.device_id, None)
-        self.assertEqual(
-            get_awaitable_result(self.auth.is_server_admin(requester)), True
-        )
-
-        # There should be no call to the introspection endpoint
-        self._rust_client.post.assert_not_called()
-
-    @override_config({"mau_stats_only": True})
-    def test_request_tracking(self) -> None:
-        """Using an access token should update the client_ips and MAU tables."""
-        # To start, there are no MAU users.
-        store = self.hs.get_datastores().main
-        mau = self.get_success(store.get_monthly_active_count())
-        self.assertEqual(mau, 0)
-
-        known_token = "token-token-GOOD-:)"
-
-        async def mock_http_client_request(
-            url: str, request_body: str, **kwargs: Any
-        ) -> bytes:
-            """Mocked auth provider response."""
-            token = parse_qs(request_body)["token"][0]
-            if token == known_token:
-                return json.dumps(
-                    {
-                        "active": True,
-                        "scope": MATRIX_USER_SCOPE,
-                        "sub": SUBJECT,
-                        "username": USERNAME,
-                    },
-                ).encode("utf-8")
-
-            return json.dumps({"active": False}).encode("utf-8")
-
-        self._rust_client.post = mock_http_client_request
-
-        EXAMPLE_IPV4_ADDR = "123.123.123.123"
-        EXAMPLE_USER_AGENT = "httprettygood"
-
-        # First test a known access token
-        channel = FakeChannel(self.site, self.reactor)
-        # type-ignore: FakeChannel is a mock of an HTTPChannel, not a proper HTTPChannel
-        req = SynapseRequest(channel, self.site, self.hs.hostname)  # type: ignore[arg-type]
-        req.client.host = EXAMPLE_IPV4_ADDR
-        req.requestHeaders.addRawHeader("Authorization", f"Bearer {known_token}")
-        req.requestHeaders.addRawHeader("User-Agent", EXAMPLE_USER_AGENT)
-        req.content = BytesIO(b"")
-        req.requestReceived(
-            b"GET",
-            b"/_matrix/client/v3/account/whoami",
-            b"1.1",
-        )
-        channel.await_result()
-        self.assertEqual(channel.code, HTTPStatus.OK, channel.json_body)
-        self.assertEqual(channel.json_body["user_id"], USER_ID, channel.json_body)
-
-        # Expect to see one MAU entry, from the first request
-        mau = self.get_success(store.get_monthly_active_count())
-        self.assertEqual(mau, 1)
-
-        conn_infos = self.get_success(
-            store.get_user_ip_and_agents(UserID.from_string(USER_ID))
-        )
-        self.assertEqual(len(conn_infos), 1, conn_infos)
-        conn_info = conn_infos[0]
-        self.assertEqual(conn_info["access_token"], known_token)
-        self.assertEqual(conn_info["ip"], EXAMPLE_IPV4_ADDR)
-        self.assertEqual(conn_info["user_agent"], EXAMPLE_USER_AGENT)
-
-        # Now test MAS making a request using the special __oidc_admin token
-        MAS_IPV4_ADDR = "127.0.0.1"
-        MAS_USER_AGENT = "masmasmas"
-
-        channel = FakeChannel(self.site, self.reactor)
-        req = SynapseRequest(channel, self.site, self.hs.hostname)  # type: ignore[arg-type]
-        req.client.host = MAS_IPV4_ADDR
-        req.requestHeaders.addRawHeader(
-            "Authorization", f"Bearer {self.auth._admin_token()}"
-        )
-        req.requestHeaders.addRawHeader("User-Agent", MAS_USER_AGENT)
-        req.content = BytesIO(b"")
-        req.requestReceived(
-            b"GET",
-            b"/_matrix/client/v3/account/whoami",
-            b"1.1",
-        )
-        channel.await_result()
-        self.assertEqual(channel.code, HTTPStatus.OK, channel.json_body)
-        self.assertEqual(
-            channel.json_body["user_id"], OIDC_ADMIN_USERID, channel.json_body
-        )
-
-        # Still expect to see one MAU entry, from the first request
-        mau = self.get_success(store.get_monthly_active_count())
-        self.assertEqual(mau, 1)
-
-        conn_infos = self.get_success(
-            store.get_user_ip_and_agents(UserID.from_string(OIDC_ADMIN_USERID))
-        )
-        self.assertEqual(conn_infos, [])