1
0

test_auth.py 23 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529
  1. # Copyright 2015 - 2016 OpenMarket 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 AsyncMock, Mock
  15. import pymacaroons
  16. from twisted.test.proto_helpers import MemoryReactor
  17. from synapse.api.auth.internal import InternalAuth
  18. from synapse.api.auth_blocking import AuthBlocking
  19. from synapse.api.constants import UserTypes
  20. from synapse.api.errors import (
  21. AuthError,
  22. Codes,
  23. InvalidClientTokenError,
  24. MissingClientTokenError,
  25. ResourceLimitError,
  26. )
  27. from synapse.appservice import ApplicationService
  28. from synapse.server import HomeServer
  29. from synapse.storage.databases.main.registration import TokenLookupResult
  30. from synapse.types import Requester, UserID
  31. from synapse.util import Clock
  32. from tests import unittest
  33. from tests.unittest import override_config
  34. from tests.utils import mock_getRawHeaders
  35. class AuthTestCase(unittest.HomeserverTestCase):
  36. def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None:
  37. self.store = Mock()
  38. # type-ignore: datastores is None until hs.setup() is called---but it'll
  39. # have been called by the HomeserverTestCase machinery.
  40. hs.datastores.main = self.store # type: ignore[union-attr]
  41. hs.get_auth_handler().store = self.store
  42. self.auth = InternalAuth(hs)
  43. # AuthBlocking reads from the hs' config on initialization. We need to
  44. # modify its config instead of the hs'
  45. self.auth_blocking = AuthBlocking(hs)
  46. self.test_user = "@foo:bar"
  47. self.test_token = b"_test_token_"
  48. # this is overridden for the appservice tests
  49. self.store.get_app_service_by_token = Mock(return_value=None)
  50. self.store.insert_client_ip = AsyncMock(return_value=None)
  51. self.store.is_support_user = AsyncMock(return_value=False)
  52. def test_get_user_by_req_user_valid_token(self) -> None:
  53. user_info = TokenLookupResult(
  54. user_id=self.test_user, token_id=5, device_id="device"
  55. )
  56. self.store.get_user_by_access_token = AsyncMock(return_value=user_info)
  57. self.store.mark_access_token_as_used = AsyncMock(return_value=None)
  58. self.store.get_user_locked_status = AsyncMock(return_value=False)
  59. request = Mock(args={})
  60. request.args[b"access_token"] = [self.test_token]
  61. request.requestHeaders.getRawHeaders = mock_getRawHeaders()
  62. requester = self.get_success(self.auth.get_user_by_req(request))
  63. self.assertEqual(requester.user.to_string(), self.test_user)
  64. def test_get_user_by_req_user_bad_token(self) -> None:
  65. self.store.get_user_by_access_token = AsyncMock(return_value=None)
  66. request = Mock(args={})
  67. request.args[b"access_token"] = [self.test_token]
  68. request.requestHeaders.getRawHeaders = mock_getRawHeaders()
  69. f = self.get_failure(
  70. self.auth.get_user_by_req(request), InvalidClientTokenError
  71. ).value
  72. self.assertEqual(f.code, 401)
  73. self.assertEqual(f.errcode, "M_UNKNOWN_TOKEN")
  74. def test_get_user_by_req_user_missing_token(self) -> None:
  75. user_info = TokenLookupResult(user_id=self.test_user, token_id=5)
  76. self.store.get_user_by_access_token = AsyncMock(return_value=user_info)
  77. request = Mock(args={})
  78. request.requestHeaders.getRawHeaders = mock_getRawHeaders()
  79. f = self.get_failure(
  80. self.auth.get_user_by_req(request), MissingClientTokenError
  81. ).value
  82. self.assertEqual(f.code, 401)
  83. self.assertEqual(f.errcode, "M_MISSING_TOKEN")
  84. def test_get_user_by_req_appservice_valid_token(self) -> None:
  85. app_service = Mock(
  86. token="foobar", url="a_url", sender=self.test_user, ip_range_whitelist=None
  87. )
  88. self.store.get_app_service_by_token = Mock(return_value=app_service)
  89. self.store.get_user_by_access_token = AsyncMock(return_value=None)
  90. request = Mock(args={})
  91. request.getClientAddress.return_value.host = "127.0.0.1"
  92. request.args[b"access_token"] = [self.test_token]
  93. request.requestHeaders.getRawHeaders = mock_getRawHeaders()
  94. requester = self.get_success(self.auth.get_user_by_req(request))
  95. self.assertEqual(requester.user.to_string(), self.test_user)
  96. def test_get_user_by_req_appservice_valid_token_good_ip(self) -> None:
  97. from netaddr import IPSet
  98. app_service = Mock(
  99. token="foobar",
  100. url="a_url",
  101. sender=self.test_user,
  102. ip_range_whitelist=IPSet(["192.168/16"]),
  103. )
  104. self.store.get_app_service_by_token = Mock(return_value=app_service)
  105. self.store.get_user_by_access_token = AsyncMock(return_value=None)
  106. request = Mock(args={})
  107. request.getClientAddress.return_value.host = "192.168.10.10"
  108. request.args[b"access_token"] = [self.test_token]
  109. request.requestHeaders.getRawHeaders = mock_getRawHeaders()
  110. requester = self.get_success(self.auth.get_user_by_req(request))
  111. self.assertEqual(requester.user.to_string(), self.test_user)
  112. def test_get_user_by_req_appservice_valid_token_bad_ip(self) -> None:
  113. from netaddr import IPSet
  114. app_service = Mock(
  115. token="foobar",
  116. url="a_url",
  117. sender=self.test_user,
  118. ip_range_whitelist=IPSet(["192.168/16"]),
  119. )
  120. self.store.get_app_service_by_token = Mock(return_value=app_service)
  121. self.store.get_user_by_access_token = AsyncMock(return_value=None)
  122. request = Mock(args={})
  123. request.getClientAddress.return_value.host = "131.111.8.42"
  124. request.args[b"access_token"] = [self.test_token]
  125. request.requestHeaders.getRawHeaders = mock_getRawHeaders()
  126. f = self.get_failure(
  127. self.auth.get_user_by_req(request), InvalidClientTokenError
  128. ).value
  129. self.assertEqual(f.code, 401)
  130. self.assertEqual(f.errcode, "M_UNKNOWN_TOKEN")
  131. def test_get_user_by_req_appservice_bad_token(self) -> None:
  132. self.store.get_app_service_by_token = Mock(return_value=None)
  133. self.store.get_user_by_access_token = AsyncMock(return_value=None)
  134. request = Mock(args={})
  135. request.args[b"access_token"] = [self.test_token]
  136. request.requestHeaders.getRawHeaders = mock_getRawHeaders()
  137. f = self.get_failure(
  138. self.auth.get_user_by_req(request), InvalidClientTokenError
  139. ).value
  140. self.assertEqual(f.code, 401)
  141. self.assertEqual(f.errcode, "M_UNKNOWN_TOKEN")
  142. def test_get_user_by_req_appservice_missing_token(self) -> None:
  143. app_service = Mock(token="foobar", url="a_url", sender=self.test_user)
  144. self.store.get_app_service_by_token = Mock(return_value=app_service)
  145. self.store.get_user_by_access_token = AsyncMock(return_value=None)
  146. request = Mock(args={})
  147. request.requestHeaders.getRawHeaders = mock_getRawHeaders()
  148. f = self.get_failure(
  149. self.auth.get_user_by_req(request), MissingClientTokenError
  150. ).value
  151. self.assertEqual(f.code, 401)
  152. self.assertEqual(f.errcode, "M_MISSING_TOKEN")
  153. def test_get_user_by_req_appservice_valid_token_valid_user_id(self) -> None:
  154. masquerading_user_id = b"@doppelganger:matrix.org"
  155. app_service = Mock(
  156. token="foobar", url="a_url", sender=self.test_user, ip_range_whitelist=None
  157. )
  158. app_service.is_interested_in_user = Mock(return_value=True)
  159. self.store.get_app_service_by_token = Mock(return_value=app_service)
  160. class FakeUserInfo:
  161. is_guest = False
  162. self.store.get_user_by_id = AsyncMock(return_value=FakeUserInfo())
  163. self.store.get_user_by_access_token = AsyncMock(return_value=None)
  164. request = Mock(args={})
  165. request.getClientAddress.return_value.host = "127.0.0.1"
  166. request.args[b"access_token"] = [self.test_token]
  167. request.args[b"user_id"] = [masquerading_user_id]
  168. request.requestHeaders.getRawHeaders = mock_getRawHeaders()
  169. requester = self.get_success(self.auth.get_user_by_req(request))
  170. self.assertEqual(
  171. requester.user.to_string(), masquerading_user_id.decode("utf8")
  172. )
  173. def test_get_user_by_req_appservice_valid_token_bad_user_id(self) -> None:
  174. masquerading_user_id = b"@doppelganger:matrix.org"
  175. app_service = Mock(
  176. token="foobar", url="a_url", sender=self.test_user, ip_range_whitelist=None
  177. )
  178. app_service.is_interested_in_user = Mock(return_value=False)
  179. self.store.get_app_service_by_token = Mock(return_value=app_service)
  180. self.store.get_user_by_access_token = AsyncMock(return_value=None)
  181. request = Mock(args={})
  182. request.getClientAddress.return_value.host = "127.0.0.1"
  183. request.args[b"access_token"] = [self.test_token]
  184. request.args[b"user_id"] = [masquerading_user_id]
  185. request.requestHeaders.getRawHeaders = mock_getRawHeaders()
  186. self.get_failure(self.auth.get_user_by_req(request), AuthError)
  187. @override_config({"experimental_features": {"msc3202_device_masquerading": True}})
  188. def test_get_user_by_req_appservice_valid_token_valid_device_id(self) -> None:
  189. """
  190. Tests that when an application service passes the device_id URL parameter
  191. with the ID of a valid device for the user in question,
  192. the requester instance tracks that device ID.
  193. """
  194. masquerading_user_id = b"@doppelganger:matrix.org"
  195. masquerading_device_id = b"DOPPELDEVICE"
  196. app_service = Mock(
  197. token="foobar", url="a_url", sender=self.test_user, ip_range_whitelist=None
  198. )
  199. app_service.is_interested_in_user = Mock(return_value=True)
  200. self.store.get_app_service_by_token = Mock(return_value=app_service)
  201. # This just needs to return a truth-y value.
  202. self.store.get_user_by_id = AsyncMock(return_value={"is_guest": False})
  203. self.store.get_user_by_access_token = AsyncMock(return_value=None)
  204. # This also needs to just return a truth-y value
  205. self.store.get_device = AsyncMock(return_value={"hidden": False})
  206. request = Mock(args={})
  207. request.getClientAddress.return_value.host = "127.0.0.1"
  208. request.args[b"access_token"] = [self.test_token]
  209. request.args[b"user_id"] = [masquerading_user_id]
  210. request.args[b"org.matrix.msc3202.device_id"] = [masquerading_device_id]
  211. request.requestHeaders.getRawHeaders = mock_getRawHeaders()
  212. requester = self.get_success(self.auth.get_user_by_req(request))
  213. self.assertEqual(
  214. requester.user.to_string(), masquerading_user_id.decode("utf8")
  215. )
  216. self.assertEqual(requester.device_id, masquerading_device_id.decode("utf8"))
  217. @override_config({"experimental_features": {"msc3202_device_masquerading": True}})
  218. def test_get_user_by_req_appservice_valid_token_invalid_device_id(self) -> None:
  219. """
  220. Tests that when an application service passes the device_id URL parameter
  221. with an ID that is not a valid device ID for the user in question,
  222. the request fails with the appropriate error code.
  223. """
  224. masquerading_user_id = b"@doppelganger:matrix.org"
  225. masquerading_device_id = b"NOT_A_REAL_DEVICE_ID"
  226. app_service = Mock(
  227. token="foobar", url="a_url", sender=self.test_user, ip_range_whitelist=None
  228. )
  229. app_service.is_interested_in_user = Mock(return_value=True)
  230. self.store.get_app_service_by_token = Mock(return_value=app_service)
  231. # This just needs to return a truth-y value.
  232. self.store.get_user_by_id = AsyncMock(return_value={"is_guest": False})
  233. self.store.get_user_by_access_token = AsyncMock(return_value=None)
  234. # This also needs to just return a falsey value
  235. self.store.get_device = AsyncMock(return_value=None)
  236. request = Mock(args={})
  237. request.getClientAddress.return_value.host = "127.0.0.1"
  238. request.args[b"access_token"] = [self.test_token]
  239. request.args[b"user_id"] = [masquerading_user_id]
  240. request.args[b"org.matrix.msc3202.device_id"] = [masquerading_device_id]
  241. request.requestHeaders.getRawHeaders = mock_getRawHeaders()
  242. failure = self.get_failure(self.auth.get_user_by_req(request), AuthError)
  243. self.assertEqual(failure.value.code, 400)
  244. self.assertEqual(failure.value.errcode, Codes.EXCLUSIVE)
  245. def test_get_user_by_req__puppeted_token__not_tracking_puppeted_mau(self) -> None:
  246. self.store.get_user_by_access_token = AsyncMock(
  247. return_value=TokenLookupResult(
  248. user_id="@baldrick:matrix.org",
  249. device_id="device",
  250. token_id=5,
  251. token_owner="@admin:matrix.org",
  252. token_used=True,
  253. )
  254. )
  255. self.store.insert_client_ip = AsyncMock(return_value=None)
  256. self.store.mark_access_token_as_used = AsyncMock(return_value=None)
  257. self.store.get_user_locked_status = AsyncMock(return_value=False)
  258. request = Mock(args={})
  259. request.getClientAddress.return_value.host = "127.0.0.1"
  260. request.args[b"access_token"] = [self.test_token]
  261. request.requestHeaders.getRawHeaders = mock_getRawHeaders()
  262. self.get_success(self.auth.get_user_by_req(request))
  263. self.store.insert_client_ip.assert_called_once()
  264. def test_get_user_by_req__puppeted_token__tracking_puppeted_mau(self) -> None:
  265. self.auth._track_puppeted_user_ips = True
  266. self.store.get_user_by_access_token = AsyncMock(
  267. return_value=TokenLookupResult(
  268. user_id="@baldrick:matrix.org",
  269. device_id="device",
  270. token_id=5,
  271. token_owner="@admin:matrix.org",
  272. token_used=True,
  273. )
  274. )
  275. self.store.get_user_locked_status = AsyncMock(return_value=False)
  276. self.store.insert_client_ip = AsyncMock(return_value=None)
  277. self.store.mark_access_token_as_used = AsyncMock(return_value=None)
  278. request = Mock(args={})
  279. request.getClientAddress.return_value.host = "127.0.0.1"
  280. request.args[b"access_token"] = [self.test_token]
  281. request.requestHeaders.getRawHeaders = mock_getRawHeaders()
  282. self.get_success(self.auth.get_user_by_req(request))
  283. self.assertEqual(self.store.insert_client_ip.call_count, 2)
  284. def test_get_user_from_macaroon(self) -> None:
  285. self.store.get_user_by_access_token = AsyncMock(return_value=None)
  286. user_id = "@baldrick:matrix.org"
  287. macaroon = pymacaroons.Macaroon(
  288. location=self.hs.config.server.server_name,
  289. identifier="key",
  290. key=self.hs.config.key.macaroon_secret_key,
  291. )
  292. # "Legacy" macaroons should not work for regular users not in the database
  293. macaroon.add_first_party_caveat("gen = 1")
  294. macaroon.add_first_party_caveat("type = access")
  295. macaroon.add_first_party_caveat("user_id = %s" % (user_id,))
  296. serialized = macaroon.serialize()
  297. self.get_failure(
  298. self.auth.get_user_by_access_token(serialized), InvalidClientTokenError
  299. )
  300. def test_get_guest_user_from_macaroon(self) -> None:
  301. class FakeUserInfo:
  302. is_guest = True
  303. self.store.get_user_by_id = AsyncMock(return_value=FakeUserInfo())
  304. self.store.get_user_by_access_token = AsyncMock(return_value=None)
  305. user_id = "@baldrick:matrix.org"
  306. macaroon = pymacaroons.Macaroon(
  307. location=self.hs.config.server.server_name,
  308. identifier="key",
  309. key=self.hs.config.key.macaroon_secret_key,
  310. )
  311. macaroon.add_first_party_caveat("gen = 1")
  312. macaroon.add_first_party_caveat("type = access")
  313. macaroon.add_first_party_caveat("user_id = %s" % (user_id,))
  314. macaroon.add_first_party_caveat("guest = true")
  315. serialized = macaroon.serialize()
  316. user_info = self.get_success(self.auth.get_user_by_access_token(serialized))
  317. self.assertEqual(user_id, user_info.user.to_string())
  318. self.assertTrue(user_info.is_guest)
  319. self.store.get_user_by_id.assert_called_with(user_id)
  320. def test_blocking_mau(self) -> None:
  321. self.auth_blocking._limit_usage_by_mau = False
  322. self.auth_blocking._max_mau_value = 50
  323. lots_of_users = 100
  324. small_number_of_users = 1
  325. # Ensure no error thrown
  326. self.get_success(self.auth_blocking.check_auth_blocking())
  327. self.auth_blocking._limit_usage_by_mau = True
  328. self.store.get_monthly_active_count = AsyncMock(return_value=lots_of_users)
  329. e = self.get_failure(
  330. self.auth_blocking.check_auth_blocking(), ResourceLimitError
  331. )
  332. self.assertEqual(e.value.admin_contact, self.hs.config.server.admin_contact)
  333. self.assertEqual(e.value.errcode, Codes.RESOURCE_LIMIT_EXCEEDED)
  334. self.assertEqual(e.value.code, 403)
  335. # Ensure does not throw an error
  336. self.store.get_monthly_active_count = AsyncMock(
  337. return_value=small_number_of_users
  338. )
  339. self.get_success(self.auth_blocking.check_auth_blocking())
  340. def test_blocking_mau__depending_on_user_type(self) -> None:
  341. self.auth_blocking._max_mau_value = 50
  342. self.auth_blocking._limit_usage_by_mau = True
  343. self.store.get_monthly_active_count = AsyncMock(return_value=100)
  344. # Support users allowed
  345. self.get_success(
  346. self.auth_blocking.check_auth_blocking(user_type=UserTypes.SUPPORT)
  347. )
  348. self.store.get_monthly_active_count = AsyncMock(return_value=100)
  349. # Bots not allowed
  350. self.get_failure(
  351. self.auth_blocking.check_auth_blocking(user_type=UserTypes.BOT),
  352. ResourceLimitError,
  353. )
  354. self.store.get_monthly_active_count = AsyncMock(return_value=100)
  355. # Real users not allowed
  356. self.get_failure(self.auth_blocking.check_auth_blocking(), ResourceLimitError)
  357. def test_blocking_mau__appservice_requester_allowed_when_not_tracking_ips(
  358. self,
  359. ) -> None:
  360. self.auth_blocking._max_mau_value = 50
  361. self.auth_blocking._limit_usage_by_mau = True
  362. self.auth_blocking._track_appservice_user_ips = False
  363. self.store.get_monthly_active_count = AsyncMock(return_value=100)
  364. self.store.user_last_seen_monthly_active = AsyncMock(return_value=None)
  365. self.store.is_trial_user = AsyncMock(return_value=False)
  366. appservice = ApplicationService(
  367. "abcd",
  368. id="1234",
  369. namespaces={
  370. "users": [{"regex": "@_appservice.*:sender", "exclusive": True}]
  371. },
  372. sender="@appservice:sender",
  373. )
  374. requester = Requester(
  375. user=UserID.from_string("@appservice:server"),
  376. access_token_id=None,
  377. device_id="FOOBAR",
  378. is_guest=False,
  379. scope=set(),
  380. shadow_banned=False,
  381. app_service=appservice,
  382. authenticated_entity="@appservice:server",
  383. )
  384. self.get_success(self.auth_blocking.check_auth_blocking(requester=requester))
  385. def test_blocking_mau__appservice_requester_disallowed_when_tracking_ips(
  386. self,
  387. ) -> None:
  388. self.auth_blocking._max_mau_value = 50
  389. self.auth_blocking._limit_usage_by_mau = True
  390. self.auth_blocking._track_appservice_user_ips = True
  391. self.store.get_monthly_active_count = AsyncMock(return_value=100)
  392. self.store.user_last_seen_monthly_active = AsyncMock(return_value=None)
  393. self.store.is_trial_user = AsyncMock(return_value=False)
  394. appservice = ApplicationService(
  395. "abcd",
  396. id="1234",
  397. namespaces={
  398. "users": [{"regex": "@_appservice.*:sender", "exclusive": True}]
  399. },
  400. sender="@appservice:sender",
  401. )
  402. requester = Requester(
  403. user=UserID.from_string("@appservice:server"),
  404. access_token_id=None,
  405. device_id="FOOBAR",
  406. is_guest=False,
  407. scope=set(),
  408. shadow_banned=False,
  409. app_service=appservice,
  410. authenticated_entity="@appservice:server",
  411. )
  412. self.get_failure(
  413. self.auth_blocking.check_auth_blocking(requester=requester),
  414. ResourceLimitError,
  415. )
  416. def test_reserved_threepid(self) -> None:
  417. self.auth_blocking._limit_usage_by_mau = True
  418. self.auth_blocking._max_mau_value = 1
  419. self.store.get_monthly_active_count = AsyncMock(return_value=2)
  420. threepid = {"medium": "email", "address": "reserved@server.com"}
  421. unknown_threepid = {"medium": "email", "address": "unreserved@server.com"}
  422. self.auth_blocking._mau_limits_reserved_threepids = [threepid]
  423. self.get_failure(self.auth_blocking.check_auth_blocking(), ResourceLimitError)
  424. self.get_failure(
  425. self.auth_blocking.check_auth_blocking(threepid=unknown_threepid),
  426. ResourceLimitError,
  427. )
  428. self.get_success(self.auth_blocking.check_auth_blocking(threepid=threepid))
  429. def test_hs_disabled(self) -> None:
  430. self.auth_blocking._hs_disabled = True
  431. self.auth_blocking._hs_disabled_message = "Reason for being disabled"
  432. e = self.get_failure(
  433. self.auth_blocking.check_auth_blocking(), ResourceLimitError
  434. )
  435. self.assertEqual(e.value.admin_contact, self.hs.config.server.admin_contact)
  436. self.assertEqual(e.value.errcode, Codes.RESOURCE_LIMIT_EXCEEDED)
  437. self.assertEqual(e.value.code, 403)
  438. def test_hs_disabled_no_server_notices_user(self) -> None:
  439. """Check that 'hs_disabled_message' works correctly when there is no
  440. server_notices user.
  441. """
  442. # this should be the default, but we had a bug where the test was doing the wrong
  443. # thing, so let's make it explicit
  444. self.auth_blocking._server_notices_mxid = None
  445. self.auth_blocking._hs_disabled = True
  446. self.auth_blocking._hs_disabled_message = "Reason for being disabled"
  447. e = self.get_failure(
  448. self.auth_blocking.check_auth_blocking(), ResourceLimitError
  449. )
  450. self.assertEqual(e.value.admin_contact, self.hs.config.server.admin_contact)
  451. self.assertEqual(e.value.errcode, Codes.RESOURCE_LIMIT_EXCEEDED)
  452. self.assertEqual(e.value.code, 403)
  453. def test_server_notices_mxid_special_cased(self) -> None:
  454. self.auth_blocking._hs_disabled = True
  455. user = "@user:server"
  456. self.auth_blocking._server_notices_mxid = user
  457. self.auth_blocking._hs_disabled_message = "Reason for being disabled"
  458. self.get_success(self.auth_blocking.check_auth_blocking(user))