Merge pull request #182 from d-grossman/master

Witness!

Author: d-grossman

poseidon/poseidonMain/poseidonMain.py

@@ -164,14 +164,21 @@ def start_monitor(self, ivalue):
 
     def stop_monitor(self, ivalue):
         ''' stop monitoring an address'''
+
+        for my_hash, my_dict in ivalue.iteritems():
+            if my_hash in self.monitoring:
+                self.monitoring.pop(my_hash)
+
+        self.logger.debug('stop_monitor:{0},{1}'.format(ivalue, type(ivalue)))
         r_exchange = 'topic-poseidon-internal'
-        r_key = 'poseidon.action.start_monitor'
+        r_key = 'poseidon.action.stop_monitor'
         r_msg = json.dumps(ivalue)
         self.rabbit_channel_local.basic_publish(exchange=r_exchange,
                                                 routing_key=r_key,
                                                 body=r_msg)
 
     def endpoint_shutdown(self, ivalue):
+        self.logger.debug('endpoint_shutdown:{0}'.format(ivalue))
         ''' shutdown an endpoint '''
         r_exchange = 'topic-poseidon-internal'
         r_key = 'poseidon.action.endpoint_shutdown'
@@ -197,13 +204,15 @@ def check_db(self, dev_hash, field):
         '''
         try:
             query = {'dev_id': dev_hash}
+            query_string = str(query).replace("\'", "\"")
             ip = self.mod_configuration['storage_interface_ip']
             port = self.mod_configuration['storage_interface_port']
             uri = 'http://' + ip + ':' + port + \
                   '/v1/storage/query/{database}/{collection}/{query_str}'.format(
                       database=self.mod_configuration['database'],
                       collection=self.mod_configuration['collection'],
-                      query_str=query)
+                      query_str=query_string)
+            self.logger.error('check_db:{0}:{1}'.format(uri, type(uri)))
             resp = requests.get(uri)
             self.logger.debug('response from db:' + resp.text)
 
@@ -241,46 +250,49 @@ def start_vent_collector(self, dev_hash, num_captures=1):
         except Exception, e:
             self.logger.debug('failed to start vent collector' + str(e))
 
+    @staticmethod
+    def just_the_hash(ivalue):
+        return ivalue.keys()[0]
+
     def handle_item(self, itype, ivalue):
         self.logger.debug('handle_item:{0}:{1}'.format(itype, ivalue))
-        ivalue = json.loads(ivalue)
+
+        # just get a string back from the ml stuff
+        if 'poseidon.algos.eval_dev_class' not in itype:
+            ivalue = json.loads(ivalue)
+
         if itype == 'poseidon.action.shutdown':
             self.logger.debug('***** shutting down')
             self.shutdown = True
         if itype == 'poseidon.action.new_machine':
             self.logger.debug('***** new machine {0}'.format(ivalue))
             # tell monitor to monitor
-            self.start_vent_collector(ivalue)
+            self.start_vent_collector(self.just_the_hash(ivalue))
             self.start_monitor(ivalue)
-        if itype == 'poseidon.analytics.results.traditional':
-            # TODO make a db call
-            # need to compare results to db
-            if False:
-                # if bad
-                self.logger.debug(
-                    '***** shutting down endpoint:{0}:{1}'.format(itype, ivalue))
-                self.endpoint_shutdown(ivalue)
-            else:
-                # if good
-                self.logger.debug(
-                    '***** allowing endpoint {0}:{1}'.format(itype,  ivalue))
-                self.stop_monitor(ivalue)
         if 'poseidon.algos.eval_dev_class' in itype:
+            # ivalue = classificationtype:<string>
             # result form eval device classifier with
             # dev hash attached to end of routing key
-            dev_hash = ivalue.split('.')[-1]
+            dev_hash = itype.split('.')[-1]
             prev_class = self.check_db(dev_hash, 'dev_classification')
+
             monitoring_id = self.monitoring[dev_hash]
-            self.stop_monitor(monitoring_id)
+            temp_d = {dev_hash: monitoring_id}
+
+            # self.stop_monitor(monitoring_id)
+            self.stop_monitor(temp_d)
+
             self.logger.debug('stopping monitoring on:' + itype)
+            self.logger.debug('classified as:{0}'.format(ivalue))
+            self.logger.debug('classified previously {0}'.format(prev_class))
             if ivalue == prev_class:
                 self.logger.debug(
-                    '***** allowing endpoint {0}:{1}'.format(itype,  ivalue))
-                self.endpoint_allow(monitoring_id)
+                    '***** allowing endpoint {0}:{1}'.format(itype,  temp_d))
+                self.endpoint_allow(temp_d)
             else:
                 self.logger.debug(
-                    '***** shutting down endpoint:{0}:{1}'.format(itype, ivalue))
-                self.endpoint_shutdown(monitoring_id)
+                    '***** shutting down endpoint:{0}:{1}'.format(itype, temp_d))
+                self.endpoint_shutdown(temp_d)
 
     def make_rabbit_connection(self, host, exchange, queue_name, keys):  # pragma: no cover
         '''

