sourcehut
/
lists.sr.ht
mirror of https://git.sr.ht/~sircmpwn/lists.sr.ht
1
0
Fork 0
Code Issues Projects Releases Wiki Activity
lists.sr.ht/listssrht-lmtp

360 lines
14 KiB
Python
Executable File
Raw Permalink Blame History

#!/usr/bin/env python3
from aiosmtpd.lmtp import SMTP, LMTP
from email.utils import parseaddr
from fnmatch import fnmatch
from grp import getgrnam
from listssrht.types.listaccess import ListAccess
from prometheus_client import Counter, start_http_server
from srht.config import cfg
import asyncio
import asyncpg
import base64
import email
import os
import signal
import sys
from listssrht.process import dispatch_message, send_error_for
loop = asyncio.new_event_loop()
emails_processed = Counter("emails_processed",
"Incremented on every incoming email")
user_errors_processed = Counter("user_errors_processed",
"Incremented on every email which results in an error response")
forwards_processed = Counter("forwards_processed",
"Incremented on every email which is forwarded to lists.sr.ht")
commands_processed = Counter("commands_processed",
"Incremented on every command email, e.g. +subscribe")
always_reject = cfg("lists.sr.ht::worker", "reject-mimetypes")
always_reject = always_reject.split(",")
posting_domain = cfg("lists.sr.ht", "posting-domain")
html_error = """Hi {}!
We received your email, but were unable to deliver it because it
contains HTML. HTML emails are not permitted. The following guide can
help you configure your client to send in plain text instead:
https://useplaintext.email
If you have any questions, please reply to this email to reach the mail
admin. We apologise for the inconvenience.
"""
forbidden_mimetype_error = """Hi {}!
We received your email, but were unable to deliver it because it
contains content which has been blacklisted by the list admin. Please
remove your {} attachments and send again.
You are also advised to configure your email client to send emails in
plain text to avoid additional errors in the future:
https://useplaintext.email
If you have any questions, please reply to this email to reach the mail
admin. We apologise for the inconvenience.
"""
text_plain_required_error = """Hi {}!
We received your email, but were unable to deliver it because there were
no text/plain parts. Our mail system requires all emails to have at
least one plain text part. The following guide can help you configure
your client to send in plain text:
https://useplaintext.email
If you have any questions, please reply to this email to reach the mail
admin. We apologise for the inconvenience.
"""
unknown_mailing_list_error = """Hi {}!
We received your email, but were unable to deliver it because the
mailing list you wrote to was not found. The correct posting addresses
are:
~username/list-name@{}
Or if your mail system has trouble sending to addresses with ~ or / in
them, you can use:
u.username.list-name@{}
If your mail system does not support our normal posting addresses, we
would appreciate it if you wrote to your mail admin to ask them to fix
their system. Our posting addresses are valid per RFC-5322.
If you have any questions, please reply to this email to reach the mail
admin. We apologise for the inconvenience.
"""
class MailHandler:
def __init__(self, pg):
self.pg = pg
async def fetch_user(self, conn):
return await conn.prepare(
'''SELECT "id" FROM "user"
WHERE username = $1''')
async def fetch_user_by_email(self, conn):
return await conn.prepare(
'''SELECT "id" FROM "user"
WHERE email = $1''')
async def fetch_list(self, conn):
return await conn.prepare(
'''SELECT
"id",
"owner_id",
"default_access",
"permit_mimetypes",
"reject_mimetypes"
FROM "list"
WHERE "owner_id" = $1 AND "name" ILIKE $2''')
async def fetch_subscription(self, conn):
return await conn.prepare(
'''SELECT "id" FROM "subscription"
WHERE (email IS NOT NULL AND email = $1) or
(user_id IS NOT NULL AND user_id = $2)''')
async def fetch_email(self, conn):
return await conn.prepare(
'''SELECT "list_id" FROM "email"
WHERE "message_id" = $1''')
async def fetch_acl_by_email(self, conn):
return await conn.prepare(
'''SELECT "permissions" FROM "access"
WHERE list_id = $1 AND email = $2''')
async def fetch_acl_by_user(self, conn):
return await conn.prepare(
'''SELECT "permissions" FROM "access"
WHERE list_id = $1 AND user_id = $2''')
async def lookup_destination(self, conn, address):
"""Looks up the list this message is addressed to and returns its ID."""
# Note: we assume postfix took care of the domain
address = address[:address.rfind("@")]
command = "post"
if "+" in address:
command = address[address.rfind("+") + 1:]
address = address[:address.rfind("+")]
if not command in ["subscribe", "unsubscribe", "post"]:
return None, None
# Get redirect if present
address = cfg("lists.sr.ht::redirects", address, default=address)
if address.startswith("~"):
# TODO: user groups
if not "/" in address:
return None, None
owner, list_name = address.split("/")
else:
address = address.split(".")
if len(address) < 3:
return None, None
prefix = address[0]
owner = address[1]
list_name = '.'.join(address[2:])
if prefix == "u":
owner = "~" + owner
else:
# TODO: user groups
return None, None
fetch_user = await self.fetch_user(conn)
owner_id = await fetch_user.fetchval(owner[1:])
if not owner_id:
return None, None
fetch_list = await self.fetch_list(conn)
result = await fetch_list.fetchrow(owner_id, list_name.replace('_', '\\_'))
return result, command
def validate(self, mail, permit_mimetypes, reject_mimetypes):
required_headers = ["From", "Subject", "Message-Id"]
prohibited_headers = ["Return-Receipt-To", "Disposition-Notification-To"]
possible_to_headers = ["To", "Cc"]
found_to_header = False
for header in possible_to_headers:
if mail.get(header):
found_to_header = True
break
if not found_to_header:
return "The To or Cc header is required."
for header in required_headers:
if not mail.get(header):
return "The {} header is required.".format(header)
for header in prohibited_headers:
if mail.get(header):
return "The {} header is prohibited.".format(header)
found_textpart = False
sender = parseaddr(mail["From"])
sender = sender[0] or sender[1]
permit_mimetypes = permit_mimetypes.split(",")
reject_mimetypes = reject_mimetypes.split(",") + always_reject
# check the email for plaintext part
for part in mail.walk():
content_type = part.get_content_type()
if content_type == "text/plain":
found_textpart = True
break
# check if the email is permitted or not
# rejection occur if any forbidden mimetype is present or if any
# non-permitted part is present, except for text/html which is ignored
# if there is also a text/plain part and it is not forbidden
for part in mail.walk():
content_type = part.get_content_type()
if fnmatch(content_type, "multipart/*"):
continue
# check whether the type is permitted
permit = False
for whitelist in permit_mimetypes:
if fnmatch(content_type, whitelist):
permit = True
break
if not permit:
if content_type == "text/html" and not found_textpart:
return html_error.format(sender)
else:
return forbidden_mimetype_error.format(sender, content_type)
for blacklist in reject_mimetypes:
if fnmatch(content_type, blacklist):
if content_type == "text/html":
return html_error.format(sender)
else:
return forbidden_mimetype_error.format(
sender, content_type)
if not found_textpart:
return text_plain_required_error.format(sender)
return None
async def handle_RCPT(self, server, session,
envelope, address, rcpt_options):
print("RCPT {}".format(address))
envelope.rcpt_tos.append(address)
return "250 OK"
async def handle_DATA(self, server, session, envelope):
emails_processed.inc()
async with self.pg.acquire() as conn:
return await self.handle_DATA_w_conn(
server, session, envelope, conn)
async def handle_DATA_w_conn(self, server, session, envelope, conn):
address = envelope.rcpt_tos[0]
print(f"DATA for {address}")
# Python's email module will refuse to return 8bit data when converting
# a message to string. Celery will refuse to transmit bytes, it only
# accepts strings. To avoid mutating the message, convert to base64.
mail_b64 = base64.b64encode(envelope.content).decode("ascii")
mail = email.message_from_bytes(envelope.content,
policy=email.policy.SMTPUTF8.clone(max_line_length=998))
dest, command = await self.lookup_destination(conn, address)
if dest is None:
sender = parseaddr(mail["From"])
sender = sender[0] or sender[1]
print("Rejected, mailing list not found")
print(envelope)
user_errors_processed.inc()
send_error_for.delay(mail_b64, unknown_mailing_list_error.format(
sender, posting_domain, posting_domain))
return "250 Mailing list not found, but sending bounce out of band"
(dest_id, owner_id, default_access,
permit_mimetypes, reject_mimetypes) = dest
default_access = ListAccess(default_access)
fetch_email = await self.fetch_email(conn)
in_reply_to = mail.get("In-Reply-To")
in_reply_to = await fetch_email.fetchval(in_reply_to)
access = ListAccess.reply if in_reply_to == dest_id else ListAccess.post
fetch_user_by_email = await self.fetch_user_by_email(conn)
_from = parseaddr(mail["From"])
user_id = await fetch_user_by_email.fetchval(_from[1])
fetch_acl_by_user = await self.fetch_acl_by_user(conn)
fetch_acl_by_email = await self.fetch_acl_by_email(conn)
if user_id:
acl = await fetch_acl_by_user.fetchrow(dest_id, user_id)
else:
acl = await fetch_acl_by_email.fetchrow(dest_id, _from[1])
if command != "post":
print("Command accepted: {}".format(mail.get("Subject")))
dispatch_message.delay(address, dest_id, mail_b64)
commands_processed.inc()
return "250 Message accepted for delivery"
err = self.validate(mail, permit_mimetypes, reject_mimetypes)
if err is not None:
print("Rejected due to validation errors")
send_error_for.delay(mail_b64, err)
user_errors_processed.inc()
return "250 Validation failed, but sending bounce out of band"
if owner_id == user_id:
print("Message accepted: {}".format(mail.get("Subject")))
dispatch_message.delay(address, dest_id, mail_b64)
forwards_processed.inc()
return "250 Message accepted for delivery"
if acl is not None:
if access not in ListAccess(acl[0]):
user_errors_processed.inc()
print("Rejected: your account is not allowed to post to this list")
return "500 Rejected. Your account is not allowed to post to this list."
else:
if access not in default_access:
user_errors_processed.inc()
print("Rejected: default ACL does not allow posting")
return "500 Rejected. You are not allowed to post to this list."
forwards_processed.inc()
print("Message accepted: {}".format(mail.get("Subject")))
dispatch_message.delay(address, dest_id, mail_b64)
return "250 Message accepted for delivery"
async def create_server():
pg = await asyncpg.create_pool(dsn=cfg("lists.sr.ht", "connection-string"))
handler = MailHandler(pg)
sock = cfg("lists.sr.ht::worker", "sock")
protocol = cfg("lists.sr.ht::worker", "protocol",
default="lmtp" if "/" in sock else "smtp")
if protocol == "smtp":
def serve():
return SMTP(handler, enable_SMTPUTF8=True)
else:
def serve():
return LMTP(handler, enable_SMTPUTF8=True)
if "/" in sock:
await loop.create_unix_server(serve, path=sock)
os.chmod(sock, 0o775)
sock_group = cfg("lists.sr.ht::worker", "sock-group", default=None)
if sock_group != None:
sock_gid = getgrnam(sock_group).gr_gid
os.chown(sock, os.getuid(), sock_gid)
print("Accepting LMTP connections")
else:
host, port = sock.split(":")
await loop.create_server(serve, host=host, port=int(port))
print("Accepting SMTP connections")
def sigint_handler():
print("Exiting due to SIGINT")
sys.exit(0)
loop.add_signal_handler(signal.SIGINT, sigint_handler)
print("Starting incoming mail daemon")
start_http_server(cfg("lists.sr.ht::worker", "lmtp-metrics-port", default=8006))
loop.run_until_complete(create_server())
loop.run_forever()
loop.close()
Reference in New Issue View Git Blame Copy Permalink