Python 100 project #50: Get Audit Report on Slack

In this project, I extended the previous project “PDF to TXT”, and now it’s posted to Slack every day.

So in short, every day the sophos XG firewall sends the security audit report(PDF) to the python powered server, and the server interpret the PDF into the text, (and of course it selects the necessary part only) and post the daily summary on slack.

Output:

Code:

import base64
from io import BytesIO
from pprint import pprint
import tempfile

import aiosmtpd.controller
import asyncio
import email

import audit_reader
import slack


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

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

        for part in msg.walk():
            if part.get_content_type().startswith("application/pdf"):
                pdf_bytes = BytesIO(part.get_payload(decode=True))
                data = audit_reader.retrieve_data(pdf_bytes)
                slack.post(data, 'security_logs', envelope.mail_from)
        print('from:', envelope.mail_from)
        return '250 OK'


async def main(loop):
    handler = CustomSMTPHandler()
    server = aiosmtpd.controller.Controller(handler,hostname='XX.XX.XX.XX', port=XXXX)
    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 io import BytesIO

from pprint import pprint

import tempfile

import aiosmtpd.controller

import asyncio

import email

import audit_reader

import slack

class CustomSMTPHandler:

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

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

for part in msg.walk():

if part.get_content_type().startswith("application/pdf"):

pdf_bytes = BytesIO(part.get_payload(decode=True))

data = audit_reader.retrieve_data(pdf_bytes)

slack.post(data, 'security_logs', envelope.mail_from)

print('from:', envelope.mail_from)

return '250 OK'

async def main(loop):

handler = CustomSMTPHandler()

server = aiosmtpd.controller.Controller(handler,hostname='XX.XX.XX.XX', port=XXXX)

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