poseidon/poseidonMonitor/NorthBoundControllerAbstraction/NorthBoundControllerAbstraction.py

@@ -96,6 +96,7 @@ def __init__(self):
         self.prev_endpoints = {}
         self.new_endpoints = {}
         self.mirroring = {}
+        self.shutdown = {}
         self.do_rabbit = True
         self.m_queue = Queue.Queue()
         self.rabbit_connection_local = None
@@ -171,7 +172,8 @@ def init_rabbit(self):  # pragma: no cover
         host = 'poseidon-rabbit'
         exchange = 'topic-poseidon-internal'
         queue_name = 'poseidon_NBCA'
-        binding_key = ['poseidon.algos.#', 'poseidon.action.#']
+        # binding_key = ['poseidon.algos.#', 'poseidon.action.#']
+        binding_key = ['poseidon.action.#']
         retval = self.make_rabbit_connection(
             host, exchange, queue_name, binding_key)
         self.rabbit_channel_local = retval[0]
@@ -244,6 +246,28 @@ def handle_item(self, item):
                 self.bcf.mirror_ip(my_dict['ip-address'])
                 self.mirroring[my_hash] = my_dict
 
+        if itype == 'poseidon.action.endpoint_shutdown':
+            self.logger.debug(
+                'endpoint_shutdown:{0}:{1}'.format(ivalue, type(ivalue)))
+            for my_hash, my_dict in ivalue.iteritems():
+                bad_ip = my_dict.get('ip-address')
+                if bad_ip is not None:
+                    self.logger.debug(
+                        '****** shutdown {0}:{1}'.format(bad_ip, ivalue))
+                    self.bcf.shutdown_ip(bad_ip)
+                    self.shutdown[my_hash] = my_dict
+
+        if itype == 'poseidon.action.stop_monitor':
+            self.logger.debug('stop_monitor:{0}:{1}'.format(itype, ivalue))
+            for my_hash, my_dict in ivalue.iteritems():
+                self.logger.debug('stop_monitor_dict:{0}'.format(my_dict))
+                my_ip = my_dict.get('ip-address')
+                if my_ip is not None:
+                    self.logger.debug('***** shutting down {0}'.format(my_ip))
+                    self.bcf.unmirror_ip(my_ip)
+                    if my_hash in self.mirroring:
+                        self.mirroring.pop(my_hash)
+
     def get_rabbit_work(self):
         '''get work item from queue if exists'''
         # type , value
@@ -285,14 +309,17 @@ def find_new_machines(self, machines):
 
     def print_state(self):
         self.logger.debug('**************PREV*****************')
-        for my_hash, my_value in self.prev_endpoints.iteritems():
-            self.logger.debug('P:{0}:{1}'.format(my_hash, my_value))
+        for my_hash, my_dict in self.prev_endpoints.iteritems():
+            self.logger.debug('P:{0}:{1}'.format(my_hash, my_dict))
         self.logger.debug('**************NEW******************')
-        for my_hash, my_value in self.new_endpoints.iteritems():
-            self.logger.debug('N:{0}:{1}'.format(my_hash, my_value))
+        for my_hash, my_dict in self.new_endpoints.iteritems():
+            self.logger.debug('N:{0}:{1}'.format(my_hash, my_dict))
         self.logger.debug('***********MIRRORING***************')
-        for my_hash, my_value in self.mirroring.iteritems():
-            self.logger.debug('M:{0}:{1}'.format(my_hash, my_value))
+        for my_hash, my_dict in self.mirroring.iteritems():
+            self.logger.debug('M:{0}:{1}'.format(my_hash, my_dict))
+        self.logger.debug('***********SHUTDOWN****************')
+        for my_hash, my_dict in self.shutdown.iteritems():
+            self.logger.debug('M:{0}:{1}'.format(my_hash, my_dict))
 
     def send_new_machines(self):
         '''send listing of new machines to main for decisions'''

poseidon/poseidonMonitor/NorthBoundControllerAbstraction/proxy/bcf/bcf.py

@@ -165,10 +165,12 @@ def shutdown_ip(self, ip_addr, shutdown=True, mac_addr=None):
             tenant = record.get('tenant')
             segment = record.get('segment')
             mac = record.get('mac')
-            name = '{0}{1}{2}'.format(tenant, segment, mac)
+            name = '{0}{1}{2}'.format(tenant, segment, mac).replace(':', '')
             if record.get('name') is not None:
                 name = record['name']
             module_logger.debug('bcf shutting down: {0}'.format(record))
+            module_logger.debug('t:{0} s:{1} n:{2} m{3} shut:{4}'.format(
+                tenant, segment, name, mac, shutdown))
             self.shutdown_endpoint(tenant, segment, name, mac, shutdown)
             shutdowns.append(record)
         return shutdowns
@@ -212,7 +214,7 @@ def get_highest(span_fabric):
             return (my_max + 1)
         else:
             module_logger.debug('noFilters online')
-            return 0
+            return 1
 
     def get_seq_by_ip(self, ip):
         my_filter = self.get_span_fabric()[0].get('filter')