path: root/server_receiver.py

#!/usr/bin/env python3
#
# Federated Electronic Memo Server written in Python.
#

from __future__ import annotations
from typing import Awaitable, Callable

from time import time
import certifi
import signal
import ssl
import os
import socket
import _thread
import logging

import ememo_common
import ememo_errors
import config


def parse_recipients_file(fn: str) -> dict[str, str]:
    local_recipients = {}
    with open(fn, "rb") as f:
        for line in f:
            recipient, raw_location = line.split(b" ", 1)
            location = raw_location.rstrip(b"\r\n").decode("utf-8")
            local_recipients[recipient.lower()] = location
    return local_recipients


def rehash(signal=None, callstack=None):
    global LOCAL_RECIPIENTS_LOWERCASED
    LOCAL_RECIPIENTS_LOWERCASED = parse_recipients_file("local-recipients.txt")


rehash()
signal.signal(signal.SIGHUP, rehash)
logging.basicConfig(level=logging.INFO)


def convey_error(conn, e):
    conn.sendall(b"E" + e.m + (b" " if e.args else b"") + b" ".join(e.args) + b"\n")


def read_specified_length(conn, expected_length):
    buffer = b""
    already_read_length = 0
    while already_read_length < expected_length:
        # TODO: A timeout is needed here to prevent slow lorris attacks and such
        newly_received = conn.recv(min(1024, expected_length - already_read_length))
        buffer += newly_received
        already_read_length += len(newly_received)
    assert already_read_length == expected_length
    return buffer


def parse_and_deliver_mail(conn, mail_buffer, incoming_server_acceptable_domains):
    received_time = str(time())
    headers, attachments_raw = ememo_common.parse_header_and_leave_rest(mail_buffer)
    # Process the parsed header data.
    try:
        from_ = headers[b"from"]
    except KeyError as e:
        print(e.args[0])
        raise ememo_errors.E_MISSING_HEADER(repr(e.args[0]).encode("utf-8"))

    to = headers.get(b"to", None)
    cc = headers.get(b"cc", None)
    bcc = headers.pop(b"bcc", None)
    mailer = headers.get(b"mailer", None)
    if not headers.get(b"message-id", None):
        raise ememo_errors.E_HEADER_ARGUMENT(b"message-id")

    if mailer:
        sender_tuple = ememo_common.parse_address(mailer)
    else:
        sender_tuple = ememo_common.parse_address(from_)

    if (
        sender_tuple[2].decode("utf-8", "surrogateescape")
        not in incoming_server_acceptable_domains
    ):
        raise ememo_errors.E_UNAUTHORIZED_SENDING_DOMAIN

    headers[b"received-server-time"] = received_time.encode("ascii")
    message_to_deliver = ememo_common.assemble_headers(headers) + attachments_raw
    print(ememo_common.assemble_headers(headers))
    print(attachments_raw)

    recipients = []
    if to:
        recipients += [ememo_common.parse_address(a) for a in to.split(b", ")]
    if cc:
        recipients += [ememo_common.parse_address(a) for a in cc.split(b", ")]
    if bcc:
        recipients += [ememo_common.parse_address(a) for a in bcc.split(b", ")]

    deliver_queue = []

    for r in recipients:
        addr = (r[1] + b"@" + r[2]).lower()
        if r[2] in config.MY_DOMAINS:
            try:
                deliver_queue.append((addr, LOCAL_RECIPIENTS_LOWERCASED[addr]))
                conn.send(b"R01 %s\n" % addr)
            except KeyError:
                conn.send(
                    b"User not found error: Local recipients table does not contain "
                    + addr
                    + b" but "
                    + r[2]
                    + b" is a local domain.\n"
                )
                conn.close()
                return
        elif addr in LOCAL_RECIPIENTS_LOWERCASED:
            deliver_queue.append((addr, LOCAL_RECIPIENTS_LOWERCASED[addr]))

    if not deliver_queue:
        raise ememo_errors.E_NOBODY_TO_DELIVER_TO

    for addr, loc in deliver_queue:
        if loc.startswith("|"):
            conn.send(b"Delivery to command standard input is not implemented.\n")
            conn.close()
            return
        elif loc.startswith("/"):
            try:
                with open(
                    os.path.join(loc, received_time) + ".emo",
                    "wb",
                ) as f:
                    f.write(message_to_deliver)
                    f.close()
            except Exception as e:
                logging.warning(repr(e) + " while trying to deliver mail.")
                conn.send(
                    b"Unable to deliver: Cannot write to mailbox.  Please contact this server's postmaster for details.\n"
                )
                conn.close()
                return
            else:
                conn.send(b"R02 %s.\n" % addr)
                logging.info("Delivered memo %s to %s." % (received_time, loc))
        else:
            raise ememo_errors.ServerFatal(
                "I do not understand the delivery location %s" % loc
            )


