Skip to content

Fix cmd2rec script #11

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
JeThWifirst wants to merge 4 commits into
base: master
Choose a base branch
from
Open

Fix cmd2rec script #11

Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
8a1a8b6
commands/cmd2rec: clean up: use new snake_case functions
JeThWifirst Feb 27, 2026
3880983
commands/cmd2rec: add missing SNMPv3 protocol choice
JeThWifirst Feb 27, 2026
274ee55
commands/cmd2rec: fix OID and value retrieval
JeThWifirst Feb 27, 2026
ed1d4b0
commands/cmd2rec: send request PDU to continue walking
JeThWifirst Feb 27, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
240 changes: 134 additions & 106 deletions snmpsim/commands/cmd2rec.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -127,9 +127,9 @@ def main():

v1arch_group.add_argument(
"--protocol-version",
choices=["1", "2c"],
choices=["1", "2c", "3"],
default="2c",
help="SNMPv1/v2c protocol version",
help="SNMPv1/v2c/v3 protocol version",
)

v1arch_group.add_argument(
Expand Down Expand Up @@ -298,7 +298,7 @@ def main():
args = parser.parse_args()

if args.debug:
pysnmp_debug.setLogger(pysnmp_debug.Debug(*args.debug))
pysnmp_debug.set_logger(pysnmp_debug.Debug(*args.debug))

if args.debug_asn1:
pyasn1_debug.setLogger(pyasn1_debug.Debug(*args.debug_asn1))
Expand Down Expand Up @@ -511,9 +511,9 @@ def main():
if isinstance(args.start_object, ObjectIdentity) or isinstance(
args.stop_object, ObjectIdentity
):
compiler.addMibCompiler(snmp_engine.getMibBuilder(), sources=args.mib_sources)
compiler.addMibCompiler(snmp_engine.get_mib_builder(), sources=args.mib_sources)

mib_view_controller = view.MibViewController(snmp_engine.getMibBuilder())
mib_view_controller = view.MibViewController(snmp_engine.get_mib_builder())

try:
if isinstance(args.start_object, ObjectIdentity):
Expand Down Expand Up @@ -620,7 +620,7 @@ def cbFun(

# initiate another SNMP walk iteration
if args.use_getbulk:
cmd_gen.sendVarBinds(
cmd_gen.send_varbinds(
snmp_engine,
"tgt",
args.v3_context_engine_id,
Expand All @@ -633,7 +633,7 @@ def cbFun(
)

else:
cmd_gen.sendVarBinds(
cmd_gen.send_varbinds(
snmp_engine,
"tgt",
args.v3_context_engine_id,
Expand All @@ -657,111 +657,139 @@ def cbFun(

# Walk var-binds
for var_bind_row in var_bind_table:
for oid, value in var_bind_row:
# EOM
if args.stop_object and oid >= args.stop_object:
stop_flag = True # stop on out of range condition

elif value is None or value.tagSet in (
rfc1905.NoSuchObject.tagSet,
rfc1905.NoSuchInstance.tagSet,
rfc1905.EndOfMibView.tagSet,
):
stop_flag = True

# remove value enumeration
if value.tagSet == rfc1902.Integer32.tagSet:
value = rfc1902.Integer32(value)

if value.tagSet == rfc1902.Unsigned32.tagSet:
value = rfc1902.Unsigned32(value)

if value.tagSet == rfc1902.Bits.tagSet:
value = rfc1902.OctetString(value)

# Build .snmprec record

context = {
"origOid": oid,
"origValue": value,
"count": cb_ctx["count"],
"total": cb_ctx["total"],
"iteration": cb_ctx["iteration"],
"reqTime": cb_ctx["reqTime"],
"args.start_object": args.start_object,
"stopOID": args.stop_object,
"stopFlag": stop_flag,
"variationModule": variation_module,
}
oid, value = var_bind_row
# EOM
if args.stop_object and oid >= args.stop_object:
stop_flag = True # stop on out of range condition

elif value is None or value.tagSet in (
rfc1905.NoSuchObject.tagSet,
rfc1905.NoSuchInstance.tagSet,
rfc1905.EndOfMibView.tagSet,
):
stop_flag = True

# remove value enumeration
if value.tagSet == rfc1902.Integer32.tagSet:
value = rfc1902.Integer32(value)

if value.tagSet == rfc1902.Unsigned32.tagSet:
value = rfc1902.Unsigned32(value)

if value.tagSet == rfc1902.Bits.tagSet:
value = rfc1902.OctetString(value)

# Build .snmprec record

context = {
"origOid": oid,
"origValue": value,
"count": cb_ctx["count"],
"total": cb_ctx["total"],
"iteration": cb_ctx["iteration"],
"reqTime": cb_ctx["reqTime"],
"args.start_object": args.start_object,
"stopOID": args.stop_object,
"stopFlag": stop_flag,
"variationModule": variation_module,
}

try:
line = data_file_handler.format(oid, value, **context)

except error.MoreDataNotification as exc:
cb_ctx["count"] = 0
cb_ctx["iteration"] += 1

more_data_notification = exc

if "period" in more_data_notification:
log.info(
"%s OIDs dumped, waiting %.2f sec(s)"
"..." % (cb_ctx["total"], more_data_notification["period"])
)

time.sleep(more_data_notification["period"])

# initiate another SNMP walk iteration
if args.use_getbulk:
cmd_gen.sendVarBinds(
snmp_engine,
"tgt",
args.v3_context_engine_id,
args.v3_context_name,
0,
args.getbulk_repetitions,
[(args.start_object, None)],
cbFun,
cb_ctx,
)

else:
cmd_gen.sendVarBinds(
snmp_engine,
"tgt",
args.v3_context_engine_id,
args.v3_context_name,
[(args.start_object, None)],
cbFun,
cb_ctx,
)

stop_flag = True # stop current iteration

except error.NoDataNotification:
pass
try:
line = data_file_handler.format(oid, value, **context)

except error.MoreDataNotification as exc:
cb_ctx["count"] = 0
cb_ctx["iteration"] += 1

more_data_notification = exc

except error.SnmpsimError as exc:
log.error(exc)
continue
if "period" in more_data_notification:
log.info(
"%s OIDs dumped, waiting %.2f sec(s)"
"..." % (cb_ctx["total"], more_data_notification["period"])
)

time.sleep(more_data_notification["period"])

# initiate another SNMP walk iteration
if args.use_getbulk:
cmd_gen.send_varbinds(
snmp_engine,
"tgt",
args.v3_context_engine_id,
args.v3_context_name,
args.getbulk_repetitions,
[(args.start_object, None)],
cbFun,
cb_ctx,
)

else:
args.output_file.write(line)
cmd_gen.send_varbinds(
snmp_engine,
"tgt",
args.v3_context_engine_id,
args.v3_context_name,
[(args.start_object, None)],
cbFun,
cb_ctx,
)

stop_flag = True # stop current iteration

except error.NoDataNotification:
pass

except error.SnmpsimError as exc:
log.error(exc)
continue

cb_ctx["count"] += 1
cb_ctx["total"] += 1
else:
args.output_file.write(line)

if cb_ctx["count"] % 100 == 0:
log.info(
"OIDs dumped: %s/%s"
% (cb_ctx["iteration"], cb_ctx["count"])
)
cb_ctx["count"] += 1
cb_ctx["total"] += 1

# Next request time
cb_ctx["reqTime"] = time.time()
if cb_ctx["count"] % 100 == 0:
log.info(
"OIDs dumped: %s/%s"
% (cb_ctx["iteration"], cb_ctx["count"])
)

# Continue walking
return not stop_flag
if not stop_flag:
# Next request time
cb_ctx["reqTime"] = time.time()

if args.use_getbulk:
cmd_gen.send_varbinds(
snmp_engine,
"tgt",
args.v3_context_engine_id,
args.v3_context_name,
args.getbulk_repetitions,
var_bind_table,
cbFun,
cb_ctx,
)

else:
cmd_gen.send_varbinds(
snmp_engine,
"tgt",
args.v3_context_engine_id,
args.v3_context_name,
var_bind_table,
cbFun,
cb_ctx,
)

return not stop_flag

# End of walk, stop the dispatcher loop and close it
if snmp_engine.transport_dispatcher.loop.is_running():
snmp_engine.transport_dispatcher.loop.stop()
snmp_engine.transport_dispatcher.close_dispatcher()

cb_ctx = {
"total": 0,
Expand All @@ -776,7 +804,7 @@ def cbFun(
if args.use_getbulk:
cmd_gen = cmdgen.BulkCommandGenerator()

cmd_gen.sendVarBinds(
cmd_gen.send_varbinds(
snmp_engine,
"tgt",
args.v3_context_engine_id,
Expand All @@ -791,7 +819,7 @@ def cbFun(
else:
cmd_gen = cmdgen.NextCommandGenerator()

cmd_gen.sendVarBinds(
cmd_gen.send_varbinds(
snmp_engine,
"tgt",
args.v3_context_engine_id,
Expand All @@ -814,7 +842,7 @@ def cbFun(
started = time.time()

try:
snmp_engine.transportDispatcher.runDispatcher()
snmp_engine.transport_dispatcher.run_dispatcher()

except KeyboardInterrupt:
log.info("Shutting down process...")
Expand Down