A simple milter that will print callbacks and macros to stdout.

debug-milter.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-

"""
debug-milter.py

A super simple Mail Filter (milter) which will print callbacks and received
macros to stdout as they are sent by the mailserver.
Ideal to understanding what a mailserver is sending to a milter for debugging
purposes.

Copyright (C) 2023 Andreas Thienemann

This program is free software: you can redistribute it and/or modify
it under the terms of the GNU General Public License as published by
the Free Software Foundation, either version 3 of the License, or
(at your option) any later version.

This program is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
GNU General Public License for more details.

You should have received a copy of the GNU General Public License
along with this program.  If not, see <https://www.gnu.org/licenses/>.
"""


from __future__ import print_function
import Milter

sock = "inet:12345@localhost"
MACRO_LIST = [
    "i",
    "j",
    "_",
    "{auth_authen}",
    "{auth_author}",
    "{auth_type}",
    "{client_addr}",
    "{client_connections}",
    "{client_name}",
    "{client_port}",
    "{client_ptr}",
    "{cert_issuer}",
    "{cert_subject}",
    "{cipher_bits}",
    "{cipher}",
    "{daemon_addr}",
    "{daemon_name}",
    "{daemon_port}",
    "{mail_addr}",
    "{mail_host}",
    "{mail_mailer}",
    "{rcpt_addr}",
    "{rcpt_host}",
    "{rcpt_mailer}",
    "{tls_version}",
    "v",
]


class DebugMilter(Milter.Base):
    def __init__(self):  # A new instance with each new connection.
        self.id = Milter.uniqueID()  # Integer incremented with each call.
        self.debug = False

    def get_macros(self):
        global MACRO_LIST
        macros = {}
        for name in MACRO_LIST:
            macros.update({name: self.getsymval(name)})
        return {k: v for k, v in macros.items() if v is not None}

    @Milter.noreply
    def connect(self, hostname, family, hostaddr):
        print(self.id, "Connect Callback called", hostname, family, hostaddr)
        print(self.id, "Macros", self.get_macros())
        return Milter.CONTINUE

    @Milter.noreply
    def hello(self, hostname):
        print(self.id, "Hello Callback called", hostname)
        print(self.id, "Macros", self.get_macros())
        return Milter.CONTINUE

    @Milter.noreply
    def envfrom(self, from_addr, *params):
        print(self.id, "MAILFROM Callback called", from_addr, params)
        print(self.id, "Macros", self.get_macros())
        return Milter.CONTINUE

    @Milter.noreply
    def envrcpt(self, to_addr, *params):
        print(self.id, "RCPTO Callback called", to_addr, params)
        print(self.id, "Macros", self.get_macros())
        return Milter.CONTINUE

    @Milter.noreply
    def header(self, field, value):
        print(self.id, "Header Callback called", field, value)
        print(self.id, "Macros", self.get_macros())
        return Milter.CONTINUE

    @Milter.noreply
    def eoh(self):
        print(self.id, "EOH Callback called")
        print(self.id, "Macros", self.get_macros())
        return Milter.CONTINUE

    @Milter.noreply
    def body(self, chunk):
        print(self.id, "Body Callback called", chunk)
        print(self.id, "Macros", self.get_macros())
        return Milter.CONTINUE

    def eom(self):
        print(self.id, "EOM Callback called")
        print(self.id, "Macros", self.get_macros())
        return Milter.CONTINUE

    def abort(self):
        print(self.id, "Abort Callback called")
        print(self.id, "Macros", self.get_macros())
        return Milter.CONTINUE

    def close(self):
        print(self.id, "Close Callback called")
        print(self.id, "Macros", self.get_macros())
        return Milter.CONTINUE


def main():
    timeout = 600
    Milter.factory = DebugMilter
    flags = Milter.CHGHDRS + Milter.ADDHDRS
    Milter.set_flags(flags)
    Milter.runmilter("debugmilter", sock, timeout)


if __name__ == "__main__":
    main()