fix: add communities for traps v1 (#1138)

Author: ajasnosz

CHANGELOG.md

@@ -7,6 +7,7 @@
 - allow to set subnets for docker network configuration from .env file
 
 ### Fixed
+- add communities for v1 traps
 
 ## [1.12.1]
 

integration_tests/test_trap_integration.py

@@ -33,10 +33,12 @@
 logger = logging.getLogger(__name__)
 
 
-def send_trap(host, port, object_identity, mib_to_load, *var_binds):
+def send_trap(
+    host, port, object_identity, mib_to_load, community, mp_model, *var_binds
+):
     iterator = sendNotification(
         SnmpEngine(),
-        CommunityData("public", mpModel=0),
+        CommunityData(community, mpModel=mp_model),
         UdpTransportTarget((host, port)),
         ContextData(),
         "trap",
@@ -73,20 +75,57 @@ def send_v3_trap(host, port, object_identity, *var_binds):
         logger.error(f"{error_indication}")
 
 
-def test_integration(request, setup_splunk):
+def test_trap_v1(request, setup_splunk):
     trap_external_ip = request.config.getoption("trap_external_ip")
     logger.info(f"I have: {trap_external_ip}")
 
     time.sleep(2)
     # send trap
     varbind1 = ("1.3.6.1.6.3.1.1.4.3.0", "1.3.6.1.4.1.20408.4.1.1.2")
-    varbind2 = ("1.3.6.1.2.1.1.1.0", OctetString("my system"))
+    varbind2 = ("1.3.6.1.2.1.1.4.0", OctetString("my contact"))
     send_trap(
-        trap_external_ip, 162, "1.3.6.1.6.3.1.1.5.2", "SNMPv2-MIB", varbind1, varbind2
+        trap_external_ip,
+        162,
+        "1.3.6.1.6.3.1.1.5.2",
+        "SNMPv2-MIB",
+        "publicv1",
+        0,
+        varbind1,
+        varbind2,
     )
 
     # wait for the message to be processed
+    time.sleep(5)
+
+    search_query = """search index="netops" sourcetype="sc4snmp:traps" earliest=-1m
+                     | head 1"""
+
+    result_count, events_count = splunk_single_search(setup_splunk, search_query)
+
+    assert result_count == 1
+
+
+def test_trap_v2(request, setup_splunk):
+    trap_external_ip = request.config.getoption("trap_external_ip")
+    logger.info(f"I have: {trap_external_ip}")
+
     time.sleep(2)
+    # send trap
+    varbind1 = ("1.3.6.1.6.3.1.1.4.3.0", "1.3.6.1.4.1.20408.4.1.1.2")
+    varbind2 = ("1.3.6.1.2.1.1.1.0", OctetString("my system"))
+    send_trap(
+        trap_external_ip,
+        162,
+        "1.3.6.1.6.3.1.1.5.2",
+        "SNMPv2-MIB",
+        "homelab",
+        1,
+        varbind1,
+        varbind2,
+    )
+
+    # wait for the message to be processed
+    time.sleep(5)
 
     search_query = """search index="netops" sourcetype="sc4snmp:traps" earliest=-1m
                      | head 1"""
@@ -103,10 +142,12 @@ def test_added_varbind(request, setup_splunk):
     time.sleep(2)
     # send trap
     varbind1 = ("1.3.6.1.2.1.1.1.0", OctetString("test_added_varbind"))
-    send_trap(trap_external_ip, 162, "1.3.6.1.2.1.2.1", "SNMPv2-MIB", varbind1)
+    send_trap(
+        trap_external_ip, 162, "1.3.6.1.2.1.2.1", "SNMPv2-MIB", "public", 1, varbind1
+    )
 
     # wait for the message to be processed
-    time.sleep(2)
+    time.sleep(5)
 
     search_query = (
         """search index="netops" "SNMPv2-MIB.sysDescr.value"="test_added_varbind" """
@@ -125,7 +166,15 @@ def test_many_traps(request, setup_splunk):
     # send trap
     varbind1 = ("1.3.6.1.2.1.1.1.0", OctetString("test_many_traps"))
     for _ in range(5):
-        send_trap(trap_external_ip, 162, "1.3.6.1.2.1.2.1", "SNMPv2-MIB", varbind1)
+        send_trap(
+            trap_external_ip,
+            162,
+            "1.3.6.1.2.1.2.1",
+            "SNMPv2-MIB",
+            "public",
+            1,
+            varbind1,
+        )
 
     # wait for the message to be processed
     time.sleep(2)
@@ -148,7 +197,14 @@ def test_more_than_one_varbind(request, setup_splunk):
     varbind1 = ("1.3.6.1.2.1.1.4.0", OctetString("test_more_than_one_varbind_contact"))
     varbind2 = ("1.3.6.1.2.1.1.1.0", OctetString("test_more_than_one_varbind"))
     send_trap(
-        trap_external_ip, 162, "1.3.6.1.2.1.2.1", "SNMPv2-MIB", varbind1, varbind2
+        trap_external_ip,
+        162,
+        "1.3.6.1.2.1.2.1",
+        "SNMPv2-MIB",
+        "public",
+        1,
+        varbind1,
+        varbind2,
     )
 
     # wait for the message to be processed
@@ -170,7 +226,13 @@ def test_loading_mibs(request, setup_splunk):
     # send trap
     varbind1 = ("1.3.6.1.6.3.1.1.4.1.0", "1.3.6.1.4.1.15597.1.1.1.1.0.1")
     send_trap(
-        trap_external_ip, 162, "1.3.6.1.4.1.15597.1.1.1.1", "SNMPv2-MIB", varbind1
+        trap_external_ip,
+        162,
+        "1.3.6.1.4.1.15597.1.1.1.1",
+        "SNMPv2-MIB",
+        "public",
+        1,
+        varbind1,
     )
 
     # wait for the message to be processed

integration_tests/traps-config.yaml

@@ -1,4 +1,7 @@
 communities:
+  1:
+    publicv1:
   2c:
     public:
+    homelab:
 usernameSecrets: []

integration_tests/values.yaml

@@ -12,6 +12,8 @@ image:
 traps:
   replicaCount: 1
   communities:
+    1:
+      - publicv1
     2c:
       - public
       - homelab

splunk_connect_for_snmp/traps.py

@@ -142,6 +142,20 @@ def authentication_observer_cb_fun(snmp_engine, execpoint, variables, contexts):
 )
 
 
+def add_communities(config_base, snmp_engine):
+    idx = 0
+    if "communities" in config_base:
+        if "2c" in config_base["communities"]:
+            for community in config_base["communities"]["2c"]:
+                idx += 1
+                config.addV1System(snmp_engine, idx, community)
+        if "1" in config_base["communities"] or 1 in config_base["communities"]:
+            v = config_base["communities"].get("1", config_base["communities"].get(1))
+            for community in v:
+                idx += 1
+                config.addV1System(snmp_engine, idx, community)
+
+
 def main():
     # Get the event loop for this thread
     loop = asyncio.new_event_loop()
@@ -176,11 +190,8 @@ def main():
 
     with open(CONFIG_PATH, encoding="utf-8") as file:
         config_base = yaml.safe_load(file)
-    idx = 0
-    if "communities" in config_base and "2c" in config_base["communities"]:
-        for community in config_base["communities"]["2c"]:
-            idx += 1
-            config.addV1System(snmp_engine, idx, community)
+
+    add_communities(config_base, snmp_engine)
 
     if "usernameSecrets" in config_base:
         for secret in config_base["usernameSecrets"]: