refactor: create is_reply_email()

app/email_utils.py

@@ -731,3 +731,7 @@ def generate_reply_email() -> str:
     # use UUID as fallback
     reply_email = f"ra+{uuid4()}@{EMAIL_DOMAIN}"
     return reply_email
+
+
+def is_reply_email(address: str) -> bool:
+    return address.startswith("reply+") or address.startswith("ra+")

app/greylisting.py

@@ -5,6 +5,7 @@ from app.config import (
     MAX_ACTIVITY_DURING_MINUTE_PER_ALIAS,
     MAX_ACTIVITY_DURING_MINUTE_PER_MAILBOX,
 )
+from app.email_utils import is_reply_email
 from app.extensions import db
 from app.log import LOG
 from app.models import Alias, EmailLog, Contact
@@ -95,7 +96,7 @@ def greylisting_needed_reply_phase(reply_email: str) -> bool:
 
 def greylisting_needed(mail_from: str, rcpt_tos: [str]) -> bool:
     for rcpt_to in rcpt_tos:
-        if rcpt_to.startswith("reply+") or rcpt_to.startswith("ra+"):
+        if is_reply_email(rcpt_to):
             if greylisting_needed_reply_phase(rcpt_to):
                 return True
         else:

email_handler.py

@@ -103,6 +103,7 @@ from app.email_utils import (
     add_header,
     get_header_unicode,
     generate_reply_email,
+    is_reply_email,
 )
 from app.extensions import db
 from app.greylisting import greylisting_needed
@@ -1691,7 +1692,7 @@ def handle(envelope: Envelope) -> str:
 
         # Reply case
         # recipient starts with "reply+" or "ra+" (ra=reverse-alias) prefix
-        if rcpt_to.startswith("reply+") or rcpt_to.startswith("ra+"):
+        if is_reply_email(rcpt_to):
             LOG.debug("Reply phase %s(%s) -> %s", mail_from, msg["From"], rcpt_to)
             is_delivered, smtp_status = handle_reply(envelope, msg, rcpt_to)
             res.append((is_delivered, smtp_status))

tests/api/test_alias.py

@@ -1,6 +1,7 @@
 from flask import url_for
 
 from app.config import PAGE_LIMIT
+from app.email_utils import is_reply_email
 from app.extensions import db
 from app.models import User, ApiKey, Alias, Contact, EmailLog, Mailbox
 from tests.utils import login
@@ -609,3 +610,9 @@ def test_get_alias(flask_client):
     assert "enabled" in res
     assert "note" in res
     assert "pinned" in res
+
+
+def test_is_reply_email(flask_client):
+    assert is_reply_email("ra+abcd@test.org")
+    assert is_reply_email("reply+abcd@test.org")
+    assert not is_reply_email("abcd@test.org")