def handle_MAIL(conn, line_split, incoming_server_acceptable_domains):
    if len(line_split) > 2:
        raise ememo_errors.E_TOO_MANY_ARGUMENTS
    elif len(line_split) < 2:
        raise ememo_errors.E_TOO_FEW_ARGUMENTS
    try:
        incoming_mail_expected_length = int(line_split[1])
    except ValueError:
        raise ememo_errors.E_SIZE_NOT_NUMBER
    incoming_mail_buffer = read_specified_length(conn, incoming_mail_expected_length)
    parse_and_deliver_mail(
        conn, incoming_mail_buffer, incoming_server_acceptable_domains
    )


def process_command_line(conn, line, incoming_server_acceptable_domains):
    if not line.strip():
        return
    line_split = line.split(b" ")
    print(line_split)
    line_split[0] = line_split[0].upper()
    if line_split[0] == b"MAIL":
        handle_MAIL(conn, line_split, incoming_server_acceptable_domains)
    elif line_split[0] == b"BYE":
        raise ememo_errors.NORMAL_CONNECTION_TERMINATION
    else:
        raise ememo_errors.E_UNKNOWN_COMMAND


def connection_session(conn, addr):
    try:
        try:
            cert = conn.getpeercert()
            print(cert)
            incoming_server_acceptable_domains = [
                d[1] for d in cert.get("subjectAltName", [])
            ] + [dict(cert["subject"][0])["commonName"]]
        except TypeError:
            if not config.NO_CERTIFICATE_MODE:
                raise
            incoming_server_acceptable_domains = []

        ememo_errors.connection_state = 0
        # 0: listening for a command line (expecting_length should be set to None)
        # 1: listening for a bulk of text (expecting_length must be known and set)

        buffer = b""
        while True:
            newly_received = conn.recv(1024)
            if newly_received == b"":
                conn.close()
                return
            tmp = newly_received.split(b"\n")
            if len(tmp) < 2:
                buffer += newly_received
                continue
            tmp[0] = buffer + tmp[0]
            buffer = tmp[-1]
            for line in tmp[0:-1]:
                try:
                    process_command_line(conn, line, incoming_server_acceptable_domains)
                except ememo_errors.CommandFatal as e:
                    convey_error(conn, e)
    except ememo_errors.ConnectionFatal as e:
        convey_error(conn, e)
    except ememo_errors.NORMAL_CONNECTION_TERMINATION:
        conn.send(b"BYE\n")
    finally:
        try:
            conn.shutdown(socket.SHUT_RDWR)
        except OSError:
            pass
        conn.close()


def raw_socket_connection_session(c, addr):
    tls_context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
    if config.NO_CERTIFICATE_MODE:
        tls_context.verify_mode = ssl.CERT_OPTIONAL
    else:
        tls_context.verify_mode = ssl.CERT_REQUIRED
    tls_context.load_cert_chain(
        config.S2S_LISTEN_CERT,
        config.S2S_LISTEN_KEY,
    )
    tls_context.load_verify_locations(cafile=certifi.where())

    try:
        conn = tls_context.wrap_socket(c, server_side=True)
    except ssl.SSLError:
        # Probably it's because the client isn't using a valid (as in
        # PKI-valid) client certificate, just drop the conneciton.
        c.close()
        return None
    return connection_session(conn, addr)


def main():
    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    s.bind((config.S2S_LISTEN_ADDR, config.S2S_LISTEN_PORT))
    s.listen(config.S2S_LISTEN_BACKLOG)
    while True:
        c, addr = s.accept()
        logging.info("Socket connection from %s port %s." % (addr[0], addr[1]))
        _thread.start_new_thread(
            raw_socket_connection_session,
            (c, addr),
        )
    s.close()


if __name__ == "__main__":
    main()