Usando Python para obtener el estado de envío de Postfix para los mensajes con un message-id

El problema: tengo un sitio web que envía correos electrónicos. Los correos electrónicos (meta) se almacenan en una base de datos antes de ser enviados al Postfix MTA. Cada email que el sitio web envía tiene un message-id único, algo así como 159766775024.77.9154479190021576437@peterspython.com. Quiero comprobar si los correos electrónicos se envían realmente y añadir esta información a los registros de la base de datos (meta) de correos electrónicos.

Postfix tiene estados como:

• status=bounced
• status=deferred
• status=sent

Primero empiezo con "enviado", después puedo añadir otros estados.

Usar Python hace nuestra vida más fácil

No pude encontrar un programa que me diera esta información, así que decidí escribirlo yo mismo. Mi servidor es un servidor Linux (Ubuntu) con ISPConfig. Usamos el /var/log/mail.log para obtener la información.

Advertencia: las diferentes versiones de Postfix pueden tener diferentes formatos de línea de registro. También en mi servidor el correo electrónico saliente es comprobado por Amavis, un filtro de contenido (spam).

Por supuesto que usamos Python para esta tarea. También me gusta la programación Bash pero cuando las cosas se vuelven un poco más complejas es mejor cambiar a un lenguaje de más alto nivel. Yo uso una mezcla aquí. Comandos Linux para reducir el tiempo de procesamiento y el tamaño del programa, y código Python para hacer algo de filtrado y generar las líneas de resultado.

Grep, Sed, Cut

Linux tiene un número de comandos que reducen enormemente nuestros esfuerzos de programación, en este caso yo uso:

• Grep
• Sed
• Cortar

Grep tiene la opción '-f' donde las líneas del archivo A pueden ser filtradas por un patrón del archivo B. Sed es muy fácil de usar si conoces las expresiones regulares. Yo lo uso para comprimir múltiples espacios en un solo espacio, añadir dos puntos a una línea, eliminar partes de una línea. Cut Se usa para escoger uno o más campos de la línea (filtrada), o columnas de un archivo.

Postfix, colas y mail.log

Postfix utiliza colas y los mensajes se asignan a colas de espera. Cuando enviamos un mensaje a Postfix, este mensaje se comprueba y si se pasa, se pone en lo que llamaré el accept_queue. Vemos una línea como:

Aug 17 16:07:52 server8 postfix/cleanup[8062]: C9DC982EEB:  message-id=<159767327280.144.3674122625658819909@peterspython.com>

Aquí el accept_queue_id es 'C9DC982EEB'. Si todo va bien, después de algún tiempo, su mensaje se pone en lo que yo llamo el send_queue. ¡Esto es un poco confuso ya que la línea contiene 'status=sent' mientras que no ha sido enviado! En este caso la línea es:

Aug 17 16:07:53 server8 postfix/smtp[8063]: C9DC982EEB: to=<peterpm@xs4all.nl>, relay=127.0.0.1[127.0.0.1]:10026, delay=0.33, delays=0.01/0.01/0.01/0.3, dsn=2.0.0,  status=sent  (250 2.0.0 from  MTA(smtp:[127.0.0.1]:10027): 250 2.0.0 Ok:  queued as 2083482EEC)

Fíjese al final de esta línea: "queued as 2083482EEC)". Para comprobar si y cuando este mensaje fue aceptado por el servidor de correo receptor debemos ahora buscar send_queue_id '2083482EEC'. Si buscamos mail.log encontraremos una línea que no sólo contiene 'status=sent' sino también '<> accepted message', donde <> es el servidor de correo receptor, ejemplo:

Aug 17 16:07:55 server8 postfix/smtp[8067]: 2083482EEC: to=<peterpm@xs4all.nl>, relay=mx3.xs4all.nl[194.109.24.134]:25, delay=2, delays=0.01/0/1.6/0.44, dsn=2.0.0,  status=sent  (250 2.0.0 mxdrop307.xs4all.net accepted message  07HE7rrG015924)

El código

Para ejecutar los comandos Linux en Python utilizo:

os.system(command)

El flujo está en el método update_message_id_and_status(). El trabajo pesado es hecho por los comandos de Linux . Generan una serie de archivos temporales. Cuando tenemos suficientes datos, usamos el código Python para facilitar el procesamiento.

Cuando dejas la depuración activada, el programa muestra las dos primeras líneas de cada archivo que se crea. Algunas líneas en el fichero resultante message_id_status_file:

159767280175.144.12615794910844491932@peterspython.com  sent 2020-08-17 16:00:05
159767326022.144.11661923945211507822@peterspython.com  sent 2020-08-17 16:07:42
159767327280.144.3674122625658819909@peterspython.com  sent 2020-08-17 16:07:55

