Du bist hier: Start › Python › SMTP DebuggingDieses Snippet kommentieren

SMTP Debugging

Dieses Script läuft auf der Kommandozeile und wartet auf Mails auf loalhost:25. Wird eine Mail empfangen (will also ein Programm eine Mail raussenden), so wird der Inhalt protokolliert, nicht aber wirklich ins Netz abgeschickt.

Das Script ist ungemein nützlich, um andere Programme wie PHP-Scripts zu testen, die Mails an Kunden versenden sollen. Damit entfällt das lästige printf($mailbody) beim Debuggen.

# SmtpMailsink Copyright 2005 Aviarc Corporation
# Written by Adam Feuer, Matt Branthwaite, and Troy Frever
# Gekuerzt und um normales Logging erweitert von Christoph.

import sys, asyncore, threading, socket, smtpd, time, StringIO

class SmtpDumpServer(smtpd.SMTPServer):
   __version__ = "Python SMTP Dump Version 0.2"

   def __init__(self, *args, **kwargs):
      smtpd.SMTPServer.__init__(self, *args, **kwargs)
      self.mailboxFile = None
      
   def process_message(self, peer, mailfrom, rcpttos, data):
      print("["+time.strftime("%d.%m.%Y %H:%M:%S")+"] Mail von "+mailfrom+" erhalten.")
      t = time.strftime("%Y%m%d_%H;%M;%S")
      f = open(t+".txt", "w")
      if f:
         #f.write("From %s\n" % mailfrom)
         f.write(data)
         f.write("\n")
         f.close()

class SmtpDumper(threading.Thread):
   TIME_TO_WAIT_BETWEEN_CHECKS_TO_STOP_SERVING = 0.001

   def __init__(self, host = "localhost", port = 25, threadName = None):   
      self.throwExceptionIfAddressIsInUse(host, port)
      self.initializeThread(threadName)
      self.initializeSmtpDumpServer(host, port)

   def throwExceptionIfAddressIsInUse(self, host, port):
      testSocket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
      testSocket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, testSocket.getsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR) | 1)
      testSocket.bind((host, port))
      testSocket.close()

   def initializeThread(self, threadName):
      self._stopevent = threading.Event()
      self.threadName = threadName
      if self.threadName is None:
         self.threadName = SmtpDumper.__class__
      threading.Thread.__init__(self, name = self.threadName)
      
   def initializeSmtpDumpServer(self, host, port):
      self.smtpDumpServer = SmtpDumpServer((host, port), None)
      smtpd.__version__   = SmtpDumpServer.__version__ 
   
   def run(self):
      while not self._stopevent.isSet():
         asyncore.loop(timeout = SmtpDumper.TIME_TO_WAIT_BETWEEN_CHECKS_TO_STOP_SERVING, count = 1)

   def stop(self, timeout = None):
      self._stopevent.set()
      threading.Thread.join(self, timeout)
      self.smtpDumpServer.close()
      
if __name__ == "__main__":
   hostname = "localhost"
   if len(sys.argv) > 1:
      hostname = sys.argv[1]
   
   dumper = SmtpDumper(host = hostname)
   dumper.start()
   
   print("Server gestartet. Warte auf Mails...")
   
   while True:
      try:
         time.sleep(1)
      except KeyboardInterrupt, e:
         print("Programm beendet.")
         sys.exit(1)

Kommentar verfassen

Fehler gefunden? Doofer Code? Ein kleines "Danke!"? Hinterlasse einfach einen Kommentar.

(muss sein)
(muss nicht sein, wird nicht angezeigt)

Dein Kommentar wird erst nach einer manuellen Prüfung angezeigt.