← Back to branch summary

~hudson-openstack/nova/trunk

« back to all changes in this revision

Viewing changes to vendor/Twisted-10.0.0/twisted/mail/relaymanager.py

  • Committer: Jesse Andrews
  • Date: 2010-05-28 06:05:26 UTC
  • Revision ID: git-v1:bf6e6e718cdc7488e2da87b21e258ccc065fe499
initial commit

Show diffs side-by-side

added added

removed removed

Lines of Context:
vendor/Twisted-10.0.0/twisted/mail/relaymanager.py
 
1
# -*- test-case-name: twisted.mail.test.test_mail -*-
 
2
# Copyright (c) 2001-2008 Twisted Matrix Laboratories.
 
3
# See LICENSE for details.
 
4
 
 
5
"""
 
6
Infrastructure for relaying mail through smart host
 
7
 
 
8
Today, internet e-mail has stopped being Peer-to-peer for many problems,
 
9
spam (unsolicited bulk mail) among them. Instead, most nodes on the
 
10
internet send all e-mail to a single computer, usually the ISP's though
 
11
sometimes other schemes, such as SMTP-after-POP, are used. This computer
 
12
is supposedly permanently up and traceable, and will do the work of
 
13
figuring out MXs and connecting to them. This kind of configuration
 
14
is usually termed "smart host", since the host we are connecting to
 
15
is "smart" (and will find MXs and connect to them) rather then just
 
16
accepting mail for a small set of domains.
 
17
 
 
18
The classes here are meant to facilitate support for such a configuration
 
19
for the twisted.mail SMTP server
 
20
"""
 
21
 
 
22
import rfc822
 
23
import os
 
24
import time
 
25
 
 
26
try:
 
27
    import cPickle as pickle
 
28
except ImportError:
 
29
    import pickle
 
30
 
 
31
from twisted.python import log
 
32
from twisted.python.failure import Failure
 
33
from twisted.python.compat import set
 
34
from twisted.mail import relay
 
35
from twisted.mail import bounce
 
36
from twisted.internet import protocol
 
37
from twisted.internet.defer import Deferred, DeferredList
 
38
from twisted.internet.error import DNSLookupError
 
39
from twisted.mail import smtp
 
40
from twisted.application import internet
 
41
 
 
42
class ManagedRelayerMixin:
 
43
    """SMTP Relayer which notifies a manager
 
44
 
 
45
    Notify the manager about successful mail, failed mail
 
46
    and broken connections
 
47
    """
 
48
 
 
49
    def __init__(self, manager):
 
50
        self.manager = manager
 
51
 
 
52
    def sentMail(self, code, resp, numOk, addresses, log):
 
53
        """called when e-mail has been sent
 
54
 
 
55
        we will always get 0 or 1 addresses.
 
56
        """
 
57
        message = self.names[0]
 
58
        if code in smtp.SUCCESS:
 
59
            self.manager.notifySuccess(self.factory, message)
 
60
        else:
 
61
            self.manager.notifyFailure(self.factory, message)
 
62
        del self.messages[0]
 
63
        del self.names[0]
 
64
 
 
65
    def connectionLost(self, reason):
 
66
        """called when connection is broken
 
67
 
 
68
        notify manager we will try to send no more e-mail
 
69
        """
 
70
        self.manager.notifyDone(self.factory)
 
71
 
 
72
class SMTPManagedRelayer(ManagedRelayerMixin, relay.SMTPRelayer):
 
73
    def __init__(self, messages, manager, *args, **kw):
 
74
        """
 
75
        @type messages: C{list} of C{str}
 
76
        @param messages: Filenames of messages to relay
 
77
 
 
78
        manager should support .notifySuccess, .notifyFailure
 
79
        and .notifyDone
 
80
        """
 
81
        ManagedRelayerMixin.__init__(self, manager)
 
82
        relay.SMTPRelayer.__init__(self, messages, *args, **kw)
 
83
 
 
84
class ESMTPManagedRelayer(ManagedRelayerMixin, relay.ESMTPRelayer):
 
85
    def __init__(self, messages, manager, *args, **kw):
 
86
        """
 
87
        @type messages: C{list} of C{str}
 
88
        @param messages: Filenames of messages to relay
 
89
 
 
90
        manager should support .notifySuccess, .notifyFailure
 
91
        and .notifyDone
 
92
        """
 
93
        ManagedRelayerMixin.__init__(self, manager)
 
