Logo Search packages:      
Sourcecode: mailping version File versions  Download package

probe.py

import os, errno
from MailPing import fileutil, mail

import smtplib
import email
from email.Utils import make_msgid
import random

def shouldSend(statedir, interval, curtime):
    last = fileutil.getTime(os.path.join(statedir, 'success'))

    try:
        pending = os.listdir(os.path.join(statedir, 'pending'))
    except OSError, e:
        if e.errno == errno.ENOENT:
            pass
        else:
            raise
    else:
        for filename in pending:
            if filename.endswith('.tmp'):
                continue
            last = max(last,
                       fileutil.getTime(os.path.join(statedir,
                                                     'pending',
                                                     filename)))

    return last + interval < curtime

MESSAGE_TEXT = """\
This is a mailping probe message sent to automatically test the
functioning of email between the addresses in the From and To
header fields.

This message should only be sent to explicitly configured addresses,
no human should ever receive this message.

This message is not spam. It is not sent by an email autoresponder.
If you are receiving this message, it means the person using
mailping misconfigured it. Please reply to this email to let them
know of this fact.
"""

def randomIdent():
    return '%08x%08x%08x%08x%08x' % (
        random.randint(0, 2**32-1),
        random.randint(0, 2**32-1),
        random.randint(0, 2**32-1),
        random.randint(0, 2**32-1),
        random.randint(0, 2**32-1),
        )

def makeProbe(ident, fromAddress, toAddress, adminAddress=None):
    msg = email.message_from_string(MESSAGE_TEXT)

    mail.setMsgField(msg, 'From', fromAddress)
    mail.setMsgField(msg, 'To', toAddress)
    if adminAddress is not None:
        mail.setMsgField(msg, 'Reply-To', adminAddress)

    mail.setID(msg, ident)

    msgid = make_msgid('mailping.%s' % ident)
    mail.setMsgField(msg, 'Message-ID', msgid)

    return msg

def send(msg, sender, recipient):
    s = smtplib.SMTP()
    s.connect()
    s.sendmail(sender, [recipient], msg.as_string())
    s.quit()

def process(statedir, interval, curtime,
            fromAddress, toAddress, adminAddress):
    if shouldSend(statedir, interval, curtime):
        ident = randomIdent()
        msg = makeProbe(ident=ident,
                        fromAddress=fromAddress,
                        toAddress=toAddress,
                        adminAddress=adminAddress)
        send(msg,
             sender=fromAddress,
             recipient=toAddress)

        pending = os.path.join(statedir, 'pending')
        try:
            os.mkdir(pending)
        except OSError, e:
            if e.errno == errno.EEXIST:
                pass
            else:
                raise
        fileutil.writeFile(os.path.join(pending, ident), '%f\n' % curtime)

Generated by  Doxygen 1.6.0   Back to index