app-MAIL-temp/tests/test_email_handler.py

269 lines
8.3 KiB
Python
Raw Normal View History

2022-04-06 12:51:04 +02:00
import random
from email.message import EmailMessage
2022-04-06 12:51:04 +02:00
from typing import List
2022-04-06 12:51:04 +02:00
import pytest
2022-03-18 15:44:07 +01:00
from aiosmtpd.smtp import Envelope
import email_handler
from app import config
2022-04-21 09:26:44 +02:00
from app.config import EMAIL_DOMAIN, ALERT_DMARC_FAILED_REPLY_PHASE
2022-04-06 12:51:04 +02:00
from app.db import Session
2022-03-18 15:44:07 +01:00
from app.email import headers, status
2022-03-30 16:09:17 +02:00
from app.email_utils import generate_verp_email
from app.models import (
Alias,
AuthorizedAddress,
IgnoredEmail,
EmailLog,
Notification,
2022-03-30 16:29:38 +02:00
VerpType,
2022-04-06 12:51:04 +02:00
Contact,
SentAlert,
)
from email_handler import (
get_mailbox_from_mail_from,
should_ignore,
is_automatic_out_of_office,
)
from tests.utils import load_eml_file, create_new_user
2020-09-28 17:41:16 +02:00
def test_get_mailbox_from_mail_from(flask_client):
user = create_new_user()
alias = Alias.create_new_random(user)
Session.commit()
2020-09-28 17:41:16 +02:00
mb = get_mailbox_from_mail_from(user.email, alias)
assert mb.email == user.email
2020-09-28 17:41:16 +02:00
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 == user.email
def test_should_ignore(flask_client):
assert should_ignore("mail_from", []) is False
assert not should_ignore("mail_from", ["rcpt_to"])
IgnoredEmail.create(mail_from="mail_from", rcpt_to="rcpt_to", commit=True)
assert should_ignore("mail_from", ["rcpt_to"])
def test_is_automatic_out_of_office():
msg = EmailMessage()
assert not is_automatic_out_of_office(msg)
msg[headers.AUTO_SUBMITTED] = "auto-replied"
assert is_automatic_out_of_office(msg)
del msg[headers.AUTO_SUBMITTED]
assert not is_automatic_out_of_office(msg)
msg[headers.AUTO_SUBMITTED] = "auto-generated"
assert is_automatic_out_of_office(msg)
2022-03-17 19:03:36 +01:00
2022-03-17 21:36:25 +01:00
2022-04-06 12:51:04 +02:00
def test_dmarc_forward_quarantine(flask_client):
user = create_new_user()
2022-03-21 12:03:11 +01:00
alias = Alias.create_new_random(user)
2022-03-21 12:31:25 +01:00
msg = load_eml_file("dmarc_quarantine.eml", {"alias_email": alias.email})
2022-03-18 15:44:07 +01:00
envelope = Envelope()
envelope.mail_from = msg["from"]
envelope.rcpt_tos = [msg["to"]]
result = email_handler.handle(envelope, msg)
2022-03-22 17:44:08 +01:00
assert result == status.E215
2022-03-18 15:44:07 +01:00
email_logs = (
EmailLog.filter_by(user_id=user.id, alias_id=alias.id)
.order_by(EmailLog.id.desc())
.all()
)
assert len(email_logs) == 1
email_log = email_logs[0]
assert email_log.blocked
assert email_log.refused_email_id
notifications = Notification.filter_by(user_id=user.id).all()
assert len(notifications) == 1
assert f"{alias.email} has a new mail in quarantine" == notifications[0].title
2022-04-08 11:28:14 +02:00
def test_gmail_dmarc_softfail(flask_client):
user = create_new_user()
2022-04-08 11:28:14 +02:00
alias = Alias.create_new_random(user)
msg = load_eml_file("dmarc_gmail_softfail.eml", {"alias_email": alias.email})
envelope = Envelope()
envelope.mail_from = msg["from"]
envelope.rcpt_tos = [msg["to"]]
result = email_handler.handle(envelope, msg)
assert result == status.E200
2022-04-11 09:28:57 +02:00
# Enable when we can verify that the actual message sent has this content
# payload = msg.get_payload()
# assert payload.find("failed anti-phishing checks") > -1
def test_prevent_5xx_from_spf(flask_client):
user = create_new_user()
alias = Alias.create_new_random(user)
msg = load_eml_file(
"5xx_overwrite_spf.eml",
{"alias_email": alias.email, "spf_result": "R_SPF_FAIL"},
)
envelope = Envelope()
2022-03-30 16:29:38 +02:00
envelope.mail_from = msg["from"]
2022-03-30 17:20:49 +02:00
# Ensure invalid email log
2022-03-30 16:29:38 +02:00
envelope.rcpt_tos = [generate_verp_email(VerpType.bounce_forward, 99999999999999)]
result = email_handler.MailHandler()._handle(envelope, msg)
2022-03-30 16:29:38 +02:00
assert status.E216 == result
def test_preserve_5xx_with_valid_spf(flask_client):
user = create_new_user()
alias = Alias.create_new_random(user)
msg = load_eml_file(
"5xx_overwrite_spf.eml",
{"alias_email": alias.email, "spf_result": "R_SPF_ALLOW"},
)
envelope = Envelope()
2022-03-30 16:29:38 +02:00
envelope.mail_from = msg["from"]
2022-03-30 17:20:49 +02:00
# Ensure invalid email log
2022-03-30 16:29:38 +02:00
envelope.rcpt_tos = [generate_verp_email(VerpType.bounce_forward, 99999999999999)]
result = email_handler.MailHandler()._handle(envelope, msg)
2022-03-30 16:29:38 +02:00
assert status.E512 == result
2022-03-29 15:59:35 +02:00
def test_preserve_5xx_with_no_header(flask_client):
user = create_new_user()
2022-03-29 15:59:35 +02:00
alias = Alias.create_new_random(user)
msg = load_eml_file(
"no_spamd_header.eml",
{"alias_email": alias.email},
)
envelope = Envelope()
2022-03-30 16:29:38 +02:00
envelope.mail_from = msg["from"]
2022-03-30 17:20:49 +02:00
# Ensure invalid email log
2022-03-30 16:29:38 +02:00
envelope.rcpt_tos = [generate_verp_email(VerpType.bounce_forward, 99999999999999)]
2022-03-29 15:59:35 +02:00
result = email_handler.MailHandler()._handle(envelope, msg)
2022-03-30 16:29:38 +02:00
assert status.E512 == result
2022-04-06 12:51:04 +02:00
def generate_dmarc_result() -> List:
return ["DMARC_POLICY_QUARANTINE", "DMARC_POLICY_REJECT", "DMARC_POLICY_SOFTFAIL"]
@pytest.mark.parametrize("dmarc_result", generate_dmarc_result())
2022-04-06 17:07:36 +02:00
def test_dmarc_reply_quarantine(flask_client, dmarc_result):
user = create_new_user()
2022-04-06 12:51:04 +02:00
alias = Alias.create_new_random(user)
Session.commit()
contact = Contact.create(
user_id=alias.user_id,
alias_id=alias.id,
website_email="random-{}@nowhere.net".format(int(random.random())),
name="Name {}".format(int(random.random())),
reply_email="random-{}@{}".format(random.random(), EMAIL_DOMAIN),
)
2022-04-06 17:07:36 +02:00
Session.commit()
2022-04-06 12:51:04 +02:00
msg = load_eml_file(
"dmarc_reply_check.eml",
{
"alias_email": alias.email,
"contact_email": contact.reply_email,
"dmarc_result": dmarc_result,
},
)
envelope = Envelope()
envelope.mail_from = msg["from"]
envelope.rcpt_tos = [msg["to"]]
result = email_handler.handle(envelope, msg)
assert result == status.E215
alerts = SentAlert.filter_by(
user_id=user.id, alert_type=ALERT_DMARC_FAILED_REPLY_PHASE
).all()
assert len(alerts) == 1
def test_add_alias_to_header_if_needed():
msg = EmailMessage()
user = create_new_user()
alias = Alias.filter_by(user_id=user.id).first()
assert msg[headers.TO] is None
email_handler.add_alias_to_header_if_needed(msg, alias)
assert msg[headers.TO] == alias.email
def test_append_alias_to_header_if_needed_existing_to():
msg = EmailMessage()
original_to = "noone@nowhere.no"
msg[headers.TO] = original_to
user = create_new_user()
alias = Alias.filter_by(user_id=user.id).first()
email_handler.add_alias_to_header_if_needed(msg, alias)
assert msg[headers.TO] == f"{original_to}, {alias.email}"
def test_avoid_add_to_header_already_present():
msg = EmailMessage()
user = create_new_user()
alias = Alias.filter_by(user_id=user.id).first()
msg[headers.TO] = alias.email
email_handler.add_alias_to_header_if_needed(msg, alias)
assert msg[headers.TO] == alias.email
def test_avoid_add_to_header_already_present_in_cc():
msg = EmailMessage()
create_new_user()
alias = Alias.first()
msg[headers.CC] = alias.email
email_handler.add_alias_to_header_if_needed(msg, alias)
assert msg[headers.TO] is None
assert msg[headers.CC] == alias.email
def test_email_sent_to_noreply(flask_client):
msg = EmailMessage()
envelope = Envelope()
envelope.mail_from = "from@domain.test"
envelope.rcpt_tos = [config.NOREPLY]
result = email_handler.handle(envelope, msg)
assert result == status.E200
def test_email_sent_to_noreplies(flask_client):
msg = EmailMessage()
envelope = Envelope()
envelope.mail_from = "from@domain.test"
config.NOREPLIES = ["other-no-reply@sl.test"]
envelope.rcpt_tos = ["other-no-reply@sl.test"]
result = email_handler.handle(envelope, msg)
assert result == status.E200
# NOREPLY isn't used anymore
envelope.rcpt_tos = [config.NOREPLY]
result = email_handler.handle(envelope, msg)
assert result == status.E515
def test_references_header(flask_client):
user = create_new_user()
alias = Alias.create_new_random(user)
msg = load_eml_file("reference_encoded.eml", {"alias_email": alias.email})
envelope = Envelope()
envelope.mail_from = "somewhere@rainbow.com"
envelope.rcpt_tos = [alias.email]
result = email_handler.handle(envelope, msg)
assert result == status.E200