parser.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import threading
from multiprocessing import get_logger
import zmq
import re


class CEFParser(object):

    def __init__(self):
        self.logger = get_logger()
        self.counter = 0

    def stats(self):
        interval = 5
        logger = get_logger()
        logger.debug('stats - processed\t%s' % self.counter)  # happens once and only once
        threading.Timer(interval, self.stats).start()

    def run(self):
        self.logger.info('CEFParser.run() is running')

        # connect to zmq rx queue device
        self.context = zmq.Context()
        self.consumer_receiver = self.context.socket(zmq.PULL)
        self.consumer_receiver.connect("tcp://127.0.0.1:5552")
        self.logger.debug('consumer_receiver = {0}'.format(self.consumer_receiver))

        # connect to zmq tx queue
        self.consumer_sender = self.context.socket(zmq.PUSH)
        self.consumer_sender.connect("tcp://127.0.0.1:5553")
        self.logger.debug('consumer_sender = {0}'.format(self.consumer_sender))

        self.stats()

        # TODO - move the regex extraction to its own method?  Create a processing chain?
        regex_header = re.compile(
            r"(?P<CEFVersion>.*?)\|"
            r"(?P<deviceVendor>.*?)\|"
            r"(?P<deviceProduct>.*?)\|"
            r"(?P<deviceVersion>.*?)\|"
            r"(?P<deviceEventClassID>.*?)\|"
            r"(?P<name>.*?)\|"
            r"(?P<severity>.*?)\|"
            r"(?P<extension>[^$]+)")
        regex_extension = re.compile(r"\b(?<=\s)([^\s=]+)=(.+?)(?=(?:\s[^\s=]+=|$))")

        while True:  #blocking forever
            self.event = self.consumer_receiver.recv()  # pull an event out of the zmq Queue device
            if self.event.startswith('CEF:0'):
                try:
                    header = regex_header.match(self.event).groupdict()  # regex parse the header portion
                    extension = dict(regex_extension.findall(header['extension']))  # regex parse the extension **CPU INTENSE!**
                    header.pop('extension', None)  # we no longer need header[extension] so remove it
                    self.cef_dict = dict(header, **extension)  # combine both dicts into one.  Found on stackoverflow
                    try:
                        self.consumer_sender.send_pyobj(self.cef_dict)  # where the dict actually gets sent to zmq tx queue
                        self.counter += 1
                    except Exception as e:
                        self.logger.debug('processed_queue.put error')
                        self.logger.debug(e)

                except:
                    self.logger.info('CEF event with bad syntax detected')
            else:
                self.logger.info('non-CEF event')