Products.ZenEvents.zenpop3

1 ########################################################################### 2 # 3 # This program is part of Zenoss Core, an open source monitoring platform. 4 # Copyright (C) 2007, 2011 Zenoss Inc. 5 # 6 # This program is free software; you can redistribute it and/or modify it 7 # under the terms of the GNU General Public License version 2 or (at your 8 # option) any later version as published by the Free Software Foundation. 9 # 10 # For complete information please visit: http://www.zenoss.com/oss/ 11 # 12 ########################################################################### 13 #! /usr/bin/env python 14 # Notes: database wants events in UTC time 15 # Events page shows local time, as determined on the server where zenoss runs 16 17 __doc__ = """zenpop3 18 19 Turn email messages obtained from POP3 accounts into events. 20 21 """ 22 23 import logging 24 import socket 25 26 import Globals 27 import zope.interface 28 29 from twisted.mail.pop3client import POP3Client 30 from twisted.internet.ssl import ClientContextFactory 31 from twisted.internet import reactor, protocol, defer, error 32 33 from Products.ZenCollector.daemon import CollectorDaemon 34 from Products.ZenCollector.interfaces import ICollector, ICollectorPreferences,\ 35 IEventService, \ 36 IScheduledTask 37 from Products.ZenCollector.tasks import NullTaskSplitter,\ 38 BaseTask, TaskStates 39 40 from Products.ZenEvents.MailProcessor import POPProcessor 41 42 43 COLLECTOR_NAME = 'zenpop3' 44 log = logging.getLogger("zen.%s" % COLLECTOR_NAME) 45 46

47 -class MailPreferences(object):

48 zope.interface.implements(ICollectorPreferences) 49

50 - def __init__(self):

51 """ 52 Constructs a new PingCollectionPreferences instance and 53 provides default values for needed attributes. 54 """ 55 self.collectorName = COLLECTOR_NAME 56 self.defaultRRDCreateCommand = None 57 self.configCycleInterval = 20 # minutes 58 self.cycleInterval = 5 * 60 # seconds 59 60 # The configurationService attribute is the fully qualified class-name 61 # of our configuration service that runs within ZenHub 62 self.configurationService = 'Products.ZenHub.services.NullConfig' 63 64 # Will be filled in based on buildOptions 65 self.options = None 66 67 self.configCycleInterval = 20*60

68

69 - def postStartupTasks(self):

70 task = MailCollectingTask(COLLECTOR_NAME, configId=COLLECTOR_NAME) 71 yield task

72

73 - def buildOptions(self, parser):

74 """ 75 Command-line options to be supported 76 """ 77 POP3_PORT = 110 78 try: 79 POP3_PORT = socket.getservbyname('pop3', 'tcp') 80 except socket.error: 81 pass 82 83 parser.add_option('--usessl', 84 dest='usessl', 85 default=False, 86 action="store_true", 87 help="Use SSL when connecting to POP server") 88 parser.add_option('--nodelete', 89 dest='nodelete', 90 default=False, 91 action="store_true", 92 help="Leave messages on POP server") 93 parser.add_option('--pophost', 94 dest='pophost', 95 default="pop.zenoss.com", 96 help="POP server from which emails are to be read") 97 parser.add_option('--popport', 98 dest='popport', 99 default=POP3_PORT, 100 type="int", 101 help="POP port from which emails are to be read") 102 parser.add_option('--popuser', 103 dest='popuser', 104 default="zenoss", 105 help="POP user") 106 parser.add_option('--poppass', 107 dest='poppass', 108 default="zenoss", 109 help="POP password") 110 parser.add_option('--cycletime', 111 dest='cycletime', 112 type="int", 113 default=60, 114 help="Frequency (in seconds) to poll the POP server") 115 parser.add_option('--eventseverity', 116 dest='eventseverity', 117 default="2", 118 type="int", 119 help="Severity for events created")

120

121 - def postStartup(self):

122 pass

123 124

125 -class POPProtocol(POP3Client):

126 """ 127 Protocol that is responsible for conversing with a POP server 128 after a connection has been established. Downloads messages (and 129 deletes them by default), and passes the messages back up to the 130 factory to process and turn into events. 131 """ 132 133 allowInsecureLogin = True 134 timeout = 15 135 totalMessages = 0 136

137 - def serverGreeting(self, unused):

138 log.debug('Server greeting received: Logging in...') 139 140 login = self.login(self.factory.user, self.factory.passwd) 141 login.addCallback(self._loggedIn) 142 login.addErrback(self.factory.deferred.errback)

143

144 - def _loggedIn(self, unused):

145 log.debug('Logged in') 146 return self.retrieveAndParse()

147

148 - def retrieveAndParse(self):

149 d = self.listSize() 150 d.addCallback(self._gotMessageSizes) 151 152 return d

153

154 - def _gotMessageSizes(self, sizes):

155 self.totalMessages = len(sizes) 156 log.info('Messages to retrieve: %d', self.totalMessages) 157 158 self.sizes = sizes 159 160 retreivers = [] 161 for i in range(len(sizes)): 162 log.debug('Retrieving message #%d...' % i) 163 d = self.retrieve(i) 164 d.addCallback(self._gotMessageLines) 165 retreivers.append(d) 166 167 deferreds = defer.DeferredList(retreivers) 168 deferreds.addCallback(self._delete) 169 return deferreds.addCallback(self.scanComplete)