94
        relay.ESMTPRelayer.__init__(self, messages, *args, **kw)
 
95
 
 
96
class SMTPManagedRelayerFactory(protocol.ClientFactory):
 
97
    protocol = SMTPManagedRelayer
 
98
 
 
99
    def __init__(self, messages, manager, *args, **kw):
 
100
        self.messages = messages
 
101
        self.manager = manager
 
102
        self.pArgs = args
 
103
        self.pKwArgs = kw
 
104
 
 
105
    def buildProtocol(self, addr):
 
106
        protocol = self.protocol(self.messages, self.manager, *self.pArgs,
 
107
            **self.pKwArgs)
 
108
        protocol.factory = self
 
109
        return protocol
 
110
 
 
111
    def clientConnectionFailed(self, connector, reason):
 
112
        """called when connection could not be made
 
113
 
 
114
        our manager should be notified that this happened,
 
115
        it might prefer some other host in that case"""
 
116
        self.manager.notifyNoConnection(self)
 
117
        self.manager.notifyDone(self)
 
118
 
 
119
class ESMTPManagedRelayerFactory(SMTPManagedRelayerFactory):
 
120
    protocol = ESMTPManagedRelayer
 
121
 
 
122
    def __init__(self, messages, manager, secret, contextFactory, *args, **kw):
 
123
        self.secret = secret
 
124
        self.contextFactory = contextFactory
 
125
        SMTPManagedRelayerFactory.__init__(self, messages, manager, *args, **kw)
 
126
 
 
127
    def buildProtocol(self, addr):
 
128
        s = self.secret and self.secret(addr)
 
129
        protocol = self.protocol(self.messages, self.manager, s,
 
130
            self.contextFactory, *self.pArgs, **self.pKwArgs)
 
131
        protocol.factory = self
 
132
        return protocol
 
133
 
 
134
class Queue:
 
135
    """A queue of ougoing emails."""
 
136
 
 
137
    noisy = True
 
138
 
 
139
    def __init__(self, directory):
 
140
        self.directory = directory
 
141
        self._init()
 
142
 
 
143
    def _init(self):
 
144
        self.n = 0
 
145
        self.waiting = {}
 
146
        self.relayed = {}
 
147
        self.readDirectory()
 
148
 
 
149
    def __getstate__(self):
 
150
        """(internal) delete volatile state"""
 
151
        return {'directory' : self.directory}
 
152
 
 
153
    def __setstate__(self, state):
 
154
        """(internal) restore volatile state"""
 
155
        self.__dict__.update(state)
 
156
        self._init()
 
157
 
 
158
    def readDirectory(self):
 
159
        """Read the messages directory.
 
160
 
 
161
        look for new messages.
 
162
        """
 
163
        for message in os.listdir(self.directory):
 
164
            # Skip non data files
 
165
            if message[-2:]!='-D':
 
166
                continue
 
167
            self.addMessage(message[:-2])
 
168
 
 
169
    def getWaiting(self):
 
170
        return self.waiting.keys()
 
171
 
 
172
    def hasWaiting(self):
 
173
        return len(self.waiting) > 0
 
174
 
 
175
    def getRelayed(self):
 
176
        return self.relayed.keys()
 
177
 
 
178
    def setRelaying(self, message):
 
179
        del self.waiting[message]
 
180
        self.relayed[message] = 1
 
181
 
 
182
    def setWaiting(self, message):
 
183
        del self.relayed[message]
 
184
        self.waiting[message] = 1
 
185
 
 
186
    def addMessage(self, message):
 
187
        if message not in self.relayed:
 
188
            self.waiting[message] = 1
 
189
            if self.noisy:
 
190
                log.msg('Set ' + message + ' waiting')
 
191
 
 
192
    def done(self, message):
 
193
        """Remove message to from queue."""
 
194
        message = os.path.basename(message)
 
195
        os.remove(self.getPath(message) + '-D')
 
196
        os.remove(self.getPath(message) + '-H')
 
197
        del self.relayed[message]
 
198
 
 
199
    def getPath(self, message):
 
200
        """Get the path in the filesystem of a message."""
 
201
        return os.path.join(self.directory, message)
 
202
 
 
203
    def getEnvelope(self, message):
 
204
        return pickle.load(self.getEnvelopeFile(message))
 
205
 
 
206
    def getEnvelopeFile(self, message):
 
207
        return open(os.path.join(self.directory, message+'-H'), 'rb')
 
208
 
 
209
    def createNewMessage(self):
 
