Python 100 project #35: Catch All Mail Server

Most of the network device has some kind of notification system, and usually notification media is either Email or SNMP. Email is way too slow if anything really happens, while SNMP the function is limited and not flexible enough. So in the later phase, I want to get those email notification posted to Slack. I can use the email integration of the Slack itself, but I decided to make one by myself so that the service can be utilized for other services as well.

First I made a simple SMTP server to catch all the email. This project simply print out the email body it receives.

Output Example:

$ python3 simple_smtpserver.py 
server running...
("<html><head><meta http-equiv='Content-Type' content='text/html; "
 "charset=UTF-8'/><title></title></head><body> Hooray!!<br><br>mail server is "
 'now working!</body></html>')
from: SEGFW_INT@Home.local

$ python3 simple_smtpserver.py

server running...

("<html><head><meta http-equiv='Content-Type' content='text/html; "

"charset=UTF-8'/><title></title></head><body> Hooray!!<br><br>mail server is "

'now working!</body></html>')

from: SEGFW_INT@Home.local

Here is the code:

import base64
from pprint import pprint

import aiosmtpd.controller
import asyncio
import email


class CustomSMTPHandler:
    async def handle_DATA(self, server, session, envelope):
        b = email.message_from_string(str(envelope.content,'utf-8'))
        if b.is_multipart():
            payloads = ""
            for payload in b.get_payload():
                payloads += payload.get_payload()
            decoded_payloads = base64.b64decode(payloads)

        else:
            decoded_payloads = base64.b64decode(b.get_payload())
        pprint(str(decoded_payloads, 'utf-8'))
        print('from:', envelope.mail_from)
        return '250 OK'


async def main(loop):
    handler = CustomSMTPHandler()
    server = aiosmtpd.controller.Controller(handler,hostname='10.10.10.2', port=8025)
    server.start()


if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    loop.create_task(main(loop=loop))
    try:
        print("server running...")
        loop.run_forever()
    except KeyboardInterrupt:
        pass

import base64

from pprint import pprint

import aiosmtpd.controller

import asyncio

import email

class CustomSMTPHandler:

async def handle_DATA(self, server, session, envelope):

b = email.message_from_string(str(envelope.content,'utf-8'))

if b.is_multipart():

payloads = ""

for payload in b.get_payload():

payloads += payload.get_payload()

decoded_payloads = base64.b64decode(payloads)

else:

decoded_payloads = base64.b64decode(b.get_payload())

pprint(str(decoded_payloads, 'utf-8'))

print('from:', envelope.mail_from)

return '250 OK'

async def main(loop):

handler = CustomSMTPHandler()

server = aiosmtpd.controller.Controller(handler,hostname='10.10.10.2', port=8025)

server.start()

if __name__ == '__main__':

loop = asyncio.get_event_loop()

loop.create_task(main(loop=loop))

try:

print("server running...")

loop.run_forever()

except KeyboardInterrupt:

pass

Share this:

Related