Handle non-ascii community values in incoming traps without crashing

changelog.d/9.fixed.md

@@ -0,0 +1 @@
+Handle non-ASCII trap message community values gracefully without crashing

src/netsnmpy/trapsession.py

@@ -4,7 +4,7 @@
 import platform
 from ipaddress import ip_address
 from socket import AF_INET, AF_INET6, inet_ntop
-from typing import Optional, Protocol
+from typing import Optional, Protocol, Union
 
 from netsnmpy import _netsnmp
 from netsnmpy.annotations import IPAddress
@@ -180,7 +180,7 @@ def __init__(
         generic_type: str,
         trap_oid: OID,
         uptime: int,
-        community: str,
+        community: Union[str, bytes],
         version: str,
         variables: VarBindList,
     ):
@@ -208,7 +208,14 @@ def from_pdu(cls, pdu: _ffi.CData) -> "SNMPTrap":
 
         source = cls.get_transport_addr(pdu)
         agent_addr = generic_type = trap_oid = uptime = None
-        community = _ffi.string(pdu.community).decode()
+        community = _ffi.string(pdu.community)
+        try:
+            community = community.decode("ascii")
+        except UnicodeDecodeError:
+            # SNMP communities are defined to be ASCII, but not every entity plays
+            # nice. If this is non-ascii, we'll keep the original bytestring as-is
+            # and leave it up to the client program to deal with it.
+            pass
 
         version = pdu.version
         if version == SNMP_VERSION_1:

tests/test_trapsession.py

@@ -0,0 +1,40 @@
+import pytest
+
+from netsnmpy.constants import SNMP_MSG_INFORM, SNMP_VERSION_2c
+from netsnmpy.netsnmp import ValueType, encode_variable, oid_to_c
+from netsnmpy.oids import OID
+from netsnmpy.trapsession import SNMPTrap, _ffi, _netsnmp
+
+_lib = _netsnmp.lib
+
+
+class TestSNMPTrap:
+    def test_given_trap_with_non_ascii_community_then_from_pdu_should_not_crash(
+        self, pdu_with_garbage_community, garbage_community
+    ):
+        trap = SNMPTrap.from_pdu(pdu_with_garbage_community)
+        assert isinstance(trap.community, bytes)
+        assert trap.community == garbage_community
+
+
+@pytest.fixture
+def pdu_with_garbage_community(garbage_community):
+    pdu = _lib.snmp_pdu_create(SNMP_MSG_INFORM)
+    pdu.version = SNMP_VERSION_2c
+
+    community_c = _ffi.new("char[]", garbage_community)
+    pdu.community = community_c
+
+    oid = oid_to_c(OID(".1.3.6.1.2.1.1.6.0"))
+    value_type = ValueType.OCTETSTRING
+    encoded_value = encode_variable(ValueType.OCTETSTRING, b"Milliways")
+    _lib.snmp_add_var(
+        pdu, oid, len(oid), value_type.value.encode("utf-8"), encoded_value
+    )
+
+    yield pdu
+
+
+@pytest.fixture
+def garbage_community():
+    yield b"foo\xcbbar"