diff --git a/email_handler.py b/email_handler.py index 3e64249c..0bff0a5d 100644 --- a/email_handler.py +++ b/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, diff --git a/tests/test_email_handler.py b/tests/test_email_handler.py new file mode 100644 index 00000000..6f575a08 --- /dev/null +++ b/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"