Removed pyasn1.compat dependency.

snmpsim/commands/responder.py

@@ -14,7 +14,6 @@
 from hashlib import md5
 
 from pyasn1 import debug as pyasn1_debug
-from pyasn1.compat.octets import null
 from pyasn1.type import univ
 from pysnmp import debug as pysnmp_debug
 from pysnmp import error
@@ -25,6 +24,7 @@
 from pysnmp.entity import engine
 from pysnmp.entity.rfc3413 import cmdrsp
 from pysnmp.entity.rfc3413 import context
+from pysnmp.proto.api import v2c
 
 from snmpsim import confdir
 from snmpsim import controller
@@ -683,19 +683,21 @@ def configure_managed_objects(
 
                 context_name = agent_name
 
-                if not args.v3_only:
-                    # snmpCommunityTable::snmpCommunityIndex can't be > 32
-                    config.addV1System(
-                        snmp_engine,
-                        agent_name,
-                        community_name,
-                        contextName=context_name,
-                    )
+                if snmp_engine:
+                    if not args.v3_only:
+                        # snmpCommunityTable::snmpCommunityIndex can't be > 32
+                        config.addV1System(
+                            snmp_engine,
+                            agent_name,
+                            community_name,
+                            contextName=context_name,
+                        )
 
-                snmp_context.registerContextName(context_name, mib_instrum)
+                if snmp_context:
+                    snmp_context.registerContextName(context_name, mib_instrum)
 
-                if len(community_name) <= 32:
-                    snmp_context.registerContextName(community_name, mib_instrum)
+                    if len(community_name) <= 32:
+                        snmp_context.registerContextName(community_name, mib_instrum)
 
                 data_index_instrum_controller.add_data_file(
                     full_path, community_name, context_name
@@ -769,7 +771,7 @@ def configure_managed_objects(
                         snmp_engine, v3_context_engine_id
                     )
                     # unregister default context
-                    snmp_context.unregisterContextName(null)
+                    snmp_context.unregisterContextName(b"")
 
                     log.info(
                         "SNMPv3 Context Engine ID: "
@@ -976,7 +978,7 @@ def configure_managed_objects(
 
                 else:
                     snmp_engine = engine.SnmpEngine(
-                        snmpEngineID=univ.OctetString(hexValue=v3_engine_id)
+                        snmpEngineID=v2c.OctetString(hexValue=v3_engine_id)
                     )
 
             except Exception as exc:
@@ -988,8 +990,7 @@ def configure_managed_objects(
 
             config.addContext(snmp_engine, "")
 
-        elif opt[0] == "--v3-context-engine-id":
-            v3_context_engine_ids.append((univ.OctetString(hexValue=opt[1]), []))
+            v3_context_engine_ids.append((v2c.OctetString(hexValue=opt[1]), []))
 
         elif opt[0] == "--data-dir":
             if v3_context_engine_ids:

snmpsim/datafile.py

@@ -9,7 +9,7 @@
 import os
 import stat
 
-from pyasn1.compat.octets import str2octs
+
 from pyasn1.type import univ
 from pysnmp.carrier.asyncio.dgram import udp
 from pysnmp.carrier.asyncio.dgram import udp6
@@ -102,6 +102,7 @@ def process_var_binds(self, var_binds, **context):
             )
         )
 
+        separator = b","
         for oid, val in var_binds:
             text_oid = str(univ.OctetString(".".join(["%s" % x for x in oid])))
 
@@ -115,7 +116,7 @@ def process_var_binds(self, var_binds, **context):
                 subtree_flag = exact_match = False
 
             else:
-                offset, subtree_flag, prev_offset = line.split(str2octs(","), 2)
+                offset, subtree_flag, prev_offset = line.split(separator, 2)
                 subtree_flag, exact_match = int(subtree_flag), True
 
             offset = int(offset)
@@ -139,7 +140,7 @@ def process_var_binds(self, var_binds, **context):
                             try:
                                 _, subtree_flag, _ = self._record_index.lookup(
                                     str(_next_oid)
-                                ).split(str2octs(","), 2)
+                                ).split(separator, 2)
 
                             except KeyError:
                                 log.error(
@@ -164,7 +165,7 @@ def process_var_binds(self, var_binds, **context):
 
                     try:
                         _, _, _prev_offset = self._record_index.lookup(str(_oid)).split(
-                            str2octs(","), 2
+                            separator, 2
                         )
 
                     except KeyError:
@@ -260,6 +261,7 @@ def get_data_files(tgt_dir, top_len=None):
     # Start processing the directory
     return process_directory(tgt_dir, top_len)
 
+
 def process_directory(tgt_dir, top_len):
     # Initialize an empty list to store directory content
     dir_content = []
@@ -283,6 +285,7 @@ def process_directory(tgt_dir, top_len):
     # Return the directory content
     return dir_content
 
+
 def process_symlink(full_path, tgt_dir):
     # Read the target of the symbolic link
     full_path = os.readlink(full_path)
@@ -294,6 +297,7 @@ def process_symlink(full_path, tgt_dir):
     # Return the full path and inode
     return full_path, inode
 
+
 def process_file(d_file, full_path, rel_path):
     # Check if the file extension matches any of the record types
     for dExt in variation.RECORD_TYPES:
@@ -303,6 +307,7 @@ def process_file(d_file, full_path, rel_path):
     # If it does not, return an empty list
     return []
 
+
 def process_file_extension(d_file, full_path, rel_path, dExt):
     # Process the relative path to create an identifier for the file
     if rel_path[0] == SELF_LABEL:
@@ -318,6 +323,7 @@ def process_file_extension(d_file, full_path, rel_path, dExt):
     # Return a tuple containing the full path, the record type, and the identifier
     return [(full_path, variation.RECORD_TYPES[dExt], ident)]
 
+
 def probe_context(transport_domain, transport_address, context_engine_id, context_name):
     """Suggest variations of context name based on request data"""
     if context_engine_id:

snmpsim/grammar/dump.py

@@ -4,7 +4,6 @@
 # Copyright (c) 2010-2019, Ilya Etingof <etingof@gmail.com>
 # License: https://www.pysnmp.com/snmpsim/license.html
 #
-from pyasn1.compat.octets import octs2str
 from pyasn1.type import univ
 from pysnmp.proto import rfc1902
 
@@ -44,7 +43,7 @@ def parse(self, line):
         filters = {"4": self._nullFilter, "6": self._unhexFilter}
 
         try:
-            oid, tag, value = octs2str(line).split("|", 2)
+            oid, tag, value = line.decode("iso-8859-1").split("|", 2)
 
         except Exception as exc:
             raise error.SnmpsimError(f"broken record <{line}>: {exc}")

snmpsim/grammar/sap.py

@@ -4,7 +4,6 @@
 # Copyright (c) 2010-2019, Ilya Etingof <etingof@gmail.com>
 # License: https://www.pysnmp.com/snmpsim/license.html
 #
-from pyasn1.compat.octets import octs2str
 from pyasn1.type import univ
 from pysnmp.proto import rfc1902
 
@@ -36,7 +35,9 @@ def parse(self, line):
         filters = {"OctetString": self._stringFilter}
 
         try:
-            oid, tag, value = (x.strip() for x in octs2str(line).split(",", 2))
+            oid, tag, value = (
+                x.strip() for x in line.decode("iso-8859-1").split(",", 2)
+            )
 
         except Exception as exc:
             raise error.SnmpsimError(f"broken record <{line}>: {exc}")

snmpsim/grammar/snmprec.py

@@ -7,7 +7,7 @@
 from string import ascii_letters
 from string import digits
 
-from pyasn1.compat.octets import octs2str, str2octs, octs2ints
+
 from pyasn1.type import univ
 from pysnmp.proto import rfc1902, rfc1905
 
@@ -16,7 +16,7 @@
 
 
 class SnmprecGrammar(AbstractGrammar):
-    ALNUMS = set(octs2ints(str2octs(ascii_letters + digits)))
+    ALNUMS = set((ascii_letters + digits).encode("iso-8859-1"))
 
     TAG_MAP = {}
 
@@ -41,13 +41,13 @@ class SnmprecGrammar(AbstractGrammar):
 
     def build(self, oid, tag, val):
         if oid and tag:
-            return str2octs(f"{oid}|{tag}|{val}\n")
+            return f"{oid}|{tag}|{val}\n".encode("iso-8859-1")
 
         raise error.SnmpsimError(f"empty OID/tag <{oid}/{tag}>")
 
     def parse(self, line):
         try:
-            oid, tag, value = octs2str(line).strip().split("|", 2)
+            oid, tag, value = line.decode("iso-8859-1").strip().split("|", 2)
 
         except Exception as exc:
             raise error.SnmpsimError(f"broken record <{line}>: {exc}")

snmpsim/grammar/walk.py

@@ -7,7 +7,6 @@
 import re
 
 from pyasn1.codec.ber import encoder
-from pyasn1.compat.octets import octs2str
 from pyasn1.type import univ
 from pysnmp.proto import rfc1902
 
@@ -191,7 +190,7 @@ def parse(self, line):
         line = line.decode("ascii", "ignore").encode()
 
         try:
-            oid, value = octs2str(line).strip().split(" = ", 1)
+            oid, value = line.decode("iso-8859-1").strip().split(" = ", 1)
 
         except Exception:
             raise error.SnmpsimError("broken record <%s>" % line)

snmpsim/record/search/file.py

@@ -4,7 +4,6 @@
 # Copyright (c) 2010-2019, Ilya Etingof <etingof@gmail.com>
 # License: https://www.pysnmp.com/snmpsim/license.html
 #
-from pyasn1.compat.octets import str2octs
 
 
 # read lines from text file ignoring #comments and blank lines
@@ -18,7 +17,7 @@ def get_record(fileObj, line_no=None, offset=0):
         tline = line.strip()
 
         # skip comment or blank line
-        if not tline or tline.startswith(str2octs("#")):
+        if not tline or tline.startswith(b"#"):
             offset += len(line)
             line = fileObj.readline()
             if line_no is not None and line:
@@ -30,7 +29,7 @@ def get_record(fileObj, line_no=None, offset=0):
     return line, line_no, offset
 
 
-def find_eol(file_obj, offset, block_size=256, eol=str2octs("\n")):
+def find_eol(file_obj, offset, block_size=256, eol=b"\n"):
     while True:
         if offset < block_size:
             offset, block_size = 0, offset

snmpsim/record/snmprec.py

@@ -10,8 +10,6 @@
 from snmpsim.grammar import snmprec
 from snmpsim.record import dump
 
-from pyasn1.compat import octets
-
 
 class SnmprecRecord(dump.DumpRecord):
     grammar = snmprec.SnmprecGrammar()
@@ -88,8 +86,8 @@ def evaluate_value(self, oid, tag, value, **context):
                 return oid, tag, self.grammar.TAG_MAP[tag](value)
 
             elif encoding_id == "x":
-                if octets.isOctetsType(value):
-                    value = octets.octs2str(value)
+                if isinstance(value, bytes):
+                    value = value.decode("iso-8859-1")
 
                 return oid, tag, self.grammar.TAG_MAP[tag](hexValue=value)
 

snmpsim/variation.py

@@ -184,7 +184,7 @@ class CompressedSnmprecRecord(SnmprecRecordMixIn, snmprec.CompressedSnmprecRecor
 RECORD_TYPES[CompressedSnmprecRecord.ext] = CompressedSnmprecRecord()
 
 
-def load_variation_modules(search_path, modules_options):
+def load_variation_modules(search_path, modules_options) -> dict:
     variation_modules = {}
     modules_options = modules_options.copy()
 

snmpsim/variation/multiplex.py

@@ -11,7 +11,7 @@
 import os
 import time
 
-from pyasn1.compat.octets import str2octs
+
 from pysnmp.proto import rfc1902
 
 from snmpsim import confdir
@@ -251,7 +251,7 @@ def variate(oid, tag, value, **context):
         exactMatch = False
 
     else:
-        offset, subtreeFlag, prevOffset = line.split(str2octs(","))
+        offset, subtreeFlag, prevOffset = line.split(b",")
         exactMatch = True
 
     text.seek(int(offset))

snmpsim/variation/redis.py

@@ -21,7 +21,7 @@
 import random
 import time
 
-from pyasn1.compat import octets
+
 from pysnmp.smi.error import WrongValueError
 
 from snmpsim import error, log
@@ -97,23 +97,23 @@ def init(**context):
 def lindex(dbConn, *args):
     ret = dbConn.lindex(*args)
     if ret is not None:
-        ret = octets.octs2str(ret)
+        ret = ret.decode("iso-8859-1")
 
     return ret
 
 
 def get(dbConn, *args):
     ret = dbConn.get(*args)
     if ret is not None:
-        ret = octets.octs2str(ret)
+        ret = ret.decode("iso-8859-1")
 
     return ret
 
 
 def evalsha(dbConn, *args):
     ret = dbConn.evalsha(*args)
     if ret is not None:
-        ret = octets.octs2str(ret)
+        ret = ret.decode("iso-8859-1")
 
     return ret