210
        """Create a new message in the queue.
 
211
 
 
212
        Return a tuple - file-like object for headers, and ISMTPMessage.
 
213
        """
 
214
        fname = "%s_%s_%s_%s" % (os.getpid(), time.time(), self.n, id(self))
 
215
        self.n = self.n + 1
 
216
        headerFile = open(os.path.join(self.directory, fname+'-H'), 'wb')
 
217
        tempFilename = os.path.join(self.directory, fname+'-C')
 
218
        finalFilename = os.path.join(self.directory, fname+'-D')
 
219
        messageFile = open(tempFilename, 'wb')
 
220
 
 
221
        from twisted.mail.mail import FileMessage
 
222
        return headerFile,FileMessage(messageFile, tempFilename, finalFilename)
 
223
 
 
224
 
 
225
class _AttemptManager(object):
 
226
    """
 
227
    Manage the state of a single attempt to flush the relay queue.
 
228
    """
 
229
    def __init__(self, manager):
 
230
        self.manager = manager
 
231
        self._completionDeferreds = []
 
232
 
 
233
 
 
234
    def getCompletionDeferred(self):
 
235
        self._completionDeferreds.append(Deferred())
 
236
        return self._completionDeferreds[-1]
 
237
 
 
238
 
 
239
    def _finish(self, relay, message):
 
240
        self.manager.managed[relay].remove(os.path.basename(message))
 
241
        self.manager.queue.done(message)
 
242
 
 
243
 
 
244
    def notifySuccess(self, relay, message):
 
245
        """a relay sent a message successfully
 
246
 
 
247
        Mark it as sent in our lists
 
248
        """
 
249
        if self.manager.queue.noisy:
 
250
            log.msg("success sending %s, removing from queue" % message)
 
251
        self._finish(relay, message)
 
252
 
 
253
 
 
254
    def notifyFailure(self, relay, message):
 
255
        """Relaying the message has failed."""
 
256
        if self.manager.queue.noisy:
 
257
            log.msg("could not relay "+message)
 
258
        # Moshe - Bounce E-mail here
 
259
        # Be careful: if it's a bounced bounce, silently
 
260
        # discard it
 
261
        message = os.path.basename(message)
 
262
        fp = self.manager.queue.getEnvelopeFile(message)
 
263
        from_, to = pickle.load(fp)
 
264
        fp.close()
 
265
        from_, to, bounceMessage = bounce.generateBounce(open(self.manager.queue.getPath(message)+'-D'), from_, to)
 
266
        fp, outgoingMessage = self.manager.queue.createNewMessage()
 
267
        pickle.dump([from_, to], fp)
 
268
        fp.close()
 
269
        for line in bounceMessage.splitlines():
 
270
             outgoingMessage.lineReceived(line)
 
271
        outgoingMessage.eomReceived()
 
272
        self._finish(relay, self.manager.queue.getPath(message))
 
273
 
 
274
 
 
275
    def notifyDone(self, relay):
 
276
        """A relaying SMTP client is disconnected.
 
277
 
 
278
        unmark all pending messages under this relay's responsibility
 
279
        as being relayed, and remove the relay.
 
280
        """
 
281
        for message in self.manager.managed.get(relay, ()):
 
282
            if self.manager.queue.noisy:
 
283
                log.msg("Setting " + message + " waiting")
 
284
            self.manager.queue.setWaiting(message)
 
285
        try:
 
286
            del self.manager.managed[relay]
 
287
        except KeyError:
 
288
            pass
 
289
        notifications = self._completionDeferreds
 
290
        self._completionDeferreds = None
 
291
        for d in notifications:
 
292
            d.callback(None)
 
293
 
 
294
 
 
295
    def notifyNoConnection(self, relay):
 
296
        """Relaying SMTP client couldn't connect.
 
297
 
 
298
        Useful because it tells us our upstream server is unavailable.
 
299
        """
 
300
        # Back off a bit
 
301
        try:
 
302
            msgs = self.manager.managed[relay]
 
303
        except KeyError:
 
304
            log.msg("notifyNoConnection passed unknown relay!")
 
305
            return
 
306
 
 
307
        if self.manager.queue.noisy:
 
308
            log.msg("Backing off on delivery of " + str(msgs))
 
309
        def setWaiting(queue, messages):
 
310
            map(queue.setWaiting, messages)
 
311
        from twisted.internet import reactor
 
