Skip to content

Instantly share code, notes, and snippets.

@parkerlreed
Last active July 23, 2022 18:55
Show Gist options
  • Save parkerlreed/0f98522fd2d4269390ec8044b306ed1e to your computer and use it in GitHub Desktop.
Save parkerlreed/0f98522fd2d4269390ec8044b306ed1e to your computer and use it in GitHub Desktop.
Basic SMTP to file saver - For when you have a 2004 multi-function printer with no network scanning!
#!/usr/bin/env python
import os
import sys
import base64
import asyncio
import subprocess
from email import message_from_bytes
from email.policy import default
from aiosmtpd.controller import Controller
i=0
p=0
if len(sys.argv) < 4:
print('Please pass in IP and port for SMTP server along with boolean for displaying files')
sys.exit()
display=sys.argv[3]
class MyHandler:
async def handle_DATA(self, server, session, envelope):
global i
global p
global display
peer = session.peer
mailfrom = envelope.mail_from
rcpttos = envelope.rcpt_tos
message = message_from_bytes(envelope.content, policy=default)
messagetostring = message.as_string()
application_pdf = None
image_jpeg = None
for part in message.walk():
if part.get_content_type() == 'application/pdf' and application_pdf is None:
application_pdf = part.get_payload()
decoded = base64.b64decode(application_pdf)
while os.path.exists('scan_%s.pdf' % i):
i += 1
with open('scan_%s.pdf' % i, 'wb') as output_file:
output_file.write(decoded)
output_file.close()
print('Wrote: ', output_file.name)
if display == 'true':
subprocess.Popen(['xdg-open', output_file.name])
if part.get_content_type() == 'image/jpeg' and image_jpeg is None:
image_jpeg = part.get_payload()
decoded = base64.b64decode(image_jpeg)
while os.path.exists('scan_%s.jpg' % p):
p += 1
with open('scan_%s.jpg' % p, 'wb') as output_file:
output_file.write(decoded)
output_file.close()
print('Wrote: ', output_file.name)
if display == 'true':
subprocess.Popen(['xdg-open', output_file.name])
return '250 OK'
my_handler = MyHandler()
async def main(loop):
my_controller = Controller(my_handler, hostname=sys.argv[1], port=sys.argv[2])
print('SMTP server running on', sys.argv[1], 'port', sys.argv[2])
print('File display is:', display)
my_controller.start()
loop = asyncio.new_event_loop()
loop.create_task(main(loop=loop))
try:
loop.run_forever()
except KeyboardInterrupt:
print("\nReceived exit, exiting")
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment