fix: Accept connections on all interfaces

Accept ipv4 and ipv6 by default Listen on all interfaces
splunk · Feb 2, 2021 · ff3448e · ff3448e
1 parent 7f9ba03
commit ff3448e
Show file tree

Hide file tree

Showing 2 changed files with 67 additions and 36 deletions.
diff --git a/config.yaml b/config.yaml
@@ -3,7 +3,7 @@
 snmp:
   port: 2162
   ipv4: True
-  ipv6: False
+  ipv6: True
   communities:
     v1:
       - public

diff --git a/splunk_connect_for_snmp_traps/manager/trap_server.py b/splunk_connect_for_snmp_traps/manager/trap_server.py
@@ -7,7 +7,9 @@
 from pysnmp.entity.rfc3413 import ntfrcv
 
 from splunk_connect_for_snmp_traps.manager.hec_config import HecConfiguration
-from splunk_connect_for_snmp_traps.manager.os_config_utils import max_allowed_working_threads
+from splunk_connect_for_snmp_traps.manager.os_config_utils import (
+    max_allowed_working_threads,
+)
 
 logger = logging.getLogger(__name__)
 
@@ -21,70 +23,99 @@ def __init__(self, server_config):
         self._thread_pool_executor = self.configure_thread_pool()
 
     def configure_thread_pool(self):
-        user_suggested_working_threads = self._server_config['thread-pool']['max-suggested-working-threads']
+        user_suggested_working_threads = self._server_config["thread-pool"][
+            "max-suggested-working-threads"
+        ]
         max_workers = max_allowed_working_threads(user_suggested_working_threads)
-        logger.debug(f'Configured a thread-pool with {max_workers} concurrent threads')
+        logger.debug(f"Configured a thread-pool with {max_workers} concurrent threads")
         return concurrent.futures.ThreadPoolExecutor(max_workers=max_workers)
 
     def configure_trap_server(self):
-        self._snmp_engine.observer.registerObserver(self.request_observer, 'rfc3412.receiveMessage:request',
-                                                    'rfc3412.returnResponsePdu')
-        snmp_config = self._server_config['snmp']
-        if snmp_config['ipv4']:
-            config.addTransport(self._snmp_engine, udp.domainName,
-                                udp.UdpTransport().openServerMode(('127.0.0.1', snmp_config['port'])))
-        if snmp_config['ipv6']:
-            config.addTransport(self._snmp_engine, udp6.domainName,
-                                udp6.Udp6Transport().openServerMode(('::1', snmp_config['port'])))
+        self._snmp_engine.observer.registerObserver(
+            self.request_observer,
+            "rfc3412.receiveMessage:request",
+            "rfc3412.returnResponsePdu",
+        )
+        snmp_config = self._server_config["snmp"]
+        if snmp_config["ipv4"]:
+            config.addTransport(
+                self._snmp_engine,
+                udp.domainName,
+                udp.UdpTransport().openServerMode(("0.0.0.0", snmp_config["port"])),
+            )
+        if snmp_config["ipv6"]:
+            config.addTransport(
+                self._snmp_engine,
+                udp6.domainName,
+                udp6.Udp6Transport().openServerMode(("::0", snmp_config["port"])),
+            )
         # SNMPv1/2c setup
         # SecurityName <-> CommunityName mapping
-        for community in snmp_config['communities']['v1']:
-            logger.info(f'Configuring V1 {community}')
+        for community in snmp_config["communities"]["v1"]:
+            logger.info(f"Configuring V1 {community}")
             config.addV1System(self._snmp_engine, community, community)
         # Register SNMP Application at the SNMP engine
         ntfrcv.NotificationReceiver(self._snmp_engine, self.snmp_callback_function)
 
     def post_trap_to_hec(self, variables_binds):
-        logger.debug('Task received, sleeping for few seconds ...')
+        logger.debug("Task received, sleeping for few seconds ...")
         for endpoint in self._hec_config.get_endpoints():
-            headers = {'Authorization': f'Splunk {self._hec_config.get_authentication_token()}'}
-            splunk_trap_data = ','.join([str(key) + str(value) for key, value in variables_binds])
-            data = {'sourcetype': 'trap-server', 'event': splunk_trap_data}
-            logger.debug(f'Posting trap to HEC using {endpoint} and headers {headers}')
-            response = requests.post(url=endpoint, json=data, headers=headers, verify=self._hec_config.is_ssl_enabled())
-            logger.debug(f'Response code is {response.status_code}')
+            headers = {
+                "Authorization": f"Splunk {self._hec_config.get_authentication_token()}"
+            }
+            splunk_trap_data = ",".join(
+                [str(key) + str(value) for key, value in variables_binds]
+            )
+            data = {"sourcetype": "trap-server", "event": splunk_trap_data}
+            logger.debug(f"Posting trap to HEC using {endpoint} and headers {headers}")
+            response = requests.post(
+                url=endpoint,
+                json=data,
+                headers=headers,
+                verify=self._hec_config.is_ssl_enabled(),
+            )
+            logger.debug(f"Response code is {response.status_code}")
 
     # Register a callback to be invoked at specified execution point of
     # SNMP Engine and passed local variables at code point's local scope
     # noinspection PyUnusedLocal,PyUnusedLocal
     def request_observer(self, snmp_engine, execution_point, variables, callback_ctx):
         logger.debug(f'Raw data is "{variables}"')
-        logger.debug('Execution point: %s' % execution_point)
+        logger.debug("Execution point: %s" % execution_point)
+        logger.debug(
+            "* transportDomain: %s"
+            % ".".join([str(x) for x in variables["transportDomain"]])
+        )
         logger.debug(
-            '* transportDomain: %s'
-            % '.'.join([str(x) for x in variables['transportDomain']])
+            "* transportAddress: %s"
+            % "@".join([str(x) for x in variables["transportAddress"]])
         )
+        logger.debug("* securityModel: %s" % variables["securityModel"])
+        logger.debug("* securityName: %s" % variables["securityName"])
+        logger.debug("* securityLevel: %s" % variables["securityLevel"])
         logger.debug(
-            '* transportAddress: %s'
-            % '@'.join([str(x) for x in variables['transportAddress']])
+            "* contextEngineId: %s" % variables["contextEngineId"].prettyPrint()
         )
-        logger.debug('* securityModel: %s' % variables['securityModel'])
-        logger.debug('* securityName: %s' % variables['securityName'])
-        logger.debug('* securityLevel: %s' % variables['securityLevel'])
-        logger.debug('* contextEngineId: %s' % variables['contextEngineId'].prettyPrint())
-        logger.debug('* contextName: %s' % variables['contextName'].prettyPrint())
-        logger.debug('* PDU: %s' % variables['pdu'].prettyPrint())
+        logger.debug("* contextName: %s" % variables["contextName"].prettyPrint())
+        logger.debug("* PDU: %s" % variables["pdu"].prettyPrint())
 
     # Callback function for receiving notifications
     # noinspection PyUnusedLocal,PyUnusedLocal,PyUnusedLocal
-    def snmp_callback_function(self, snmp_engine, state_reference, context_engine_id, context_name, var_binds,
-                               callback_ctx):
+    def snmp_callback_function(
+        self,
+        snmp_engine,
+        state_reference,
+        context_engine_id,
+        context_name,
+        var_binds,
+        callback_ctx,
+    ):
         logger.debug(
             'Notification from ContextEngineId "%s", ContextName "%s"'
             % (context_engine_id.prettyPrint(), context_name.prettyPrint())
         )
         for name, val in var_binds:
-            logger.debug('%s = %s' % (name.prettyPrint(), val.prettyPrint()))
+            logger.debug("%s = %s" % (name.prettyPrint(), val.prettyPrint()))
         self._thread_pool_executor.submit(self.post_trap_to_hec, var_binds)
 
     def run_trap_server(self):