312
        reactor.callLater(30, setWaiting, self.manager.queue, msgs)
 
313
        del self.manager.managed[relay]
 
314
 
 
315
 
 
316
 
 
317
class SmartHostSMTPRelayingManager:
 
318
    """Manage SMTP Relayers
 
319
 
 
320
    Manage SMTP relayers, keeping track of the existing connections,
 
321
    each connection's responsibility in term of messages. Create
 
322
    more relayers if the need arises.
 
323
 
 
324
    Someone should press .checkState periodically
 
325
 
 
326
    @ivar fArgs: Additional positional arguments used to instantiate
 
327
    C{factory}.
 
328
 
 
329
    @ivar fKwArgs: Additional keyword arguments used to instantiate
 
330
    C{factory}.
 
331
 
 
332
    @ivar factory: A callable which returns a ClientFactory suitable for
 
333
    making SMTP connections.
 
334
    """
 
335
 
 
336
    factory = SMTPManagedRelayerFactory
 
337
 
 
338
    PORT = 25
 
339
 
 
340
    mxcalc = None
 
341
 
 
342
    def __init__(self, queue, maxConnections=2, maxMessagesPerConnection=10):
 
343
        """
 
344
        @type queue: Any implementor of C{IQueue}
 
345
        @param queue: The object used to queue messages on their way to
 
346
        delivery.
 
347
 
 
348
        @type maxConnections: C{int}
 
349
        @param maxConnections: The maximum number of SMTP connections to
 
350
        allow to be opened at any given time.
 
351
 
 
352
        @type maxMessagesPerConnection: C{int}
 
353
        @param maxMessagesPerConnection: The maximum number of messages a
 
354
        relayer will be given responsibility for.
 
355
 
 
356
        Default values are meant for a small box with 1-5 users.
 
357
        """
 
358
        self.maxConnections = maxConnections
 
359
        self.maxMessagesPerConnection = maxMessagesPerConnection
 
360
        self.managed = {} # SMTP clients we're managing
 
361
        self.queue = queue
 
362
        self.fArgs = ()
 
363
        self.fKwArgs = {}
 
364
 
 
365
    def __getstate__(self):
 
366
        """(internal) delete volatile state"""
 
367
        dct = self.__dict__.copy()
 
368
        del dct['managed']
 
369
        return dct
 
370
 
 
371
    def __setstate__(self, state):
 
372
        """(internal) restore volatile state"""
 
373
        self.__dict__.update(state)
 
374
        self.managed = {}
 
375
 
 
376
    def checkState(self):
 
377
        """
 
378
        Synchronize with the state of the world, and maybe launch a new
 
379
        relay.
 
380
 
 
381
        Call me periodically to check I am still up to date.
 
382
 
 
383
        @return: None or a Deferred which fires when all of the SMTP clients
 
384
        started by this call have disconnected.
 
385
        """
 
386
        self.queue.readDirectory()
 
387
        if (len(self.managed) >= self.maxConnections):
 
388
            return
 
389
        if  not self.queue.hasWaiting():
 
390
            return
 
391
 
 
392
        return self._checkStateMX()
 
393
 
 
394
    def _checkStateMX(self):
 
395
        nextMessages = self.queue.getWaiting()
 
396
        nextMessages.reverse()
 
397
 
 
398
        exchanges = {}
 
399
        for msg in nextMessages:
 
400
            from_, to = self.queue.getEnvelope(msg)
 
401
            name, addr = rfc822.parseaddr(to)
 
402
            parts = addr.split('@', 1)
 
403
            if len(parts) != 2:
 
404
                log.err("Illegal message destination: " + to)
 
405
                continue
 
406
            domain = parts[1]
 
407
 
 
408
            self.queue.setRelaying(msg)
 
409
            exchanges.setdefault(domain, []).append(self.queue.getPath(msg))
 
410
            if len(exchanges) >= (self.maxConnections - len(self.managed)):
 
411
                break
 
412
 
 
413
        if self.mxcalc is None:
 
414
            self.mxcalc = MXCalculator()
 
415
 
 
416
        relays = []
 
417
        for (domain, msgs) in exchanges.iteritems():
 
418
            manager = _AttemptManager(self)
 
419
            factory = self.factory(msgs, manager, *self.fArgs, **self.fKwArgs)
 
420
            self.managed[factory] = map(os.path.basename, msgs)
 
421
            relayAttemptDeferred = manager.getCompletionDeferred()
 
422
            connectSetupDeferred = self.mxcalc.getMX(domain)
 
