Add get_mailbox_from_mail_from()

email_handler.py

@@ -43,7 +43,7 @@ from email.mime.multipart import MIMEMultipart
 from email.utils import formataddr, make_msgid
 from io import BytesIO
 from smtplib import SMTP, SMTPRecipientsRefused
-from typing import List, Tuple
+from typing import List, Tuple, Optional
 
 import aiosmtpd
 import aiospamc
@@ -914,6 +914,24 @@ async def handle_reply(envelope, smtp: SMTP, msg: Message, rcpt_to: str) -> (boo
     return True, "250 Message accepted for delivery"
 
 
+def get_mailbox_from_mail_from(mail_from: str, alias) -> Optional[Mailbox]:
+    """return the corresponding mailbox given the mail_from and alias
+    Usually the mail_from=mailbox.email but it can also be one of the authorized address
+    """
+    for mailbox in alias.mailboxes:
+        if mailbox.email == mail_from:
+            return mailbox
+
+        for address in mailbox.authorized_addresses:
+            if address.email == mail_from:
+                LOG.debug(
+                    "Found an authorized address for %s %s %s", alias, mailbox, address
+                )
+                return mailbox
+
+    return None
+
+
 def spf_pass(
     ip: str,
     envelope,

tests/test_email_handler.py

@@ -0,0 +1,34 @@
+from app.models import User, Alias, AuthorizedAddress
+from email_handler import get_mailbox_from_mail_from
+
+
+def test_get_mailbox_from_mail_from(flask_client):
+    user = User.create(
+        email="a@b.c",
+        password="password",
+        name="Test User",
+        activated=True,
+        commit=True,
+    )
+    alias = Alias.create(
+        user_id=user.id,
+        email="first@d1.test",
+        mailbox_id=user.default_mailbox_id,
+        commit=True,
+    )
+
+    mb = get_mailbox_from_mail_from("a@b.c", alias)
+    assert mb.email == "a@b.c"
+
+    mb = get_mailbox_from_mail_from("unauthorized@gmail.com", alias)
+    assert mb is None
+
+    # authorized address
+    AuthorizedAddress.create(
+        user_id=user.id,
+        mailbox_id=user.default_mailbox_id,
+        email="unauthorized@gmail.com",
+        commit=True,
+    )
+    mb = get_mailbox_from_mail_from("unauthorized@gmail.com", alias)
+    assert mb.email == "a@b.c"