|
- # Copyright 2022 Matrix.org Foundation C.I.C.
- #
- # Licensed under the Apache License, Version 2.0 (the "License");
- # you may not use this file except in compliance with the License.
- # You may obtain a copy of the License at
- #
- # http://www.apache.org/licenses/LICENSE-2.0
- #
- # Unless required by applicable law or agreed to in writing, software
- # distributed under the License is distributed on an "AS IS" BASIS,
- # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- # See the License for the specific language governing permissions and
- # limitations under the License.
-
- from http import HTTPStatus
- from io import BytesIO
- from typing import Any, Dict, Optional, Union
- from unittest.mock import ANY, AsyncMock, Mock
- from urllib.parse import parse_qs
-
- from signedjson.key import (
- encode_verify_key_base64,
- generate_signing_key,
- get_verify_key,
- )
- from signedjson.sign import sign_json
-
- from twisted.test.proto_helpers import MemoryReactor
- from twisted.web.http_headers import Headers
- from twisted.web.iweb import IResponse
-
- from synapse.api.errors import (
- AuthError,
- Codes,
- InvalidClientTokenError,
- OAuthInsufficientScopeError,
- SynapseError,
- )
- 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.util import Clock
-
- from tests.server import FakeChannel
- from tests.test_utils import FakeResponse, get_awaitable_result
- from tests.unittest import HomeserverTestCase, override_config, skip_unless
- from tests.utils import HAS_AUTHLIB, checked_cast, mock_getRawHeaders
-
- # These are a few constants that are used as config parameters in the tests.
- SERVER_NAME = "test"
- ISSUER = "https://issuer/"
- CLIENT_ID = "test-client-id"
- CLIENT_SECRET = "test-client-secret"
- BASE_URL = "https://synapse/"
- SCOPES = ["openid"]
-
- AUTHORIZATION_ENDPOINT = ISSUER + "authorize"
- TOKEN_ENDPOINT = ISSUER + "token"
- USERINFO_ENDPOINT = ISSUER + "userinfo"
- WELL_KNOWN = ISSUER + ".well-known/openid-configuration"
- JWKS_URI = ISSUER + ".well-known/jwks.json"
- INTROSPECTION_ENDPOINT = ISSUER + "introspect"
-
- SYNAPSE_ADMIN_SCOPE = "urn:synapse:admin:*"
- MATRIX_USER_SCOPE = "urn:matrix:org.matrix.msc2967.client:api:*"
- MATRIX_GUEST_SCOPE = "urn:matrix:org.matrix.msc2967.client:api:guest"
- MATRIX_DEVICE_SCOPE_PREFIX = "urn:matrix:org.matrix.msc2967.client:device:"
- DEVICE = "AABBCCDD"
- MATRIX_DEVICE_SCOPE = MATRIX_DEVICE_SCOPE_PREFIX + DEVICE
- SUBJECT = "abc-def-ghi"
- USERNAME = "test-user"
- USER_ID = "@" + USERNAME + ":" + SERVER_NAME
- OIDC_ADMIN_USERID = f"@__oidc_admin:{SERVER_NAME}"
-
-
- async def get_json(url: str) -> JsonDict:
- # Mock get_json calls to handle jwks & oidc discovery endpoints
- if url == WELL_KNOWN:
- # Minimal discovery document, as defined in OpenID.Discovery
- # https://openid.net/specs/openid-connect-discovery-1_0.html#ProviderMetadata
- return {
- "issuer": ISSUER,
- "authorization_endpoint": AUTHORIZATION_ENDPOINT,
- "token_endpoint": TOKEN_ENDPOINT,
- "jwks_uri": JWKS_URI,
- "userinfo_endpoint": USERINFO_ENDPOINT,
- "introspection_endpoint": INTROSPECTION_ENDPOINT,
- "response_types_supported": ["code"],
- "subject_types_supported": ["public"],
- "id_token_signing_alg_values_supported": ["RS256"],
- }
- elif url == JWKS_URI:
- return {"keys": []}
-
- return {}
-
-
- @skip_unless(HAS_AUTHLIB, "requires authlib")
- 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]:
- config = super().default_config()
- config["public_baseurl"] = BASE_URL
- config["disable_registration"] = True
- config["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",
- }
- }
- return config
-
- def make_homeserver(self, reactor: MemoryReactor, clock: Clock) -> HomeServer:
- self.http_client = Mock(spec=["get_json"])
- self.http_client.get_json.side_effect = get_json
- self.http_client.user_agent = b"Synapse Test"
-
- hs = self.setup_test_homeserver(proxied_http_client=self.http_client)
-
- # Import this here so that we've checked that authlib is available.
- from synapse.api.auth.msc3861_delegated import MSC3861DelegatedAuth
-
- self.auth = checked_cast(MSC3861DelegatedAuth, hs.get_auth())
-
- return hs
-
- def _assertParams(self) -> None:
- """Assert that the request parameters are correct."""
- params = parse_qs(self.http_client.request.call_args[1]["data"].decode("utf-8"))
- self.assertEqual(params["token"], ["mockAccessToken"])
- self.assertEqual(params["client_id"], [CLIENT_ID])
- self.assertEqual(params["client_secret"], [CLIENT_SECRET])
-
- def test_inactive_token(self) -> None:
- """The handler should return a 403 where the token is inactive."""
-
- self.http_client.request = AsyncMock(
- return_value=FakeResponse.json(
- code=200,
- payload={"active": False},
- )
- )
- request = Mock(args={})
- request.args[b"access_token"] = [b"mockAccessToken"]
- request.requestHeaders.getRawHeaders = mock_getRawHeaders()
- self.get_failure(self.auth.get_user_by_req(request), InvalidClientTokenError)
- self.http_client.get_json.assert_called_once_with(WELL_KNOWN)
- self.http_client.request.assert_called_once_with(
- method="POST", uri=INTROSPECTION_ENDPOINT, data=ANY, headers=ANY
- )
- self._assertParams()
-
- def test_active_no_scope(self) -> None:
- """The handler should return a 403 where no scope is given."""
-
- self.http_client.request = AsyncMock(
- return_value=FakeResponse.json(
- code=200,
- payload={"active": True},
- )
- )
- request = Mock(args={})
- request.args[b"access_token"] = [b"mockAccessToken"]
- request.requestHeaders.getRawHeaders = mock_getRawHeaders()
- self.get_failure(self.auth.get_user_by_req(request), InvalidClientTokenError)
- self.http_client.get_json.assert_called_once_with(WELL_KNOWN)
- self.http_client.request.assert_called_once_with(
- method="POST", uri=INTROSPECTION_ENDPOINT, data=ANY, headers=ANY
- )
- self._assertParams()
-
- def test_active_user_no_subject(self) -> None:
- """The handler should return a 500 when no subject is present."""
-
- self.http_client.request = AsyncMock(
- return_value=FakeResponse.json(
- code=200,
- payload={"active": True, "scope": " ".join([MATRIX_USER_SCOPE])},
- )
- )
- request = Mock(args={})
- request.args[b"access_token"] = [b"mockAccessToken"]
- request.requestHeaders.getRawHeaders = mock_getRawHeaders()
- self.get_failure(self.auth.get_user_by_req(request), InvalidClientTokenError)
- self.http_client.get_json.assert_called_once_with(WELL_KNOWN)
- self.http_client.request.assert_called_once_with(
- method="POST", uri=INTROSPECTION_ENDPOINT, data=ANY, headers=ANY
- )
- self._assertParams()
-
- def test_active_no_user_scope(self) -> None:
- """The handler should return a 500 when no subject is present."""
-
- self.http_client.request = AsyncMock(
- return_value=FakeResponse.json(
- code=200,
- payload={
- "active": True,
- "sub": SUBJECT,
- "scope": " ".join([MATRIX_DEVICE_SCOPE]),
- },
- )
- )
- request = Mock(args={})
- request.args[b"access_token"] = [b"mockAccessToken"]
- request.requestHeaders.getRawHeaders = mock_getRawHeaders()
- self.get_failure(self.auth.get_user_by_req(request), InvalidClientTokenError)
- self.http_client.get_json.assert_called_once_with(WELL_KNOWN)
- self.http_client.request.assert_called_once_with(
- method="POST", uri=INTROSPECTION_ENDPOINT, data=ANY, headers=ANY
- )
- self._assertParams()
-
- def test_active_admin_not_user(self) -> None:
- """The handler should raise when the scope has admin right but not user."""
-
- self.http_client.request = AsyncMock(
- return_value=FakeResponse.json(
- code=200,
- payload={
- "active": True,
- "sub": SUBJECT,
- "scope": " ".join([SYNAPSE_ADMIN_SCOPE]),
- "username": USERNAME,
- },
- )
- )
- request = Mock(args={})
- request.args[b"access_token"] = [b"mockAccessToken"]
- request.requestHeaders.getRawHeaders = mock_getRawHeaders()
- self.get_failure(self.auth.get_user_by_req(request), InvalidClientTokenError)
- self.http_client.get_json.assert_called_once_with(WELL_KNOWN)
- self.http_client.request.assert_called_once_with(
- method="POST", uri=INTROSPECTION_ENDPOINT, data=ANY, headers=ANY
- )
- self._assertParams()
-
- def test_active_admin(self) -> None:
- """The handler should return a requester with admin rights."""
-
- self.http_client.request = AsyncMock(
- return_value=FakeResponse.json(
- code=200,
- payload={
- "active": True,
- "sub": SUBJECT,
- "scope": " ".join([SYNAPSE_ADMIN_SCOPE, MATRIX_USER_SCOPE]),
- "username": USERNAME,
- },
- )
- )
- request = Mock(args={})
- request.args[b"access_token"] = [b"mockAccessToken"]
- request.requestHeaders.getRawHeaders = mock_getRawHeaders()
- requester = self.get_success(self.auth.get_user_by_req(request))
- self.http_client.get_json.assert_called_once_with(WELL_KNOWN)
- self.http_client.request.assert_called_once_with(
- method="POST", uri=INTROSPECTION_ENDPOINT, data=ANY, headers=ANY
- )
- self._assertParams()
- self.assertEqual(requester.user.to_string(), "@%s:%s" % (USERNAME, SERVER_NAME))
- self.assertEqual(requester.is_guest, False)
- self.assertEqual(requester.device_id, None)
- self.assertEqual(
- get_awaitable_result(self.auth.is_server_admin(requester)), True
- )
-
- def test_active_admin_highest_privilege(self) -> None:
- """The handler should resolve to the most permissive scope."""
-
- self.http_client.request = AsyncMock(
- return_value=FakeResponse.json(
- code=200,
- payload={
- "active": True,
- "sub": SUBJECT,
- "scope": " ".join(
- [SYNAPSE_ADMIN_SCOPE, MATRIX_USER_SCOPE, MATRIX_GUEST_SCOPE]
- ),
- "username": USERNAME,
- },
- )
- )
- request = Mock(args={})
- request.args[b"access_token"] = [b"mockAccessToken"]
- request.requestHeaders.getRawHeaders = mock_getRawHeaders()
- requester = self.get_success(self.auth.get_user_by_req(request))
- self.http_client.get_json.assert_called_once_with(WELL_KNOWN)
- self.http_client.request.assert_called_once_with(
- method="POST", uri=INTROSPECTION_ENDPOINT, data=ANY, headers=ANY
- )
- self._assertParams()
- self.assertEqual(requester.user.to_string(), "@%s:%s" % (USERNAME, SERVER_NAME))
- self.assertEqual(requester.is_guest, False)
- self.assertEqual(requester.device_id, None)
- self.assertEqual(
- get_awaitable_result(self.auth.is_server_admin(requester)), True
- )
-
- def test_active_user(self) -> None:
- """The handler should return a requester with normal user rights."""
-
- self.http_client.request = AsyncMock(
- return_value=FakeResponse.json(
- code=200,
- payload={
- "active": True,
- "sub": SUBJECT,
- "scope": " ".join([MATRIX_USER_SCOPE]),
- "username": USERNAME,
- },
- )
- )
- request = Mock(args={})
- request.args[b"access_token"] = [b"mockAccessToken"]
- request.requestHeaders.getRawHeaders = mock_getRawHeaders()
- requester = self.get_success(self.auth.get_user_by_req(request))
- self.http_client.get_json.assert_called_once_with(WELL_KNOWN)
- self.http_client.request.assert_called_once_with(
- method="POST", uri=INTROSPECTION_ENDPOINT, data=ANY, headers=ANY
- )
- self._assertParams()
- self.assertEqual(requester.user.to_string(), "@%s:%s" % (USERNAME, SERVER_NAME))
- self.assertEqual(requester.is_guest, False)
- self.assertEqual(requester.device_id, None)
- self.assertEqual(
- get_awaitable_result(self.auth.is_server_admin(requester)), False
- )
-
- def test_active_user_with_device(self) -> None:
- """The handler should return a requester with normal user rights and a device ID."""
-
- self.http_client.request = AsyncMock(
- return_value=FakeResponse.json(
- code=200,
- payload={
- "active": True,
- "sub": SUBJECT,
- "scope": " ".join([MATRIX_USER_SCOPE, MATRIX_DEVICE_SCOPE]),
- "username": USERNAME,
- },
- )
- )
- request = Mock(args={})
- request.args[b"access_token"] = [b"mockAccessToken"]
- request.requestHeaders.getRawHeaders = mock_getRawHeaders()
- requester = self.get_success(self.auth.get_user_by_req(request))
- self.http_client.get_json.assert_called_once_with(WELL_KNOWN)
- self.http_client.request.assert_called_once_with(
- method="POST", uri=INTROSPECTION_ENDPOINT, data=ANY, headers=ANY
- )
- self._assertParams()
- self.assertEqual(requester.user.to_string(), "@%s:%s" % (USERNAME, SERVER_NAME))
- self.assertEqual(requester.is_guest, False)
- self.assertEqual(
- get_awaitable_result(self.auth.is_server_admin(requester)), False
- )
- self.assertEqual(requester.device_id, DEVICE)
-
- def test_multiple_devices(self) -> None:
- """The handler should raise an error if multiple devices are found in the scope."""
-
- self.http_client.request = AsyncMock(
- return_value=FakeResponse.json(
- code=200,
- payload={
- "active": True,
- "sub": SUBJECT,
- "scope": " ".join(
- [
- MATRIX_USER_SCOPE,
- f"{MATRIX_DEVICE_SCOPE_PREFIX}AABBCC",
- f"{MATRIX_DEVICE_SCOPE_PREFIX}DDEEFF",
- ]
- ),
- "username": USERNAME,
- },
- )
- )
- request = Mock(args={})
- request.args[b"access_token"] = [b"mockAccessToken"]
- request.requestHeaders.getRawHeaders = mock_getRawHeaders()
- self.get_failure(self.auth.get_user_by_req(request), AuthError)
-
- def test_active_guest_not_allowed(self) -> None:
- """The handler should return an insufficient scope error."""
-
- self.http_client.request = AsyncMock(
- return_value=FakeResponse.json(
- code=200,
- payload={
- "active": True,
- "sub": SUBJECT,
- "scope": " ".join([MATRIX_GUEST_SCOPE, MATRIX_DEVICE_SCOPE]),
- "username": USERNAME,
- },
- )
- )
- request = Mock(args={})
- request.args[b"access_token"] = [b"mockAccessToken"]
- request.requestHeaders.getRawHeaders = mock_getRawHeaders()
- error = self.get_failure(
- self.auth.get_user_by_req(request), OAuthInsufficientScopeError
- )
- self.http_client.get_json.assert_called_once_with(WELL_KNOWN)
- self.http_client.request.assert_called_once_with(
- method="POST", uri=INTROSPECTION_ENDPOINT, data=ANY, headers=ANY
- )
- self._assertParams()
- self.assertEqual(
- getattr(error.value, "headers", {})["WWW-Authenticate"],
- 'Bearer error="insufficient_scope", scope="urn:matrix:org.matrix.msc2967.client:api:*"',
- )
-
- def test_active_guest_allowed(self) -> None:
- """The handler should return a requester with guest user rights and a device ID."""
-
- self.http_client.request = AsyncMock(
- return_value=FakeResponse.json(
- code=200,
- payload={
- "active": True,
- "sub": SUBJECT,
- "scope": " ".join([MATRIX_GUEST_SCOPE, MATRIX_DEVICE_SCOPE]),
- "username": USERNAME,
- },
- )
- )
- request = Mock(args={})
- request.args[b"access_token"] = [b"mockAccessToken"]
- request.requestHeaders.getRawHeaders = mock_getRawHeaders()
- requester = self.get_success(
- self.auth.get_user_by_req(request, allow_guest=True)
- )
- self.http_client.get_json.assert_called_once_with(WELL_KNOWN)
- self.http_client.request.assert_called_once_with(
- method="POST", uri=INTROSPECTION_ENDPOINT, data=ANY, headers=ANY
- )
- self._assertParams()
- self.assertEqual(requester.user.to_string(), "@%s:%s" % (USERNAME, SERVER_NAME))
- self.assertEqual(requester.is_guest, True)
- self.assertEqual(
- get_awaitable_result(self.auth.is_server_admin(requester)), False
- )
- self.assertEqual(requester.device_id, DEVICE)
-
- def test_unavailable_introspection_endpoint(self) -> None:
- """The handler should return an internal server error."""
- request = Mock(args={})
- request.args[b"access_token"] = [b"mockAccessToken"]
- request.requestHeaders.getRawHeaders = mock_getRawHeaders()
-
- # The introspection endpoint is returning an error.
- self.http_client.request = AsyncMock(
- return_value=FakeResponse(code=500, body=b"Internal Server Error")
- )
- error = self.get_failure(self.auth.get_user_by_req(request), SynapseError)
- self.assertEqual(error.value.code, 503)
-
- # The introspection endpoint request fails.
- self.http_client.request = AsyncMock(side_effect=Exception())
- error = self.get_failure(self.auth.get_user_by_req(request), SynapseError)
- self.assertEqual(error.value.code, 503)
-
- # The introspection endpoint does not return a JSON object.
- self.http_client.request = AsyncMock(
- return_value=FakeResponse.json(
- code=200, payload=["this is an array", "not an object"]
- )
- )
- error = self.get_failure(self.auth.get_user_by_req(request), SynapseError)
- self.assertEqual(error.value.code, 503)
-
- # The introspection endpoint does not return valid JSON.
- self.http_client.request = AsyncMock(
- return_value=FakeResponse(code=200, body=b"this is not valid JSON")
- )
- error = self.get_failure(self.auth.get_user_by_req(request), SynapseError)
- self.assertEqual(error.value.code, 503)
-
- def make_device_keys(self, user_id: str, device_id: str) -> JsonDict:
- # We only generate a master key to simplify the test.
- master_signing_key = generate_signing_key(device_id)
- master_verify_key = encode_verify_key_base64(get_verify_key(master_signing_key))
-
- return {
- "master_key": sign_json(
- {
- "user_id": user_id,
- "usage": ["master"],
- "keys": {"ed25519:" + master_verify_key: master_verify_key},
- },
- user_id,
- master_signing_key,
- ),
- }
-
- def test_cross_signing(self) -> None:
- """Try uploading device keys with OAuth delegation enabled."""
-
- self.http_client.request = AsyncMock(
- return_value=FakeResponse.json(
- code=200,
- payload={
- "active": True,
- "sub": SUBJECT,
- "scope": " ".join([MATRIX_USER_SCOPE, MATRIX_DEVICE_SCOPE]),
- "username": USERNAME,
- },
- )
- )
- keys_upload_body = self.make_device_keys(USER_ID, DEVICE)
- channel = self.make_request(
- "POST",
- "/_matrix/client/v3/keys/device_signing/upload",
- keys_upload_body,
- access_token="mockAccessToken",
- )
-
- self.assertEqual(channel.code, 200, channel.json_body)
-
- channel = self.make_request(
- "POST",
- "/_matrix/client/v3/keys/device_signing/upload",
- keys_upload_body,
- access_token="mockAccessToken",
- )
-
- self.assertEqual(channel.code, HTTPStatus.NOT_IMPLEMENTED, channel.json_body)
-
- def expect_unauthorized(
- self, method: str, path: str, content: Union[bytes, str, JsonDict] = ""
- ) -> None:
- channel = self.make_request(method, path, content, shorthand=False)
-
- self.assertEqual(channel.code, 401, channel.json_body)
-
- def expect_unrecognized(
- self, method: str, path: str, content: Union[bytes, str, JsonDict] = ""
- ) -> None:
- channel = self.make_request(method, path, content)
-
- self.assertEqual(channel.code, 404, channel.json_body)
- self.assertEqual(
- channel.json_body["errcode"], Codes.UNRECOGNIZED, channel.json_body
- )
-
- def test_uia_endpoints(self) -> None:
- """Test that endpoints that were removed in MSC2964 are no longer available."""
-
- # This is just an endpoint that should remain visible (but requires auth):
- self.expect_unauthorized("GET", "/_matrix/client/v3/devices")
-
- # This remains usable, but will require a uia scope:
- self.expect_unauthorized(
- "POST", "/_matrix/client/v3/keys/device_signing/upload"
- )
-
- def test_3pid_endpoints(self) -> None:
- """Test that 3pid account management endpoints that were removed in MSC2964 are no longer available."""
-
- # Remains and requires auth:
- self.expect_unauthorized("GET", "/_matrix/client/v3/account/3pid")
- self.expect_unauthorized(
- "POST",
- "/_matrix/client/v3/account/3pid/bind",
- {
- "client_secret": "foo",
- "id_access_token": "bar",
- "id_server": "foo",
- "sid": "bar",
- },
- )
- self.expect_unauthorized("POST", "/_matrix/client/v3/account/3pid/unbind", {})
-
- # These are gone:
- self.expect_unrecognized(
- "POST", "/_matrix/client/v3/account/3pid"
- ) # deprecated
- self.expect_unrecognized("POST", "/_matrix/client/v3/account/3pid/add")
- self.expect_unrecognized("POST", "/_matrix/client/v3/account/3pid/delete")
- self.expect_unrecognized(
- "POST", "/_matrix/client/v3/account/3pid/email/requestToken"
- )
- self.expect_unrecognized(
- "POST", "/_matrix/client/v3/account/3pid/msisdn/requestToken"
- )
-
- def test_account_management_endpoints_removed(self) -> None:
- """Test that account management endpoints that were removed in MSC2964 are no longer available."""
- self.expect_unrecognized("POST", "/_matrix/client/v3/account/deactivate")
- self.expect_unrecognized("POST", "/_matrix/client/v3/account/password")
- self.expect_unrecognized(
- "POST", "/_matrix/client/v3/account/password/email/requestToken"
- )
- self.expect_unrecognized(
- "POST", "/_matrix/client/v3/account/password/msisdn/requestToken"
- )
-
- def test_registration_endpoints_removed(self) -> None:
- """Test that registration endpoints that were removed in MSC2964 are no longer available."""
- self.expect_unrecognized(
- "GET", "/_matrix/client/v1/register/m.login.registration_token/validity"
- )
- # This is still available for AS registrations
- # self.expect_unrecognized("POST", "/_matrix/client/v3/register")
- self.expect_unrecognized("GET", "/_matrix/client/v3/register/available")
- self.expect_unrecognized(
- "POST", "/_matrix/client/v3/register/email/requestToken"
- )
- self.expect_unrecognized(
- "POST", "/_matrix/client/v3/register/msisdn/requestToken"
- )
-
- def test_session_management_endpoints_removed(self) -> None:
- """Test that session management endpoints that were removed in MSC2964 are no longer available."""
- self.expect_unrecognized("GET", "/_matrix/client/v3/login")
- self.expect_unrecognized("POST", "/_matrix/client/v3/login")
- self.expect_unrecognized("GET", "/_matrix/client/v3/login/sso/redirect")
- self.expect_unrecognized("POST", "/_matrix/client/v3/logout")
- self.expect_unrecognized("POST", "/_matrix/client/v3/logout/all")
- self.expect_unrecognized("POST", "/_matrix/client/v3/refresh")
- self.expect_unrecognized("GET", "/_matrix/static/client/login")
-
- def test_device_management_endpoints_removed(self) -> None:
- """Test that device management endpoints that were removed in MSC2964 are no longer available."""
- self.expect_unrecognized("POST", "/_matrix/client/v3/delete_devices")
- self.expect_unrecognized("DELETE", "/_matrix/client/v3/devices/{DEVICE}")
-
- def test_openid_endpoints_removed(self) -> None:
- """Test that OpenID id_token endpoints that were removed in MSC2964 are no longer available."""
- self.expect_unrecognized(
- "POST", "/_matrix/client/v3/user/{USERNAME}/openid/request_token"
- )
-
- def test_admin_api_endpoints_removed(self) -> None:
- """Test that admin API endpoints that were removed in MSC2964 are no longer available."""
- self.expect_unrecognized("GET", "/_synapse/admin/v1/registration_tokens")
- self.expect_unrecognized("POST", "/_synapse/admin/v1/registration_tokens/new")
- self.expect_unrecognized("GET", "/_synapse/admin/v1/registration_tokens/abcd")
- self.expect_unrecognized("PUT", "/_synapse/admin/v1/registration_tokens/abcd")
- self.expect_unrecognized(
- "DELETE", "/_synapse/admin/v1/registration_tokens/abcd"
- )
- self.expect_unrecognized("POST", "/_synapse/admin/v1/reset_password/foo")
- self.expect_unrecognized("POST", "/_synapse/admin/v1/users/foo/login")
- self.expect_unrecognized("GET", "/_synapse/admin/v1/register")
- self.expect_unrecognized("POST", "/_synapse/admin/v1/register")
- 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.http_client.request = AsyncMock(
- return_value=FakeResponse.json(code=200, payload={"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.http_client.request.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(
- method: str,
- uri: str,
- data: Optional[bytes] = None,
- headers: Optional[Headers] = None,
- ) -> IResponse:
- """Mocked auth provider response."""
- assert method == "POST"
- token = parse_qs(data)[b"token"][0].decode("utf-8")
- if token == known_token:
- return FakeResponse.json(
- code=200,
- payload={
- "active": True,
- "scope": MATRIX_USER_SCOPE,
- "sub": SUBJECT,
- "username": USERNAME,
- },
- )
-
- return FakeResponse.json(code=200, payload={"active": False})
-
- self.http_client.request = 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) # 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) # 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, [])
|