test_email.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
'''
Criticality: High
Check that emails can be sent.
'''
import os
import subprocess
import sys
import random
import time
import imp

try:
    import spf
except ImportError:
    subprocess.check_call(['apt-get', '-qq', '-y', 'install', 'python3-spf'])
    import spf
if subprocess.call(['which', 'netstat']) != 0:
    subprocess.check_call(['apt-get', '-qq', '-y', 'install', 'net-tools'])

YELLOW = '\033[93m'
GREEN = '\033[92m'
RED = '\033[91m'
DEF = '\033[0m'

os.chdir(os.path.dirname(__file__))
if not os.path.isfile('../utils.py'):
    print('The envsetup configuration was not found.')
    sys.exit(1)
else:
    es_utils = imp.load_source('es_utils', '../utils.py')
    conf = es_utils.load_conf()


def print_color(txt, col):
    print('%s%s%s' % (col, txt, DEF))


def print_yellow(txt):
    print_color(txt, YELLOW)


def print_red(txt):
    print_color(txt, RED)


def print_green(txt):
    print_color(txt, GREEN)


def check_listening_port():
    # check that postfix listens the port 25 correctly
    status, out = subprocess.getstatusoutput('netstat -pant | grep master | grep ":25"')
    if status != 0:
        print_red('The port 25 is not listened by postfix "master" process.')
        return 1
    print_green('Postfix "master" process is listening port 25 correctly.')
    if '127.0.0.1:25' not in out:
        print_red('Postfix "master" process is not listening address 127.0.0.1, please check postfix configuration.')
        print('Postfix should listen address 127.0.0.1 to be sure that this server cannot be used as an SMTP relay by external services.')
        return 1
    print_green('Postfix "master" process is listening address 127.0.0.1 correctly.')
    return 0


def check_relay():
    print('Checking if SMTP relay conforms to conf.')
    status, out = subprocess.getstatusoutput('grep relayhost /etc/postfix/main.cf')
    if status == 0:
        configured_relay = out[len('relayhost'):].strip(' \t=').replace('[', '').replace(']', '')
    else:
        configured_relay = ''
    if not configured_relay:
        # no relay configured, check relayless situations
        ip = conf.get('NETWORK_IP_NAT')
        if not ip:
            ip = conf.get('NETWORK_IP')
        if not ip:
            print_yellow('Cannot determine public IP address')
            return 3
        with open('/etc/mailname', 'r') as f:
            d = f.read().strip()
        if d not in ('ubicast.tv', 'ubicast.eu'):
            print_yellow('/etc/mailname does not contain ubicast.eu or ubicast.tv, mails will probably not be received on ubicast mailing lists')
            return 3
        # check spf
        result, explain = spf.check2(i=ip, s='support@ubicast.eu', h='')
        if result != 'pass':
            print_red('ip %s is not in ubicast.eu SPF and emails sent from support@ubicast.eu may be treated as spam' % ip)
            return 3
    conf_relay = conf.get('EMAIL_SMTP_SERVER', '').replace('[', '').replace(']', '')
    if conf_relay != configured_relay:
        print_red('Configured STMP relay (%s) does not match the expected value (%s).' % (configured_relay, conf_relay))
        return 3
    else:
        print_green('STMP relay is properly set.')
        return 0


def send_test_email():
    email = 'noreply+%s-%s@ubicast.eu' % (time.time(), random.randint(0, 1000))
    print('Sending test email to "%s".' % email)
    cmd = 'echo "This is a test email" | mail -s "Test email from `cat /etc/hostname`" %s' % email
    subprocess.getstatusoutput(cmd)
    timeout = 120
    waited = 1
    delay = 2
    print('Timeout to find email sending log is set to %s seconds.' % timeout)
    out = ''
    if os.path.isfile('/var/log/mail.log'):
        cmd = 'grep "%s" /var/log/mail.log' % email
    else:
        print('/var/log/mail.log not found, trying journalctl')
        cmd = 'journalctl -u postfix | grep %s' % email
    time.sleep(1)
    while True:
        status, out = subprocess.getstatusoutput(cmd)
        if status == 0:
            out = out.strip().split('\n')[-1]
            if 'status=deferred' not in out:
                print('Log entry found after %s seconds.' % waited)
                break
        if waited >= timeout:
            print_red('Failed to send email.')
            print('No log entry found after %s seconds using command: %s' % (waited, cmd))
            return 1
        print('No log entry found after %s seconds, waiting %s more seconds...' % (waited, delay))
        time.sleep(delay)
        waited += delay
        delay *= 2
    if 'bounced' not in out or 'The email account that you tried to reach does not exist.' in out:
        print_green('Email sent.')
        return 0
    else:
        print_red('Failed to send email.')
        print('Sending log line:\n%s' % out)
        return 1


if not os.path.exists('/etc/postfix'):
    print_red('Postfix dir does not exists, please install postfix.')
    sys.exit(1)

rc = check_listening_port()
if rc == 0:
    rc = check_relay()
    if send_test_email() == 1:
        rc = 1

sys.exit(rc)