423
            connectSetupDeferred.addCallback(lambda mx: str(mx.name))
 
424
            connectSetupDeferred.addCallback(self._cbExchange, self.PORT, factory)
 
425
            connectSetupDeferred.addErrback(lambda err: (relayAttemptDeferred.errback(err), err)[1])
 
426
            connectSetupDeferred.addErrback(self._ebExchange, factory, domain)
 
427
            relays.append(relayAttemptDeferred)
 
428
        return DeferredList(relays)
 
429
 
 
430
 
 
431
    def _cbExchange(self, address, port, factory):
 
432
        from twisted.internet import reactor
 
433
        reactor.connectTCP(address, port, factory)
 
434
 
 
435
    def _ebExchange(self, failure, factory, domain):
 
436
        log.err('Error setting up managed relay factory for ' + domain)
 
437
        log.err(failure)
 
438
        def setWaiting(queue, messages):
 
439
            map(queue.setWaiting, messages)
 
440
        from twisted.internet import reactor
 
441
        reactor.callLater(30, setWaiting, self.queue, self.managed[factory])
 
442
        del self.managed[factory]
 
443
 
 
444
class SmartHostESMTPRelayingManager(SmartHostSMTPRelayingManager):
 
445
    factory = ESMTPManagedRelayerFactory
 
446
 
 
447
def _checkState(manager):
 
448
    manager.checkState()
 
449
 
 
450
def RelayStateHelper(manager, delay):
 
451
    return internet.TimerService(delay, _checkState, manager)
 
452
 
 
453
 
 
454
 
 
455
class CanonicalNameLoop(Exception):
 
456
    """
 
457
    When trying to look up the MX record for a host, a set of CNAME records was
 
458
    found which form a cycle and resolution was abandoned.
 
459
    """
 
460
 
 
461
 
 
462
class CanonicalNameChainTooLong(Exception):
 
463
    """
 
464
    When trying to look up the MX record for a host, too many CNAME records
 
465
    which point to other CNAME records were encountered and resolution was
 
466
    abandoned.
 
467
    """
 
468
 
 
469
 
 
470
class MXCalculator:
 
471
    """
 
472
    A utility for looking up mail exchange hosts and tracking whether they are
 
473
    working or not.
 
474
 
 
475
    @ivar clock: L{IReactorTime} provider which will be used to decide when to
 
476
        retry mail exchanges which have not been working.
 
477
    """
 
478
    timeOutBadMX = 60 * 60 # One hour
 
479
    fallbackToDomain = True
 
480
 
 
481
    def __init__(self, resolver=None, clock=None):
 
482
        self.badMXs = {}
 
483
        if resolver is None:
 
484
            from twisted.names.client import createResolver
 
485
            resolver = createResolver()
 
486
        self.resolver = resolver
 
487
        if clock is None:
 
488
            from twisted.internet import reactor as clock
 
489
        self.clock = clock
 
490
 
 
491
 
 
492
    def markBad(self, mx):
 
493
        """Indicate a given mx host is not currently functioning.
 
494
 
 
495
        @type mx: C{str}
 
496
        @param mx: The hostname of the host which is down.
 
497
        """
 
498
        self.badMXs[str(mx)] = self.clock.seconds() + self.timeOutBadMX
 
499
 
 
500
    def markGood(self, mx):
 
501
        """Indicate a given mx host is back online.
 
502
 
 
503
        @type mx: C{str}
 
504
        @param mx: The hostname of the host which is up.
 
505
        """
 
506
        try:
 
507
            del self.badMXs[mx]
 
508
        except KeyError:
 
509
            pass
 
510
 
 
511
    def getMX(self, domain, maximumCanonicalChainLength=3):
 
512
        """
 
513
        Find an MX record for the given domain.
 
514
 
 
515
        @type domain: C{str}
 
516
        @param domain: The domain name for which to look up an MX record.
 
517
 
 
518
        @type maximumCanonicalChainLength: C{int}
 
519
        @param maximumCanonicalChainLength: The maximum number of unique CNAME
 
520
            records to follow while looking up the MX record.
 
521
 
 
522
        @return: A L{Deferred} which is called back with a string giving the
 
523
            name in the found MX record or which is errbacked if no MX record
 
524
            can be found.
 
525
        """
 
526
        mailExchangeDeferred = self.resolver.lookupMailExchange(domain)
 
527
        mailExchangeDeferred.addCallback(self._filterRecords)
 
