Break using an alias as a mailbox loop in the email_handler.py (#1624)

* Do not allow to use email alias as account email when linking

* Add missing status

* Remove TODO

* Also break contact as email loop

* Better test names

* Allow a reverse alias to send an email to an alias

* Ident fix

* Removed invalid test

---------

Co-authored-by: Adrià Casajús <adria.casajus@proton.ch>

app/account_linking.py

@@ -9,13 +9,17 @@ from newrelic import agent
 from app.db import Session
 from app.email_utils import send_welcome_email
 from app.utils import sanitize_email
-from app.errors import AccountAlreadyLinkedToAnotherPartnerException
+from app.errors import (
+    AccountAlreadyLinkedToAnotherPartnerException,
+    AccountAlreadyLinkedToAnotherUserException,
+)
 from app.log import LOG
 from app.models import (
     PartnerSubscription,
     Partner,
     PartnerUser,
     User,
+    Alias,
 )
 from app.utils import random_string
 
@@ -192,11 +196,18 @@ def get_login_strategy(
     return ExistingUnlinkedUserStrategy(link_request, user, partner)
 
 
+def check_alias(email: str) -> bool:
+    alias = Alias.get_by(email=email)
+    if alias is not None:
+        raise AccountAlreadyLinkedToAnotherUserException()
+
+
 def process_login_case(
     link_request: PartnerLinkRequest, partner: Partner
 ) -> LinkResult:
     # Sanitize email just in case
     link_request.email = sanitize_email(link_request.email)
+    check_alias(link_request.email)
     # Try to find a SimpleLogin user registered with that partner user id
     partner_user = PartnerUser.get_by(
         partner_id=partner.id, external_user_id=link_request.external_user_id

app/email/status.py

@@ -60,4 +60,5 @@ E522 = (
 )
 E523 = "550 SL E523 Unknown error"
 E524 = "550 SL E524 Wrong use of reverse-alias"
+E525 = "550 SL E525 Alias loop"
 # endregion

app/errors.py

@@ -71,7 +71,7 @@ class ErrContactErrorUpgradeNeeded(SLException):
     """raised when user cannot create a contact because the plan doesn't allow it"""
 
     def error_for_user(self) -> str:
-        return f"Please upgrade to premium to create reverse-alias"
+        return "Please upgrade to premium to create reverse-alias"
 
 
 class ErrAddressInvalid(SLException):
@@ -108,3 +108,8 @@ class AccountAlreadyLinkedToAnotherPartnerException(LinkException):
 class AccountAlreadyLinkedToAnotherUserException(LinkException):
     def __init__(self):
         super().__init__("This account is linked to another user")
+
+
+class AccountIsUsingAliasAsEmail(LinkException):
+    def __init__(self):
+        super().__init__("Your account has an alias as it's email address")

email_handler.py

@@ -693,6 +693,36 @@ def handle_forward(envelope, msg: Message, rcpt_to: str) -> List[Tuple[bool, str
             LOG.d("%s unverified, do not forward", mailbox)
             ret.append((False, status.E517))
         else:
+            # Check if the mailbox is also an alias and stop the loop
+            mailbox_as_alias = Alias.get_by(email=mailbox.email)
+            if mailbox_as_alias is not None:
+                LOG.info(
+                    f"Mailbox {mailbox.id} has email {mailbox.email} that is also alias {alias.id}. Stopping loop"
+                )
+                mailbox.verified = False
+                Session.commit()
+                mailbox_url = f"{URL}/dashboard/mailbox/{mailbox.id}/"
+                send_email_with_rate_control(
+                    user,
+                    ALERT_MAILBOX_IS_ALIAS,
+                    user.email,
+                    f"Your mailbox {mailbox.email} is an alias",
+                    render(
+                        "transactional/mailbox-invalid.txt.jinja2",
+                        mailbox=mailbox,
+                        mailbox_url=mailbox_url,
+                        alias=alias,
+                    ),
+                    render(
+                        "transactional/mailbox-invalid.html",
+                        mailbox=mailbox,
+                        mailbox_url=mailbox_url,
+                        alias=alias,
+                    ),
+                    max_nb_alert=1,
+                )
+                ret.append((False, status.E525))
+                continue
             # create a copy of message for each forward
             ret.append(
                 forward_email_to_mailbox(

tests/test_email_handler.py

@@ -368,3 +368,19 @@ def test_send_email_from_non_canonical_matches_already_existing_user(flask_clien
     assert len(email_logs) == 1
     assert email_logs[0].alias_id == alias.id
     assert email_logs[0].mailbox_id == user.default_mailbox_id
+
+
+@mail_sender.store_emails_test_decorator
+def test_break_loop_alias_as_mailbox(flask_client):
+    user = create_new_user()
+    alias = Alias.create_new_random(user)
+    user.default_mailbox.email = alias.email
+    Session.commit()
+    envelope = Envelope()
+    envelope.mail_from = random_email()
+    envelope.rcpt_tos = [alias.email]
+    msg = EmailMessage()
+    msg[headers.TO] = alias.email
+    msg[headers.SUBJECT] = random_string()
+    result = email_handler.handle(envelope, msg)
+    assert result == status.E525

tests/test_utils.py

@@ -66,7 +66,7 @@ def canonicalize_email_cases():
         yield (f"a@{domain}", f"a@{domain}")
         yield (f"a.b@{domain}", f"ab@{domain}")
         yield (f"a.b+c@{domain}", f"ab@{domain}")
-    yield (f"a.b+c@other.com", f"a.b+c@other.com")
+        yield ("a.b+c@other.com", "a.b+c@other.com")
 
 
 @pytest.mark.parametrize("dirty,clean", canonicalize_email_cases())

tests/utils.py

@@ -17,7 +17,7 @@ def create_new_user(email: Optional[str] = None, name: Optional[str] = None) ->
     if not email:
         email = f"user_{random_token(10)}@mailbox.test"
     if not name:
-        name = f"Test User"
+        name = "Test User"
     # new user has a different email address
     user = User.create(
         email=email,