mirror of
https://mau.dev/maunium/synapse.git
synced 2024-11-18 16:02:15 +01:00
2ec8ca5e60
SynapseRequest is in danger of becoming a bit of a dumping-ground for "useful stuff relating to Requests", which isn't really its intention (its purpose is to override render, finished and connectionLost to set up the LoggingContext and write the right entries to the request log). Putting utility functions inside SynapseRequest means that lots of our code ends up requiring a SynapseRequest when there is nothing synapse-specific about the Request at all, and any old twisted.web.iweb.IRequest will do. This increases code coupling and makes testing more difficult. In short: move get_user_agent out to a utility function.
121 lines
4.2 KiB
Python
121 lines
4.2 KiB
Python
# Copyright 2020 The Matrix.org Foundation C.I.C.
|
|
#
|
|
# Licensed under the Apache License, Version 2.0 (the "License");
|
|
# you may not use this file except in compliance with the License.
|
|
# 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 mock import Mock
|
|
|
|
from synapse.handlers.cas_handler import CasResponse
|
|
|
|
from tests.test_utils import simple_async_mock
|
|
from tests.unittest import HomeserverTestCase
|
|
|
|
# These are a few constants that are used as config parameters in the tests.
|
|
BASE_URL = "https://synapse/"
|
|
SERVER_URL = "https://issuer/"
|
|
|
|
|
|
class CasHandlerTestCase(HomeserverTestCase):
|
|
def default_config(self):
|
|
config = super().default_config()
|
|
config["public_baseurl"] = BASE_URL
|
|
cas_config = {
|
|
"enabled": True,
|
|
"server_url": SERVER_URL,
|
|
"service_url": BASE_URL,
|
|
}
|
|
config["cas_config"] = cas_config
|
|
|
|
return config
|
|
|
|
def make_homeserver(self, reactor, clock):
|
|
hs = self.setup_test_homeserver()
|
|
|
|
self.handler = hs.get_cas_handler()
|
|
|
|
# Reduce the number of attempts when generating MXIDs.
|
|
sso_handler = hs.get_sso_handler()
|
|
sso_handler._MAP_USERNAME_RETRIES = 3
|
|
|
|
return hs
|
|
|
|
def test_map_cas_user_to_user(self):
|
|
"""Ensure that mapping the CAS user returned from a provider to an MXID works properly."""
|
|
|
|
# stub out the auth handler
|
|
auth_handler = self.hs.get_auth_handler()
|
|
auth_handler.complete_sso_login = simple_async_mock()
|
|
|
|
cas_response = CasResponse("test_user", {})
|
|
request = _mock_request()
|
|
self.get_success(
|
|
self.handler._handle_cas_response(request, cas_response, "redirect_uri", "")
|
|
)
|
|
|
|
# check that the auth handler got called as expected
|
|
auth_handler.complete_sso_login.assert_called_once_with(
|
|
"@test_user:test", request, "redirect_uri", None
|
|
)
|
|
|
|
def test_map_cas_user_to_existing_user(self):
|
|
"""Existing users can log in with CAS account."""
|
|
store = self.hs.get_datastore()
|
|
self.get_success(
|
|
store.register_user(user_id="@test_user:test", password_hash=None)
|
|
)
|
|
|
|
# stub out the auth handler
|
|
auth_handler = self.hs.get_auth_handler()
|
|
auth_handler.complete_sso_login = simple_async_mock()
|
|
|
|
# Map a user via SSO.
|
|
cas_response = CasResponse("test_user", {})
|
|
request = _mock_request()
|
|
self.get_success(
|
|
self.handler._handle_cas_response(request, cas_response, "redirect_uri", "")
|
|
)
|
|
|
|
# check that the auth handler got called as expected
|
|
auth_handler.complete_sso_login.assert_called_once_with(
|
|
"@test_user:test", request, "redirect_uri", None
|
|
)
|
|
|
|
# Subsequent calls should map to the same mxid.
|
|
auth_handler.complete_sso_login.reset_mock()
|
|
self.get_success(
|
|
self.handler._handle_cas_response(request, cas_response, "redirect_uri", "")
|
|
)
|
|
auth_handler.complete_sso_login.assert_called_once_with(
|
|
"@test_user:test", request, "redirect_uri", None
|
|
)
|
|
|
|
def test_map_cas_user_to_invalid_localpart(self):
|
|
"""CAS automaps invalid characters to base-64 encoding."""
|
|
|
|
# stub out the auth handler
|
|
auth_handler = self.hs.get_auth_handler()
|
|
auth_handler.complete_sso_login = simple_async_mock()
|
|
|
|
cas_response = CasResponse("föö", {})
|
|
request = _mock_request()
|
|
self.get_success(
|
|
self.handler._handle_cas_response(request, cas_response, "redirect_uri", "")
|
|
)
|
|
|
|
# check that the auth handler got called as expected
|
|
auth_handler.complete_sso_login.assert_called_once_with(
|
|
"@f=c3=b6=c3=b6:test", request, "redirect_uri", None
|
|
)
|
|
|
|
|
|
def _mock_request():
|
|
"""Returns a mock which will stand in as a SynapseRequest"""
|
|
return Mock(spec=["getClientIP", "getHeader"])
|