mirror of
https://github.com/chatmail/relay.git
synced 2026-05-19 12:28:06 +00:00
passes the test
This commit is contained in:
@@ -34,6 +34,41 @@ def check_encrypted(message):
|
|||||||
return True
|
return True
|
||||||
|
|
||||||
|
|
||||||
|
def check_mdn(message, envelope):
|
||||||
|
if len(envelope.rcpt_tos) != 1:
|
||||||
|
return False
|
||||||
|
|
||||||
|
for name in ["auto-submitted", "chat-version"]:
|
||||||
|
if not message.get(name):
|
||||||
|
return False
|
||||||
|
|
||||||
|
if message.get_content_type() != "multipart/report":
|
||||||
|
return False
|
||||||
|
|
||||||
|
body = message.get_body()
|
||||||
|
if body.get_content_type() != "text/plain":
|
||||||
|
return False
|
||||||
|
|
||||||
|
if list(body.iter_attachments()) or list(body.iter_parts()):
|
||||||
|
return False
|
||||||
|
|
||||||
|
# even with all mime-structural checks an attacker
|
||||||
|
# could try to abuse the subject or body to contain links or other
|
||||||
|
# annoyance -- we only check for http links for now
|
||||||
|
# and reasonable sizes
|
||||||
|
|
||||||
|
subject = message.get("subject")
|
||||||
|
if "http" in subject or len(subject) > 50:
|
||||||
|
return False # actually could serve as a flag for malicious attempt
|
||||||
|
|
||||||
|
text = body.get_payload()
|
||||||
|
# how long the read-receipt can become?
|
||||||
|
if len(text) > 500 or "http" in text:
|
||||||
|
return False
|
||||||
|
|
||||||
|
return True
|
||||||
|
|
||||||
|
|
||||||
class SMTPController(Controller):
|
class SMTPController(Controller):
|
||||||
def factory(self):
|
def factory(self):
|
||||||
return SMTP(self.handler, **self.SMTP_kwargs)
|
return SMTP(self.handler, **self.SMTP_kwargs)
|
||||||
@@ -82,6 +117,9 @@ def check_DATA(envelope):
|
|||||||
if envelope.mail_from.lower() != from_addr.lower():
|
if envelope.mail_from.lower() != from_addr.lower():
|
||||||
return f"500 Invalid FROM <{from_addr!r}> for <{envelope.mail_from!r}>"
|
return f"500 Invalid FROM <{from_addr!r}> for <{envelope.mail_from!r}>"
|
||||||
|
|
||||||
|
if not mail_encrypted and check_mdn(message, envelope):
|
||||||
|
return
|
||||||
|
|
||||||
envelope_from_domain = from_addr.split("@").pop()
|
envelope_from_domain = from_addr.split("@").pop()
|
||||||
for recipient in envelope.rcpt_tos:
|
for recipient in envelope.rcpt_tos:
|
||||||
if envelope.mail_from == recipient:
|
if envelope.mail_from == recipient:
|
||||||
|
|||||||
@@ -1,4 +1,4 @@
|
|||||||
from chatmaild.filtermail import check_encrypted, check_DATA, SendRateLimiter
|
from chatmaild.filtermail import check_encrypted, check_DATA, SendRateLimiter, check_mdn
|
||||||
import pytest
|
import pytest
|
||||||
|
|
||||||
|
|
||||||
@@ -41,8 +41,33 @@ def test_filtermail_encryption_detection(maildata):
|
|||||||
assert not check_encrypted(msg)
|
assert not check_encrypted(msg)
|
||||||
|
|
||||||
|
|
||||||
def test_filtermail_mdn_is_not_encrypted(maildata):
|
def test_filtermail_is_mdn(maildata, gencreds):
|
||||||
assert not check_encrypted(maildata("mdn.eml"))
|
from_addr = gencreds()[0]
|
||||||
|
to_addr = gencreds()[0] + ".other"
|
||||||
|
msg = maildata("mdn.eml", from_addr, to_addr)
|
||||||
|
|
||||||
|
class env:
|
||||||
|
mail_from = from_addr
|
||||||
|
rcpt_tos = [to_addr]
|
||||||
|
content = msg.as_bytes()
|
||||||
|
|
||||||
|
assert check_mdn(msg, env)
|
||||||
|
print(msg.as_string())
|
||||||
|
assert not check_DATA(env)
|
||||||
|
|
||||||
|
|
||||||
|
def test_filtermail_to_multiple_recipients_no_mdn(maildata, gencreds):
|
||||||
|
from_addr = gencreds()[0]
|
||||||
|
to_addr = gencreds()[0] + ".other"
|
||||||
|
thirdaddr = gencreds()[0]
|
||||||
|
msg = maildata("mdn.eml", from_addr, to_addr)
|
||||||
|
|
||||||
|
class env:
|
||||||
|
mail_from = from_addr
|
||||||
|
rcpt_tos = [to_addr, thirdaddr]
|
||||||
|
content = msg.as_bytes()
|
||||||
|
|
||||||
|
assert not check_mdn(msg, env)
|
||||||
|
|
||||||
|
|
||||||
def test_send_rate_limiter():
|
def test_send_rate_limiter():
|
||||||
|
|||||||
@@ -290,7 +290,7 @@ class Remote:
|
|||||||
def maildata(request, gencreds):
|
def maildata(request, gencreds):
|
||||||
datadir = conftestdir.joinpath("mail-data")
|
datadir = conftestdir.joinpath("mail-data")
|
||||||
|
|
||||||
def maildata(name, parsed=True, from_addr=None, to_addr=None):
|
def maildata(name, from_addr=None, to_addr=None):
|
||||||
if from_addr is None:
|
if from_addr is None:
|
||||||
from_addr = gencreds()[0]
|
from_addr = gencreds()[0]
|
||||||
if to_addr is None:
|
if to_addr is None:
|
||||||
|
|||||||
@@ -1,6 +1,6 @@
|
|||||||
Subject: Message opened
|
Subject: Message opened
|
||||||
From: <barbaz@c2.testrun.org>
|
From: <{from_addr}>
|
||||||
To: <foobar@c2.testrun.org>
|
To: <{to_addr}>
|
||||||
Date: Sun, 15 Oct 2023 16:43:25 +0000
|
Date: Sun, 15 Oct 2023 16:43:25 +0000
|
||||||
Message-ID: <Mr.78MWtlV7RAi.goCFzBhCYfy@c2.testrun.org>
|
Message-ID: <Mr.78MWtlV7RAi.goCFzBhCYfy@c2.testrun.org>
|
||||||
Auto-Submitted: auto-replied
|
Auto-Submitted: auto-replied
|
||||||
|
|||||||
Reference in New Issue
Block a user