test_email.py

#!/usr/bin/env python3

"""
Criticality: High
Checks that emails can be sent.
"""

from ipaddress import ip_address
from pathlib import Path
import random
import socket
import subprocess
import sys
import time

import spf

sys.path.append(str(Path(__file__).parents[1].resolve()))

# pylint: disable=wrong-import-position
from envsetup import utils as u  # noqa: E402


def check_listen() -> tuple:
    """Check that Postfix is listening on 127.0.0.1:25.

    :return: Exit return codes
    :rtype: tuple
    """

    warnings = 0
    errors = 0

    # get listening state from ss
    status, out = subprocess.getstatusoutput("ss -pant | grep master | grep ':25'")

    if status != 0 or ("127.0.0.1:25" not in out and "[::1]:25" not in out):
        u.error("Postfix is not listening on localhost:25")
        errors += 1

    u.success("Postfix is listening on localhost:25")

    return warnings, errors


def check_relay(relay: str, domain: str) -> tuple:
    """Check that Postfix is not an open relay.

    :param relay: Hostname or IP address of relay host
    :type relay: str
    :param domain: Domain name under which mails will be send
    :type domain: str
    :return: Exit return codes
    :rtype: tuple
    """

    warnings = 0
    errors = 0

    # get relayhost value from Postfix config
    status, out = subprocess.getstatusoutput("grep relayhost /etc/postfix/main.cf")

    if status == 0:
        configured_relay = (
            out.replace("relayhost", "").strip(" \t=").replace("[", "").replace("]", "")
        )
    else:
        configured_relay = ""

    if not configured_relay:
        # check domain origin
        if Path("/etc/mailname").exists():
            with open("/etc/mailname", "r") as mailname:
                myorigin = mailname.read().strip()
        else:
            out = subprocess.getoutput("grep myorigin /etc/postfix/main.cf")
            myorigin = out.replace("myorigin", "").strip()
        # possible origin names
        origins = set(
            (domain or None, u.exec_cmd("hostname", log_output=False)[1] or None)
        )
        if myorigin not in origins:
            u.warning('"myorigin" setting does not contain a valid domain')
            warnings += 1

    if relay != configured_relay:
        u.error("STMP relay must be {}".format(relay))
        errors += 1

    if not errors and not warnings:
        u.success("STMP relay is properly set")

    return warnings, errors


def check_send(sender: str) -> tuple:
    """Check that Postfix can send email.

    :param sender: Sender mail address
    :type sender: str
    :return: Exit return codes
    :rtype: tuple
    """

    warnings = 0
    errors = 0

    # send email
    email = "noreply+{}-{}@ubicast.eu".format(time.time(), random.randint(0, 1000))
    if sender:
        sender = "-a 'From: {}' ".format(sender)
    else:
        u.warning("Sender address is not set")
        warnings += 1
    cmd = "echo 'test email' | mail -s 'Email used to test configuration.' {}{}".format(
        sender, email
    )
    subprocess.getoutput(cmd)

    # init vars
    timeout = 120
    waited = 1
    delay = 2
    out = ""

    # find logs
    if Path("/var/log/mail.log").is_file():
        cmd = "grep '{}' /var/log/mail.log".format(email)
    else:
        u.info("/var/log/mail.log not found, trying journalctl")
        cmd = "journalctl -t postfix/smtp | grep {}".format(email)

    # logs polling
    time.sleep(1)
    while True:
        status, out = subprocess.getstatusoutput(cmd)
        # found
        if status == 0:
            out = out.strip().split("\n")[-1]
            if "status=deferred" not in out:
                break
        # timeout
        if waited >= timeout:
            u.error("Failed to send email.")
            u.info("> no log entry found using command: {}".format(cmd))
            errors += 1
            break
        # wait before next iteration
        time.sleep(delay)
        waited += delay
        delay *= 2

    # check output for errors
    if "bounced" in out or "you tried to reach does not exist" in out:
        u.error("Failed to send email")
        u.info("> sending log line:\n{}".format(out))
        errors += 1

    if not errors:
        u.success("Can send email")

    return warnings, errors


def check_spf(ip_addr: str, sender: str, domain: str) -> tuple:
    """Check that SPF records passes.

    :param ip_addr: Host ip address of server or relay
    :type ip_addr: str
    :param sender: Sender mail address
    :type sender: str
    :param domain: Domain name under which mails will be send
    :type domain: str
    :return: Exit return codes
    :rtype: tuple
    """

    warnings = 0
    errors = 0

    if ip_address(ip_addr).is_private:
        u.info("{} is a private address, cannot check SPF".format(ip_addr))
    elif ip_addr and sender:
        # check spf
        result, _ = spf.check2(i=ip_addr, s=domain, h="")
        if result in ("pass", "neutral"):
            u.success("SPF for {} in {}: {}".format(ip_addr, domain, result))
        elif result == "none":
            u.info("SPF for {} in {}: {}".format(ip_addr, domain, result))
        else:
            u.warning("SPF for {} in {}: {}".format(ip_addr, domain, result))
            warnings += 1
    else:
        u.info("IP or sender not set, cannot check SPF")

    return warnings, errors


def main():
    """Run all checks and exits with corresponding exit code."""

    warnings = 0
    errors = 0

    print("Checking email settings:")

    if not Path("/etc/postfix").exists():
        u.info("postfix is not installed")
        exit(2)

    # get settings
    conf = u.load_conf()
    relay = conf.get("EMAIL_SMTP_SERVER", "").replace("[", "").replace("]", "")
    ip_addr = (
        (socket.gethostbyname(relay) if relay else None)
        or conf.get("NETWORK_IP_NAT")
        or conf.get("NETWORK_IP")
        or u.get_ip()
        or None
    )
    sender = conf.get("EMAIL_SENDER") or ""
    if not sender and Path("/etc/postfix/generic").exists():
        with open("/etc/postfix/generic") as sender_fo:
            sender = sender_fo.readline().split()[-1]
    domain = sender.split("@")[-1] or None

    # check that we are not an open relay
    check_warn, check_err = check_listen()
    warnings += check_warn if check_warn else warnings
    errors += check_err if check_err else errors

    # check that relayhost is correct
    check_warn, check_err = check_relay(relay, domain)
    warnings += check_warn if check_warn else warnings
    errors += check_err if check_err else errors

    # check that we can send emails
    check_warn, check_err = check_send(sender)
    warnings += check_warn if check_warn else warnings
    errors += check_err if check_err else errors

    # check that spf record is correct
    check_warn, check_err = check_spf(ip_addr, sender, domain)
    warnings += check_warn if check_warn else warnings
    errors += check_err if check_err else errors

    if errors:
        exit(1)
    elif warnings:
        exit(3)

    exit(0)


if __name__ == "__main__":
    main()