# Copyright 2020 Quentin Gliech # # 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. import os from typing import Any, Awaitable, ContextManager, Dict, Optional, Tuple from unittest.mock import ANY, Mock, patch from urllib.parse import parse_qs, urlparse import pymacaroons from twisted.test.proto_helpers import MemoryReactor from synapse.handlers.sso import MappingException from synapse.http.site import SynapseRequest from synapse.server import HomeServer from synapse.types import JsonDict, UserID from synapse.util import Clock from synapse.util.macaroons import get_value_from_macaroon from synapse.util.stringutils import random_string from tests.test_utils import FakeResponse, get_awaitable_result, simple_async_mock from tests.test_utils.oidc import FakeAuthorizationGrant, FakeOidcServer from tests.unittest import HomeserverTestCase, override_config try: import authlib # noqa: F401 from authlib.oidc.core import UserInfo from authlib.oidc.discovery import OpenIDProviderMetadata from synapse.handlers.oidc import Token, UserAttributeDict HAS_OIDC = True except ImportError: HAS_OIDC = False # These are a few constants that are used as config parameters in the tests. ISSUER = "https://issuer/" CLIENT_ID = "test-client-id" CLIENT_SECRET = "test-client-secret" BASE_URL = "https://synapse/" CALLBACK_URL = BASE_URL + "_synapse/client/oidc/callback" SCOPES = ["openid"] # config for common cases DEFAULT_CONFIG = { "enabled": True, "client_id": CLIENT_ID, "client_secret": CLIENT_SECRET, "issuer": ISSUER, "scopes": SCOPES, "user_mapping_provider": {"module": __name__ + ".TestMappingProvider"}, } # extends the default config with explicit OAuth2 endpoints instead of using discovery EXPLICIT_ENDPOINT_CONFIG = { **DEFAULT_CONFIG, "discover": False, "authorization_endpoint": ISSUER + "authorize", "token_endpoint": ISSUER + "token", "jwks_uri": ISSUER + "jwks", } class TestMappingProvider: @staticmethod def parse_config(config: JsonDict) -> None: return None def __init__(self, config: None): pass def get_remote_user_id(self, userinfo: "UserInfo") -> str: return userinfo["sub"] async def map_user_attributes( self, userinfo: "UserInfo", token: "Token" ) -> "UserAttributeDict": # This is testing not providing the full map. return {"localpart": userinfo["username"], "display_name": None} # type: ignore[typeddict-item] # Do not include get_extra_attributes to test backwards compatibility paths. class TestMappingProviderExtra(TestMappingProvider): async def get_extra_attributes( self, userinfo: "UserInfo", token: "Token" ) -> JsonDict: return {"phone": userinfo["phone"]} class TestMappingProviderFailures(TestMappingProvider): # Superclass is testing the legacy interface for map_user_attributes. async def map_user_attributes( # type: ignore[override] self, userinfo: "UserInfo", token: "Token", failures: int ) -> "UserAttributeDict": return { # type: ignore[typeddict-item] "localpart": userinfo["username"] + (str(failures) if failures else ""), "display_name": None, } def _key_file_path() -> str: """path to a file containing the private half of a test key""" # this key was generated with: # openssl ecparam -name prime256v1 -genkey -noout | # openssl pkcs8 -topk8 -nocrypt -out oidc_test_key.p8 # # we use PKCS8 rather than SEC-1 (which is what openssl ecparam spits out), because # that's what Apple use, and we want to be sure that we work with Apple's keys. # # (For the record: both PKCS8 and SEC-1 specify (different) ways of representing # keys using ASN.1. Both are then typically formatted using PEM, which says: use the # base64-encoded DER encoding of ASN.1, with headers and footers. But we don't # really need to care about any of that.) return os.path.join(os.path.dirname(__file__), "oidc_test_key.p8") def _public_key_file_path() -> str: """path to a file containing the public half of a test key""" # this was generated with: # openssl ec -in oidc_test_key.p8 -pubout -out oidc_test_key.pub.pem # # See above about where oidc_test_key.p8 came from return os.path.join(os.path.dirname(__file__), "oidc_test_key.pub.pem") class OidcHandlerTestCase(HomeserverTestCase): if not HAS_OIDC: skip = "requires OIDC" def default_config(self) -> Dict[str, Any]: config = super().default_config() config["public_baseurl"] = BASE_URL return config def make_homeserver(self, reactor: MemoryReactor, clock: Clock) -> HomeServer: self.fake_server = FakeOidcServer(clock=clock, issuer=ISSUER) hs = self.setup_test_homeserver() self.hs_patcher = self.fake_server.patch_homeserver(hs=hs) self.hs_patcher.start() # type: ignore[attr-defined] self.handler = hs.get_oidc_handler() self.provider = self.handler._providers["oidc"] sso_handler = hs.get_sso_handler() # Mock the render error method. self.render_error = Mock(return_value=None) sso_handler.render_error = self.render_error # type: ignore[assignment] # Reduce the number of attempts when generating MXIDs. sso_handler._MAP_USERNAME_RETRIES = 3 auth_handler = hs.get_auth_handler() # Mock the complete SSO login method. self.complete_sso_login = simple_async_mock() auth_handler.complete_sso_login = self.complete_sso_login # type: ignore[assignment] return hs def tearDown(self) -> None: self.hs_patcher.stop() # type: ignore[attr-defined] return super().tearDown() def reset_mocks(self) -> None: """Reset all the Mocks.""" self.fake_server.reset_mocks() self.render_error.reset_mock() self.complete_sso_login.reset_mock() def metadata_edit(self, values: dict) -> ContextManager[Mock]: """Modify the result that will be returned by the well-known query""" metadata = self.fake_server.get_metadata() metadata.update(values) return patch.object(self.fake_server, "get_metadata", return_value=metadata) def start_authorization( self, userinfo: dict, client_redirect_url: str = "http://client/redirect", scope: str = "openid", with_sid: bool = False, ) -> Tuple[SynapseRequest, FakeAuthorizationGrant]: """Start an authorization request, and get the callback request back.""" nonce = random_string(10) state = random_string(10) code, grant = self.fake_server.start_authorization( userinfo=userinfo, scope=scope, client_id=self.provider._client_auth.client_id, redirect_uri=self.provider._callback_url, nonce=nonce, with_sid=with_sid, ) session = self._generate_oidc_session_token(state, nonce, client_redirect_url) return _build_callback_request(code, state, session), grant def assertRenderedError( self, error: str, error_description: Optional[str] = None ) -> Tuple[Any, ...]: self.render_error.assert_called_once() args = self.render_error.call_args[0] self.assertEqual(args[1], error) if error_description is not None: self.assertEqual(args[2], error_description) # Reset the render_error mock self.render_error.reset_mock() return args @override_config({"oidc_config": DEFAULT_CONFIG}) def test_config(self) -> None: """Basic config correctly sets up the callback URL and client auth correctly.""" self.assertEqual(self.provider._callback_url, CALLBACK_URL) self.assertEqual(self.provider._client_auth.client_id, CLIENT_ID) self.assertEqual(self.provider._client_auth.client_secret, CLIENT_SECRET) @override_config({"oidc_config": {**DEFAULT_CONFIG, "discover": True}}) def test_discovery(self) -> None: """The handler should discover the endpoints from OIDC discovery document.""" # This would throw if some metadata were invalid metadata = self.get_success(self.provider.load_metadata()) self.fake_server.get_metadata_handler.assert_called_once() self.assertEqual(metadata.issuer, self.fake_server.issuer) self.assertEqual( metadata.authorization_endpoint, self.fake_server.authorization_endpoint, ) self.assertEqual(metadata.token_endpoint, self.fake_server.token_endpoint) self.assertEqual(metadata.jwks_uri, self.fake_server.jwks_uri) # It seems like authlib does not have that defined in its metadata models self.assertEqual( metadata.get("userinfo_endpoint"), self.fake_server.userinfo_endpoint, ) # subsequent calls should be cached self.reset_mocks() self.get_success(self.provider.load_metadata()) self.fake_server.get_metadata_handler.assert_not_called() @override_config({"oidc_config": EXPLICIT_ENDPOINT_CONFIG}) def test_no_discovery(self) -> None: """When discovery is disabled, it should not try to load from discovery document.""" self.get_success(self.provider.load_metadata()) self.fake_server.get_metadata_handler.assert_not_called() @override_config({"oidc_config": DEFAULT_CONFIG}) def test_load_jwks(self) -> None: """JWKS loading is done once (then cached) if used.""" jwks = self.get_success(self.provider.load_jwks()) self.fake_server.get_jwks_handler.assert_called_once() self.assertEqual(jwks, self.fake_server.get_jwks()) # subsequent calls should be cached… self.reset_mocks() self.get_success(self.provider.load_jwks()) self.fake_server.get_jwks_handler.assert_not_called() # …unless forced self.reset_mocks() self.get_success(self.provider.load_jwks(force=True)) self.fake_server.get_jwks_handler.assert_called_once() with self.metadata_edit({"jwks_uri": None}): # If we don't do this, the load_metadata call will throw because of the # missing jwks_uri self.provider._user_profile_method = "userinfo_endpoint" self.get_success(self.provider.load_metadata(force=True)) self.get_failure(self.provider.load_jwks(force=True), RuntimeError) @override_config({"oidc_config": DEFAULT_CONFIG}) def test_validate_config(self) -> None: """Provider metadatas are extensively validated.""" h = self.provider def force_load_metadata() -> Awaitable[None]: async def force_load() -> "OpenIDProviderMetadata": return await h.load_metadata(force=True) return get_awaitable_result(force_load()) # Default test config does not throw force_load_metadata() with self.metadata_edit({"issuer": None}): self.assertRaisesRegex(ValueError, "issuer", force_load_metadata) with self.metadata_edit({"issuer": "http://insecure/"}): self.assertRaisesRegex(ValueError, "issuer", force_load_metadata) with self.metadata_edit({"issuer": "https://invalid/?because=query"}): self.assertRaisesRegex(ValueError, "issuer", force_load_metadata) with self.metadata_edit({"authorization_endpoint": None}): self.assertRaisesRegex( ValueError, "authorization_endpoint", force_load_metadata ) with self.metadata_edit({"authorization_endpoint": "http://insecure/auth"}): self.assertRaisesRegex( ValueError, "authorization_endpoint", force_load_metadata ) with self.metadata_edit({"token_endpoint": None}): self.assertRaisesRegex(ValueError, "token_endpoint", force_load_metadata) with self.metadata_edit({"token_endpoint": "http://insecure/token"}): self.assertRaisesRegex(ValueError, "token_endpoint", force_load_metadata) with self.metadata_edit({"jwks_uri": None}): self.assertRaisesRegex(ValueError, "jwks_uri", force_load_metadata) with self.metadata_edit({"jwks_uri": "http://insecure/jwks.json"}): self.assertRaisesRegex(ValueError, "jwks_uri", force_load_metadata) with self.metadata_edit({"response_types_supported": ["id_token"]}): self.assertRaisesRegex( ValueError, "response_types_supported", force_load_metadata ) with self.metadata_edit( {"token_endpoint_auth_methods_supported": ["client_secret_basic"]} ): # should not throw, as client_secret_basic is the default auth method force_load_metadata() with self.metadata_edit( {"token_endpoint_auth_methods_supported": ["client_secret_post"]} ): self.assertRaisesRegex( ValueError, "token_endpoint_auth_methods_supported", force_load_metadata, ) # Tests for configs that require the userinfo endpoint self.assertFalse(h._uses_userinfo) self.assertEqual(h._user_profile_method, "auto") h._user_profile_method = "userinfo_endpoint" self.assertTrue(h._uses_userinfo) # Revert the profile method and do not request the "openid" scope: this should # mean that we check for a userinfo endpoint h._user_profile_method = "auto" h._scopes = [] self.assertTrue(h._uses_userinfo) with self.metadata_edit({"userinfo_endpoint": None}): self.assertRaisesRegex(ValueError, "userinfo_endpoint", force_load_metadata) with self.metadata_edit({"jwks_uri": None}): # Shouldn't raise with a valid userinfo, even without jwks force_load_metadata() @override_config({"oidc_config": {**DEFAULT_CONFIG, "skip_verification": True}}) def test_skip_verification(self) -> None: """Provider metadata validation can be disabled by config.""" with self.metadata_edit({"issuer": "http://insecure"}): # This should not throw get_awaitable_result(self.provider.load_metadata()) @override_config({"oidc_config": DEFAULT_CONFIG}) def test_redirect_request(self) -> None: """The redirect request has the right arguments & generates a valid session cookie.""" req = Mock(spec=["cookies"]) req.cookies = [] url = urlparse( self.get_success( self.provider.handle_redirect_request(req, b"http://client/redirect") ) ) auth_endpoint = urlparse(self.fake_server.authorization_endpoint) self.assertEqual(url.scheme, auth_endpoint.scheme) self.assertEqual(url.netloc, auth_endpoint.netloc) self.assertEqual(url.path, auth_endpoint.path) params = parse_qs(url.query) self.assertEqual(params["redirect_uri"], [CALLBACK_URL]) self.assertEqual(params["response_type"], ["code"]) self.assertEqual(params["scope"], [" ".join(SCOPES)]) self.assertEqual(params["client_id"], [CLIENT_ID]) self.assertEqual(len(params["state"]), 1) self.assertEqual(len(params["nonce"]), 1) self.assertNotIn("code_challenge", params) # Check what is in the cookies self.assertEqual(len(req.cookies), 2) # two cookies cookie_header = req.cookies[0] # The cookie name and path don't really matter, just that it has to be coherent # between the callback & redirect handlers. parts = [p.strip() for p in cookie_header.split(b";")] self.assertIn(b"Path=/_synapse/client/oidc", parts) name, cookie = parts[0].split(b"=") self.assertEqual(name, b"oidc_session") macaroon = pymacaroons.Macaroon.deserialize(cookie) state = get_value_from_macaroon(macaroon, "state") nonce = get_value_from_macaroon(macaroon, "nonce") code_verifier = get_value_from_macaroon(macaroon, "code_verifier") redirect = get_value_from_macaroon(macaroon, "client_redirect_url") self.assertEqual(params["state"], [state]) self.assertEqual(params["nonce"], [nonce]) self.assertEqual(code_verifier, "") self.assertEqual(redirect, "http://client/redirect") @override_config({"oidc_config": DEFAULT_CONFIG}) def test_redirect_request_with_code_challenge(self) -> None: """The redirect request has the right arguments & generates a valid session cookie.""" req = Mock(spec=["cookies"]) req.cookies = [] with self.metadata_edit({"code_challenge_methods_supported": ["S256"]}): url = urlparse( self.get_success( self.provider.handle_redirect_request( req, b"http://client/redirect" ) ) ) # Ensure the code_challenge param is added to the redirect. params = parse_qs(url.query) self.assertEqual(len(params["code_challenge"]), 1) # Check what is in the cookies self.assertEqual(len(req.cookies), 2) # two cookies cookie_header = req.cookies[0] # The cookie name and path don't really matter, just that it has to be coherent # between the callback & redirect handlers. parts = [p.strip() for p in cookie_header.split(b";")] self.assertIn(b"Path=/_synapse/client/oidc", parts) name, cookie = parts[0].split(b"=") self.assertEqual(name, b"oidc_session") # Ensure the code_verifier is set in the cookie. macaroon = pymacaroons.Macaroon.deserialize(cookie) code_verifier = get_value_from_macaroon(macaroon, "code_verifier") self.assertNotEqual(code_verifier, "") @override_config({"oidc_config": {**DEFAULT_CONFIG, "pkce_method": "always"}}) def test_redirect_request_with_forced_code_challenge(self) -> None: """The redirect request has the right arguments & generates a valid session cookie.""" req = Mock(spec=["cookies"]) req.cookies = [] url = urlparse( self.get_success( self.provider.handle_redirect_request(req, b"http://client/redirect") ) ) # Ensure the code_challenge param is added to the redirect. params = parse_qs(url.query) self.assertEqual(len(params["code_challenge"]), 1) # Check what is in the cookies self.assertEqual(len(req.cookies), 2) # two cookies cookie_header = req.cookies[0] # The cookie name and path don't really matter, just that it has to be coherent # between the callback & redirect handlers. parts = [p.strip() for p in cookie_header.split(b";")] self.assertIn(b"Path=/_synapse/client/oidc", parts) name, cookie = parts[0].split(b"=") self.assertEqual(name, b"oidc_session") # Ensure the code_verifier is set in the cookie. macaroon = pymacaroons.Macaroon.deserialize(cookie) code_verifier = get_value_from_macaroon(macaroon, "code_verifier") self.assertNotEqual(code_verifier, "") @override_config({"oidc_config": {**DEFAULT_CONFIG, "pkce_method": "never"}}) def test_redirect_request_with_disabled_code_challenge(self) -> None: """The redirect request has the right arguments & generates a valid session cookie.""" req = Mock(spec=["cookies"]) req.cookies = [] # The metadata should state that PKCE is enabled. with self.metadata_edit({"code_challenge_methods_supported": ["S256"]}): url = urlparse( self.get_success( self.provider.handle_redirect_request( req, b"http://client/redirect" ) ) ) # Ensure the code_challenge param is added to the redirect. params = parse_qs(url.query) self.assertNotIn("code_challenge", params) # Check what is in the cookies self.assertEqual(len(req.cookies), 2) # two cookies cookie_header = req.cookies[0] # The cookie name and path don't really matter, just that it has to be coherent # between the callback & redirect handlers. parts = [p.strip() for p in cookie_header.split(b";")] self.assertIn(b"Path=/_synapse/client/oidc", parts) name, cookie = parts[0].split(b"=") self.assertEqual(name, b"oidc_session") # Ensure the code_verifier is blank in the cookie. macaroon = pymacaroons.Macaroon.deserialize(cookie) code_verifier = get_value_from_macaroon(macaroon, "code_verifier") self.assertEqual(code_verifier, "") @override_config({"oidc_config": DEFAULT_CONFIG}) def test_callback_error(self) -> None: """Errors from the provider returned in the callback are displayed.""" request = Mock(args={}) request.args[b"error"] = [b"invalid_client"] self.get_success(self.handler.handle_oidc_callback(request)) self.assertRenderedError("invalid_client", "") request.args[b"error_description"] = [b"some description"] self.get_success(self.handler.handle_oidc_callback(request)) self.assertRenderedError("invalid_client", "some description") @override_config({"oidc_config": DEFAULT_CONFIG}) def test_callback(self) -> None: """Code callback works and display errors if something went wrong. A lot of scenarios are tested here: - when the callback works, with userinfo from ID token - when the user mapping fails - when ID token verification fails - when the callback works, with userinfo fetched from the userinfo endpoint - when the userinfo fetching fails - when the code exchange fails """ # ensure that we are correctly testing the fallback when "get_extra_attributes" # is not implemented. mapping_provider = self.provider._user_mapping_provider with self.assertRaises(AttributeError): _ = mapping_provider.get_extra_attributes username = "bar" userinfo = { "sub": "foo", "username": username, } expected_user_id = "@%s:%s" % (username, self.hs.hostname) client_redirect_url = "http://client/redirect" request, _ = self.start_authorization( userinfo, client_redirect_url=client_redirect_url ) self.get_success(self.handler.handle_oidc_callback(request)) self.complete_sso_login.assert_called_once_with( expected_user_id, self.provider.idp_id, request, client_redirect_url, None, new_user=True, auth_provider_session_id=None, ) self.fake_server.post_token_handler.assert_called_once() self.fake_server.get_userinfo_handler.assert_not_called() self.render_error.assert_not_called() # Handle mapping errors request, _ = self.start_authorization(userinfo) with patch.object( self.provider, "_remote_id_from_userinfo", new=Mock(side_effect=MappingException()), ): self.get_success(self.handler.handle_oidc_callback(request)) self.assertRenderedError("mapping_error") # Handle ID token errors request, _ = self.start_authorization(userinfo) with self.fake_server.id_token_override({"iss": "https://bad.issuer/"}): self.get_success(self.handler.handle_oidc_callback(request)) self.assertRenderedError("invalid_token") self.reset_mocks() # With userinfo fetching self.provider._user_profile_method = "userinfo_endpoint" # Without the "openid" scope, the FakeProvider does not generate an id_token request, _ = self.start_authorization(userinfo, scope="") self.get_success(self.handler.handle_oidc_callback(request)) self.complete_sso_login.assert_called_once_with( expected_user_id, self.provider.idp_id, request, ANY, None, new_user=False, auth_provider_session_id=None, ) self.fake_server.post_token_handler.assert_called_once() self.fake_server.get_userinfo_handler.assert_called_once() self.render_error.assert_not_called() self.reset_mocks() # With an ID token, userinfo fetching and sid in the ID token self.provider._user_profile_method = "userinfo_endpoint" request, grant = self.start_authorization(userinfo, with_sid=True) self.assertIsNotNone(grant.sid) self.get_success(self.handler.handle_oidc_callback(request)) self.complete_sso_login.assert_called_once_with( expected_user_id, self.provider.idp_id, request, ANY, None, new_user=False, auth_provider_session_id=grant.sid, ) self.fake_server.post_token_handler.assert_called_once() self.fake_server.get_userinfo_handler.assert_called_once() self.render_error.assert_not_called() # Handle userinfo fetching error request, _ = self.start_authorization(userinfo) with self.fake_server.buggy_endpoint(userinfo=True): self.get_success(self.handler.handle_oidc_callback(request)) self.assertRenderedError("fetch_error") request, _ = self.start_authorization(userinfo) with self.fake_server.buggy_endpoint(token=True): self.get_success(self.handler.handle_oidc_callback(request)) self.assertRenderedError("server_error") @override_config({"oidc_config": DEFAULT_CONFIG}) def test_callback_session(self) -> None: """The callback verifies the session presence and validity""" request = Mock(spec=["args", "getCookie", "cookies"]) # Missing cookie request.args = {} request.getCookie.return_value = None self.get_success(self.handler.handle_oidc_callback(request)) self.assertRenderedError("missing_session", "No session cookie found") # Missing session parameter request.args = {} request.getCookie.return_value = "session" self.get_success(self.handler.handle_oidc_callback(request)) self.assertRenderedError("invalid_request", "State parameter is missing") # Invalid cookie request.args = {} request.args[b"state"] = [b"state"] request.getCookie.return_value = "session" self.get_success(self.handler.handle_oidc_callback(request)) self.assertRenderedError("invalid_session") # Mismatching session session = self._generate_oidc_session_token( state="state", nonce="nonce", client_redirect_url="http://client/redirect", ) request.args = {} request.args[b"state"] = [b"mismatching state"] request.getCookie.return_value = session self.get_success(self.handler.handle_oidc_callback(request)) self.assertRenderedError("mismatching_session") # Valid session request.args = {} request.args[b"state"] = [b"state"] request.getCookie.return_value = session self.get_success(self.handler.handle_oidc_callback(request)) self.assertRenderedError("invalid_request") @override_config( {"oidc_config": {**DEFAULT_CONFIG, "client_auth_method": "client_secret_post"}} ) def test_exchange_code(self) -> None: """Code exchange behaves correctly and handles various error scenarios.""" token = { "type": "Bearer", "access_token": "aabbcc", } self.fake_server.post_token_handler.side_effect = None self.fake_server.post_token_handler.return_value = FakeResponse.json( payload=token ) code = "code" ret = self.get_success(self.provider._exchange_code(code, code_verifier="")) kwargs = self.fake_server.request.call_args[1] self.assertEqual(ret, token) self.assertEqual(kwargs["method"], "POST") self.assertEqual(kwargs["uri"], self.fake_server.token_endpoint) args = parse_qs(kwargs["data"].decode("utf-8")) self.assertEqual(args["grant_type"], ["authorization_code"]) self.assertEqual(args["code"], [code]) self.assertEqual(args["client_id"], [CLIENT_ID]) self.assertEqual(args["client_secret"], [CLIENT_SECRET]) self.assertEqual(args["redirect_uri"], [CALLBACK_URL]) # Test providing a code verifier. code_verifier = "code_verifier" ret = self.get_success( self.provider._exchange_code(code, code_verifier=code_verifier) ) kwargs = self.fake_server.request.call_args[1] self.assertEqual(ret, token) self.assertEqual(kwargs["method"], "POST") self.assertEqual(kwargs["uri"], self.fake_server.token_endpoint) args = parse_qs(kwargs["data"].decode("utf-8")) self.assertEqual(args["grant_type"], ["authorization_code"]) self.assertEqual(args["code"], [code]) self.assertEqual(args["client_id"], [CLIENT_ID]) self.assertEqual(args["client_secret"], [CLIENT_SECRET]) self.assertEqual(args["redirect_uri"], [CALLBACK_URL]) self.assertEqual(args["code_verifier"], [code_verifier]) # Test error handling self.fake_server.post_token_handler.return_value = FakeResponse.json( code=400, payload={"error": "foo", "error_description": "bar"} ) from synapse.handlers.oidc import OidcError exc = self.get_failure( self.provider._exchange_code(code, code_verifier=""), OidcError ) self.assertEqual(exc.value.error, "foo") self.assertEqual(exc.value.error_description, "bar") # Internal server error with no JSON body self.fake_server.post_token_handler.return_value = FakeResponse( code=500, body=b"Not JSON" ) exc = self.get_failure( self.provider._exchange_code(code, code_verifier=""), OidcError ) self.assertEqual(exc.value.error, "server_error") # Internal server error with JSON body self.fake_server.post_token_handler.return_value = FakeResponse.json( code=500, payload={"error": "internal_server_error"} ) exc = self.get_failure( self.provider._exchange_code(code, code_verifier=""), OidcError ) self.assertEqual(exc.value.error, "internal_server_error") # 4xx error without "error" field self.fake_server.post_token_handler.return_value = FakeResponse.json( code=400, payload={} ) exc = self.get_failure( self.provider._exchange_code(code, code_verifier=""), OidcError ) self.assertEqual(exc.value.error, "server_error") # 2xx error with "error" field self.fake_server.post_token_handler.return_value = FakeResponse.json( code=200, payload={"error": "some_error"} ) exc = self.get_failure( self.provider._exchange_code(code, code_verifier=""), OidcError ) self.assertEqual(exc.value.error, "some_error") @override_config( { "oidc_config": { "enabled": True, "client_id": CLIENT_ID, "issuer": ISSUER, "client_auth_method": "client_secret_post", "client_secret_jwt_key": { "key_file": _key_file_path(), "jwt_header": {"alg": "ES256", "kid": "ABC789"}, "jwt_payload": {"iss": "DEFGHI"}, }, } } ) def test_exchange_code_jwt_key(self) -> None: """Test that code exchange works with a JWK client secret.""" from authlib.jose import jwt token = { "type": "Bearer", "access_token": "aabbcc", } self.fake_server.post_token_handler.side_effect = None self.fake_server.post_token_handler.return_value = FakeResponse.json( payload=token ) code = "code" # advance the clock a bit before we start, so we aren't working with zero # timestamps. self.reactor.advance(1000) start_time = self.reactor.seconds() ret = self.get_success(self.provider._exchange_code(code, code_verifier="")) self.assertEqual(ret, token) # the request should have hit the token endpoint kwargs = self.fake_server.request.call_args[1] self.assertEqual(kwargs["method"], "POST") self.assertEqual(kwargs["uri"], self.fake_server.token_endpoint) # the client secret provided to the should be a jwt which can be checked with # the public key args = parse_qs(kwargs["data"].decode("utf-8")) secret = args["client_secret"][0] with open(_public_key_file_path()) as f: key = f.read() claims = jwt.decode(secret, key) self.assertEqual(claims.header["kid"], "ABC789") self.assertEqual(claims["aud"], ISSUER) self.assertEqual(claims["iss"], "DEFGHI") self.assertEqual(claims["sub"], CLIENT_ID) self.assertEqual(claims["iat"], start_time) self.assertGreater(claims["exp"], start_time) # check the rest of the POSTed data self.assertEqual(args["grant_type"], ["authorization_code"]) self.assertEqual(args["code"], [code]) self.assertEqual(args["client_id"], [CLIENT_ID]) self.assertEqual(args["redirect_uri"], [CALLBACK_URL]) @override_config( { "oidc_config": { "enabled": True, "client_id": CLIENT_ID, "issuer": ISSUER, "client_auth_method": "none", } } ) def test_exchange_code_no_auth(self) -> None: """Test that code exchange works with no client secret.""" token = { "type": "Bearer", "access_token": "aabbcc", } self.fake_server.post_token_handler.side_effect = None self.fake_server.post_token_handler.return_value = FakeResponse.json( payload=token ) code = "code" ret = self.get_success(self.provider._exchange_code(code, code_verifier="")) self.assertEqual(ret, token) # the request should have hit the token endpoint kwargs = self.fake_server.request.call_args[1] self.assertEqual(kwargs["method"], "POST") self.assertEqual(kwargs["uri"], self.fake_server.token_endpoint) # check the POSTed data args = parse_qs(kwargs["data"].decode("utf-8")) self.assertEqual(args["grant_type"], ["authorization_code"]) self.assertEqual(args["code"], [code]) self.assertEqual(args["client_id"], [CLIENT_ID]) self.assertEqual(args["redirect_uri"], [CALLBACK_URL]) @override_config( { "oidc_config": { **DEFAULT_CONFIG, "user_mapping_provider": { "module": __name__ + ".TestMappingProviderExtra" }, } } ) def test_extra_attributes(self) -> None: """ Login while using a mapping provider that implements get_extra_attributes. """ userinfo = { "sub": "foo", "username": "foo", "phone": "1234567", } request, _ = self.start_authorization(userinfo) self.get_success(self.handler.handle_oidc_callback(request)) self.complete_sso_login.assert_called_once_with( "@foo:test", self.provider.idp_id, request, ANY, {"phone": "1234567"}, new_user=True, auth_provider_session_id=None, ) @override_config({"oidc_config": {**DEFAULT_CONFIG, "enable_registration": True}}) def test_map_userinfo_to_user(self) -> None: """Ensure that mapping the userinfo returned from a provider to an MXID works properly.""" userinfo: dict = { "sub": "test_user", "username": "test_user", } request, _ = self.start_authorization(userinfo) self.get_success(self.handler.handle_oidc_callback(request)) self.complete_sso_login.assert_called_once_with( "@test_user:test", self.provider.idp_id, request, ANY, None, new_user=True, auth_provider_session_id=None, ) self.reset_mocks() # Some providers return an integer ID. userinfo = { "sub": 1234, "username": "test_user_2", } request, _ = self.start_authorization(userinfo) self.get_success(self.handler.handle_oidc_callback(request)) self.complete_sso_login.assert_called_once_with( "@test_user_2:test", self.provider.idp_id, request, ANY, None, new_user=True, auth_provider_session_id=None, ) self.reset_mocks() # Test if the mxid is already taken store = self.hs.get_datastores().main user3 = UserID.from_string("@test_user_3:test") self.get_success( store.register_user(user_id=user3.to_string(), password_hash=None) ) userinfo = {"sub": "test3", "username": "test_user_3"} request, _ = self.start_authorization(userinfo) self.get_success(self.handler.handle_oidc_callback(request)) self.complete_sso_login.assert_not_called() self.assertRenderedError( "mapping_error", "Mapping provider does not support de-duplicating Matrix IDs", ) @override_config({"oidc_config": {**DEFAULT_CONFIG, "enable_registration": False}}) def test_map_userinfo_to_user_does_not_register_new_user(self) -> None: """Ensures new users are not registered if the enabled registration flag is disabled.""" userinfo: dict = { "sub": "test_user", "username": "test_user", } request, _ = self.start_authorization(userinfo) self.get_success(self.handler.handle_oidc_callback(request)) self.complete_sso_login.assert_not_called() self.assertRenderedError( "mapping_error", "User does not exist and registrations are disabled", ) @override_config({"oidc_config": {**DEFAULT_CONFIG, "allow_existing_users": True}}) def test_map_userinfo_to_existing_user(self) -> None: """Existing users can log in with OpenID Connect when allow_existing_users is True.""" store = self.hs.get_datastores().main user = UserID.from_string("@test_user:test") self.get_success( store.register_user(user_id=user.to_string(), password_hash=None) ) # Map a user via SSO. userinfo = { "sub": "test", "username": "test_user", } request, _ = self.start_authorization(userinfo) self.get_success(self.handler.handle_oidc_callback(request)) self.complete_sso_login.assert_called_once_with( user.to_string(), self.provider.idp_id, request, ANY, None, new_user=False, auth_provider_session_id=None, ) self.reset_mocks() # Subsequent calls should map to the same mxid. request, _ = self.start_authorization(userinfo) self.get_success(self.handler.handle_oidc_callback(request)) self.complete_sso_login.assert_called_once_with( user.to_string(), self.provider.idp_id, request, ANY, None, new_user=False, auth_provider_session_id=None, ) self.reset_mocks() # Note that a second SSO user can be mapped to the same Matrix ID. (This # requires a unique sub, but something that maps to the same matrix ID, # in this case we'll just use the same username. A more realistic example # would be subs which are email addresses, and mapping from the localpart # of the email, e.g. bob@foo.com and bob@bar.com -> @bob:test.) userinfo = { "sub": "test1", "username": "test_user", } request, _ = self.start_authorization(userinfo) self.get_success(self.handler.handle_oidc_callback(request)) self.complete_sso_login.assert_called_once_with( user.to_string(), self.provider.idp_id, request, ANY, None, new_user=False, auth_provider_session_id=None, ) self.reset_mocks() # Register some non-exact matching cases. user2 = UserID.from_string("@TEST_user_2:test") self.get_success( store.register_user(user_id=user2.to_string(), password_hash=None) ) user2_caps = UserID.from_string("@test_USER_2:test") self.get_success( store.register_user(user_id=user2_caps.to_string(), password_hash=None) ) # Attempting to login without matching a name exactly is an error. userinfo = { "sub": "test2", "username": "TEST_USER_2", } request, _ = self.start_authorization(userinfo) self.get_success(self.handler.handle_oidc_callback(request)) self.complete_sso_login.assert_not_called() args = self.assertRenderedError("mapping_error") self.assertTrue( args[2].startswith( "Attempted to login as '@TEST_USER_2:test' but it matches more than one user inexactly:" ) ) # Logging in when matching a name exactly should work. user2 = UserID.from_string("@TEST_USER_2:test") self.get_success( store.register_user(user_id=user2.to_string(), password_hash=None) ) request, _ = self.start_authorization(userinfo) self.get_success(self.handler.handle_oidc_callback(request)) self.complete_sso_login.assert_called_once_with( "@TEST_USER_2:test", self.provider.idp_id, request, ANY, None, new_user=False, auth_provider_session_id=None, ) @override_config({"oidc_config": DEFAULT_CONFIG}) def test_map_userinfo_to_invalid_localpart(self) -> None: """If the mapping provider generates an invalid localpart it should be rejected.""" userinfo = {"sub": "test2", "username": "föö"} request, _ = self.start_authorization(userinfo) self.get_success(self.handler.handle_oidc_callback(request)) self.assertRenderedError("mapping_error", "localpart is invalid: föö") @override_config( { "oidc_config": { **DEFAULT_CONFIG, "user_mapping_provider": { "module": __name__ + ".TestMappingProviderFailures" }, } } ) def test_map_userinfo_to_user_retries(self) -> None: """The mapping provider can retry generating an MXID if the MXID is already in use.""" store = self.hs.get_datastores().main self.get_success( store.register_user(user_id="@test_user:test", password_hash=None) ) userinfo = { "sub": "test", "username": "test_user", } request, _ = self.start_authorization(userinfo) self.get_success(self.handler.handle_oidc_callback(request)) # test_user is already taken, so test_user1 gets registered instead. self.complete_sso_login.assert_called_once_with( "@test_user1:test", self.provider.idp_id, request, ANY, None, new_user=True, auth_provider_session_id=None, ) self.reset_mocks() # Register all of the potential mxids for a particular OIDC username. self.get_success( store.register_user(user_id="@tester:test", password_hash=None) ) for i in range(1, 3): self.get_success( store.register_user(user_id="@tester%d:test" % i, password_hash=None) ) # Now attempt to map to a username, this will fail since all potential usernames are taken. userinfo = { "sub": "tester", "username": "tester", } request, _ = self.start_authorization(userinfo) self.get_success(self.handler.handle_oidc_callback(request)) self.complete_sso_login.assert_not_called() self.assertRenderedError( "mapping_error", "Unable to generate a Matrix ID from the SSO response" ) @override_config({"oidc_config": DEFAULT_CONFIG}) def test_empty_localpart(self) -> None: """Attempts to map onto an empty localpart should be rejected.""" userinfo = { "sub": "tester", "username": "", } request, _ = self.start_authorization(userinfo) self.get_success(self.handler.handle_oidc_callback(request)) self.assertRenderedError("mapping_error", "localpart is invalid: ") @override_config( { "oidc_config": { **DEFAULT_CONFIG, "user_mapping_provider": { "config": {"localpart_template": "{{ user.username }}"} }, } } ) def test_null_localpart(self) -> None: """Mapping onto a null localpart via an empty OIDC attribute should be rejected""" userinfo = { "sub": "tester", "username": None, } request, _ = self.start_authorization(userinfo) self.get_success(self.handler.handle_oidc_callback(request)) self.assertRenderedError("mapping_error", "localpart is invalid: ") @override_config( { "oidc_config": { **DEFAULT_CONFIG, "attribute_requirements": [{"attribute": "test", "value": "foobar"}], } } ) def test_attribute_requirements(self) -> None: """The required attributes must be met from the OIDC userinfo response.""" # userinfo lacking "test": "foobar" attribute should fail. userinfo = { "sub": "tester", "username": "tester", } request, _ = self.start_authorization(userinfo) self.get_success(self.handler.handle_oidc_callback(request)) self.complete_sso_login.assert_not_called() # userinfo with "test": "foobar" attribute should succeed. userinfo = { "sub": "tester", "username": "tester", "test": "foobar", } request, _ = self.start_authorization(userinfo) self.get_success(self.handler.handle_oidc_callback(request)) # check that the auth handler got called as expected self.complete_sso_login.assert_called_once_with( "@tester:test", self.provider.idp_id, request, ANY, None, new_user=True, auth_provider_session_id=None, ) @override_config( { "oidc_config": { **DEFAULT_CONFIG, "attribute_requirements": [{"attribute": "test", "value": "foobar"}], } } ) def test_attribute_requirements_contains(self) -> None: """Test that auth succeeds if userinfo attribute CONTAINS required value""" # userinfo with "test": ["foobar", "foo", "bar"] attribute should succeed. userinfo = { "sub": "tester", "username": "tester", "test": ["foobar", "foo", "bar"], } request, _ = self.start_authorization(userinfo) self.get_success(self.handler.handle_oidc_callback(request)) # check that the auth handler got called as expected self.complete_sso_login.assert_called_once_with( "@tester:test", self.provider.idp_id, request, ANY, None, new_user=True, auth_provider_session_id=None, ) @override_config( { "oidc_config": { **DEFAULT_CONFIG, "attribute_requirements": [{"attribute": "test", "value": "foobar"}], } } ) def test_attribute_requirements_mismatch(self) -> None: """ Test that auth fails if attributes exist but don't match, or are non-string values. """ # userinfo with "test": "not_foobar" attribute should fail userinfo: dict = { "sub": "tester", "username": "tester", "test": "not_foobar", } request, _ = self.start_authorization(userinfo) self.get_success(self.handler.handle_oidc_callback(request)) self.complete_sso_login.assert_not_called() # userinfo with "test": ["foo", "bar"] attribute should fail userinfo = { "sub": "tester", "username": "tester", "test": ["foo", "bar"], } request, _ = self.start_authorization(userinfo) self.get_success(self.handler.handle_oidc_callback(request)) self.complete_sso_login.assert_not_called() # userinfo with "test": False attribute should fail # this is largely just to ensure we don't crash here userinfo = { "sub": "tester", "username": "tester", "test": False, } request, _ = self.start_authorization(userinfo) self.get_success(self.handler.handle_oidc_callback(request)) self.complete_sso_login.assert_not_called() # userinfo with "test": None attribute should fail # a value of None breaks the OIDC spec, but it's important to not crash here userinfo = { "sub": "tester", "username": "tester", "test": None, } request, _ = self.start_authorization(userinfo) self.get_success(self.handler.handle_oidc_callback(request)) self.complete_sso_login.assert_not_called() # userinfo with "test": 1 attribute should fail # this is largely just to ensure we don't crash here userinfo = { "sub": "tester", "username": "tester", "test": 1, } request, _ = self.start_authorization(userinfo) self.get_success(self.handler.handle_oidc_callback(request)) self.complete_sso_login.assert_not_called() # userinfo with "test": 3.14 attribute should fail # this is largely just to ensure we don't crash here userinfo = { "sub": "tester", "username": "tester", "test": 3.14, } request, _ = self.start_authorization(userinfo) self.get_success(self.handler.handle_oidc_callback(request)) self.complete_sso_login.assert_not_called() def _generate_oidc_session_token( self, state: str, nonce: str, client_redirect_url: str, ui_auth_session_id: str = "", ) -> str: from synapse.handlers.oidc import OidcSessionData return self.handler._macaroon_generator.generate_oidc_session_token( state=state, session_data=OidcSessionData( idp_id=self.provider.idp_id, nonce=nonce, client_redirect_url=client_redirect_url, ui_auth_session_id=ui_auth_session_id, code_verifier="", ), ) def _build_callback_request( code: str, state: str, session: str, ip_address: str = "10.0.0.1", ) -> Mock: """Builds a fake SynapseRequest to mock the browser callback Returns a Mock object which looks like the SynapseRequest we get from a browser after SSO (before we return to the client) Args: code: the authorization code which would have been returned by the OIDC provider state: the "state" param which would have been passed around in the query param. Should be the same as was embedded in the session in _build_oidc_session. session: the "session" which would have been passed around in the cookie. ip_address: the IP address to pretend the request came from """ request = Mock( spec=[ "args", "getCookie", "cookies", "requestHeaders", "getClientAddress", "getHeader", ] ) request.cookies = [] request.getCookie.return_value = session request.args = {} request.args[b"code"] = [code.encode("utf-8")] request.args[b"state"] = [state.encode("utf-8")] request.getClientAddress.return_value.host = ip_address return request