Create send_email_with_rate_control(): same as send_email() but with rate control

app/config.py

@@ -254,3 +254,15 @@ with open(get_abs_path(DISPOSABLE_FILE_PATH), "r") as f:
 APPLE_API_SECRET = os.environ.get("APPLE_API_SECRET")
 # for Mac App
 MACAPP_APPLE_API_SECRET = os.environ.get("MACAPP_APPLE_API_SECRET")
+
+# maximal number of alerts that can be sent to the same email in 24h
+MAX_ALERT_24H = 4
+
+# When a reverse-alias receives emails from un unknown mailbox
+ALERT_REVERSE_ALIAS_UNKNOWN_MAILBOX = "reverse_alias_unknown_mailbox"
+
+# When a forwarding email is bounced
+ALERT_BOUNCE_EMAIL = "bounce"
+
+# When a forwarding email is detected as spam
+ALERT_SPAM_EMAIL = "spam"

app/email_utils.py

@@ -8,6 +8,7 @@ from email.utils import make_msgid, formatdate, parseaddr
 from smtplib import SMTP
 from typing import Optional
 
+import arrow
 import dkim
 from jinja2 import Environment, FileSystemLoader
 
@@ -24,10 +25,12 @@ from app.config import (
     POSTFIX_SUBMISSION_TLS,
     MAX_NB_EMAIL_FREE_PLAN,
     DISPOSABLE_EMAIL_DOMAINS,
+    MAX_ALERT_24H,
 )
 from app.dns_utils import get_mx_domains
+from app.extensions import db
 from app.log import LOG
-from app.models import Mailbox, User
+from app.models import Mailbox, User, SentAlert
 
 
 def render(template_name, **kwargs) -> str:
@@ -235,6 +238,43 @@ def send_email(
     smtp.sendmail(SUPPORT_EMAIL, to_email, msg_raw)
 
 
+def send_email_with_rate_control(
+    user: User,
+    alert_type: str,
+    to_email: str,
+    subject,
+    plaintext,
+    html=None,
+    bounced_email: Optional[Message] = None,
+) -> bool:
+    """Same as send_email with rate control over alert_type.
+    For now no more than _MAX_ALERT_24h alert can be sent in the last 24h
+
+    Return true if the email is sent, otherwise False
+    """
+    to_email = to_email.lower().strip()
+    one_day_ago = arrow.now().shift(days=-1)
+    nb_alert = (
+        SentAlert.query.filter_by(alert_type=alert_type, to_email=to_email)
+        .filter(SentAlert.created_at > one_day_ago)
+        .count()
+    )
+
+    if nb_alert > MAX_ALERT_24H:
+        LOG.error(
+            "%s emails were sent to %s in the last 24h, alert type %s",
+            nb_alert,
+            to_email,
+            alert_type,
+        )
+        return False
+
+    SentAlert.create(user_id=user.id, alert_type=alert_type, to_email=to_email)
+    db.session.commit()
+    send_email(to_email, subject, plaintext, html, bounced_email)
+    return True
+
+
 def get_email_local_part(address):
     """
     Get the local part from email

tests/test_email_utils.py

@@ -1,5 +1,6 @@
 from email.message import EmailMessage
 
+from app.config import MAX_ALERT_24H
 from app.email_utils import (
     get_email_domain_part,
     email_belongs_to_alias_domains,
@@ -7,6 +8,7 @@ from app.email_utils import (
     delete_header,
     add_or_replace_header,
     parseaddr_unicode,
+    send_email_with_rate_control,
 )
 from app.extensions import db
 from app.models import User, CustomDomain
@@ -101,3 +103,18 @@ def test_parseaddr_unicode():
         "pöstal",
         "abcd@gmail.com",
     )
+
+
+def test_send_email_with_rate_control(flask_client):
+    user = User.create(
+        email="a@b.c", password="password", name="Test User", activated=True
+    )
+    db.session.commit()
+
+    for _ in range(MAX_ALERT_24H + 1):
+        assert send_email_with_rate_control(
+            user, "test alert type", "abcd@gmail.com", "subject", "plaintext"
+        )
+    assert not send_email_with_rate_control(
+        user, "test alert type", "abcd@gmail.com", "subject", "plaintext"
+    )