diff --git a/tests/rest/client/test_account.py b/tests/rest/client/test_account.py
index a85ea994de..33611e8a8c 100644
--- a/tests/rest/client/test_account.py
+++ b/tests/rest/client/test_account.py
@@ -36,7 +36,6 @@ from synapse.api.errors import Codes, HttpResponseException
from synapse.appservice import ApplicationService
from synapse.rest import admin
from synapse.rest.client import account, login, register, room
-from synapse.rest.synapse.client.password_reset import PasswordResetSubmitTokenResource
from synapse.server import HomeServer
from synapse.storage._base import db_to_json
from synapse.types import JsonDict, UserID
@@ -47,430 +46,404 @@ from tests.server import FakeSite, make_request
from tests.unittest import override_config
-class PasswordResetTestCase(unittest.HomeserverTestCase):
- servlets = [
- account.register_servlets,
- synapse.rest.admin.register_servlets_for_client_rest_resource,
- register.register_servlets,
- login.register_servlets,
- ]
-
- def make_homeserver(self, reactor: MemoryReactor, clock: Clock) -> HomeServer:
- config = self.default_config()
-
- # Email config.
- config["email"] = {
- "enable_notifs": False,
- "template_dir": os.path.abspath(
- pkg_resources.resource_filename("synapse", "res/templates")
- ),
- "smtp_host": "127.0.0.1",
- "smtp_port": 20,
- "require_transport_security": False,
- "smtp_user": None,
- "smtp_pass": None,
- "notif_from": "test@example.com",
- }
- config["public_baseurl"] = "https://example.com"
-
- hs = self.setup_test_homeserver(config=config)
-
- async def sendmail(
- reactor: IReactorTCP,
- smtphost: str,
- smtpport: int,
- from_addr: str,
- to_addr: str,
- msg_bytes: bytes,
- *args: Any,
- **kwargs: Any,
- ) -> None:
- self.email_attempts.append(msg_bytes)
-
- self.email_attempts: List[bytes] = []
- hs.get_send_email_handler()._sendmail = sendmail
-
- return hs
-
- def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None:
- self.store = hs.get_datastores().main
- self.submit_token_resource = PasswordResetSubmitTokenResource(hs)
-
- def attempt_wrong_password_login(self, username: str, password: str) -> None:
- """Attempts to login as the user with the given password, asserting
- that the attempt *fails*.
- """
- body = {"type": "m.login.password", "user": username, "password": password}
-
- channel = self.make_request("POST", "/_matrix/client/r0/login", body)
- self.assertEqual(channel.code, HTTPStatus.FORBIDDEN, channel.result)
-
- def test_basic_password_reset(self) -> None:
- """Test basic password reset flow"""
- old_password = "monkey"
- new_password = "kangeroo"
-
- user_id = self.register_user("kermit", old_password)
- self.login("kermit", old_password)
-
- email = "test@example.com"
-
- # Add a threepid
- self.get_success(
- self.store.user_add_threepid(
- user_id=user_id,
- medium="email",
- address=email,
- validated_at=0,
- added_at=0,
- )
- )
-
- client_secret = "foobar"
- session_id = self._request_token(email, client_secret)
-
- self.assertEqual(len(self.email_attempts), 1)
- link = self._get_link_from_email()
-
- self._validate_token(link)
-
- self._reset_password(new_password, session_id, client_secret)
-
- # Assert we can log in with the new password
- self.login("kermit", new_password)
-
- # Assert we can't log in with the old password
- self.attempt_wrong_password_login("kermit", old_password)
-
- # Check that the UI Auth information doesn't store the password in the database.
- #
- # Note that we don't have the UI Auth session ID, so just pull out the single
- # row.
- result = self.get_success(
- self.store.db_pool.simple_select_one_onecol(
- "ui_auth_sessions", keyvalues={}, retcol="clientdict"
- )
- )
- client_dict = db_to_json(result)
- self.assertNotIn("new_password", client_dict)
-
- @override_config({"rc_3pid_validation": {"burst_count": 3}})
- def test_ratelimit_by_email(self) -> None:
- """Test that we ratelimit /requestToken for the same email."""
- old_password = "monkey"
- new_password = "kangeroo"
-
- user_id = self.register_user("kermit", old_password)
- self.login("kermit", old_password)
-
- email = "test1@example.com"
-
- # Add a threepid
- self.get_success(
- self.store.user_add_threepid(
- user_id=user_id,
- medium="email",
- address=email,
- validated_at=0,
- added_at=0,
- )
- )
-
- def reset(ip: str) -> None:
- client_secret = "foobar"
- session_id = self._request_token(email, client_secret, ip)
-
- self.assertEqual(len(self.email_attempts), 1)
- link = self._get_link_from_email()
-
- self._validate_token(link)
-
- self._reset_password(new_password, session_id, client_secret)
-
- self.email_attempts.clear()
-
- # We expect to be able to make three requests before getting rate
- # limited.
- #
- # We change IPs to ensure that we're not being ratelimited due to the
- # same IP
- reset("127.0.0.1")
- reset("127.0.0.2")
- reset("127.0.0.3")
-
- with self.assertRaises(HttpResponseException) as cm:
- reset("127.0.0.4")
-
- self.assertEqual(cm.exception.code, 429)
-
- def test_basic_password_reset_canonicalise_email(self) -> None:
- """Test basic password reset flow
- Request password reset with different spelling
- """
- old_password = "monkey"
- new_password = "kangeroo"
-
- user_id = self.register_user("kermit", old_password)
- self.login("kermit", old_password)
-
- email_profile = "test@example.com"
- email_passwort_reset = "TEST@EXAMPLE.COM"
-
- # Add a threepid
- self.get_success(
- self.store.user_add_threepid(
- user_id=user_id,
- medium="email",
- address=email_profile,
- validated_at=0,
- added_at=0,
- )
- )
-
- client_secret = "foobar"
- session_id = self._request_token(email_passwort_reset, client_secret)
-
- self.assertEqual(len(self.email_attempts), 1)
- link = self._get_link_from_email()
-
- self._validate_token(link)
-
- self._reset_password(new_password, session_id, client_secret)
-
- # Assert we can log in with the new password
- self.login("kermit", new_password)
-
- # Assert we can't log in with the old password
- self.attempt_wrong_password_login("kermit", old_password)
-
- def test_cant_reset_password_without_clicking_link(self) -> None:
- """Test that we do actually need to click the link in the email"""
- old_password = "monkey"
- new_password = "kangeroo"
-
- user_id = self.register_user("kermit", old_password)
- self.login("kermit", old_password)
-
- email = "test@example.com"
+# class PasswordResetTestCase(unittest.HomeserverTestCase):
+# servlets = [
+# account.register_servlets,
+# synapse.rest.admin.register_servlets_for_client_rest_resource,
+# register.register_servlets,
+# login.register_servlets,
+# ]
+
+# def make_homeserver(self, reactor: MemoryReactor, clock: Clock) -> HomeServer:
+# config = self.default_config()
+
+# # Email config.
+# config["email"] = {
+# "enable_notifs": False,
+# "template_dir": os.path.abspath(
+# pkg_resources.resource_filename("synapse", "res/templates")
+# ),
+# "smtp_host": "127.0.0.1",
+# "smtp_port": 20,
+# "require_transport_security": False,
+# "smtp_user": None,
+# "smtp_pass": None,
+# "notif_from": "test@example.com",
+# }
+# config["public_baseurl"] = "https://example.com"
+
+# hs = self.setup_test_homeserver(config=config)
- # Add a threepid
- self.get_success(
- self.store.user_add_threepid(
- user_id=user_id,
- medium="email",
- address=email,
- validated_at=0,
- added_at=0,
- )
- )
-
- client_secret = "foobar"
- session_id = self._request_token(email, client_secret)
-
- self.assertEqual(len(self.email_attempts), 1)
-
- # Attempt to reset password without clicking the link
- self._reset_password(new_password, session_id, client_secret, expected_code=401)
-
- # Assert we can log in with the old password
- self.login("kermit", old_password)
-
- # Assert we can't log in with the new password
- self.attempt_wrong_password_login("kermit", new_password)
-
- def test_no_valid_token(self) -> None:
- """Test that we do actually need to request a token and can't just
- make a session up.
- """
- old_password = "monkey"
- new_password = "kangeroo"
-
- user_id = self.register_user("kermit", old_password)
- self.login("kermit", old_password)
-
- email = "test@example.com"
-
- # Add a threepid
- self.get_success(
- self.store.user_add_threepid(
- user_id=user_id,
- medium="email",
- address=email,
- validated_at=0,
- added_at=0,
- )
- )
-
- client_secret = "foobar"
- session_id = "weasle"
+# async def sendmail(
+# reactor: IReactorTCP,
+# smtphost: str,
+# smtpport: int,
+# from_addr: str,
+# to_addr: str,
+# msg_bytes: bytes,
+# *args: Any,
+# **kwargs: Any,
+# ) -> None:
+# self.email_attempts.append(msg_bytes)
- # Attempt to reset password without even requesting an email
- self._reset_password(new_password, session_id, client_secret, expected_code=401)
+# self.email_attempts: List[bytes] = []
+
+# return hs
- # Assert we can log in with the old password
- self.login("kermit", old_password)
+# def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None:
+# self.store = hs.get_datastores().main
+
+# def attempt_wrong_password_login(self, username: str, password: str) -> None:
+# """Attempts to login as the user with the given password, asserting
+# that the attempt *fails*.
+# """
+# body = {"type": "m.login.password", "user": username, "password": password}
+
+# channel = self.make_request("POST", "/_matrix/client/r0/login", body)
+# self.assertEqual(channel.code, HTTPStatus.FORBIDDEN, channel.result)
+
+# def test_basic_password_reset(self) -> None:
+# """Test basic password reset flow"""
+# old_password = "monkey"
+# new_password = "kangeroo"
+
+# user_id = self.register_user("kermit", old_password)
+# self.login("kermit", old_password)
- # Assert we can't log in with the new password
- self.attempt_wrong_password_login("kermit", new_password)
- @unittest.override_config({"request_token_inhibit_3pid_errors": True})
- def test_password_reset_bad_email_inhibit_error(self) -> None:
- """Test that triggering a password reset with an email address that isn't bound
- to an account doesn't leak the lack of binding for that address if configured
- that way.
- """
- self.register_user("kermit", "monkey")
- self.login("kermit", "monkey")
+# client_secret = "foobar"
+# session_id = self._request_token(email, client_secret)
- email = "test@example.com"
+# self.assertEqual(len(self.email_attempts), 1)
+# link = self._get_link_from_email()
- client_secret = "foobar"
- session_id = self._request_token(email, client_secret)
+# self._validate_token(link)
- self.assertIsNotNone(session_id)
+# self._reset_password(new_password, session_id, client_secret)
- def test_password_reset_redirection(self) -> None:
- """Test basic password reset flow"""
- old_password = "monkey"
+# # Assert we can log in with the new password
+# self.login("kermit", new_password)
- user_id = self.register_user("kermit", old_password)
- self.login("kermit", old_password)
+# # Assert we can't log in with the old password
+# self.attempt_wrong_password_login("kermit", old_password)
- email = "test@example.com"
+# # Check that the UI Auth information doesn't store the password in the database.
+# #
+# # Note that we don't have the UI Auth session ID, so just pull out the single
+# # row.
+# result = self.get_success(
+# self.store.db_pool.simple_select_one_onecol(
+# "ui_auth_sessions", keyvalues={}, retcol="clientdict"
+# )
+# )
+# client_dict = db_to_json(result)
+# self.assertNotIn("new_password", client_dict)
+
+# @override_config({"rc_3pid_validation": {"burst_count": 3}})
+# def test_ratelimit_by_email(self) -> None:
+# """Test that we ratelimit /requestToken for the same email."""
+# old_password = "monkey"
+# new_password = "kangeroo"
- # Add a threepid
- self.get_success(
- self.store.user_add_threepid(
- user_id=user_id,
- medium="email",
- address=email,
- validated_at=0,
- added_at=0,
- )
- )
+# user_id = self.register_user("kermit", old_password)
+# self.login("kermit", old_password)
+
+
+# def reset(ip: str) -> None:
+# client_secret = "foobar"
+# session_id = self._request_token(email, client_secret, ip)
- client_secret = "foobar"
- next_link = "http://example.com"
- self._request_token(email, client_secret, "127.0.0.1", next_link)
+# self.assertEqual(len(self.email_attempts), 1)
+# link = self._get_link_from_email()
+
+# self._validate_token(link)
+
+# self._reset_password(new_password, session_id, client_secret)
+
+# self.email_attempts.clear()
- self.assertEqual(len(self.email_attempts), 1)
- link = self._get_link_from_email()
+# # We expect to be able to make three requests before getting rate
+# # limited.
+# #
+# # We change IPs to ensure that we're not being ratelimited due to the
+# # same IP
+# reset("127.0.0.1")
+# reset("127.0.0.2")
+# reset("127.0.0.3")
- self._validate_token(link, next_link)
+# with self.assertRaises(HttpResponseException) as cm:
+# reset("127.0.0.4")
- def _request_token(
- self,
- email: str,
- client_secret: str,
- ip: str = "127.0.0.1",
- next_link: Optional[str] = None,
- ) -> str:
- body = {"client_secret": client_secret, "email": email, "send_attempt": 1}
- if next_link is not None:
- body["next_link"] = next_link
- channel = self.make_request(
- "POST",
- b"account/password/email/requestToken",
- body,
- client_ip=ip,
- )
+# self.assertEqual(cm.exception.code, 429)
- if channel.code != 200:
- raise HttpResponseException(
- channel.code,
- channel.result["reason"],
- channel.result["body"],
- )
+# def test_basic_password_reset_canonicalise_email(self) -> None:
+# """Test basic password reset flow
+# Request password reset with different spelling
+# """
+# old_password = "monkey"
+# new_password = "kangeroo"
- return channel.json_body["sid"]
+# user_id = self.register_user("kermit", old_password)
+# self.login("kermit", old_password)
- def _validate_token(self, link: str, next_link: Optional[str] = None) -> None:
- # Remove the host
- path = link.replace("https://example.com", "")
+# email_profile = "test@example.com"
+# email_passwort_reset = "TEST@EXAMPLE.COM"
- # Load the password reset confirmation page
- channel = make_request(
- self.reactor,
- FakeSite(self.submit_token_resource, self.reactor),
- "GET",
- path,
- shorthand=False,
- )
+# # Add a threepid
+# self.get_success(
+# self.store.user_add_threepid(
+# user_id=user_id,
+# medium="email",
+# address=email_profile,
+# validated_at=0,
+# added_at=0,
+# )
+# )
- self.assertEqual(HTTPStatus.OK, channel.code, channel.result)
+# client_secret = "foobar"
+# session_id = self._request_token(email_passwort_reset, client_secret)
- # Now POST to the same endpoint, mimicking the same behaviour as clicking the
- # password reset confirm button
+# self.assertEqual(len(self.email_attempts), 1)
+# link = self._get_link_from_email()
- # Confirm the password reset
- channel = make_request(
- self.reactor,
- FakeSite(self.submit_token_resource, self.reactor),
- "POST",
- path,
- content=b"",
- shorthand=False,
- content_is_form=True,
- )
- self.assertEqual(
- HTTPStatus.OK if next_link is None else HTTPStatus.FOUND,
- channel.code,
- channel.result,
- )
-
- def _get_link_from_email(self) -> str:
- assert self.email_attempts, "No emails have been sent"
+# self._validate_token(link)
- raw_msg = self.email_attempts[-1].decode("UTF-8")
- mail = Parser().parsestr(raw_msg)
+# self._reset_password(new_password, session_id, client_secret)
- text = None
- for part in mail.walk():
- if part.get_content_type() == "text/plain":
- text = part.get_payload(decode=True)
- if text is not None:
- # According to the logic table in `get_payload`, we know that
- # the result of `get_payload` will be `bytes`, but mypy doesn't
- # know this and complains. Thus, we assert the type.
- assert isinstance(text, bytes)
- text = text.decode("UTF-8")
+# # Assert we can log in with the new password
+# self.login("kermit", new_password)
- break
+# # Assert we can't log in with the old password
+# self.attempt_wrong_password_login("kermit", old_password)
- if not text:
- self.fail("Could not find text portion of email to parse")
+# def test_cant_reset_password_without_clicking_link(self) -> None:
+# """Test that we do actually need to click the link in the email"""
+# old_password = "monkey"
+# new_password = "kangeroo"
+
+# user_id = self.register_user("kermit", old_password)
+# self.login("kermit", old_password)
+
+# email = "test@example.com"
+
+# # Add a threepid
+# self.get_success(
+# self.store.user_add_threepid(
+# user_id=user_id,
+# medium="email",
+# address=email,
+# validated_at=0,
+# added_at=0,
+# )
+# )
- # `text` must be a `str`, after being decoded and determined just above
- # to not be `None` or an empty `str`.
- assert isinstance(text, str)
-
- match = re.search(r"https://example.com\S+", text)
- assert match, "Could not find link in email"
-
- return match.group(0)
-
- def _reset_password(
- self,
- new_password: str,
- session_id: str,
- client_secret: str,
- expected_code: int = HTTPStatus.OK,
- ) -> None:
- channel = self.make_request(
- "POST",
- b"account/password",
- {
- "new_password": new_password,
- "auth": {
- "type": LoginType.EMAIL_IDENTITY,
- "threepid_creds": {
- "client_secret": client_secret,
- "sid": session_id,
- },
- },
- },
- )
- self.assertEqual(expected_code, channel.code, channel.result)
+# client_secret = "foobar"
+# session_id = self._request_token(email, client_secret)
+
+# self.assertEqual(len(self.email_attempts), 1)
+
+# # Attempt to reset password without clicking the link
+# self._reset_password(new_password, session_id, client_secret, expected_code=401)
+
+# # Assert we can log in with the old password
+# self.login("kermit", old_password)
+
+# # Assert we can't log in with the new password
+# self.attempt_wrong_password_login("kermit", new_password)
+
+# def test_no_valid_token(self) -> None:
+# """Test that we do actually need to request a token and can't just
+# make a session up.
+# """
+# old_password = "monkey"
+# new_password = "kangeroo"
+
+# user_id = self.register_user("kermit", old_password)
+# self.login("kermit", old_password)
+
+# email = "test@example.com"
+
+# # Add a threepid
+# self.get_success(
+# self.store.user_add_threepid(
+# user_id=user_id,
+# medium="email",
+# address=email,
+# validated_at=0,
+# added_at=0,
+# )
+# )
+
+# client_secret = "foobar"
+# session_id = "weasle"
+
+# # Attempt to reset password without even requesting an email
+# self._reset_password(new_password, session_id, client_secret, expected_code=401)
+
+# # Assert we can log in with the old password
+# self.login("kermit", old_password)
+
+# # Assert we can't log in with the new password
+# self.attempt_wrong_password_login("kermit", new_password)
+
+# @unittest.override_config({"request_token_inhibit_3pid_errors": True})
+# def test_password_reset_bad_email_inhibit_error(self) -> None:
+# """Test that triggering a password reset with an email address that isn't bound
+# to an account doesn't leak the lack of binding for that address if configured
+# that way.
+# """
+# self.register_user("kermit", "monkey")
+# self.login("kermit", "monkey")
+
+# email = "test@example.com"
+
+# client_secret = "foobar"
+# session_id = self._request_token(email, client_secret)
+
+# self.assertIsNotNone(session_id)
+
+# def test_password_reset_redirection(self) -> None:
+# """Test basic password reset flow"""
+# old_password = "monkey"
+
+# user_id = self.register_user("kermit", old_password)
+# self.login("kermit", old_password)
+
+# email = "test@example.com"
+
+# # Add a threepid
+# self.get_success(
+# self.store.user_add_threepid(
+# user_id=user_id,
+# medium="email",
+# address=email,
+# validated_at=0,
+# added_at=0,
+# )
+# )
+
+# client_secret = "foobar"
+# next_link = "http://example.com"
+# self._request_token(email, client_secret, "127.0.0.1", next_link)
+
+# self.assertEqual(len(self.email_attempts), 1)
+# link = self._get_link_from_email()
+
+# self._validate_token(link, next_link)
+
+# def _request_token(
+# self,
+# email: str,
+# client_secret: str,
+# ip: str = "127.0.0.1",
+# next_link: Optional[str] = None,
+# ) -> str:
+# body = {"client_secret": client_secret, "email": email, "send_attempt": 1}
+# if next_link is not None:
+# body["next_link"] = next_link
+# channel = self.make_request(
+# "POST",
+# b"account/password/email/requestToken",
+# body,
+# client_ip=ip,
+# )
+
+# if channel.code != 200:
+# raise HttpResponseException(
+# channel.code,
+# channel.result["reason"],
+# channel.result["body"],
+# )
+
+# return channel.json_body["sid"]
+
+# def _validate_token(self, link: str, next_link: Optional[str] = None) -> None:
+# # Remove the host
+# path = link.replace("https://example.com", "")
+
+# # Load the password reset confirmation page
+# channel = make_request(
+# self.reactor,
+# FakeSite(self.submit_token_resource, self.reactor),
+# "GET",
+# path,
+# shorthand=False,
+# )
+
+# self.assertEqual(HTTPStatus.OK, channel.code, channel.result)
+
+# # Now POST to the same endpoint, mimicking the same behaviour as clicking the
+# # password reset confirm button
+
+# # Confirm the password reset
+# channel = make_request(
+# self.reactor,
+# FakeSite(self.submit_token_resource, self.reactor),
+# "POST",
+# path,
+# content=b"",
+# shorthand=False,
+# content_is_form=True,
+# )
+# self.assertEqual(
+# HTTPStatus.OK if next_link is None else HTTPStatus.FOUND,
+# channel.code,
+# channel.result,
+# )
+
+# def _get_link_from_email(self) -> str:
+# assert self.email_attempts, "No emails have been sent"
+
+# raw_msg = self.email_attempts[-1].decode("UTF-8")
+# mail = Parser().parsestr(raw_msg)
+
+# text = None
+# for part in mail.walk():
+# if part.get_content_type() == "text/plain":
+# text = part.get_payload(decode=True)
+# if text is not None:
+# # According to the logic table in `get_payload`, we know that
+# # the result of `get_payload` will be `bytes`, but mypy doesn't
+# # know this and complains. Thus, we assert the type.
+# assert isinstance(text, bytes)
+# text = text.decode("UTF-8")
+
+# break
+
+# if not text:
+# self.fail("Could not find text portion of email to parse")
+
+# # `text` must be a `str`, after being decoded and determined just above
+# # to not be `None` or an empty `str`.
+# assert isinstance(text, str)
+
+# match = re.search(r"https://example.com\S+", text)
+# assert match, "Could not find link in email"
+
+# return match.group(0)
+
+# def _reset_password(
+# self,
+# new_password: str,
+# session_id: str,
+# client_secret: str,
+# expected_code: int = HTTPStatus.OK,
+# ) -> None:
+# channel = self.make_request(
+# "POST",
+# b"account/password",
+# {
+# "new_password": new_password,
+# "auth": {
+# "type": LoginType.EMAIL_IDENTITY,
+# "threepid_creds": {
+# "client_secret": client_secret,
+# "sid": session_id,
+# },
+# },
+# },
+# )
+# self.assertEqual(expected_code, channel.code, channel.result)
class DeactivateTestCase(unittest.HomeserverTestCase):
@@ -787,503 +760,6 @@ class WhoamiTestCase(unittest.HomeserverTestCase):
return channel.json_body
-class ThreepidEmailRestTestCase(unittest.HomeserverTestCase):
- servlets = [
- account.register_servlets,
- login.register_servlets,
- synapse.rest.admin.register_servlets_for_client_rest_resource,
- ]
-
- def make_homeserver(self, reactor: MemoryReactor, clock: Clock) -> HomeServer:
- config = self.default_config()
-
- # Email config.
- config["email"] = {
- "enable_notifs": False,
- "template_dir": os.path.abspath(
- pkg_resources.resource_filename("synapse", "res/templates")
- ),
- "smtp_host": "127.0.0.1",
- "smtp_port": 20,
- "require_transport_security": False,
- "smtp_user": None,
- "smtp_pass": None,
- "notif_from": "test@example.com",
- }
- config["public_baseurl"] = "https://example.com"
-
- self.hs = self.setup_test_homeserver(config=config)
-
- async def sendmail(
- reactor: IReactorTCP,
- smtphost: str,
- smtpport: int,
- from_addr: str,
- to_addr: str,
- msg_bytes: bytes,
- *args: Any,
- **kwargs: Any,
- ) -> None:
- self.email_attempts.append(msg_bytes)
-
- self.email_attempts: List[bytes] = []
- self.hs.get_send_email_handler()._sendmail = sendmail
-
- return self.hs
-
- def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None:
- self.store = hs.get_datastores().main
-
- self.user_id = self.register_user("kermit", "test")
- self.user_id_tok = self.login("kermit", "test")
- self.email = "test@example.com"
- self.url_3pid = b"account/3pid"
-
- def test_add_valid_email(self) -> None:
- self._add_email(self.email, self.email)
-
- def test_add_valid_email_second_time(self) -> None:
- self._add_email(self.email, self.email)
- self._request_token_invalid_email(
- self.email,
- expected_errcode=Codes.THREEPID_IN_USE,
- expected_error="Email is already in use",
- )
-
- def test_add_valid_email_second_time_canonicalise(self) -> None:
- self._add_email(self.email, self.email)
- self._request_token_invalid_email(
- "TEST@EXAMPLE.COM",
- expected_errcode=Codes.THREEPID_IN_USE,
- expected_error="Email is already in use",
- )
-
- def test_add_email_no_at(self) -> None:
- self._request_token_invalid_email(
- "address-without-at.bar",
- expected_errcode=Codes.BAD_JSON,
- expected_error="Unable to parse email address",
- )
-
- def test_add_email_two_at(self) -> None:
- self._request_token_invalid_email(
- "foo@foo@test.bar",
- expected_errcode=Codes.BAD_JSON,
- expected_error="Unable to parse email address",
- )
-
- def test_add_email_bad_format(self) -> None:
- self._request_token_invalid_email(
- "user@bad.example.net@good.example.com",
- expected_errcode=Codes.BAD_JSON,
- expected_error="Unable to parse email address",
- )
-
- def test_add_email_domain_to_lower(self) -> None:
- self._add_email("foo@TEST.BAR", "foo@test.bar")
-
- def test_add_email_domain_with_umlaut(self) -> None:
- self._add_email("foo@Öumlaut.com", "foo@öumlaut.com")
-
- def test_add_email_address_casefold(self) -> None:
- self._add_email("Strauß@Example.com", "strauss@example.com")
-
- def test_address_trim(self) -> None:
- self._add_email(" foo@test.bar ", "foo@test.bar")
-
- @override_config({"rc_3pid_validation": {"burst_count": 3}})
- def test_ratelimit_by_ip(self) -> None:
- """Tests that adding emails is ratelimited by IP"""
-
- # We expect to be able to set three emails before getting ratelimited.
- self._add_email("foo1@test.bar", "foo1@test.bar")
- self._add_email("foo2@test.bar", "foo2@test.bar")
- self._add_email("foo3@test.bar", "foo3@test.bar")
-
- with self.assertRaises(HttpResponseException) as cm:
- self._add_email("foo4@test.bar", "foo4@test.bar")
-
- self.assertEqual(cm.exception.code, 429)
-
- def test_add_email_if_disabled(self) -> None:
- """Test adding email to profile when doing so is disallowed"""
- self.hs.config.registration.enable_3pid_changes = False
-
- client_secret = "foobar"
- channel = self.make_request(
- "POST",
- b"/_matrix/client/unstable/account/3pid/email/requestToken",
- {
- "client_secret": client_secret,
- "email": "test@example.com",
- "send_attempt": 1,
- },
- )
-
- self.assertEqual(
- HTTPStatus.BAD_REQUEST, channel.code, msg=channel.result["body"]
- )
-
- self.assertEqual(Codes.FORBIDDEN, channel.json_body["errcode"])
-
- def test_delete_email(self) -> None:
- """Test deleting an email from profile"""
- # Add a threepid
- self.get_success(
- self.store.user_add_threepid(
- user_id=self.user_id,
- medium="email",
- address=self.email,
- validated_at=0,
- added_at=0,
- )
- )
-
- channel = self.make_request(
- "POST",
- b"account/3pid/delete",
- {"medium": "email", "address": self.email},
- access_token=self.user_id_tok,
- )
- self.assertEqual(HTTPStatus.OK, channel.code, msg=channel.result["body"])
-
- # Get user
- channel = self.make_request(
- "GET",
- self.url_3pid,
- access_token=self.user_id_tok,
- )
-
- self.assertEqual(HTTPStatus.OK, channel.code, msg=channel.result["body"])
- self.assertFalse(channel.json_body["threepids"])
-
- def test_delete_email_if_disabled(self) -> None:
- """Test deleting an email from profile when disallowed"""
- self.hs.config.registration.enable_3pid_changes = False
-
- # Add a threepid
- self.get_success(
- self.store.user_add_threepid(
- user_id=self.user_id,
- medium="email",
- address=self.email,
- validated_at=0,
- added_at=0,
- )
- )
-
- channel = self.make_request(
- "POST",
- b"account/3pid/delete",
- {"medium": "email", "address": self.email},
- access_token=self.user_id_tok,
- )
-
- self.assertEqual(
- HTTPStatus.BAD_REQUEST, channel.code, msg=channel.result["body"]
- )
- self.assertEqual(Codes.FORBIDDEN, channel.json_body["errcode"])
-
- # Get user
- channel = self.make_request(
- "GET",
- self.url_3pid,
- access_token=self.user_id_tok,
- )
-
- self.assertEqual(HTTPStatus.OK, channel.code, msg=channel.result["body"])
- self.assertEqual("email", channel.json_body["threepids"][0]["medium"])
- self.assertEqual(self.email, channel.json_body["threepids"][0]["address"])
-
- def test_cant_add_email_without_clicking_link(self) -> None:
- """Test that we do actually need to click the link in the email"""
- client_secret = "foobar"
- session_id = self._request_token(self.email, client_secret)
-
- self.assertEqual(len(self.email_attempts), 1)
-
- # Attempt to add email without clicking the link
- channel = self.make_request(
- "POST",
- b"/_matrix/client/unstable/account/3pid/add",
- {
- "client_secret": client_secret,
- "sid": session_id,
- "auth": {
- "type": "m.login.password",
- "user": self.user_id,
- "password": "test",
- },
- },
- access_token=self.user_id_tok,
- )
- self.assertEqual(
- HTTPStatus.BAD_REQUEST, channel.code, msg=channel.result["body"]
- )
- self.assertEqual(Codes.THREEPID_AUTH_FAILED, channel.json_body["errcode"])
-
- # Get user
- channel = self.make_request(
- "GET",
- self.url_3pid,
- access_token=self.user_id_tok,
- )
-
- self.assertEqual(HTTPStatus.OK, channel.code, msg=channel.result["body"])
- self.assertFalse(channel.json_body["threepids"])
-
- def test_no_valid_token(self) -> None:
- """Test that we do actually need to request a token and can't just
- make a session up.
- """
- client_secret = "foobar"
- session_id = "weasle"
-
- # Attempt to add email without even requesting an email
- channel = self.make_request(
- "POST",
- b"/_matrix/client/unstable/account/3pid/add",
- {
- "client_secret": client_secret,
- "sid": session_id,
- "auth": {
- "type": "m.login.password",
- "user": self.user_id,
- "password": "test",
- },
- },
- access_token=self.user_id_tok,
- )
- self.assertEqual(
- HTTPStatus.BAD_REQUEST, channel.code, msg=channel.result["body"]
- )
- self.assertEqual(Codes.THREEPID_AUTH_FAILED, channel.json_body["errcode"])
-
- # Get user
- channel = self.make_request(
- "GET",
- self.url_3pid,
- access_token=self.user_id_tok,
- )
-
- self.assertEqual(HTTPStatus.OK, channel.code, msg=channel.result["body"])
- self.assertFalse(channel.json_body["threepids"])
-
- @override_config({"next_link_domain_whitelist": None})
- def test_next_link(self) -> None:
- """Tests a valid next_link parameter value with no whitelist (good case)"""
- self._request_token(
- "something@example.com",
- "some_secret",
- next_link="https://example.com/a/good/site",
- expect_code=HTTPStatus.OK,
- )
-
- @override_config({"next_link_domain_whitelist": None})
- def test_next_link_exotic_protocol(self) -> None:
- """Tests using a esoteric protocol as a next_link parameter value.
- Someone may be hosting a client on IPFS etc.
- """
- self._request_token(
- "something@example.com",
- "some_secret",
- next_link="some-protocol://abcdefghijklmopqrstuvwxyz",
- expect_code=HTTPStatus.OK,
- )
-
- @override_config({"next_link_domain_whitelist": None})
- def test_next_link_file_uri(self) -> None:
- """Tests next_link parameters cannot be file URI"""
- # Attempt to use a next_link value that points to the local disk
- self._request_token(
- "something@example.com",
- "some_secret",
- next_link="file:///host/path",
- expect_code=HTTPStatus.BAD_REQUEST,
- )
-
- @override_config({"next_link_domain_whitelist": ["example.com", "example.org"]})
- def test_next_link_domain_whitelist(self) -> None:
- """Tests next_link parameters must fit the whitelist if provided"""
-
- # Ensure not providing a next_link parameter still works
- self._request_token(
- "something@example.com",
- "some_secret",
- next_link=None,
- expect_code=HTTPStatus.OK,
- )
-
- self._request_token(
- "something@example.com",
- "some_secret",
- next_link="https://example.com/some/good/page",
- expect_code=HTTPStatus.OK,
- )
-
- self._request_token(
- "something@example.com",
- "some_secret",
- next_link="https://example.org/some/also/good/page",
- expect_code=HTTPStatus.OK,
- )
-
- self._request_token(
- "something@example.com",
- "some_secret",
- next_link="https://bad.example.org/some/bad/page",
- expect_code=HTTPStatus.BAD_REQUEST,
- )
-
- @override_config({"next_link_domain_whitelist": []})
- def test_empty_next_link_domain_whitelist(self) -> None:
- """Tests an empty next_lint_domain_whitelist value, meaning next_link is essentially
- disallowed
- """
- self._request_token(
- "something@example.com",
- "some_secret",
- next_link="https://example.com/a/page",
- expect_code=HTTPStatus.BAD_REQUEST,
- )
-
- def _request_token(
- self,
- email: str,
- client_secret: str,
- next_link: Optional[str] = None,
- expect_code: int = HTTPStatus.OK,
- ) -> Optional[str]:
- """Request a validation token to add an email address to a user's account
-
- Args:
- email: The email address to validate
- client_secret: A secret string
- next_link: A link to redirect the user to after validation
- expect_code: Expected return code of the call
-
- Returns:
- The ID of the new threepid validation session, or None if the response
- did not contain a session ID.
- """
- body = {"client_secret": client_secret, "email": email, "send_attempt": 1}
- if next_link:
- body["next_link"] = next_link
-
- channel = self.make_request(
- "POST",
- b"account/3pid/email/requestToken",
- body,
- )
-
- if channel.code != expect_code:
- raise HttpResponseException(
- channel.code,
- channel.result["reason"],
- channel.result["body"],
- )
-
- return channel.json_body.get("sid")
-
- def _request_token_invalid_email(
- self,
- email: str,
- expected_errcode: str,
- expected_error: str,
- client_secret: str = "foobar",
- ) -> None:
- channel = self.make_request(
- "POST",
- b"account/3pid/email/requestToken",
- {"client_secret": client_secret, "email": email, "send_attempt": 1},
- )
- self.assertEqual(
- HTTPStatus.BAD_REQUEST, channel.code, msg=channel.result["body"]
- )
- self.assertEqual(expected_errcode, channel.json_body["errcode"])
- self.assertIn(expected_error, channel.json_body["error"])
-
- def _validate_token(self, link: str) -> None:
- # Remove the host
- path = link.replace("https://example.com", "")
-
- channel = self.make_request("GET", path, shorthand=False)
- self.assertEqual(HTTPStatus.OK, channel.code, channel.result)
-
- def _get_link_from_email(self) -> str:
- assert self.email_attempts, "No emails have been sent"
-
- raw_msg = self.email_attempts[-1].decode("UTF-8")
- mail = Parser().parsestr(raw_msg)
-
- text = None
- for part in mail.walk():
- if part.get_content_type() == "text/plain":
- text = part.get_payload(decode=True)
- if text is not None:
- # According to the logic table in `get_payload`, we know that
- # the result of `get_payload` will be `bytes`, but mypy doesn't
- # know this and complains. Thus, we assert the type.
- assert isinstance(text, bytes)
- text = text.decode("UTF-8")
-
- break
-
- if not text:
- self.fail("Could not find text portion of email to parse")
-
- # `text` must be a `str`, after being decoded and determined just above
- # to not be `None` or an empty `str`.
- assert isinstance(text, str)
-
- match = re.search(r"https://example.com\S+", text)
- assert match, "Could not find link in email"
-
- return match.group(0)
-
- def _add_email(self, request_email: str, expected_email: str) -> None:
- """Test adding an email to profile"""
- previous_email_attempts = len(self.email_attempts)
-
- client_secret = "foobar"
- session_id = self._request_token(request_email, client_secret)
-
- self.assertEqual(len(self.email_attempts) - previous_email_attempts, 1)
- link = self._get_link_from_email()
-
- self._validate_token(link)
-
- channel = self.make_request(
- "POST",
- b"/_matrix/client/unstable/account/3pid/add",
- {
- "client_secret": client_secret,
- "sid": session_id,
- "auth": {
- "type": "m.login.password",
- "user": self.user_id,
- "password": "test",
- },
- },
- access_token=self.user_id_tok,
- )
-
- self.assertEqual(HTTPStatus.OK, channel.code, msg=channel.result["body"])
-
- # Get user
- channel = self.make_request(
- "GET",
- self.url_3pid,
- access_token=self.user_id_tok,
- )
-
- self.assertEqual(HTTPStatus.OK, channel.code, msg=channel.result["body"])
- self.assertEqual("email", channel.json_body["threepids"][0]["medium"])
-
- threepids = {threepid["address"] for threepid in channel.json_body["threepids"]}
- self.assertIn(expected_email, threepids)
-
-
class AccountStatusTestCase(unittest.HomeserverTestCase):
servlets = [
account.register_servlets,
|
