forked from MirrorHub/synapse
Fix inconsistent handling of upper and lower cases of email addresses. (#7021)
fixes #7016
This commit is contained in:
parent
8097659f6e
commit
21a212f8e5
1
changelog.d/7021.bugfix
Normal file
1
changelog.d/7021.bugfix
Normal file
|
@ -0,0 +1 @@
|
||||||
|
Fix inconsistent handling of upper and lower case in email addresses when used as identifiers for login, etc. Contributed by @dklimpel.
|
|
@ -45,6 +45,7 @@ from synapse.metrics.background_process_metrics import run_as_background_process
|
||||||
from synapse.module_api import ModuleApi
|
from synapse.module_api import ModuleApi
|
||||||
from synapse.push.mailer import load_jinja2_templates
|
from synapse.push.mailer import load_jinja2_templates
|
||||||
from synapse.types import Requester, UserID
|
from synapse.types import Requester, UserID
|
||||||
|
from synapse.util.threepids import canonicalise_email
|
||||||
|
|
||||||
from ._base import BaseHandler
|
from ._base import BaseHandler
|
||||||
|
|
||||||
|
@ -928,7 +929,7 @@ class AuthHandler(BaseHandler):
|
||||||
# for the presence of an email address during password reset was
|
# for the presence of an email address during password reset was
|
||||||
# case sensitive).
|
# case sensitive).
|
||||||
if medium == "email":
|
if medium == "email":
|
||||||
address = address.lower()
|
address = canonicalise_email(address)
|
||||||
|
|
||||||
await self.store.user_add_threepid(
|
await self.store.user_add_threepid(
|
||||||
user_id, medium, address, validated_at, self.hs.get_clock().time_msec()
|
user_id, medium, address, validated_at, self.hs.get_clock().time_msec()
|
||||||
|
@ -956,7 +957,7 @@ class AuthHandler(BaseHandler):
|
||||||
|
|
||||||
# 'Canonicalise' email addresses as per above
|
# 'Canonicalise' email addresses as per above
|
||||||
if medium == "email":
|
if medium == "email":
|
||||||
address = address.lower()
|
address = canonicalise_email(address)
|
||||||
|
|
||||||
identity_handler = self.hs.get_handlers().identity_handler
|
identity_handler = self.hs.get_handlers().identity_handler
|
||||||
result = await identity_handler.try_unbind_threepid(
|
result = await identity_handler.try_unbind_threepid(
|
||||||
|
|
|
@ -28,6 +28,7 @@ from synapse.rest.client.v2_alpha._base import client_patterns
|
||||||
from synapse.rest.well_known import WellKnownBuilder
|
from synapse.rest.well_known import WellKnownBuilder
|
||||||
from synapse.types import UserID
|
from synapse.types import UserID
|
||||||
from synapse.util.msisdn import phone_number_to_msisdn
|
from synapse.util.msisdn import phone_number_to_msisdn
|
||||||
|
from synapse.util.threepids import canonicalise_email
|
||||||
|
|
||||||
logger = logging.getLogger(__name__)
|
logger = logging.getLogger(__name__)
|
||||||
|
|
||||||
|
@ -206,11 +207,14 @@ class LoginRestServlet(RestServlet):
|
||||||
if medium is None or address is None:
|
if medium is None or address is None:
|
||||||
raise SynapseError(400, "Invalid thirdparty identifier")
|
raise SynapseError(400, "Invalid thirdparty identifier")
|
||||||
|
|
||||||
if medium == "email":
|
# For emails, canonicalise the address.
|
||||||
# For emails, transform the address to lowercase.
|
# We store all email addresses canonicalised in the DB.
|
||||||
# We store all email addreses as lowercase in the DB.
|
|
||||||
# (See add_threepid in synapse/handlers/auth.py)
|
# (See add_threepid in synapse/handlers/auth.py)
|
||||||
address = address.lower()
|
if medium == "email":
|
||||||
|
try:
|
||||||
|
address = canonicalise_email(address)
|
||||||
|
except ValueError as e:
|
||||||
|
raise SynapseError(400, str(e))
|
||||||
|
|
||||||
# We also apply account rate limiting using the 3PID as a key, as
|
# We also apply account rate limiting using the 3PID as a key, as
|
||||||
# otherwise using 3PID bypasses the ratelimiting based on user ID.
|
# otherwise using 3PID bypasses the ratelimiting based on user ID.
|
||||||
|
|
|
@ -30,7 +30,7 @@ from synapse.http.servlet import (
|
||||||
from synapse.push.mailer import Mailer, load_jinja2_templates
|
from synapse.push.mailer import Mailer, load_jinja2_templates
|
||||||
from synapse.util.msisdn import phone_number_to_msisdn
|
from synapse.util.msisdn import phone_number_to_msisdn
|
||||||
from synapse.util.stringutils import assert_valid_client_secret, random_string
|
from synapse.util.stringutils import assert_valid_client_secret, random_string
|
||||||
from synapse.util.threepids import check_3pid_allowed
|
from synapse.util.threepids import canonicalise_email, check_3pid_allowed
|
||||||
|
|
||||||
from ._base import client_patterns, interactive_auth_handler
|
from ._base import client_patterns, interactive_auth_handler
|
||||||
|
|
||||||
|
@ -83,7 +83,15 @@ class EmailPasswordRequestTokenRestServlet(RestServlet):
|
||||||
client_secret = body["client_secret"]
|
client_secret = body["client_secret"]
|
||||||
assert_valid_client_secret(client_secret)
|
assert_valid_client_secret(client_secret)
|
||||||
|
|
||||||
email = body["email"]
|
# Canonicalise the email address. The addresses are all stored canonicalised
|
||||||
|
# in the database. This allows the user to reset his password without having to
|
||||||
|
# know the exact spelling (eg. upper and lower case) of address in the database.
|
||||||
|
# Stored in the database "foo@bar.com"
|
||||||
|
# User requests with "FOO@bar.com" would raise a Not Found error
|
||||||
|
try:
|
||||||
|
email = canonicalise_email(body["email"])
|
||||||
|
except ValueError as e:
|
||||||
|
raise SynapseError(400, str(e))
|
||||||
send_attempt = body["send_attempt"]
|
send_attempt = body["send_attempt"]
|
||||||
next_link = body.get("next_link") # Optional param
|
next_link = body.get("next_link") # Optional param
|
||||||
|
|
||||||
|
@ -94,6 +102,10 @@ class EmailPasswordRequestTokenRestServlet(RestServlet):
|
||||||
Codes.THREEPID_DENIED,
|
Codes.THREEPID_DENIED,
|
||||||
)
|
)
|
||||||
|
|
||||||
|
# The email will be sent to the stored address.
|
||||||
|
# This avoids a potential account hijack by requesting a password reset to
|
||||||
|
# an email address which is controlled by the attacker but which, after
|
||||||
|
# canonicalisation, matches the one in our database.
|
||||||
existing_user_id = await self.hs.get_datastore().get_user_id_by_threepid(
|
existing_user_id = await self.hs.get_datastore().get_user_id_by_threepid(
|
||||||
"email", email
|
"email", email
|
||||||
)
|
)
|
||||||
|
@ -274,10 +286,13 @@ class PasswordRestServlet(RestServlet):
|
||||||
if "medium" not in threepid or "address" not in threepid:
|
if "medium" not in threepid or "address" not in threepid:
|
||||||
raise SynapseError(500, "Malformed threepid")
|
raise SynapseError(500, "Malformed threepid")
|
||||||
if threepid["medium"] == "email":
|
if threepid["medium"] == "email":
|
||||||
# For emails, transform the address to lowercase.
|
# For emails, canonicalise the address.
|
||||||
# We store all email addreses as lowercase in the DB.
|
# We store all email addresses canonicalised in the DB.
|
||||||
# (See add_threepid in synapse/handlers/auth.py)
|
# (See add_threepid in synapse/handlers/auth.py)
|
||||||
threepid["address"] = threepid["address"].lower()
|
try:
|
||||||
|
threepid["address"] = canonicalise_email(threepid["address"])
|
||||||
|
except ValueError as e:
|
||||||
|
raise SynapseError(400, str(e))
|
||||||
# if using email, we must know about the email they're authing with!
|
# if using email, we must know about the email they're authing with!
|
||||||
threepid_user_id = await self.datastore.get_user_id_by_threepid(
|
threepid_user_id = await self.datastore.get_user_id_by_threepid(
|
||||||
threepid["medium"], threepid["address"]
|
threepid["medium"], threepid["address"]
|
||||||
|
@ -392,7 +407,16 @@ class EmailThreepidRequestTokenRestServlet(RestServlet):
|
||||||
client_secret = body["client_secret"]
|
client_secret = body["client_secret"]
|
||||||
assert_valid_client_secret(client_secret)
|
assert_valid_client_secret(client_secret)
|
||||||
|
|
||||||
email = body["email"]
|
# Canonicalise the email address. The addresses are all stored canonicalised
|
||||||
|
# in the database.
|
||||||
|
# This ensures that the validation email is sent to the canonicalised address
|
||||||
|
# as it will later be entered into the database.
|
||||||
|
# Otherwise the email will be sent to "FOO@bar.com" and stored as
|
||||||
|
# "foo@bar.com" in database.
|
||||||
|
try:
|
||||||
|
email = canonicalise_email(body["email"])
|
||||||
|
except ValueError as e:
|
||||||
|
raise SynapseError(400, str(e))
|
||||||
send_attempt = body["send_attempt"]
|
send_attempt = body["send_attempt"]
|
||||||
next_link = body.get("next_link") # Optional param
|
next_link = body.get("next_link") # Optional param
|
||||||
|
|
||||||
|
@ -403,9 +427,7 @@ class EmailThreepidRequestTokenRestServlet(RestServlet):
|
||||||
Codes.THREEPID_DENIED,
|
Codes.THREEPID_DENIED,
|
||||||
)
|
)
|
||||||
|
|
||||||
existing_user_id = await self.store.get_user_id_by_threepid(
|
existing_user_id = await self.store.get_user_id_by_threepid("email", email)
|
||||||
"email", body["email"]
|
|
||||||
)
|
|
||||||
|
|
||||||
if existing_user_id is not None:
|
if existing_user_id is not None:
|
||||||
if self.config.request_token_inhibit_3pid_errors:
|
if self.config.request_token_inhibit_3pid_errors:
|
||||||
|
|
|
@ -47,7 +47,7 @@ from synapse.push.mailer import load_jinja2_templates
|
||||||
from synapse.util.msisdn import phone_number_to_msisdn
|
from synapse.util.msisdn import phone_number_to_msisdn
|
||||||
from synapse.util.ratelimitutils import FederationRateLimiter
|
from synapse.util.ratelimitutils import FederationRateLimiter
|
||||||
from synapse.util.stringutils import assert_valid_client_secret, random_string
|
from synapse.util.stringutils import assert_valid_client_secret, random_string
|
||||||
from synapse.util.threepids import check_3pid_allowed
|
from synapse.util.threepids import canonicalise_email, check_3pid_allowed
|
||||||
|
|
||||||
from ._base import client_patterns, interactive_auth_handler
|
from ._base import client_patterns, interactive_auth_handler
|
||||||
|
|
||||||
|
@ -116,7 +116,14 @@ class EmailRegisterRequestTokenRestServlet(RestServlet):
|
||||||
client_secret = body["client_secret"]
|
client_secret = body["client_secret"]
|
||||||
assert_valid_client_secret(client_secret)
|
assert_valid_client_secret(client_secret)
|
||||||
|
|
||||||
email = body["email"]
|
# For emails, canonicalise the address.
|
||||||
|
# We store all email addresses canonicalised in the DB.
|
||||||
|
# (See on_POST in EmailThreepidRequestTokenRestServlet
|
||||||
|
# in synapse/rest/client/v2_alpha/account.py)
|
||||||
|
try:
|
||||||
|
email = canonicalise_email(body["email"])
|
||||||
|
except ValueError as e:
|
||||||
|
raise SynapseError(400, str(e))
|
||||||
send_attempt = body["send_attempt"]
|
send_attempt = body["send_attempt"]
|
||||||
next_link = body.get("next_link") # Optional param
|
next_link = body.get("next_link") # Optional param
|
||||||
|
|
||||||
|
@ -128,7 +135,7 @@ class EmailRegisterRequestTokenRestServlet(RestServlet):
|
||||||
)
|
)
|
||||||
|
|
||||||
existing_user_id = await self.hs.get_datastore().get_user_id_by_threepid(
|
existing_user_id = await self.hs.get_datastore().get_user_id_by_threepid(
|
||||||
"email", body["email"]
|
"email", email
|
||||||
)
|
)
|
||||||
|
|
||||||
if existing_user_id is not None:
|
if existing_user_id is not None:
|
||||||
|
@ -552,6 +559,15 @@ class RegisterRestServlet(RestServlet):
|
||||||
if login_type in auth_result:
|
if login_type in auth_result:
|
||||||
medium = auth_result[login_type]["medium"]
|
medium = auth_result[login_type]["medium"]
|
||||||
address = auth_result[login_type]["address"]
|
address = auth_result[login_type]["address"]
|
||||||
|
# For emails, canonicalise the address.
|
||||||
|
# We store all email addresses canonicalised in the DB.
|
||||||
|
# (See on_POST in EmailThreepidRequestTokenRestServlet
|
||||||
|
# in synapse/rest/client/v2_alpha/account.py)
|
||||||
|
if medium == "email":
|
||||||
|
try:
|
||||||
|
address = canonicalise_email(address)
|
||||||
|
except ValueError as e:
|
||||||
|
raise SynapseError(400, str(e))
|
||||||
|
|
||||||
existing_user_id = await self.store.get_user_id_by_threepid(
|
existing_user_id = await self.store.get_user_id_by_threepid(
|
||||||
medium, address
|
medium, address
|
||||||
|
|
|
@ -48,3 +48,26 @@ def check_3pid_allowed(hs, medium, address):
|
||||||
return True
|
return True
|
||||||
|
|
||||||
return False
|
return False
|
||||||
|
|
||||||
|
|
||||||
|
def canonicalise_email(address: str) -> str:
|
||||||
|
"""'Canonicalise' email address
|
||||||
|
Case folding of local part of email address and lowercase domain part
|
||||||
|
See MSC2265, https://github.com/matrix-org/matrix-doc/pull/2265
|
||||||
|
|
||||||
|
Args:
|
||||||
|
address: email address to be canonicalised
|
||||||
|
Returns:
|
||||||
|
The canonical form of the email address
|
||||||
|
Raises:
|
||||||
|
ValueError if the address could not be parsed.
|
||||||
|
"""
|
||||||
|
|
||||||
|
address = address.strip()
|
||||||
|
|
||||||
|
parts = address.split("@")
|
||||||
|
if len(parts) != 2:
|
||||||
|
logger.debug("Couldn't parse email address %s", address)
|
||||||
|
raise ValueError("Unable to parse email address")
|
||||||
|
|
||||||
|
return parts[0].casefold() + "@" + parts[1].lower()
|
||||||
|
|
|
@ -108,6 +108,46 @@ class PasswordResetTestCase(unittest.HomeserverTestCase):
|
||||||
# Assert we can't log in with the old password
|
# Assert we can't log in with the old password
|
||||||
self.attempt_wrong_password_login("kermit", old_password)
|
self.attempt_wrong_password_login("kermit", old_password)
|
||||||
|
|
||||||
|
def test_basic_password_reset_canonicalise_email(self):
|
||||||
|
"""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.assertEquals(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):
|
def test_cant_reset_password_without_clicking_link(self):
|
||||||
"""Test that we do actually need to click the link in the email
|
"""Test that we do actually need to click the link in the email
|
||||||
"""
|
"""
|
||||||
|
@ -386,44 +426,67 @@ class ThreepidEmailRestTestCase(unittest.HomeserverTestCase):
|
||||||
self.email = "test@example.com"
|
self.email = "test@example.com"
|
||||||
self.url_3pid = b"account/3pid"
|
self.url_3pid = b"account/3pid"
|
||||||
|
|
||||||
def test_add_email(self):
|
def test_add_valid_email(self):
|
||||||
"""Test adding an email to profile
|
self.get_success(self._add_email(self.email, self.email))
|
||||||
"""
|
|
||||||
client_secret = "foobar"
|
|
||||||
session_id = self._request_token(self.email, client_secret)
|
|
||||||
|
|
||||||
self.assertEquals(len(self.email_attempts), 1)
|
def test_add_valid_email_second_time(self):
|
||||||
link = self._get_link_from_email()
|
self.get_success(self._add_email(self.email, self.email))
|
||||||
|
self.get_success(
|
||||||
self._validate_token(link)
|
self._request_token_invalid_email(
|
||||||
|
self.email,
|
||||||
request, channel = self.make_request(
|
expected_errcode=Codes.THREEPID_IN_USE,
|
||||||
"POST",
|
expected_error="Email is already in use",
|
||||||
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.render(request)
|
def test_add_valid_email_second_time_canonicalise(self):
|
||||||
self.assertEqual(200, int(channel.result["code"]), msg=channel.result["body"])
|
self.get_success(self._add_email(self.email, self.email))
|
||||||
|
self.get_success(
|
||||||
# Get user
|
self._request_token_invalid_email(
|
||||||
request, channel = self.make_request(
|
"TEST@EXAMPLE.COM",
|
||||||
"GET", self.url_3pid, access_token=self.user_id_tok,
|
expected_errcode=Codes.THREEPID_IN_USE,
|
||||||
|
expected_error="Email is already in use",
|
||||||
|
)
|
||||||
)
|
)
|
||||||
self.render(request)
|
|
||||||
|
|
||||||
self.assertEqual(200, int(channel.result["code"]), msg=channel.result["body"])
|
def test_add_email_no_at(self):
|
||||||
self.assertEqual("email", channel.json_body["threepids"][0]["medium"])
|
self.get_success(
|
||||||
self.assertEqual(self.email, channel.json_body["threepids"][0]["address"])
|
self._request_token_invalid_email(
|
||||||
|
"address-without-at.bar",
|
||||||
|
expected_errcode=Codes.UNKNOWN,
|
||||||
|
expected_error="Unable to parse email address",
|
||||||
|
)
|
||||||
|
)
|
||||||
|
|
||||||
|
def test_add_email_two_at(self):
|
||||||
|
self.get_success(
|
||||||
|
self._request_token_invalid_email(
|
||||||
|
"foo@foo@test.bar",
|
||||||
|
expected_errcode=Codes.UNKNOWN,
|
||||||
|
expected_error="Unable to parse email address",
|
||||||
|
)
|
||||||
|
)
|
||||||
|
|
||||||
|
def test_add_email_bad_format(self):
|
||||||
|
self.get_success(
|
||||||
|
self._request_token_invalid_email(
|
||||||
|
"user@bad.example.net@good.example.com",
|
||||||
|
expected_errcode=Codes.UNKNOWN,
|
||||||
|
expected_error="Unable to parse email address",
|
||||||
|
)
|
||||||
|
)
|
||||||
|
|
||||||
|
def test_add_email_domain_to_lower(self):
|
||||||
|
self.get_success(self._add_email("foo@TEST.BAR", "foo@test.bar"))
|
||||||
|
|
||||||
|
def test_add_email_domain_with_umlaut(self):
|
||||||
|
self.get_success(self._add_email("foo@Öumlaut.com", "foo@öumlaut.com"))
|
||||||
|
|
||||||
|
def test_add_email_address_casefold(self):
|
||||||
|
self.get_success(self._add_email("Strauß@Example.com", "strauss@example.com"))
|
||||||
|
|
||||||
|
def test_address_trim(self):
|
||||||
|
self.get_success(self._add_email(" foo@test.bar ", "foo@test.bar"))
|
||||||
|
|
||||||
def test_add_email_if_disabled(self):
|
def test_add_email_if_disabled(self):
|
||||||
"""Test adding email to profile when doing so is disallowed
|
"""Test adding email to profile when doing so is disallowed
|
||||||
|
@ -616,6 +679,19 @@ class ThreepidEmailRestTestCase(unittest.HomeserverTestCase):
|
||||||
|
|
||||||
return channel.json_body["sid"]
|
return channel.json_body["sid"]
|
||||||
|
|
||||||
|
def _request_token_invalid_email(
|
||||||
|
self, email, expected_errcode, expected_error, client_secret="foobar",
|
||||||
|
):
|
||||||
|
request, channel = self.make_request(
|
||||||
|
"POST",
|
||||||
|
b"account/3pid/email/requestToken",
|
||||||
|
{"client_secret": client_secret, "email": email, "send_attempt": 1},
|
||||||
|
)
|
||||||
|
self.render(request)
|
||||||
|
self.assertEqual(400, int(channel.result["code"]), msg=channel.result["body"])
|
||||||
|
self.assertEqual(expected_errcode, channel.json_body["errcode"])
|
||||||
|
self.assertEqual(expected_error, channel.json_body["error"])
|
||||||
|
|
||||||
def _validate_token(self, link):
|
def _validate_token(self, link):
|
||||||
# Remove the host
|
# Remove the host
|
||||||
path = link.replace("https://example.com", "")
|
path = link.replace("https://example.com", "")
|
||||||
|
@ -643,3 +719,42 @@ class ThreepidEmailRestTestCase(unittest.HomeserverTestCase):
|
||||||
assert match, "Could not find link in email"
|
assert match, "Could not find link in email"
|
||||||
|
|
||||||
return match.group(0)
|
return match.group(0)
|
||||||
|
|
||||||
|
def _add_email(self, request_email, expected_email):
|
||||||
|
"""Test adding an email to profile
|
||||||
|
"""
|
||||||
|
client_secret = "foobar"
|
||||||
|
session_id = self._request_token(request_email, client_secret)
|
||||||
|
|
||||||
|
self.assertEquals(len(self.email_attempts), 1)
|
||||||
|
link = self._get_link_from_email()
|
||||||
|
|
||||||
|
self._validate_token(link)
|
||||||
|
|
||||||
|
request, 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.render(request)
|
||||||
|
self.assertEqual(200, int(channel.result["code"]), msg=channel.result["body"])
|
||||||
|
|
||||||
|
# Get user
|
||||||
|
request, channel = self.make_request(
|
||||||
|
"GET", self.url_3pid, access_token=self.user_id_tok,
|
||||||
|
)
|
||||||
|
self.render(request)
|
||||||
|
|
||||||
|
self.assertEqual(200, int(channel.result["code"]), msg=channel.result["body"])
|
||||||
|
self.assertEqual("email", channel.json_body["threepids"][0]["medium"])
|
||||||
|
self.assertEqual(expected_email, channel.json_body["threepids"][0]["address"])
|
||||||
|
|
49
tests/util/test_threepids.py
Normal file
49
tests/util/test_threepids.py
Normal file
|
@ -0,0 +1,49 @@
|
||||||
|
# -*- coding: utf-8 -*-
|
||||||
|
# Copyright 2020 Dirk Klimpel
|
||||||
|
#
|
||||||
|
# Licensed under the Apache License, Version 2.0 (the "License");
|
||||||
|
# you may not use this file except in compliance with the License.
|
||||||
|
# You may obtain a copy of the License at
|
||||||
|
#
|
||||||
|
# http://www.apache.org/licenses/LICENSE-2.0
|
||||||
|
#
|
||||||
|
# Unless required by applicable law or agreed to in writing, software
|
||||||
|
# distributed under the License is distributed on an "AS IS" BASIS,
|
||||||
|
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||||
|
# See the License for the specific language governing permissions and
|
||||||
|
# limitations under the License.
|
||||||
|
|
||||||
|
from synapse.util.threepids import canonicalise_email
|
||||||
|
|
||||||
|
from tests.unittest import HomeserverTestCase
|
||||||
|
|
||||||
|
|
||||||
|
class CanonicaliseEmailTests(HomeserverTestCase):
|
||||||
|
def test_no_at(self):
|
||||||
|
with self.assertRaises(ValueError):
|
||||||
|
canonicalise_email("address-without-at.bar")
|
||||||
|
|
||||||
|
def test_two_at(self):
|
||||||
|
with self.assertRaises(ValueError):
|
||||||
|
canonicalise_email("foo@foo@test.bar")
|
||||||
|
|
||||||
|
def test_bad_format(self):
|
||||||
|
with self.assertRaises(ValueError):
|
||||||
|
canonicalise_email("user@bad.example.net@good.example.com")
|
||||||
|
|
||||||
|
def test_valid_format(self):
|
||||||
|
self.assertEqual(canonicalise_email("foo@test.bar"), "foo@test.bar")
|
||||||
|
|
||||||
|
def test_domain_to_lower(self):
|
||||||
|
self.assertEqual(canonicalise_email("foo@TEST.BAR"), "foo@test.bar")
|
||||||
|
|
||||||
|
def test_domain_with_umlaut(self):
|
||||||
|
self.assertEqual(canonicalise_email("foo@Öumlaut.com"), "foo@öumlaut.com")
|
||||||
|
|
||||||
|
def test_address_casefold(self):
|
||||||
|
self.assertEqual(
|
||||||
|
canonicalise_email("Strauß@Example.com"), "strauss@example.com"
|
||||||
|
)
|
||||||
|
|
||||||
|
def test_address_trim(self):
|
||||||
|
self.assertEqual(canonicalise_email(" foo@test.bar "), "foo@test.bar")
|
Loading…
Reference in a new issue