test_account.py 49 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155115611571158115911601161116211631164116511661167116811691170117111721173117411751176117711781179118011811182118311841185118611871188118911901191119211931194119511961197119811991200120112021203120412051206120712081209121012111212121312141215121612171218121912201221122212231224122512261227122812291230123112321233123412351236123712381239124012411242124312441245124612471248124912501251125212531254125512561257125812591260126112621263126412651266126712681269127012711272127312741275127612771278127912801281128212831284128512861287128812891290129112921293129412951296129712981299130013011302130313041305130613071308130913101311131213131314131513161317131813191320132113221323132413251326132713281329133013311332133313341335133613371338133913401341134213431344134513461347134813491350135113521353135413551356135713581359136013611362136313641365136613671368136913701371137213731374137513761377137813791380138113821383138413851386138713881389139013911392139313941395139613971398139914001401140214031404140514061407140814091410141114121413141414151416141714181419142014211422142314241425142614271428142914301431143214331434143514361437143814391440144114421443144414451446144714481449145014511452145314541455145614571458145914601461146214631464146514661467146814691470147114721473147414751476
  1. # Copyright 2022 The Matrix.org Foundation C.I.C.
  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. import os
  15. import re
  16. from email.parser import Parser
  17. from http import HTTPStatus
  18. from typing import Any, Dict, List, Optional, Union
  19. from unittest.mock import Mock
  20. import pkg_resources
  21. from twisted.internet.interfaces import IReactorTCP
  22. from twisted.test.proto_helpers import MemoryReactor
  23. import synapse.rest.admin
  24. from synapse.api.constants import LoginType, Membership
  25. from synapse.api.errors import Codes, HttpResponseException
  26. from synapse.appservice import ApplicationService
  27. from synapse.rest import admin
  28. from synapse.rest.client import account, login, register, room
  29. from synapse.rest.synapse.client.password_reset import PasswordResetSubmitTokenResource
  30. from synapse.server import HomeServer
  31. from synapse.storage._base import db_to_json
  32. from synapse.types import JsonDict, UserID
  33. from synapse.util import Clock
  34. from tests import unittest
  35. from tests.server import FakeSite, make_request
  36. from tests.unittest import override_config
  37. class PasswordResetTestCase(unittest.HomeserverTestCase):
  38. servlets = [
  39. account.register_servlets,
  40. synapse.rest.admin.register_servlets_for_client_rest_resource,
  41. register.register_servlets,
  42. login.register_servlets,
  43. ]
  44. def make_homeserver(self, reactor: MemoryReactor, clock: Clock) -> HomeServer:
  45. config = self.default_config()
  46. # Email config.
  47. config["email"] = {
  48. "enable_notifs": False,
  49. "template_dir": os.path.abspath(
  50. pkg_resources.resource_filename("synapse", "res/templates")
  51. ),
  52. "smtp_host": "127.0.0.1",
  53. "smtp_port": 20,
  54. "require_transport_security": False,
  55. "smtp_user": None,
  56. "smtp_pass": None,
  57. "notif_from": "test@example.com",
  58. }
  59. config["public_baseurl"] = "https://example.com"
  60. hs = self.setup_test_homeserver(config=config)
  61. async def sendmail(
  62. reactor: IReactorTCP,
  63. smtphost: str,
  64. smtpport: int,
  65. from_addr: str,
  66. to_addr: str,
  67. msg_bytes: bytes,
  68. *args: Any,
  69. **kwargs: Any,
  70. ) -> None:
  71. self.email_attempts.append(msg_bytes)
  72. self.email_attempts: List[bytes] = []
  73. hs.get_send_email_handler()._sendmail = sendmail
  74. return hs
  75. def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None:
  76. self.store = hs.get_datastores().main
  77. self.submit_token_resource = PasswordResetSubmitTokenResource(hs)
  78. def attempt_wrong_password_login(self, username: str, password: str) -> None:
  79. """Attempts to login as the user with the given password, asserting
  80. that the attempt *fails*.
  81. """
  82. body = {"type": "m.login.password", "user": username, "password": password}
  83. channel = self.make_request("POST", "/_matrix/client/r0/login", body)
  84. self.assertEqual(channel.code, HTTPStatus.FORBIDDEN, channel.result)
  85. def test_basic_password_reset(self) -> None:
  86. """Test basic password reset flow"""
  87. old_password = "monkey"
  88. new_password = "kangeroo"
  89. user_id = self.register_user("kermit", old_password)
  90. self.login("kermit", old_password)
  91. email = "test@example.com"
  92. # Add a threepid
  93. self.get_success(
  94. self.store.user_add_threepid(
  95. user_id=user_id,
  96. medium="email",
  97. address=email,
  98. validated_at=0,
  99. added_at=0,
  100. )
  101. )
  102. client_secret = "foobar"
  103. session_id = self._request_token(email, client_secret)
  104. self.assertEqual(len(self.email_attempts), 1)
  105. link = self._get_link_from_email()
  106. self._validate_token(link)
  107. self._reset_password(new_password, session_id, client_secret)
  108. # Assert we can log in with the new password
  109. self.login("kermit", new_password)
  110. # Assert we can't log in with the old password
  111. self.attempt_wrong_password_login("kermit", old_password)
  112. # Check that the UI Auth information doesn't store the password in the database.
  113. #
  114. # Note that we don't have the UI Auth session ID, so just pull out the single
  115. # row.
  116. result = self.get_success(
  117. self.store.db_pool.simple_select_one_onecol(
  118. "ui_auth_sessions", keyvalues={}, retcol="clientdict"
  119. )
  120. )
  121. client_dict = db_to_json(result)
  122. self.assertNotIn("new_password", client_dict)
  123. @override_config({"rc_3pid_validation": {"burst_count": 3}})
  124. def test_ratelimit_by_email(self) -> None:
  125. """Test that we ratelimit /requestToken for the same email."""
  126. old_password = "monkey"
  127. new_password = "kangeroo"
  128. user_id = self.register_user("kermit", old_password)
  129. self.login("kermit", old_password)
  130. email = "test1@example.com"
  131. # Add a threepid
  132. self.get_success(
  133. self.store.user_add_threepid(
  134. user_id=user_id,
  135. medium="email",
  136. address=email,
  137. validated_at=0,
  138. added_at=0,
  139. )
  140. )
  141. def reset(ip: str) -> None:
  142. client_secret = "foobar"
  143. session_id = self._request_token(email, client_secret, ip)
  144. self.assertEqual(len(self.email_attempts), 1)
  145. link = self._get_link_from_email()
  146. self._validate_token(link)
  147. self._reset_password(new_password, session_id, client_secret)
  148. self.email_attempts.clear()
  149. # We expect to be able to make three requests before getting rate
  150. # limited.
  151. #
  152. # We change IPs to ensure that we're not being ratelimited due to the
  153. # same IP
  154. reset("127.0.0.1")
  155. reset("127.0.0.2")
  156. reset("127.0.0.3")
  157. with self.assertRaises(HttpResponseException) as cm:
  158. reset("127.0.0.4")
  159. self.assertEqual(cm.exception.code, 429)
  160. def test_basic_password_reset_canonicalise_email(self) -> None:
  161. """Test basic password reset flow
  162. Request password reset with different spelling
  163. """
  164. old_password = "monkey"
  165. new_password = "kangeroo"
  166. user_id = self.register_user("kermit", old_password)
  167. self.login("kermit", old_password)
  168. email_profile = "test@example.com"
  169. email_passwort_reset = "TEST@EXAMPLE.COM"
  170. # Add a threepid
  171. self.get_success(
  172. self.store.user_add_threepid(
  173. user_id=user_id,
  174. medium="email",
  175. address=email_profile,
  176. validated_at=0,
  177. added_at=0,
  178. )
  179. )
  180. client_secret = "foobar"
  181. session_id = self._request_token(email_passwort_reset, client_secret)
  182. self.assertEqual(len(self.email_attempts), 1)
  183. link = self._get_link_from_email()
  184. self._validate_token(link)
  185. self._reset_password(new_password, session_id, client_secret)
  186. # Assert we can log in with the new password
  187. self.login("kermit", new_password)
  188. # Assert we can't log in with the old password
  189. self.attempt_wrong_password_login("kermit", old_password)
  190. def test_cant_reset_password_without_clicking_link(self) -> None:
  191. """Test that we do actually need to click the link in the email"""
  192. old_password = "monkey"
  193. new_password = "kangeroo"
  194. user_id = self.register_user("kermit", old_password)
  195. self.login("kermit", old_password)
  196. email = "test@example.com"
  197. # Add a threepid
  198. self.get_success(
  199. self.store.user_add_threepid(
  200. user_id=user_id,
  201. medium="email",
  202. address=email,
  203. validated_at=0,
  204. added_at=0,
  205. )
  206. )
  207. client_secret = "foobar"
  208. session_id = self._request_token(email, client_secret)
  209. self.assertEqual(len(self.email_attempts), 1)
  210. # Attempt to reset password without clicking the link
  211. self._reset_password(new_password, session_id, client_secret, expected_code=401)
  212. # Assert we can log in with the old password
  213. self.login("kermit", old_password)
  214. # Assert we can't log in with the new password
  215. self.attempt_wrong_password_login("kermit", new_password)
  216. def test_no_valid_token(self) -> None:
  217. """Test that we do actually need to request a token and can't just
  218. make a session up.
  219. """
  220. old_password = "monkey"
  221. new_password = "kangeroo"
  222. user_id = self.register_user("kermit", old_password)
  223. self.login("kermit", old_password)
  224. email = "test@example.com"
  225. # Add a threepid
  226. self.get_success(
  227. self.store.user_add_threepid(
  228. user_id=user_id,
  229. medium="email",
  230. address=email,
  231. validated_at=0,
  232. added_at=0,
  233. )
  234. )
  235. client_secret = "foobar"
  236. session_id = "weasle"
  237. # Attempt to reset password without even requesting an email
  238. self._reset_password(new_password, session_id, client_secret, expected_code=401)
  239. # Assert we can log in with the old password
  240. self.login("kermit", old_password)
  241. # Assert we can't log in with the new password
  242. self.attempt_wrong_password_login("kermit", new_password)
  243. @unittest.override_config({"request_token_inhibit_3pid_errors": True})
  244. def test_password_reset_bad_email_inhibit_error(self) -> None:
  245. """Test that triggering a password reset with an email address that isn't bound
  246. to an account doesn't leak the lack of binding for that address if configured
  247. that way.
  248. """
  249. self.register_user("kermit", "monkey")
  250. self.login("kermit", "monkey")
  251. email = "test@example.com"
  252. client_secret = "foobar"
  253. session_id = self._request_token(email, client_secret)
  254. self.assertIsNotNone(session_id)
  255. def _request_token(
  256. self,
  257. email: str,
  258. client_secret: str,
  259. ip: str = "127.0.0.1",
  260. ) -> str:
  261. channel = self.make_request(
  262. "POST",
  263. b"account/password/email/requestToken",
  264. {"client_secret": client_secret, "email": email, "send_attempt": 1},
  265. client_ip=ip,
  266. )
  267. if channel.code != 200:
  268. raise HttpResponseException(
  269. channel.code,
  270. channel.result["reason"],
  271. channel.result["body"],
  272. )
  273. return channel.json_body["sid"]
  274. def _validate_token(self, link: str) -> None:
  275. # Remove the host
  276. path = link.replace("https://example.com", "")
  277. # Load the password reset confirmation page
  278. channel = make_request(
  279. self.reactor,
  280. FakeSite(self.submit_token_resource, self.reactor),
  281. "GET",
  282. path,
  283. shorthand=False,
  284. )
  285. self.assertEqual(HTTPStatus.OK, channel.code, channel.result)
  286. # Now POST to the same endpoint, mimicking the same behaviour as clicking the
  287. # password reset confirm button
  288. # Confirm the password reset
  289. channel = make_request(
  290. self.reactor,
  291. FakeSite(self.submit_token_resource, self.reactor),
  292. "POST",
  293. path,
  294. content=b"",
  295. shorthand=False,
  296. content_is_form=True,
  297. )
  298. self.assertEqual(HTTPStatus.OK, channel.code, channel.result)
  299. def _get_link_from_email(self) -> str:
  300. assert self.email_attempts, "No emails have been sent"
  301. raw_msg = self.email_attempts[-1].decode("UTF-8")
  302. mail = Parser().parsestr(raw_msg)
  303. text = None
  304. for part in mail.walk():
  305. if part.get_content_type() == "text/plain":
  306. text = part.get_payload(decode=True).decode("UTF-8")
  307. break
  308. if not text:
  309. self.fail("Could not find text portion of email to parse")
  310. assert text is not None
  311. match = re.search(r"https://example.com\S+", text)
  312. assert match, "Could not find link in email"
  313. return match.group(0)
  314. def _reset_password(
  315. self,
  316. new_password: str,
  317. session_id: str,
  318. client_secret: str,
  319. expected_code: int = HTTPStatus.OK,
  320. ) -> None:
  321. channel = self.make_request(
  322. "POST",
  323. b"account/password",
  324. {
  325. "new_password": new_password,
  326. "auth": {
  327. "type": LoginType.EMAIL_IDENTITY,
  328. "threepid_creds": {
  329. "client_secret": client_secret,
  330. "sid": session_id,
  331. },
  332. },
  333. },
  334. )
  335. self.assertEqual(expected_code, channel.code, channel.result)
  336. class DeactivateTestCase(unittest.HomeserverTestCase):
  337. servlets = [
  338. synapse.rest.admin.register_servlets_for_client_rest_resource,
  339. login.register_servlets,
  340. account.register_servlets,
  341. room.register_servlets,
  342. ]
  343. def make_homeserver(self, reactor: MemoryReactor, clock: Clock) -> HomeServer:
  344. self.hs = self.setup_test_homeserver()
  345. return self.hs
  346. def test_deactivate_account(self) -> None:
  347. user_id = self.register_user("kermit", "test")
  348. tok = self.login("kermit", "test")
  349. self.deactivate(user_id, tok)
  350. store = self.hs.get_datastores().main
  351. # Check that the user has been marked as deactivated.
  352. self.assertTrue(self.get_success(store.get_user_deactivated_status(user_id)))
  353. # Check that this access token has been invalidated.
  354. channel = self.make_request("GET", "account/whoami", access_token=tok)
  355. self.assertEqual(channel.code, 401)
  356. def test_pending_invites(self) -> None:
  357. """Tests that deactivating a user rejects every pending invite for them."""
  358. store = self.hs.get_datastores().main
  359. inviter_id = self.register_user("inviter", "test")
  360. inviter_tok = self.login("inviter", "test")
  361. invitee_id = self.register_user("invitee", "test")
  362. invitee_tok = self.login("invitee", "test")
  363. # Make @inviter:test invite @invitee:test in a new room.
  364. room_id = self.helper.create_room_as(inviter_id, tok=inviter_tok)
  365. self.helper.invite(
  366. room=room_id, src=inviter_id, targ=invitee_id, tok=inviter_tok
  367. )
  368. # Make sure the invite is here.
  369. pending_invites = self.get_success(
  370. store.get_invited_rooms_for_local_user(invitee_id)
  371. )
  372. self.assertEqual(len(pending_invites), 1, pending_invites)
  373. self.assertEqual(pending_invites[0].room_id, room_id, pending_invites)
  374. # Deactivate @invitee:test.
  375. self.deactivate(invitee_id, invitee_tok)
  376. # Check that the invite isn't there anymore.
  377. pending_invites = self.get_success(
  378. store.get_invited_rooms_for_local_user(invitee_id)
  379. )
  380. self.assertEqual(len(pending_invites), 0, pending_invites)
  381. # Check that the membership of @invitee:test in the room is now "leave".
  382. memberships = self.get_success(
  383. store.get_rooms_for_local_user_where_membership_is(
  384. invitee_id, [Membership.LEAVE]
  385. )
  386. )
  387. self.assertEqual(len(memberships), 1, memberships)
  388. self.assertEqual(memberships[0].room_id, room_id, memberships)
  389. def test_deactivate_account_deletes_server_side_backup_keys(self) -> None:
  390. key_handler = self.hs.get_e2e_room_keys_handler()
  391. room_keys = {
  392. "rooms": {
  393. "!abc:matrix.org": {
  394. "sessions": {
  395. "c0ff33": {
  396. "first_message_index": 1,
  397. "forwarded_count": 1,
  398. "is_verified": False,
  399. "session_data": "SSBBTSBBIEZJU0gK",
  400. }
  401. }
  402. }
  403. }
  404. }
  405. user_id = self.register_user("missPiggy", "test")
  406. tok = self.login("missPiggy", "test")
  407. # add some backup keys/versions
  408. version = self.get_success(
  409. key_handler.create_version(
  410. user_id,
  411. {
  412. "algorithm": "m.megolm_backup.v1",
  413. "auth_data": "first_version_auth_data",
  414. },
  415. )
  416. )
  417. self.get_success(key_handler.upload_room_keys(user_id, version, room_keys))
  418. version2 = self.get_success(
  419. key_handler.create_version(
  420. user_id,
  421. {
  422. "algorithm": "m.megolm_backup.v1",
  423. "auth_data": "second_version_auth_data",
  424. },
  425. )
  426. )
  427. self.get_success(key_handler.upload_room_keys(user_id, version2, room_keys))
  428. self.deactivate(user_id, tok)
  429. store = self.hs.get_datastores().main
  430. # Check that the user has been marked as deactivated.
  431. self.assertTrue(self.get_success(store.get_user_deactivated_status(user_id)))
  432. # Check that there are no entries in 'e2e_room_keys` and `e2e_room_keys_versions`
  433. res = self.get_success(
  434. self.hs.get_datastores().main.db_pool.simple_select_list(
  435. "e2e_room_keys", {"user_id": user_id}, "*", "simple_select"
  436. )
  437. )
  438. self.assertEqual(len(res), 0)
  439. res2 = self.get_success(
  440. self.hs.get_datastores().main.db_pool.simple_select_list(
  441. "e2e_room_keys_versions", {"user_id": user_id}, "*", "simple_select"
  442. )
  443. )
  444. self.assertEqual(len(res2), 0)
  445. def test_background_update_deletes_deactivated_users_server_side_backup_keys(
  446. self,
  447. ) -> None:
  448. key_handler = self.hs.get_e2e_room_keys_handler()
  449. room_keys = {
  450. "rooms": {
  451. "!abc:matrix.org": {
  452. "sessions": {
  453. "c0ff33": {
  454. "first_message_index": 1,
  455. "forwarded_count": 1,
  456. "is_verified": False,
  457. "session_data": "SSBBTSBBIEZJU0gK",
  458. }
  459. }
  460. }
  461. }
  462. }
  463. self.store = self.hs.get_datastores().main
  464. # create a bunch of users and add keys for them
  465. users = []
  466. for i in range(20):
  467. user_id = self.register_user("missPiggy" + str(i), "test")
  468. users.append((user_id,))
  469. # add some backup keys/versions
  470. version = self.get_success(
  471. key_handler.create_version(
  472. user_id,
  473. {
  474. "algorithm": "m.megolm_backup.v1",
  475. "auth_data": str(i) + "_version_auth_data",
  476. },
  477. )
  478. )
  479. self.get_success(key_handler.upload_room_keys(user_id, version, room_keys))
  480. version2 = self.get_success(
  481. key_handler.create_version(
  482. user_id,
  483. {
  484. "algorithm": "m.megolm_backup.v1",
  485. "auth_data": str(i) + "_version_auth_data",
  486. },
  487. )
  488. )
  489. self.get_success(key_handler.upload_room_keys(user_id, version2, room_keys))
  490. # deactivate most of the users by editing DB
  491. self.get_success(
  492. self.store.db_pool.simple_update_many(
  493. table="users",
  494. key_names=("name",),
  495. key_values=users[0:18],
  496. value_names=("deactivated",),
  497. value_values=[(1,) for i in range(1, 19)],
  498. desc="",
  499. )
  500. )
  501. # run background update
  502. self.get_success(
  503. self.store.db_pool.simple_insert(
  504. "background_updates",
  505. {
  506. "update_name": "delete_e2e_backup_keys_for_deactivated_users",
  507. "progress_json": "{}",
  508. },
  509. )
  510. )
  511. self.store.db_pool.updates._all_done = False
  512. self.wait_for_background_updates()
  513. # check that keys are deleted for the deactivated users but not the others
  514. res = self.get_success(
  515. self.hs.get_datastores().main.db_pool.simple_select_list(
  516. "e2e_room_keys", None, ("user_id",), "simple_select"
  517. )
  518. )
  519. self.assertEqual(len(res), 4)
  520. res2 = self.get_success(
  521. self.hs.get_datastores().main.db_pool.simple_select_list(
  522. "e2e_room_keys_versions", None, ("user_id",), "simple_select"
  523. )
  524. )
  525. self.assertEqual(len(res2), 4)
  526. def deactivate(self, user_id: str, tok: str) -> None:
  527. request_data = {
  528. "auth": {
  529. "type": "m.login.password",
  530. "user": user_id,
  531. "password": "test",
  532. },
  533. "erase": False,
  534. }
  535. channel = self.make_request(
  536. "POST", "account/deactivate", request_data, access_token=tok
  537. )
  538. self.assertEqual(channel.code, 200, channel.json_body)
  539. class WhoamiTestCase(unittest.HomeserverTestCase):
  540. servlets = [
  541. synapse.rest.admin.register_servlets_for_client_rest_resource,
  542. login.register_servlets,
  543. account.register_servlets,
  544. register.register_servlets,
  545. ]
  546. def default_config(self) -> Dict[str, Any]:
  547. config = super().default_config()
  548. config["allow_guest_access"] = True
  549. return config
  550. def test_GET_whoami(self) -> None:
  551. device_id = "wouldgohere"
  552. user_id = self.register_user("kermit", "test")
  553. tok = self.login("kermit", "test", device_id=device_id)
  554. whoami = self._whoami(tok)
  555. self.assertEqual(
  556. whoami,
  557. {
  558. "user_id": user_id,
  559. "device_id": device_id,
  560. "is_guest": False,
  561. },
  562. )
  563. def test_GET_whoami_guests(self) -> None:
  564. channel = self.make_request(
  565. b"POST", b"/_matrix/client/r0/register?kind=guest", b"{}"
  566. )
  567. tok = channel.json_body["access_token"]
  568. user_id = channel.json_body["user_id"]
  569. device_id = channel.json_body["device_id"]
  570. whoami = self._whoami(tok)
  571. self.assertEqual(
  572. whoami,
  573. {
  574. "user_id": user_id,
  575. "device_id": device_id,
  576. "is_guest": True,
  577. },
  578. )
  579. def test_GET_whoami_appservices(self) -> None:
  580. user_id = "@as:test"
  581. as_token = "i_am_an_app_service"
  582. appservice = ApplicationService(
  583. as_token,
  584. id="1234",
  585. namespaces={"users": [{"regex": user_id, "exclusive": True}]},
  586. sender=user_id,
  587. )
  588. self.hs.get_datastores().main.services_cache.append(appservice)
  589. whoami = self._whoami(as_token)
  590. self.assertEqual(
  591. whoami,
  592. {
  593. "user_id": user_id,
  594. "is_guest": False,
  595. },
  596. )
  597. self.assertFalse(hasattr(whoami, "device_id"))
  598. def _whoami(self, tok: str) -> JsonDict:
  599. channel = self.make_request("GET", "account/whoami", {}, access_token=tok)
  600. self.assertEqual(channel.code, 200)
  601. return channel.json_body
  602. class ThreepidEmailRestTestCase(unittest.HomeserverTestCase):
  603. servlets = [
  604. account.register_servlets,
  605. login.register_servlets,
  606. synapse.rest.admin.register_servlets_for_client_rest_resource,
  607. ]
  608. def make_homeserver(self, reactor: MemoryReactor, clock: Clock) -> HomeServer:
  609. config = self.default_config()
  610. # Email config.
  611. config["email"] = {
  612. "enable_notifs": False,
  613. "template_dir": os.path.abspath(
  614. pkg_resources.resource_filename("synapse", "res/templates")
  615. ),
  616. "smtp_host": "127.0.0.1",
  617. "smtp_port": 20,
  618. "require_transport_security": False,
  619. "smtp_user": None,
  620. "smtp_pass": None,
  621. "notif_from": "test@example.com",
  622. }
  623. config["public_baseurl"] = "https://example.com"
  624. self.hs = self.setup_test_homeserver(config=config)
  625. async def sendmail(
  626. reactor: IReactorTCP,
  627. smtphost: str,
  628. smtpport: int,
  629. from_addr: str,
  630. to_addr: str,
  631. msg_bytes: bytes,
  632. *args: Any,
  633. **kwargs: Any,
  634. ) -> None:
  635. self.email_attempts.append(msg_bytes)
  636. self.email_attempts: List[bytes] = []
  637. self.hs.get_send_email_handler()._sendmail = sendmail
  638. return self.hs
  639. def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None:
  640. self.store = hs.get_datastores().main
  641. self.user_id = self.register_user("kermit", "test")
  642. self.user_id_tok = self.login("kermit", "test")
  643. self.email = "test@example.com"
  644. self.url_3pid = b"account/3pid"
  645. def test_add_valid_email(self) -> None:
  646. self._add_email(self.email, self.email)
  647. def test_add_valid_email_second_time(self) -> None:
  648. self._add_email(self.email, self.email)
  649. self._request_token_invalid_email(
  650. self.email,
  651. expected_errcode=Codes.THREEPID_IN_USE,
  652. expected_error="Email is already in use",
  653. )
  654. def test_add_valid_email_second_time_canonicalise(self) -> None:
  655. self._add_email(self.email, self.email)
  656. self._request_token_invalid_email(
  657. "TEST@EXAMPLE.COM",
  658. expected_errcode=Codes.THREEPID_IN_USE,
  659. expected_error="Email is already in use",
  660. )
  661. def test_add_email_no_at(self) -> None:
  662. self._request_token_invalid_email(
  663. "address-without-at.bar",
  664. expected_errcode=Codes.BAD_JSON,
  665. expected_error="Unable to parse email address",
  666. )
  667. def test_add_email_two_at(self) -> None:
  668. self._request_token_invalid_email(
  669. "foo@foo@test.bar",
  670. expected_errcode=Codes.BAD_JSON,
  671. expected_error="Unable to parse email address",
  672. )
  673. def test_add_email_bad_format(self) -> None:
  674. self._request_token_invalid_email(
  675. "user@bad.example.net@good.example.com",
  676. expected_errcode=Codes.BAD_JSON,
  677. expected_error="Unable to parse email address",
  678. )
  679. def test_add_email_domain_to_lower(self) -> None:
  680. self._add_email("foo@TEST.BAR", "foo@test.bar")
  681. def test_add_email_domain_with_umlaut(self) -> None:
  682. self._add_email("foo@Öumlaut.com", "foo@öumlaut.com")
  683. def test_add_email_address_casefold(self) -> None:
  684. self._add_email("Strauß@Example.com", "strauss@example.com")
  685. def test_address_trim(self) -> None:
  686. self._add_email(" foo@test.bar ", "foo@test.bar")
  687. @override_config({"rc_3pid_validation": {"burst_count": 3}})
  688. def test_ratelimit_by_ip(self) -> None:
  689. """Tests that adding emails is ratelimited by IP"""
  690. # We expect to be able to set three emails before getting ratelimited.
  691. self._add_email("foo1@test.bar", "foo1@test.bar")
  692. self._add_email("foo2@test.bar", "foo2@test.bar")
  693. self._add_email("foo3@test.bar", "foo3@test.bar")
  694. with self.assertRaises(HttpResponseException) as cm:
  695. self._add_email("foo4@test.bar", "foo4@test.bar")
  696. self.assertEqual(cm.exception.code, 429)
  697. def test_add_email_if_disabled(self) -> None:
  698. """Test adding email to profile when doing so is disallowed"""
  699. self.hs.config.registration.enable_3pid_changes = False
  700. client_secret = "foobar"
  701. channel = self.make_request(
  702. "POST",
  703. b"/_matrix/client/unstable/account/3pid/email/requestToken",
  704. {
  705. "client_secret": client_secret,
  706. "email": "test@example.com",
  707. "send_attempt": 1,
  708. },
  709. )
  710. self.assertEqual(
  711. HTTPStatus.BAD_REQUEST, channel.code, msg=channel.result["body"]
  712. )
  713. self.assertEqual(Codes.FORBIDDEN, channel.json_body["errcode"])
  714. def test_delete_email(self) -> None:
  715. """Test deleting an email from profile"""
  716. # Add a threepid
  717. self.get_success(
  718. self.store.user_add_threepid(
  719. user_id=self.user_id,
  720. medium="email",
  721. address=self.email,
  722. validated_at=0,
  723. added_at=0,
  724. )
  725. )
  726. channel = self.make_request(
  727. "POST",
  728. b"account/3pid/delete",
  729. {"medium": "email", "address": self.email},
  730. access_token=self.user_id_tok,
  731. )
  732. self.assertEqual(HTTPStatus.OK, channel.code, msg=channel.result["body"])
  733. # Get user
  734. channel = self.make_request(
  735. "GET",
  736. self.url_3pid,
  737. access_token=self.user_id_tok,
  738. )
  739. self.assertEqual(HTTPStatus.OK, channel.code, msg=channel.result["body"])
  740. self.assertFalse(channel.json_body["threepids"])
  741. def test_delete_email_if_disabled(self) -> None:
  742. """Test deleting an email from profile when disallowed"""
  743. self.hs.config.registration.enable_3pid_changes = False
  744. # Add a threepid
  745. self.get_success(
  746. self.store.user_add_threepid(
  747. user_id=self.user_id,
  748. medium="email",
  749. address=self.email,
  750. validated_at=0,
  751. added_at=0,
  752. )
  753. )
  754. channel = self.make_request(
  755. "POST",
  756. b"account/3pid/delete",
  757. {"medium": "email", "address": self.email},
  758. access_token=self.user_id_tok,
  759. )
  760. self.assertEqual(
  761. HTTPStatus.BAD_REQUEST, channel.code, msg=channel.result["body"]
  762. )
  763. self.assertEqual(Codes.FORBIDDEN, channel.json_body["errcode"])
  764. # Get user
  765. channel = self.make_request(
  766. "GET",
  767. self.url_3pid,
  768. access_token=self.user_id_tok,
  769. )
  770. self.assertEqual(HTTPStatus.OK, channel.code, msg=channel.result["body"])
  771. self.assertEqual("email", channel.json_body["threepids"][0]["medium"])
  772. self.assertEqual(self.email, channel.json_body["threepids"][0]["address"])
  773. def test_cant_add_email_without_clicking_link(self) -> None:
  774. """Test that we do actually need to click the link in the email"""
  775. client_secret = "foobar"
  776. session_id = self._request_token(self.email, client_secret)
  777. self.assertEqual(len(self.email_attempts), 1)
  778. # Attempt to add email without clicking the link
  779. channel = self.make_request(
  780. "POST",
  781. b"/_matrix/client/unstable/account/3pid/add",
  782. {
  783. "client_secret": client_secret,
  784. "sid": session_id,
  785. "auth": {
  786. "type": "m.login.password",
  787. "user": self.user_id,
  788. "password": "test",
  789. },
  790. },
  791. access_token=self.user_id_tok,
  792. )
  793. self.assertEqual(
  794. HTTPStatus.BAD_REQUEST, channel.code, msg=channel.result["body"]
  795. )
  796. self.assertEqual(Codes.THREEPID_AUTH_FAILED, channel.json_body["errcode"])
  797. # Get user
  798. channel = self.make_request(
  799. "GET",
  800. self.url_3pid,
  801. access_token=self.user_id_tok,
  802. )
  803. self.assertEqual(HTTPStatus.OK, channel.code, msg=channel.result["body"])
  804. self.assertFalse(channel.json_body["threepids"])
  805. def test_no_valid_token(self) -> None:
  806. """Test that we do actually need to request a token and can't just
  807. make a session up.
  808. """
  809. client_secret = "foobar"
  810. session_id = "weasle"
  811. # Attempt to add email without even requesting an email
  812. channel = self.make_request(
  813. "POST",
  814. b"/_matrix/client/unstable/account/3pid/add",
  815. {
  816. "client_secret": client_secret,
  817. "sid": session_id,
  818. "auth": {
  819. "type": "m.login.password",
  820. "user": self.user_id,
  821. "password": "test",
  822. },
  823. },
  824. access_token=self.user_id_tok,
  825. )
  826. self.assertEqual(
  827. HTTPStatus.BAD_REQUEST, channel.code, msg=channel.result["body"]
  828. )
  829. self.assertEqual(Codes.THREEPID_AUTH_FAILED, channel.json_body["errcode"])
  830. # Get user
  831. channel = self.make_request(
  832. "GET",
  833. self.url_3pid,
  834. access_token=self.user_id_tok,
  835. )
  836. self.assertEqual(HTTPStatus.OK, channel.code, msg=channel.result["body"])
  837. self.assertFalse(channel.json_body["threepids"])
  838. @override_config({"next_link_domain_whitelist": None})
  839. def test_next_link(self) -> None:
  840. """Tests a valid next_link parameter value with no whitelist (good case)"""
  841. self._request_token(
  842. "something@example.com",
  843. "some_secret",
  844. next_link="https://example.com/a/good/site",
  845. expect_code=HTTPStatus.OK,
  846. )
  847. @override_config({"next_link_domain_whitelist": None})
  848. def test_next_link_exotic_protocol(self) -> None:
  849. """Tests using a esoteric protocol as a next_link parameter value.
  850. Someone may be hosting a client on IPFS etc.
  851. """
  852. self._request_token(
  853. "something@example.com",
  854. "some_secret",
  855. next_link="some-protocol://abcdefghijklmopqrstuvwxyz",
  856. expect_code=HTTPStatus.OK,
  857. )
  858. @override_config({"next_link_domain_whitelist": None})
  859. def test_next_link_file_uri(self) -> None:
  860. """Tests next_link parameters cannot be file URI"""
  861. # Attempt to use a next_link value that points to the local disk
  862. self._request_token(
  863. "something@example.com",
  864. "some_secret",
  865. next_link="file:///host/path",
  866. expect_code=HTTPStatus.BAD_REQUEST,
  867. )
  868. @override_config({"next_link_domain_whitelist": ["example.com", "example.org"]})
  869. def test_next_link_domain_whitelist(self) -> None:
  870. """Tests next_link parameters must fit the whitelist if provided"""
  871. # Ensure not providing a next_link parameter still works
  872. self._request_token(
  873. "something@example.com",
  874. "some_secret",
  875. next_link=None,
  876. expect_code=HTTPStatus.OK,
  877. )
  878. self._request_token(
  879. "something@example.com",
  880. "some_secret",
  881. next_link="https://example.com/some/good/page",
  882. expect_code=HTTPStatus.OK,
  883. )
  884. self._request_token(
  885. "something@example.com",
  886. "some_secret",
  887. next_link="https://example.org/some/also/good/page",
  888. expect_code=HTTPStatus.OK,
  889. )
  890. self._request_token(
  891. "something@example.com",
  892. "some_secret",
  893. next_link="https://bad.example.org/some/bad/page",
  894. expect_code=HTTPStatus.BAD_REQUEST,
  895. )
  896. @override_config({"next_link_domain_whitelist": []})
  897. def test_empty_next_link_domain_whitelist(self) -> None:
  898. """Tests an empty next_lint_domain_whitelist value, meaning next_link is essentially
  899. disallowed
  900. """
  901. self._request_token(
  902. "something@example.com",
  903. "some_secret",
  904. next_link="https://example.com/a/page",
  905. expect_code=HTTPStatus.BAD_REQUEST,
  906. )
  907. def _request_token(
  908. self,
  909. email: str,
  910. client_secret: str,
  911. next_link: Optional[str] = None,
  912. expect_code: int = HTTPStatus.OK,
  913. ) -> Optional[str]:
  914. """Request a validation token to add an email address to a user's account
  915. Args:
  916. email: The email address to validate
  917. client_secret: A secret string
  918. next_link: A link to redirect the user to after validation
  919. expect_code: Expected return code of the call
  920. Returns:
  921. The ID of the new threepid validation session, or None if the response
  922. did not contain a session ID.
  923. """
  924. body = {"client_secret": client_secret, "email": email, "send_attempt": 1}
  925. if next_link:
  926. body["next_link"] = next_link
  927. channel = self.make_request(
  928. "POST",
  929. b"account/3pid/email/requestToken",
  930. body,
  931. )
  932. if channel.code != expect_code:
  933. raise HttpResponseException(
  934. channel.code,
  935. channel.result["reason"],
  936. channel.result["body"],
  937. )
  938. return channel.json_body.get("sid")
  939. def _request_token_invalid_email(
  940. self,
  941. email: str,
  942. expected_errcode: str,
  943. expected_error: str,
  944. client_secret: str = "foobar",
  945. ) -> None:
  946. channel = self.make_request(
  947. "POST",
  948. b"account/3pid/email/requestToken",
  949. {"client_secret": client_secret, "email": email, "send_attempt": 1},
  950. )
  951. self.assertEqual(
  952. HTTPStatus.BAD_REQUEST, channel.code, msg=channel.result["body"]
  953. )
  954. self.assertEqual(expected_errcode, channel.json_body["errcode"])
  955. self.assertIn(expected_error, channel.json_body["error"])
  956. def _validate_token(self, link: str) -> None:
  957. # Remove the host
  958. path = link.replace("https://example.com", "")
  959. channel = self.make_request("GET", path, shorthand=False)
  960. self.assertEqual(HTTPStatus.OK, channel.code, channel.result)
  961. def _get_link_from_email(self) -> str:
  962. assert self.email_attempts, "No emails have been sent"
  963. raw_msg = self.email_attempts[-1].decode("UTF-8")
  964. mail = Parser().parsestr(raw_msg)
  965. text = None
  966. for part in mail.walk():
  967. if part.get_content_type() == "text/plain":
  968. text = part.get_payload(decode=True).decode("UTF-8")
  969. break
  970. if not text:
  971. self.fail("Could not find text portion of email to parse")
  972. assert text is not None
  973. match = re.search(r"https://example.com\S+", text)
  974. assert match, "Could not find link in email"
  975. return match.group(0)
  976. def _add_email(self, request_email: str, expected_email: str) -> None:
  977. """Test adding an email to profile"""
  978. previous_email_attempts = len(self.email_attempts)
  979. client_secret = "foobar"
  980. session_id = self._request_token(request_email, client_secret)
  981. self.assertEqual(len(self.email_attempts) - previous_email_attempts, 1)
  982. link = self._get_link_from_email()
  983. self._validate_token(link)
  984. channel = self.make_request(
  985. "POST",
  986. b"/_matrix/client/unstable/account/3pid/add",
  987. {
  988. "client_secret": client_secret,
  989. "sid": session_id,
  990. "auth": {
  991. "type": "m.login.password",
  992. "user": self.user_id,
  993. "password": "test",
  994. },
  995. },
  996. access_token=self.user_id_tok,
  997. )
  998. self.assertEqual(HTTPStatus.OK, channel.code, msg=channel.result["body"])
  999. # Get user
  1000. channel = self.make_request(
  1001. "GET",
  1002. self.url_3pid,
  1003. access_token=self.user_id_tok,
  1004. )
  1005. self.assertEqual(HTTPStatus.OK, channel.code, msg=channel.result["body"])
  1006. self.assertEqual("email", channel.json_body["threepids"][0]["medium"])
  1007. threepids = {threepid["address"] for threepid in channel.json_body["threepids"]}
  1008. self.assertIn(expected_email, threepids)
  1009. class AccountStatusTestCase(unittest.HomeserverTestCase):
  1010. servlets = [
  1011. account.register_servlets,
  1012. admin.register_servlets,
  1013. login.register_servlets,
  1014. ]
  1015. url = "/_matrix/client/unstable/org.matrix.msc3720/account_status"
  1016. def make_homeserver(self, reactor: MemoryReactor, clock: Clock) -> HomeServer:
  1017. config = self.default_config()
  1018. config["experimental_features"] = {"msc3720_enabled": True}
  1019. return self.setup_test_homeserver(config=config)
  1020. def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None:
  1021. self.requester = self.register_user("requester", "password")
  1022. self.requester_tok = self.login("requester", "password")
  1023. self.server_name = hs.config.server.server_name
  1024. def test_missing_mxid(self) -> None:
  1025. """Tests that not providing any MXID raises an error."""
  1026. self._test_status(
  1027. users=None,
  1028. expected_status_code=HTTPStatus.BAD_REQUEST,
  1029. expected_errcode=Codes.MISSING_PARAM,
  1030. )
  1031. def test_invalid_mxid(self) -> None:
  1032. """Tests that providing an invalid MXID raises an error."""
  1033. self._test_status(
  1034. users=["bad:test"],
  1035. expected_status_code=HTTPStatus.BAD_REQUEST,
  1036. expected_errcode=Codes.INVALID_PARAM,
  1037. )
  1038. def test_local_user_not_exists(self) -> None:
  1039. """Tests that the account status endpoints correctly reports that a user doesn't
  1040. exist.
  1041. """
  1042. user = "@unknown:" + self.hs.config.server.server_name
  1043. self._test_status(
  1044. users=[user],
  1045. expected_statuses={
  1046. user: {
  1047. "exists": False,
  1048. },
  1049. },
  1050. expected_failures=[],
  1051. )
  1052. def test_local_user_exists(self) -> None:
  1053. """Tests that the account status endpoint correctly reports that a user doesn't
  1054. exist.
  1055. """
  1056. user = self.register_user("someuser", "password")
  1057. self._test_status(
  1058. users=[user],
  1059. expected_statuses={
  1060. user: {
  1061. "exists": True,
  1062. "deactivated": False,
  1063. },
  1064. },
  1065. expected_failures=[],
  1066. )
  1067. def test_local_user_deactivated(self) -> None:
  1068. """Tests that the account status endpoint correctly reports a deactivated user."""
  1069. user = self.register_user("someuser", "password")
  1070. self.get_success(
  1071. self.hs.get_datastores().main.set_user_deactivated_status(
  1072. user, deactivated=True
  1073. )
  1074. )
  1075. self._test_status(
  1076. users=[user],
  1077. expected_statuses={
  1078. user: {
  1079. "exists": True,
  1080. "deactivated": True,
  1081. },
  1082. },
  1083. expected_failures=[],
  1084. )
  1085. def test_mixed_local_and_remote_users(self) -> None:
  1086. """Tests that if some users are remote the account status endpoint correctly
  1087. merges the remote responses with the local result.
  1088. """
  1089. # We use 3 users: one doesn't exist but belongs on the local homeserver, one is
  1090. # deactivated and belongs on one remote homeserver, and one belongs to another
  1091. # remote homeserver that didn't return any result (the federation code should
  1092. # mark that user as a failure).
  1093. users = [
  1094. "@unknown:" + self.hs.config.server.server_name,
  1095. "@deactivated:remote",
  1096. "@failed:otherremote",
  1097. "@bad:badremote",
  1098. ]
  1099. async def post_json(
  1100. destination: str,
  1101. path: str,
  1102. data: Optional[JsonDict] = None,
  1103. *a: Any,
  1104. **kwa: Any,
  1105. ) -> Union[JsonDict, list]:
  1106. if destination == "remote":
  1107. return {
  1108. "account_statuses": {
  1109. users[1]: {
  1110. "exists": True,
  1111. "deactivated": True,
  1112. },
  1113. }
  1114. }
  1115. elif destination == "badremote":
  1116. # badremote tries to overwrite the status of a user that doesn't belong
  1117. # to it (i.e. users[1]) with false data, which Synapse is expected to
  1118. # ignore.
  1119. return {
  1120. "account_statuses": {
  1121. users[3]: {
  1122. "exists": False,
  1123. },
  1124. users[1]: {
  1125. "exists": False,
  1126. },
  1127. }
  1128. }
  1129. # if destination == "otherremote"
  1130. else:
  1131. return {}
  1132. # Register a mock that will return the expected result depending on the remote.
  1133. self.hs.get_federation_http_client().post_json = Mock(side_effect=post_json) # type: ignore[method-assign]
  1134. # Check that we've got the correct response from the client-side endpoint.
  1135. self._test_status(
  1136. users=users,
  1137. expected_statuses={
  1138. users[0]: {
  1139. "exists": False,
  1140. },
  1141. users[1]: {
  1142. "exists": True,
  1143. "deactivated": True,
  1144. },
  1145. users[3]: {
  1146. "exists": False,
  1147. },
  1148. },
  1149. expected_failures=[users[2]],
  1150. )
  1151. @unittest.override_config(
  1152. {
  1153. "use_account_validity_in_account_status": True,
  1154. }
  1155. )
  1156. def test_no_account_validity(self) -> None:
  1157. """Tests that if we decide to include account validity in the response but no
  1158. account validity 'is_user_expired' callback is provided, we default to marking all
  1159. users as not expired.
  1160. """
  1161. user = self.register_user("someuser", "password")
  1162. self._test_status(
  1163. users=[user],
  1164. expected_statuses={
  1165. user: {
  1166. "exists": True,
  1167. "deactivated": False,
  1168. "org.matrix.expired": False,
  1169. },
  1170. },
  1171. expected_failures=[],
  1172. )
  1173. @unittest.override_config(
  1174. {
  1175. "use_account_validity_in_account_status": True,
  1176. }
  1177. )
  1178. def test_account_validity_expired(self) -> None:
  1179. """Test that if we decide to include account validity in the response and the user
  1180. is expired, we return the correct info.
  1181. """
  1182. user = self.register_user("someuser", "password")
  1183. async def is_expired(user_id: str) -> bool:
  1184. # We can't blindly say everyone is expired, otherwise the request to get the
  1185. # account status will fail.
  1186. return UserID.from_string(user_id).localpart == "someuser"
  1187. account_validity_callbacks = self.hs.get_module_api_callbacks().account_validity
  1188. account_validity_callbacks.is_user_expired_callbacks.append(is_expired)
  1189. self._test_status(
  1190. users=[user],
  1191. expected_statuses={
  1192. user: {
  1193. "exists": True,
  1194. "deactivated": False,
  1195. "org.matrix.expired": True,
  1196. },
  1197. },
  1198. expected_failures=[],
  1199. )
  1200. def _test_status(
  1201. self,
  1202. users: Optional[List[str]],
  1203. expected_status_code: int = HTTPStatus.OK,
  1204. expected_statuses: Optional[Dict[str, Dict[str, bool]]] = None,
  1205. expected_failures: Optional[List[str]] = None,
  1206. expected_errcode: Optional[str] = None,
  1207. ) -> None:
  1208. """Send a request to the account status endpoint and check that the response
  1209. matches with what's expected.
  1210. Args:
  1211. users: The account(s) to request the status of, if any. If set to None, no
  1212. `user_id` query parameter will be included in the request.
  1213. expected_status_code: The expected HTTP status code.
  1214. expected_statuses: The expected account statuses, if any.
  1215. expected_failures: The expected failures, if any.
  1216. expected_errcode: The expected Matrix error code, if any.
  1217. """
  1218. content = {}
  1219. if users is not None:
  1220. content["user_ids"] = users
  1221. channel = self.make_request(
  1222. method="POST",
  1223. path=self.url,
  1224. content=content,
  1225. access_token=self.requester_tok,
  1226. )
  1227. self.assertEqual(channel.code, expected_status_code)
  1228. if expected_statuses is not None:
  1229. self.assertEqual(channel.json_body["account_statuses"], expected_statuses)
  1230. if expected_failures is not None:
  1231. self.assertEqual(channel.json_body["failures"], expected_failures)
  1232. if expected_errcode is not None:
  1233. self.assertEqual(channel.json_body["errcode"], expected_errcode)