cqrenet/relay
3
0
Fork 0
mirror of https://github.com/chatmail/relay.git synced 2026-05-10 16:04:37 +00:00
Code Issues Packages Projects Releases Wiki Activity

Turn filtermail into a beforequeue handler and implement rate limit

Browse Source
This commit is contained in:
link2xt
2023-10-19 03:04:00 +00:00
committed by GitHub
parent 329b845c79 1cdc5d1351
commit 89311063f8
7 changed files with 128 additions and 123 deletions
Download Patch File Download Diff File Expand all files Collapse all files

123
chatmaild/src/chatmaild/filtermail.py
View File

@@ -1,12 +1,14 @@
#!/usr/bin/env python3
import asyncio
import logging
import time
import sys
from email.parser import BytesParser
from email import policy
from email.utils import parseaddr
from aiosmtpd.lmtp import LMTP
from aiosmtpd.controller import UnixSocketController
from aiosmtpd.smtp import SMTP
from aiosmtpd.controller import Controller
from smtplib import SMTP as SMTPClient
@@ -32,94 +34,93 @@ def check_encrypted(message):
return True
class ExampleController(UnixSocketController):
class SMTPController(Controller):
def factory(self):
return LMTP(self.handler, **self.SMTP_kwargs)
return SMTP(self.handler, **self.SMTP_kwargs)
class ExampleHandler:
async def handle_RCPT(self, server, session, envelope, address, rcpt_options):
envelope.rcpt_tos.append(address)
class BeforeQueueHandler:
def __init__(self):
self.send_rate_limiter = SendRateLimiter()
async def handle_MAIL(self, server, session, envelope, address, mail_options):
logging.info(f"handle_MAIL from {address}")
envelope.mail_from = address
if not self.send_rate_limiter.is_sending_allowed(address):
return f"450 4.7.1: Too much mail from {address}"
parts = envelope.mail_from.split("@")
if len(parts) != 2:
return f"500 Invalid from address <{envelope.mail_from!r}>"
return "250 OK"
async def handle_DATA(self, server, session, envelope):
valid_recipients, res = lmtp_handle_DATA(envelope)
# Reinject the mail back into Postfix.
if valid_recipients:
logging.info("Reinjecting the mail")
client = SMTPClient("localhost", "10026")
client.sendmail(envelope.mail_from, valid_recipients, envelope.content)
else:
logging.info("no valid recipients, ignoring mail")
return "\r\n".join(res)
logging.info("handle_DATA before-queue")
error = check_DATA(envelope)
if error:
return error
logging.info("re-injecting the mail that passed checks")
client = SMTPClient("localhost", "10025")
client.sendmail(envelope.mail_from, envelope.rcpt_tos, envelope.content)
return "250 OK"
async def asyncmain(loop):
controller = ExampleController(
ExampleHandler(), unix_socket="/var/spool/postfix/private/filtermail"
)
controller.start()
async def asyncmain_beforequeue(port):
Controller(BeforeQueueHandler(), hostname="127.0.0.1", port=port).start()
def lmtp_handle_DATA(envelope):
def check_DATA(envelope):
"""the central filtering function for e-mails."""
logging.info(f"Processing DATA message from {envelope.mail_from}")
message = BytesParser(policy=policy.default).parsebytes(envelope.content)
mail_encrypted = check_encrypted(message)
valid_recipients = []
res = []
_, from_addr = parseaddr(message.get("from").strip())
logging.info(f"mime-from: {from_addr} envelope-from: {envelope.mail_from!r}")
if envelope.mail_from.lower() != from_addr.lower():
return f"500 Invalid FROM <{from_addr!r}> for <{envelope.mail_from!r}>"
envelope_from_domain = from_addr.split("@").pop()
for recipient in envelope.rcpt_tos:
my_local_domain = envelope.mail_from.split("@")
if len(my_local_domain) != 2:
res += [f"500 Invalid from address <{envelope.mail_from}>"]
continue
_, from_addr = parseaddr(message.get("from").strip())
logging.info(f"mime-from: {from_addr} envelope-from: {envelope.mail_from}")
if envelope.mail_from.lower() != from_addr.lower():
res += [f"500 Invalid FROM <{from_addr}> for <{envelope.mail_from}>"]
continue
if envelope.mail_from == recipient:
# Always allow sending emails to self.
valid_recipients += [recipient]
res += ["250 OK"]
continue
res = recipient.split("@")
if len(res) != 2:
return f"500 Invalid address <{recipient}>"
_recipient_addr, recipient_domain = res
recipient_local_domain = recipient.split("@")
if len(recipient_local_domain) != 2:
res += [f"500 Invalid address <{recipient}>"]
continue
is_outgoing = recipient_domain != envelope_from_domain
if is_outgoing and not mail_encrypted:
is_securejoin = message.get("secure-join") in ["vc-request", "vg-request"]
if not is_securejoin:
return f"500 Invalid unencrypted mail to <{recipient}>"
is_outgoing = recipient_local_domain[1] != my_local_domain[1]
if (
is_outgoing
and not mail_encrypted
and message.get("secure-join") != "vc-request"
and message.get("secure-join") != "vg-request"
):
res += ["500 Outgoing mail must be encrypted"]
continue
class SendRateLimiter:
MAX_USER_SEND_PER_MINUTE = 80
valid_recipients += [recipient]
res += ["250 OK"]
def __init__(self):
self.addr2timestamps = {}
assert len(envelope.rcpt_tos) == len(res)
assert len(valid_recipients) <= len(res)
return valid_recipients, res
def is_sending_allowed(self, mail_from):
last = self.addr2timestamps.setdefault(mail_from, [])
now = time.time()
last[:] = [ts for ts in last if ts >= (now - 60)]
if len(last) <= self.MAX_USER_SEND_PER_MINUTE:
last.append(now)
return True
return False
def main():
args = sys.argv[1:]
assert len(args) == 1
logging.basicConfig(level=logging.INFO)
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
loop.create_task(asyncmain(loop=loop))
task = asyncmain_beforequeue(port=int(args[0]))
loop.create_task(task)
loop.run_forever()
if __name__ == "__main__":
main()