528
        mailExchangeDeferred.addCallback(
 
529
            self._cbMX, domain, maximumCanonicalChainLength)
 
530
        mailExchangeDeferred.addErrback(self._ebMX, domain)
 
531
        return mailExchangeDeferred
 
532
 
 
533
 
 
534
    def _filterRecords(self, records):
 
535
        """
 
536
        Convert a DNS response (a three-tuple of lists of RRHeaders) into a
 
537
        mapping from record names to lists of corresponding record payloads.
 
538
        """
 
539
        recordBag = {}
 
540
        for answer in records[0]:
 
541
            recordBag.setdefault(str(answer.name), []).append(answer.payload)
 
542
        return recordBag
 
543
 
 
544
 
 
545
    def _cbMX(self, answers, domain, cnamesLeft):
 
546
        """
 
547
        Try to find the MX host from the given DNS information.
 
548
 
 
549
        This will attempt to resolve CNAME results.  It can recognize loops
 
550
        and will give up on non-cyclic chains after a specified number of
 
551
        lookups.
 
552
        """
 
553
        # Do this import here so that relaymanager.py doesn't depend on
 
554
        # twisted.names, only MXCalculator will.
 
555
        from twisted.names import dns, error
 
556
 
 
557
        seenAliases = set()
 
558
        exchanges = []
 
559
        # Examine the answers for the domain we asked about
 
560
        pertinentRecords = answers.get(domain, [])
 
561
        while pertinentRecords:
 
562
            record = pertinentRecords.pop()
 
563
 
 
564
            # If it's a CNAME, we'll need to do some more processing
 
565
            if record.TYPE == dns.CNAME:
 
566
 
 
567
                # Remember that this name was an alias.
 
568
                seenAliases.add(domain)
 
569
 
 
570
                canonicalName = str(record.name)
 
571
                # See if we have some local records which might be relevant.
 
572
                if canonicalName in answers:
 
573
 
 
574
                    # Make sure it isn't a loop contained entirely within the
 
575
                    # results we have here.
 
576
                    if canonicalName in seenAliases:
 
577
                        return Failure(CanonicalNameLoop(record))
 
578
 
 
579
                    pertinentRecords = answers[canonicalName]
 
580
                    exchanges = []
 
581
                else:
 
582
                    if cnamesLeft:
 
583
                        # Request more information from the server.
 
584
                        return self.getMX(canonicalName, cnamesLeft - 1)
 
585
                    else:
 
586
                        # Give up.
 
587
                        return Failure(CanonicalNameChainTooLong(record))
 
588
 
 
589
            # If it's an MX, collect it.
 
590
            if record.TYPE == dns.MX:
 
591
                exchanges.append((record.preference, record))
 
592
 
 
593
        if exchanges:
 
594
            exchanges.sort()
 
595
            for (preference, record) in exchanges:
 
596
                host = str(record.name)
 
597
                if host not in self.badMXs:
 
598
                    return record
 
599
                t = self.clock.seconds() - self.badMXs[host]
 
600
                if t >= 0:
 
601
                    del self.badMXs[host]
 
602
                    return record
 
603
            return exchanges[0][1]
 
604
        else:
 
605
            # Treat no answers the same as an error - jump to the errback to try
 
606
            # to look up an A record.  This provides behavior described as a
 
607
            # special case in RFC 974 in the section headed I{Interpreting the
 
608
            # List of MX RRs}.
 
609
            return Failure(
 
610
                error.DNSNameError("No MX records for %r" % (domain,)))
 
611
 
 
612
 
 
613
    def _ebMX(self, failure, domain):
 
614
        from twisted.names import error, dns
 
615
 
 
616
        if self.fallbackToDomain:
 
617
            failure.trap(error.DNSNameError)
 
618
            log.msg("MX lookup failed; attempting to use hostname (%s) directly" % (domain,))
 
619
 
 
620
            # Alright, I admit, this is a bit icky.
 
621
            d = self.resolver.getHostByName(domain)
 
622
            def cbResolved(addr):
 
623
                return dns.Record_MX(name=addr)
 
624
            def ebResolved(err):
 
625
                err.trap(error.DNSNameError)
 
626
                raise DNSLookupError()
 
627
            d.addCallbacks(cbResolved, ebResolved)
 
628
            return d
 
629
        elif failure.check(error.DNSNameError):
 
630
            raise IOError("No MX found for %r" % (domain,))
 
631
        return failure