Package Products :: Package ZenEvents :: Module zenmail |
[hide private]
[frames] | no frames]

# Source Code for Module Products.ZenEvents.zenmail

      1  ##############################################################################
      2  #
      3  # Copyright (C) Zenoss, Inc. 2007, 2011, all rights reserved.
      4  #
      5  # This content is made available according to terms specified in
      6  # License.zenoss under the directory where your Zenoss product is installed.
      7  #
      8  ##############################################################################
      9
     10
     11  # Notes: database wants events in UTC time
     12  # Events page shows local time, as determined on the server where zenoss runs
     13
     14  __doc__ = """zenmail
     15
     16  Listen on the SMTP port and convert email messages into events.
     17
     18  To test:
     19   # Test pre-reqs
     20   yum -y install mailx sendmail
     21
     22   # Mail to the local zenmail instance
     23  mail -s "Hello world" bogo@localhost
     24  Happy happy, joy, joy
     25  .
     26  Cc:
     27
     28  """
     29
     30  import logging
     31  from email.Header import Header
     32  import email
     33  import os
     34  import socket
     35
     36  import Globals
     37  import zope.interface
     38  import zope.component
     39  from zope.interface import implements
     40
     41  from twisted.mail import smtp
     42  from twisted.internet import reactor, protocol, defer
     43
     44  from Products.ZenCollector.daemon import CollectorDaemon
     45  from Products.ZenCollector.interfaces import ICollector, ICollectorPreferences,\
     46                                               IEventService, \
     47                                               IScheduledTask
     48  from Products.ZenCollector.tasks import NullTaskSplitter,\
     49                                          BaseTask, TaskStates
     50
     51  # Invalidation issues arise if we don't import
     52  from Products.ZenCollector.services.config import DeviceProxy
     53
     54  from Products.ZenEvents.MailProcessor import MailProcessor
     55  from Products.ZenUtils.Utils import unused
     56  unused(Globals, DeviceProxy)
     57
     58
     59  COLLECTOR_NAME = 'zenmail'
     60  log = logging.getLogger("zen.%s" % COLLECTOR_NAME)
     61
     62
     63 -class MailPreferences(object):
     64      zope.interface.implements(ICollectorPreferences)
     65
     66 -    def __init__(self):
     67          """
     68          Constructs a new PingCollectionPreferences instance and
     69          provides default values for needed attributes.
     70          """
     71          self.collectorName = COLLECTOR_NAME
     72          self.defaultRRDCreateCommand = None
     73          self.configCycleInterval = 20 # minutes
     74          self.cycleInterval = 5 * 60 # seconds
     75
     76          # The configurationService attribute is the fully qualified class-name
     77          # of our configuration service that runs within ZenHub
     78          self.configurationService = 'Products.ZenHub.services.NullConfig'
     79
     80          # Will be filled in based on buildOptions
     81          self.options = None
     82
     83          self.configCycleInterval = 20*60
     84
     85 -    def postStartupTasks(self):
     86          task = MailListeningTask(COLLECTOR_NAME, configId=COLLECTOR_NAME)
     87          yield task
     88
     89 -    def buildOptions(self, parser):
     90          """
     91          Command-line options to be supported
     92          """
     93          SMTP_PORT = 25
     94          try:
     95              SMTP_PORT = socket.getservbyname('smtp', 'tcp')
     96          except socket.error:
     97              pass
     98
     99          parser.add_option('--useFileDescriptor',
    100              dest='useFileDescriptor',
    101              default=-1,
    102              type="int",
    103              help="File descriptor to use for listening")
    104          parser.add_option('--listenPort',
    105              dest='listenPort',
    106              default=SMTP_PORT,
    107              type="int",
    108              help="Alternative listen port to use (default %default)")
    109          parser.add_option('--eventseverity',
    110              dest='eventseverity',
    111              default="2",
    112              type="int",
    113              help="Severity for events created")
    114          parser.add_option('--listenip',
    115              dest='listenip',
    116              default='0.0.0.0',
    117              help='IP address to listen on. Default is 0.0.0.0')
    118
    119 -    def postStartup(self):
    120          pass
    121
    122
    123 -class ZenossEventPoster(object):
    124      """
    125      Implementation of interface definition for messages
    126      that can be sent via SMTP.
    127      """
    128      implements(smtp.IMessage)
    129
    130 -    def __init__(self, processor):
    131          self.lines = []
    132          self.processor = processor
    133
    134 -    def lineReceived(self, line):
    135          self.lines.append(line)
    136
    137 -    def postEvent(self, messageStr):
    138          email.message_from_string(messageStr)
    139          self.processor.process(messageStr)
    140
    141 -    def eomReceived(self):
    142          log.info('Message data completed %s.', self.lines)
    143          self.lines.append('')
    144          messageData = '\n'.join(self.lines)
    145
    146          self.postEvent(messageData)
    147
    148          return defer.succeed("Received End Of Message marker")
    149
    150 -    def connectionLost(self):
    151          log.info('Connection lost unexpectedly')
    152          del(self.lines)
    153
    154
    155 -class ZenossDelivery(object):
    156      implements(smtp.IMessageDelivery)
    157
    158 -    def __init__(self, processor):
    159          self.processor = processor
    160
    161 -    def receivedHeader(self, helo, unused, ignored):
    162          myHostname, self.clientIP = helo
    163          date = smtp.rfc822date()
    164
    165          headerValue = 'by %s from %s with ESMTP ; %s' % (
    166              myHostname, self.clientIP, date)
    167
    168          log.info('Relayed (or sent directly) from: %s', self.clientIP)
    169
    170          header = 'Received: %s' % Header(headerValue)
    171          return header
    172
    173 -    def validateTo(self, user):
    174          log.info('to: %s', user.dest)
    175          return self.makePoster
    176
    177 -    def makePoster(self):
    178          return ZenossEventPoster(self.processor)
    179
    180 -    def validateFrom(self, unused, originAddress):
    181          log.info("from: %s", originAddress)
    182          return originAddress
    183
    184
    185 -class SMTPFactory(protocol.ServerFactory):
    186 -    def __init__(self, processor):
    187          self.processor = processor
    188
    189 -    def buildProtocol(self, unused):
    190          delivery = ZenossDelivery(self.processor)
    191          smtpProtocol = smtp.SMTP(delivery)
    192          smtpProtocol.factory = self
    193          return smtpProtocol
    194
    195
    196 -class MailListeningTask(BaseTask):
    197      zope.interface.implements(IScheduledTask)
    198
    199 -    def __init__(self, taskName, configId,
    200                   scheduleIntervalSeconds=3600, taskConfig=None):
    201          BaseTask.__init__(self, taskName, configId,
    202                   scheduleIntervalSeconds, taskConfig)
    203          self.log = log
    204
    205          # Needed for interface
    206          self.name = taskName
    207          self.configId = configId
    208          self.state = TaskStates.STATE_IDLE
    209          self.interval = scheduleIntervalSeconds
    210          self._preferences = taskConfig
    211          self._daemon = zope.component.getUtility(ICollector)
    212          self._eventService = zope.component.queryUtility(IEventService)
    213          self._preferences = self._daemon
    214
    215          self.options = self._daemon.options
    216
    217          # Allow MailProcessor to work unmodified
    218          self.sendEvent = self._eventService.sendEvent
    219
    220          if (self.options.useFileDescriptor < 0 and \
    221              self.options.listenPort < 1024):
    222              self._daemon.openPrivilegedPort('--listen',
    223                  '--proto=tcp', '--port=%s:%d' % (
    224                      self.options.listenip, self.options.listenPort))
    225
    226          self._daemon.changeUser()
    227          self.processor = MailProcessor(self, self.options.eventseverity)
    228
    229          self.factory = SMTPFactory(self.processor)
    230
    231          log.info("listening on %s:%d" % (
    232              self.options.listenip, self.options.listenPort))
    233          if self.options.useFileDescriptor != -1:
    234              self.useTcpFileDescriptor(int(self.options.useFileDescriptor),
    235                                        self.factory)
    236          else:
    237              reactor.listenTCP(self.options.listenPort, self.factory,
    238                  interface=self.options.listenip)
    239
    240 -    def doTask(self):
    241          """
    242          This is a wait-around task since we really are called
    243          asynchronously.
    244          """
    245          return defer.succeed("Waiting for SMTP messages...")
    246
    247 -    def useTcpFileDescriptor(self, fd, factory):
    248          for i in range(19800, 19999):
    249              try:
    250                  p = reactor.listenTCP(i, factory)
    251                  os.dup2(fd, p.socket.fileno())
    252                  p.socket.listen(p.backlog)
    253                  p.socket.setblocking(False)
    254                  p.socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    255                  os.close(fd)
    256                  return p
    257              except socket.error:
    258                  pass
    259          raise socket.error("Unable to find an open socket to listen on")
    260
    261 -    def cleanup(self):
    262          pass
    263
    264
    265 -class MailDaemon(CollectorDaemon):
    266
    267      _frameworkFactoryName = "nosip"
    268
    269
    270  if __name__=='__main__':
    271      myPreferences = MailPreferences()
    272      myTaskSplitter = NullTaskSplitter()
    273      daemon = MailDaemon(myPreferences, myTaskSplitter)
    274      daemon.run()
    275