test_terms_auth.py 4.8 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129
  1. # Copyright 2018 New Vector Ltd
  2. #
  3. # Licensed under the Apache License, Version 2.0 (the "License");
  4. # you may not use this file except in compliance with the License.
  5. # You may obtain a copy of the License at
  6. #
  7. # http://www.apache.org/licenses/LICENSE-2.0
  8. #
  9. # Unless required by applicable law or agreed to in writing, software
  10. # distributed under the License is distributed on an "AS IS" BASIS,
  11. # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  12. # See the License for the specific language governing permissions and
  13. # limitations under the License.
  14. from unittest.mock import Mock
  15. from twisted.internet.interfaces import IReactorTime
  16. from twisted.test.proto_helpers import MemoryReactor, MemoryReactorClock
  17. from synapse.rest.client.register import register_servlets
  18. from synapse.server import HomeServer
  19. from synapse.types import JsonDict
  20. from synapse.util import Clock
  21. from tests import unittest
  22. class TermsTestCase(unittest.HomeserverTestCase):
  23. servlets = [register_servlets]
  24. def default_config(self) -> JsonDict:
  25. config = super().default_config()
  26. config.update(
  27. {
  28. "public_baseurl": "https://example.org/",
  29. "user_consent": {
  30. "version": "1.0",
  31. "policy_name": "My Cool Privacy Policy",
  32. "template_dir": "/",
  33. "require_at_registration": True,
  34. },
  35. }
  36. )
  37. return config
  38. def prepare(
  39. self, reactor: MemoryReactor, clock: Clock, homeserver: HomeServer
  40. ) -> None:
  41. # type-ignore: mypy-zope doesn't seem to recognise that MemoryReactorClock
  42. # implements IReactorTime, via inheritance from twisted.internet.testing.Clock
  43. self.clock: IReactorTime = MemoryReactorClock() # type: ignore[assignment]
  44. self.hs_clock = Clock(self.clock)
  45. self.url = "/_matrix/client/r0/register"
  46. self.registration_handler = Mock()
  47. self.auth_handler = Mock()
  48. self.device_handler = Mock()
  49. def test_ui_auth(self) -> None:
  50. # Do a UI auth request
  51. request_data: JsonDict = {"username": "kermit", "password": "monkey"}
  52. channel = self.make_request(b"POST", self.url, request_data)
  53. self.assertEqual(channel.code, 401, channel.result)
  54. self.assertTrue(channel.json_body is not None)
  55. self.assertIsInstance(channel.json_body["session"], str)
  56. self.assertIsInstance(channel.json_body["flows"], list)
  57. for flow in channel.json_body["flows"]:
  58. self.assertIsInstance(flow["stages"], list)
  59. self.assertTrue(len(flow["stages"]) > 0)
  60. self.assertTrue("m.login.terms" in flow["stages"])
  61. expected_params = {
  62. "m.login.terms": {
  63. "policies": {
  64. "privacy_policy": {
  65. "en": {
  66. "name": "My Cool Privacy Policy",
  67. "url": "https://example.org/_matrix/consent?v=1.0",
  68. },
  69. "version": "1.0",
  70. }
  71. }
  72. }
  73. }
  74. self.assertIsInstance(channel.json_body["params"], dict)
  75. self.assertLessEqual(
  76. channel.json_body["params"].items(), expected_params.items()
  77. )
  78. # We have to complete the dummy auth stage before completing the terms stage
  79. request_data = {
  80. "username": "kermit",
  81. "password": "monkey",
  82. "auth": {
  83. "session": channel.json_body["session"],
  84. "type": "m.login.dummy",
  85. },
  86. }
  87. self.registration_handler.check_username = Mock(return_value=True)
  88. channel = self.make_request(b"POST", self.url, request_data)
  89. # We don't bother checking that the response is correct - we'll leave that to
  90. # other tests. We just want to make sure we're on the right path.
  91. self.assertEqual(channel.code, 401, channel.result)
  92. # Finish the UI auth for terms
  93. request_data = {
  94. "username": "kermit",
  95. "password": "monkey",
  96. "auth": {
  97. "session": channel.json_body["session"],
  98. "type": "m.login.terms",
  99. },
  100. }
  101. channel = self.make_request(b"POST", self.url, request_data)
  102. # We're interested in getting a response that looks like a successful
  103. # registration, not so much that the details are exactly what we want.
  104. self.assertEqual(channel.code, 200, channel.result)
  105. self.assertTrue(channel.json_body is not None)
  106. self.assertIsInstance(channel.json_body["user_id"], str)
  107. self.assertIsInstance(channel.json_body["access_token"], str)
  108. self.assertIsInstance(channel.json_body["device_id"], str)