Aquí está el script Python :

import  datetime
import inspect
import os
import csv
import subprocess 

MAIL_LOG_FILE='/var/log/mail.log'
MAIL_LOG_FILE='./mail.log' 
TMP_FILE_DIR='/tmp'
POSTFIX_MESSAGE_ID_STATUS_FILE_DIR='.'
POSTFIX_MESSAGE_ID_STATUS_FILENAME='message_id_status.txt'

class  PostfixMessageIdStatus:

    def __init__(self):
        self.mail_log_file = MAIL_LOG_FILE
        self.message_id_status_file = os.path.join(POSTFIX_MESSAGE_ID_STATUS_FILE_DIR,  POSTFIX_MESSAGE_ID_STATUS_FILENAME)
        self.tmp_message_ids_file = os.path.join(TMP_FILE_DIR, 'tmp_message_ids.txt')
        self.tmp_accept_queue_queue_id_message_id_all_file = os.path.join(TMP_FILE_DIR, 'tmp_accept_queue_queue_id_message_id_all.txt')
        self.tmp_accept_queue_queue_id_message_id_file = os.path.join(TMP_FILE_DIR, 'tmp_accept_queue_queue_id_message_id.txt')
        self.tmp_accept_queue_ids_file = os.path.join(TMP_FILE_DIR, 'tmp_accept_queue_ids.txt')
        self.tmp_accept_queue_id_send_queue_id_file = os.path.join(TMP_FILE_DIR, 'tmp_accept_queue_id_send_queue_id.txt')
        self.tmp_send_queue_ids_file = os.path.join(TMP_FILE_DIR, 'tmp_send_queue_ids.txt')
        self.tmp_sent_lines_file = os.path.join(TMP_FILE_DIR, 'self.tmp_send_lines.txt')

        self.my_message_id_domains = [
            '@peterspython.com'
        ]
        self.class_name = 'PostfixMessageIdStatus'
        self.dbg = False

    def get_fname(self):
        frame = inspect.stack()[1]
        function_name = inspect.currentframe().f_back.f_code.co_name
        fname = self.class_name  +  '::'  +  function_name
        return fname

    def print_line_count_and_few_lines_of_file(self, f): 

        if not os.path.isfile(f):
            raise FileNotFoundError(fname, ': file {} does not exist'.format(f))
        result = subprocess.check_output(['wc', '-l', f]).decode("utf-8") 
        line_count = int(result.split(' ')[0]) 
        print('file: {}'.format(f))
        print('line_count: {}'.format(line_count))
        command = "head --lines=2 "  +  f
        os.system(command)

    def create_empty_message_id_status_file_if_not_exists(self, f):
        fname = self.get_fname()
        dbg = self.dbg

        if not os.path.exists(f):
            print(fname  +  ': file {} does not exist so create it'.format(f))
            open(f, 'a').close()

            if not os.path.exists(f):
                raise Exception(fname  +  ': file {} could not be created'.format(f))

    def copy_message_ids_from_message_id_status_file_into_tmp_message_ids_file(self):
        fname = self.get_fname()
        dbg = self.dbg
        if dbg:
            print(fname  +  '()')

        command = 'cut -d " " -f 1 '  +  self.message_id_status_file  +  ' > '  +  self.tmp_message_ids_file
        if dbg:
            print(fname  +  ': command = {}'.format(command))
        os.system(command)

    def filter_mail_log_to_get_all_accept_queue_id_and_message_ids(self):
        fname = self.get_fname()
        dbg = self.dbg
        if dbg:
            print(fname  +  '()')

        # only include our own message_ids using the domain part
        message_id_domains_filter = '-e '  +  ' -e '.join(self.my_message_id_domains)

        # do not remove ':' from the queue_id, we need this later
        command = "grep 'message-id=' "  +  self.mail_log_file  +  " | grep "  +  message_id_domains_filter  +  " | sed 's/\s\s*/ /g' | sed 's/message-id\=<//g' | sed 's/>//g' | cut -d ' ' -f 6,7 | sort -u > "  +  self.tmp_accept_queue_queue_id_message_id_all_file
        if dbg:
            print(fname  +  ': command = {}'.format(command))
        os.system(command)

    def filter_tmp_accept_queue_queue_id_message_id_all_file_with_tmp_message_ids_file(self):
        fname = self.get_fname()
        dbg = self.dbg
        if dbg:
            print(fname  +  '()')

        command = "grep -v -f "  +  self.tmp_message_ids_file  +  " "  +  self.tmp_accept_queue_queue_id_message_id_all_file  +  " > "  +  self.tmp_accept_queue_queue_id_message_id_file
        if dbg:
            print(fname  +  ': command = {}'.format(command))
        os.system(command)

    def create_tmp_accept_queue_ids_file(self):
        fname = self.get_fname()
        dbg = self.dbg
        if dbg:
            print(fname  +  '()')

        command = "cut -d ' ' -f 1 "  +  self.tmp_accept_queue_queue_id_message_id_all_file  +  " > "  +  self.tmp_accept_queue_ids_file
        if dbg:
            print(fname  +  ': command = {}'.format(command))
        os.system(command)

    def filter_mail_log_by_accept_queue_ids_and_queued_as_to_get_send_queue_ids(self):
        fname = self.get_fname()
        dbg = self.dbg
        if dbg:
            print(fname  +  '()')

        command = "grep -f "  +  self.tmp_accept_queue_ids_file  +  ' '   +  self.mail_log_file  +  " | grep 'status=sent' | grep 'queued as' | sed 's/to.*queued as/ /g' | sed 's/)//g' | sed 's/\s\s*/ /g' |  sed 's/$/:/g' | cut -d ' ' -f 6,7 > "  +  self.tmp_accept_queue_id_send_queue_id_file
        if dbg:
            print(fname  +  ': command = {}'.format(command))
        os.system(command)

    def create_tmp_send_queue_ids_file(self):
        fname = self.get_fname()
        dbg = self.dbg
        if dbg:
            print(fname  +  '()')

        command = "cut -d ' ' -f 2 "  +  self.tmp_accept_queue_id_send_queue_id_file  +  " > "  +  self.tmp_send_queue_ids_file
        if dbg:
            print(fname  +  ': command = {}'.format(command))
        os.system(command)

    def filter_mail_log_by_send_queue_ids_and_accepted_message_to_get_sent_lines(self):
        fname = self.get_fname()
        dbg = self.dbg
        if dbg:
            print(fname  +  '()')

        command = "grep -f "  +  self.tmp_send_queue_ids_file  +  ' '   +  self.mail_log_file  +  " | grep 'status=sent' | grep 'accepted message'| sed 's/\s\s*/ /g' > "  +  self.tmp_sent_lines_file
        if dbg:
            print(fname  +  ': command = {}'.format(command))
        os.system(command)

    def  update_message_id_and_status(self):
        fname = self.get_fname()
        dbg = self.dbg
        dbg = True

        if dbg:
            print(fname  +  '()')

        # we use grep, sed, cut to do heavy work

        self.create_empty_message_id_status_file_if_not_exists(self.message_id_status_file)

        self.copy_message_ids_from_message_id_status_file_into_tmp_message_ids_file()
        if dbg:
            self.print_line_count_and_few_lines_of_file(self.tmp_message_ids_file)

        # after accepting a message, postfix passes the message to cleanup
        # example postfix line:
        # Aug 15 14:51:47 server8 postfix/cleanup[12794]: 23EFC803C7:  message-id=<135717442.1972837.1597495901382@mail.yahoo.com>

        # get all  accept_queue  queue_id and message_id 
        self.filter_mail_log_to_get_all_accept_queue_id_and_message_ids()
        if dbg:
            self.print_line_count_and_few_lines_of_file(self.tmp_accept_queue_queue_id_message_id_all_file)

        # remove lines from tmp_accept_queue_queue_id_message_id_all_file that contain message_ids we already have
        self.filter_tmp_accept_queue_queue_id_message_id_all_file_with_tmp_message_ids_file()
        if dbg:
            self.print_line_count_and_few_lines_of_file(self.tmp_accept_queue_queue_id_message_id_file)

        # we now have a file with lines:  accept_queue_id  message_id
        # example:
        # E0EB882BF4: 159777360230.7.2488114074420651363@peterspython.com

        # create from the above a file with only  accept_queue_ids
        self.create_tmp_accept_queue_ids_file()
        if dbg:
            self.print_line_count_and_few_lines_of_file(self.tmp_accept_queue_ids_file)

        # we now have a file with  accept_queue_ids
        # example:
        # E0EB882BF4:

        # use this file to filter the  mail.log  file to get lines ending with 'queued as D80D582C38)'
        # Aug 15 15:00:01 server8 postfix/smtp[13089]: 82E3D82C26: to=<peterpm@xs4all.nl>, relay=127.0.0.1[127.0.0.1]:10026, delay=0.39, delays=0.03/0.01/0.01/0.33, dsn=2.0.0,  status=sent  (250 2.0.0 from  MTA(smtp:[127.0.0.1]:10027): 250 2.0.0 Ok:  queued as D80D582C38)
        self.filter_mail_log_by_accept_queue_ids_and_queued_as_to_get_send_queue_ids()
        if dbg:
            self.print_line_count_and_few_lines_of_file(self.tmp_accept_queue_id_send_queue_id_file)

        # we now have a file with  accept_queue_ids and  send_queue_ids
        # example:
        # 9168182C54: 8A6FF82CB8:

        # create from the above a file with only  send_queue_ids
        self.create_tmp_send_queue_ids_file()
        if dbg:
            self.print_line_count_and_few_lines_of_file(self.tmp_send_queue_ids_file)

        # we now have a file with  send_queue_ids
        # example:
        # 5346D82C08:

        # use this file to filter the  mail.log  file to get lines ending with 'accepted message'
        # Aug 15 15:00:03 server8 postfix/smtp[13029]: D80D582C38: to=<peterpm@xs4all.nl>, relay=mx1.xs4all.nl[194.109.24.132]:25, delay=2, delays=0.01/0.01/1.6/0.42, dsn=2.0.0,  status=sent  (250 2.0.0 mxdrop301.xs4all.net accepted message  07FD01qu014687)
        self.filter_mail_log_by_send_queue_ids_and_accepted_message_to_get_sent_lines()
        if dbg:
            self.print_line_count_and_few_lines_of_file(self.tmp_sent_lines_file)

        # we now have a file with sent lines
        # example:
        # Aug 16 07:00:04 server8 postfix/smtp[12993]: EB13A82DD3: to=<peterpm@xs4all.nl>, relay=mx1.xs4all.nl[194.109.24.132]:25, delay=2, delays=0.01/0.01/1.6/0.46, dsn=2.0.0,  status=sent  (250 2.0.0 mxdrop306.xs4all.net accepted message  07G50207012529)

        # from here we use python to process the sent lines and update our message_id and status file

        # slurp and process self.tmp_accept_queue_id_send_queue_id_file
         accept_queue_id2message_ids = {}
        with open(self.tmp_accept_queue_queue_id_message_id_file) as fh_csv:
            rows = csv.reader(fh_csv, delimiter=' ')
            for row in rows:
                 accept_queue_id2message_ids[row[0]] = row[1]
        if dbg:
            print(fname  +  ':  accept_queue_id2message_ids = {}'.format(accept_queue_id2message_ids))

        # slurp and process self.tmp_accept_queue_id_send_queue_id_file
         send_queue_id2accept_queue_ids = {}
        with open(self.tmp_accept_queue_id_send_queue_id_file) as fh_csv:
            rows = csv.reader(fh_csv, delimiter=' ')
            for row in rows:
                 send_queue_id2accept_queue_ids[row[1]] = row[0]
        if dbg:
            print(fname  +  ':  send_queue_id2accept_queue_ids = {}'.format(send_queue_id2accept_queue_ids))

        # iterate sent lines
        message_id_status_lines = []
        with open(self.tmp_sent_lines_file) as fh:
            for line in fh:
                line_parts = line.split(' ')

                log_dt = '2020 '  +  ' '.join(line_parts[:3])
                log_dt_obj =  datetime.datetime.strptime(log_dt, '%Y %b %d %H:%M:%S')
                log_dt_iso = log_dt_obj.strftime("%Y-%m-%d %H:%M:%S")

                 send_queue_id  = line_parts[5]
                if  send_queue_id  not in  send_queue_id2accept_queue_ids:
                    continue

                 accept_queue_id  =  send_queue_id2accept_queue_ids[send_queue_id]
                if  accept_queue_id  not in  accept_queue_id2message_ids:
                    continue

                message_id =  accept_queue_id2message_ids[accept_queue_id]

                message_id_status_line = '{} sent {}'.format(message_id, log_dt_iso)
                if dbg:
                    print(fname  +  ': {}'.format(message_id_status_line))

                message_id_status_lines.append(message_id_status_line)

        # update message_id_status_file
        if len(message_id_status_lines) > 0:
            with open(self.message_id_status_file, 'a') as fh:
                fh.write('\n'.join(message_id_status_lines)  +  '\n')

# start
postfix_message_id_status =  PostfixMessageIdStatus()
postfix_message_id_status.update_message_id_and_status()

Resumen

Ejecutar los comandos Linux desde Python hace que tu programa sea pequeño y rápido. Una vez que tenemos nuestros datos, usamos el código Python para hacer cosas que son más difíciles de hacer en Bash. Procesar el archivo mail.log es un reto, pero una vez que sabes lo que está pasando, no es tan difícil. Y finalmente tengo un archivo que contiene los message-ids que realmente han sido enviados, lo que significa, aceptados por el servidor de correo receptor. Puedo usar este archivo para actualizar los registros de correo electrónico (meta) en la base de datos.

Enlaces / créditos

Postfix
http://www.postfix.org/