4
chatmaild/src/chatmaild/filtermail.service
View File

@@ -1,8 +1,8 @@
[Unit]
Description=Email filter for chatmail servers
Description=Chatmail Postfix BeforeQeue filter
[Service]
ExecStart=/usr/local/bin/filtermail
ExecStart=/usr/local/bin/filtermail 10080
Restart=always
RestartSec=30

24
chatmaild/src/chatmaild/test_filtermail.py
View File

@@ -1,6 +1,7 @@
from .filtermail import check_encrypted, lmtp_handle_DATA
from .filtermail import check_encrypted, check_DATA, SendRateLimiter
from email.parser import BytesParser
from email import policy
import pytest
def test_reject_forged_from():
@@ -30,15 +31,12 @@ def test_reject_forged_from():
# test that the filter lets good mail through
envelope.content = makemail(envelope.mail_from).as_bytes()
valid_recipients, res = lmtp_handle_DATA(envelope=envelope)
assert valid_recipients == envelope.rcpt_tos
assert len(res) == 1 and "250" in res[0]
assert not check_DATA(envelope=envelope)
# test that the filter rejects forged mail
envelope.content = makemail("forged@c3.testrun.org").as_bytes()
valid_recipients, res = lmtp_handle_DATA(envelope=envelope)
assert not valid_recipients
assert len(res) == 1 and "500" in res[0]
error = check_DATA(envelope=envelope)
assert "500" in error
def test_filtermail():
@@ -326,3 +324,15 @@ def test_filtermail():
]
).encode()
)
def test_send_rate_limiter():
limiter = SendRateLimiter()
for i in range(100):
if limiter.is_sending_allowed("some@example.org"):
if i <= SendRateLimiter.MAX_USER_SEND_PER_MINUTE:
continue
pytest.fail("limiter didn't work")
else:
assert i == SendRateLimiter.MAX_USER_SEND_PER_MINUTE + 1
break

59
deploy-chatmail/src/deploy_chatmail/__init__.py
View File

@@ -34,43 +34,28 @@ def _install_chatmaild() -> None:
commands=[f"pip install --break-system-packages {remote_path}"],
)
files.put(
name="upload doveauth-dictproxy.service",
src=importlib.resources.files("chatmaild")
.joinpath("doveauth-dictproxy.service")
.open("rb"),
dest="/etc/systemd/system/doveauth-dictproxy.service",
user="root",
group="root",
mode="644",
)
systemd.service(
name="Setup doveauth-dictproxy service",
service="doveauth-dictproxy.service",
running=True,
enabled=True,
restarted=True,
daemon_reload=True,
)
files.put(
name="upload filtermail.service",
src=importlib.resources.files("chatmaild")
.joinpath("filtermail.service")
.open("rb"),
dest="/etc/systemd/system/filtermail.service",
user="root",
group="root",
mode="644",
)
systemd.service(
name="Setup filtermail service",
service="filtermail.service",
running=True,
enabled=True,
restarted=True,
daemon_reload=True,
)
for fn in (
"doveauth-dictproxy",
"filtermail",
):
files.put(
name=f"Upload {fn}.service",
src=importlib.resources.files("chatmaild")
.joinpath(f"{fn}.service")
.open("rb"),
dest=f"/etc/systemd/system/{fn}.service",
user="root",
group="root",
mode="644",
)
systemd.service(
name=f"Setup {fn} service",
service=f"{fn}.service",
running=True,
enabled=True,
restarted=True,
daemon_reload=True,
)
def _configure_opendkim(domain: str, dkim_selector: str) -> bool:

6
deploy-chatmail/src/deploy_chatmail/postfix/master.cf.j2
View File

@@ -32,7 +32,7 @@ submission inet n - y - - smtpd
-o smtpd_recipient_restrictions=
-o smtpd_relay_restrictions=permit_sasl_authenticated,reject
-o milter_macro_daemon_name=ORIGINATING
-o content_filter=filter:unix:private/filtermail
-o smtpd_proxy_filter=127.0.0.1:10080
smtps inet n - y - - smtpd
-o syslog_name=postfix/smtps
-o smtpd_tls_wrappermode=yes
@@ -47,7 +47,7 @@ smtps inet n - y - - smtpd
-o smtpd_recipient_restrictions=
-o smtpd_relay_restrictions=permit_sasl_authenticated,reject
-o milter_macro_daemon_name=ORIGINATING
-o content_filter=filter:unix:private/filtermail
-o smtpd_proxy_filter=127.0.0.1:10080
#628 inet n - y - - qmqpd
pickup unix n - y 60 1 pickup
cleanup unix n - y - 0 cleanup
@@ -76,5 +76,5 @@ scache unix - - y - 1 scache
postlog unix-dgram n - n - 1 postlogd
filter unix - n n - - lmtp
# Local SMTP server for reinjecting filered mail.
localhost:10026 inet n - n - 10 smtpd
localhost:10025 inet n - n - 10 smtpd
-o content_filter=

34
online-tests/test_0_basic.py
View File

@@ -20,7 +20,7 @@ def test_use_two_chatmailservers(cmfactory, maildomain2):
@pytest.mark.parametrize("forgeaddr", ["internal", "someone@example.org"])
def test_reject_forged_from(cmsetup, mailgen, lp, remote, forgeaddr):
def test_reject_forged_from(cmsetup, mailgen, lp, forgeaddr):
user1, user3 = cmsetup.gen_users(2)
lp.sec("send encrypted message with forged from")
@@ -36,17 +36,25 @@ def test_reject_forged_from(cmsetup, mailgen, lp, remote, forgeaddr):
print(f" {line}")
lp.sec("Send forged mail and check remote postfix lmtp processing result")
remote_log = remote.iter_output("journalctl -t postfix/lmtp")
user1.smtp.sendmail(from_addr=user1.addr, to_addrs=[user3.addr], msg=msg)
for line in remote_log:
# print(line)
if "500 invalid from" in line and user3.addr in line:
break
else:
pytest.fail("remote postfix/filtermail failed to reject message")
with pytest.raises(smtplib.SMTPException) as e:
user1.smtp.sendmail(from_addr=user1.addr, to_addrs=[user3.addr], msg=msg)
assert "500" in str(e.value)
# check that the logged in user (who sent the forged msg) got a non-delivery notice
for message in user1.imap.fetch_all_messages():
if "Invalid FROM" in message and addr_to_forge in message:
@pytest.mark.slow
def test_exceed_rate_limit(cmsetup, gencreds, mailgen):
"""Test that the per-account send-mail limit is exceeded."""
user1, user2 = cmsetup.gen_users(2)
mail = mailgen.get_encrypted(user1.addr, user2.addr)
for i in range(100):
print("Sending mail", str(i))
try:
user1.smtp.sendmail(user1.addr, [user2.addr], mail)
except smtplib.SMTPException as e:
if i < 80:
pytest.fail(f"rate limit was exceeded too early with msg {i}")
outcome = e.recipients[user2.addr]
assert outcome[0] == 450
assert b'4.7.1: Too much mail from' in outcome[1]
return
pytest.fail(f"forged From={addr_to_forge} did not cause non-delivery notice")
pytest.fail("Rate limit was not exceeded")

1
online-tests/test_0_login.py
View File

@@ -1,4 +1,5 @@
import pytest
import smtplib
def test_login_basic_functioning(imap_or_smtp, gencreds, lp):