170

171 - def _gotMessageLines(self, messageLines):

172 log.debug('Passing message up to factory') 173 self.factory.handleMessage("\r\n".join(messageLines))

174

175 - def _delete(self, unused):

176 deleters = [] 177 if not self.factory.nodelete: 178 for index in range(len(self.sizes)): 179 log.info('Deleting message #%d...' % index) 180 d = self.delete(index) 181 deleters.append(d) 182 183 deferreds = defer.DeferredList(deleters) 184 return deferreds

185

186 - def scanComplete(self, unused):

187 log.debug("Scan complete") 188 self.quit()

189 190

191 -class POPFactory(protocol.ClientFactory):

192 """ 193 Factory that stores the configuration the protocol uses to do 194 its job. 195 """ 196 protocol = POPProtocol 197

198 - def __init__(self, user, passwd, processor, nodelete):

199 self.user = user 200 self.passwd = passwd 201 self.processor = processor 202 self.deferred = defer.Deferred() 203 self.nodelete = nodelete

204

205 - def handleMessage(self, messageData):

206 self.processor.process(messageData)

207

208 - def clientConnectionFailed(self, unused, reason):

209 self.deferred.errback(reason)

210 211

212 -class MailCollectingTask(BaseTask):

213 zope.interface.implements(IScheduledTask) 214 215 STATE_COLLECTING = 'COLLECTING' 216

217 - def __init__(self, taskName, configId, 218 scheduleIntervalSeconds=60, taskConfig=None):

219 BaseTask.__init__(self, taskName, configId, 220 scheduleIntervalSeconds, taskConfig) 221 self.log = log 222 223 # Needed for interface 224 self.name = taskName 225 self.configId = configId 226 self.state = TaskStates.STATE_IDLE 227 self._preferences = taskConfig 228 self._daemon = zope.component.getUtility(ICollector) 229 self._eventService = zope.component.queryUtility(IEventService) 230 self._preferences = self._daemon 231 232 self.options = self._daemon.options 233 234 # This will take a bit to catch up, but.... 235 self.interval = self.options.cycletime 236 237 # Allow MailProcessor to work unmodified 238 self.sendEvent = self._eventService.sendEvent 239 240 self._daemon.changeUser() 241 self.processor = POPProcessor(self,self.options.eventseverity) 242 self._connection = None

243

244 - def doTask(self):

245 d = defer.maybeDeferred(self.checkForMessages) 246 return d

247

248 - def makeFactory(self):

249 self.factory = POPFactory(self.options.popuser, self.options.poppass, 250 self.processor, self.options.nodelete) 251 self.factory.deferred.addErrback(self.handleError)

252

253 - def checkForMessages(self):

254 self.state = MailCollectingTask.STATE_COLLECTING 255 256 self.makeFactory() 257 if self.options.usessl: 258 log.debug("Connecting to server %s:%s using SSL as %s", 259 self.options.pophost, self.options.popport, self.options.popuser) 260 self._connection = reactor.connectSSL(self.options.pophost, self.options.popport, 261 self.factory, ClientContextFactory()) 262 else: 263 log.debug("Connecting to server %s:%s using plaintext as %s", 264 self.options.pophost, self.options.popport, self.options.popuser) 265 self._connection = reactor.connectTCP(self.options.pophost, self.options.popport, 266 self.factory) 267 return defer.succeed("Connected to server %s:%s" % ( 268 self.options.pophost, self.options.popport))

269

270 - def _finished(self, result=None):

271 if self._connection: 272 self._connection.disconnect() 273 if self.factory: 274 message = "Last retrieved %d messages" % self.factory.protocl.totalMessages 275 else: 276 message = "Completed" 277 return defer.succeed(message)

278

279 - def handleError(self, err):

280 if err.type == error.TimeoutError: 281 message = "Timed out connecting to %s:%d" % ( 282 self.options.pophost, self.options.popport) 283 284 elif err.type == error.ConnectionRefusedError: 285 message = "Connection refused by %s:%d" % ( 286 self.options.pophost, self.options.popport) 287 288 elif err.type == error.ConnectError: 289 message = "Connection failed to %s:%d" % ( 290 self.options.pophost, self.options.popport) 291 else: 292 message = err.getErrorMessage() 293 self.sendEvent(dict( 294 device=socket.getfqdn(), 295 component=COLLECTOR_NAME, 296 severity=5, 297 summary="Fatal error in %s" % COLLECTOR_NAME, 298 message=message, 299 )) 300 301 # Force the task to quit 302 self.state = TaskStates.STATE_COMPLETED 303 304 log.error(message) 305 return defer.succeed(message)

306

307 - def cleanup(self):

308 self._finished()

309 310 311 if __name__=='__main__': 312 myPreferences = MailPreferences() 313 myTaskSplitter = NullTaskSplitter() 314 daemon = CollectorDaemon(myPreferences, myTaskSplitter) 315 daemon.run() 316

Source Code for Module Products.ZenEvents.zenpop3