# src/netgear_switch/protocols/snmp/parse.py
"""Pure SNMP-row -> models.py parsers. No I/O."""
from __future__ import annotations
import string
from types import MappingProxyType
from typing import TYPE_CHECKING
from ...models import (
IpMode,
LLDPNeighbor,
MacEntry,
MgmtIpConfig,
PoEDetect,
PoEStatus,
PortStats,
PortStatus,
Sensor,
SyslogConfig,
SyslogServer,
VLANInfo,
)
from .client import SnmpError, SnmpRow
if TYPE_CHECKING:
from collections.abc import Mapping, Sequence
from ...registry import SwitchModel
def _suffix(row: SnmpRow, base: str) -> str | None:
prefix = base + "."
if not row.oid.startswith(prefix):
return None
return row.oid[len(prefix) :]
[docs]
def index_int_column(rows: Sequence[SnmpRow], base_oid: str) -> dict[int, int]:
"""Map a single-int-index column walk to {index: int_value}.
Raises SnmpError on a value that is not an integer under base_oid: the
walk is pinned to one column, so a non-integer means the table drifted.
"""
out: dict[int, int] = {}
for row in rows:
suffix = _suffix(row, base_oid)
if suffix is None or "." in suffix:
continue
try:
idx = int(suffix)
except ValueError as exc:
raise SnmpError(f"malformed index {suffix!r} at {row.oid}") from exc
try:
value = int(row.value)
except ValueError as exc:
raise SnmpError(f"non-integer value {row.value!r} at {row.oid}") from exc
out[idx] = value
return out
[docs]
def index_str_column(rows: Sequence[SnmpRow], base_oid: str) -> dict[int, str]:
"""Map a single-index column walk to {index: str_value}.
An absent column (no rows under base_oid) yields an empty dict. But a row
that IS present under base_oid with a single, non-integer index component is
table drift, not absence, and raises SnmpError naming the offending OID —
consistent with index_int_column. (A multi-component suffix belongs to a
different, deeper column and is skipped.)
A text-name OCTET STRING (ifName/ifAlias/dot1qVlanStaticName) can
legitimately arrive as ``str`` from the CLI transport or ``bytes`` from
the pysnmp transport (its non-printable heuristic picks Hex-STRING for
values with any byte outside the printable-ASCII range, e.g. a name with
a trailing NUL). Both are valid text here: ``bytes`` is decoded to
``str`` (utf-8, replacing undecodable bytes) so the two transports yield
the same model value. Any other type (e.g. an int) is a genuine wrong-type
reply and still raises.
"""
out: dict[int, str] = {}
for row in rows:
suffix = _suffix(row, base_oid)
if suffix is None or "." in suffix:
continue
try:
idx = int(suffix)
except ValueError as exc:
raise SnmpError(f"non-integer index {suffix!r} at {row.oid}") from exc
value = row.value
if isinstance(value, bytes):
value = value.decode("utf-8", "replace")
elif not isinstance(value, str):
raise SnmpError(f"non-string value {value!r} at {row.oid}")
out[idx] = value
return out
ETHERNET_CSMACD = 6 # ifType value for a physical Ethernet port
[docs]
def physical_ports(if_types: Sequence[SnmpRow]) -> set[int] | None:
"""The set of physical (ethernetCsmacd) ifIndexes from an ifType walk.
Returns ``None`` when the walk is EMPTY -- the caller then keeps every
interface, so a transport/mock that does not surface ifType is unchanged.
When ifType IS present (every real switch), non-physical interfaces are
excluded: the M4300 ifTable carries 128 ieee8023adLag(161) + a CPU(1) + a
VLAN(135) interface alongside its 16 ethernetCsmacd(6) ports, none of which
the web UI's port pages list -- so filtering here makes SNMP get_ports/
get_pvids agree field-for-field with the HTTP backend."""
from . import oids
type_map = index_int_column(if_types, oids.IF_TYPE)
if not type_map:
return None
return {idx for idx, t in type_map.items() if t == ETHERNET_CSMACD}
#: dot3StatsDuplexStatus -> PortStatus.full_duplex. 1 (unknown) maps to None
#: rather than False: "the agent does not know" and "this link is half duplex"
#: are different answers. MEASURED on the GS728TPP -- every link-up port reads
#: 3, every link-down port reads 1.
_DUPLEX_STATUS = {2: False, 3: True}
[docs]
def parse_port_status(
admin: Sequence[SnmpRow],
oper: Sequence[SnmpRow],
speed: Sequence[SnmpRow],
names: Sequence[SnmpRow],
aliases: Sequence[SnmpRow],
if_types: Sequence[SnmpRow] = (),
duplex: Sequence[SnmpRow] = (),
pause: Sequence[SnmpRow] = (),
) -> list[PortStatus]:
"""Per-port status. ``duplex``/``pause`` are the EtherLike-MIB columns.
Both are optional because they are genuinely absent on some agents (see
``oids.DOT3_STATS_DUPLEX_STATUS``): the GS728TPP serves them, the GSM7252PS
does not publish those columns at all. A port with no row stays ``None``,
which is what the CLI backend already reports where its own table omits the
value -- absence is never rendered as False.
"""
from . import oids
admin_map = index_int_column(admin, oids.IF_ADMIN_STATUS)
oper_map = index_int_column(oper, oids.IF_OPER_STATUS)
speed_map = index_int_column(speed, oids.IF_HIGH_SPEED)
name_map = index_str_column(names, oids.IF_NAME)
# ifAlias (operator-set description): distinct column from ifName above.
# An absent row for a port (or an empty-string alias) both mean "no
# description set" -> honest None, never a fabricated "".
alias_map = index_str_column(aliases, oids.IF_ALIAS)
duplex_map = index_int_column(duplex, oids.DOT3_STATS_DUPLEX_STATUS)
pause_map = index_int_column(pause, oids.DOT3_PAUSE_OPER_MODE)
physical = physical_ports(if_types)
ports = sorted(set(admin_map) | set(oper_map))
if physical is not None:
ports = [p for p in ports if p in physical]
result: list[PortStatus] = []
for p in ports:
link_up = oper_map.get(p) == 1
mbps = speed_map.get(p)
pause_mode = pause_map.get(p)
result.append(
PortStatus(
port=p,
name=name_map.get(p) or None,
admin_enabled=admin_map.get(p) == 1,
link_up=link_up,
full_duplex=_DUPLEX_STATUS.get(duplex_map.get(p, 0)),
# dot3PauseOperMode 1 is "disabled"; 2/3/4 are the enabled
# directions, and any of them means pause frames are in use.
flow_control=None if pause_mode is None else pause_mode != 1,
# A DOWN port has no operational speed: ifHighSpeed keeps
# reporting the configured rate on a down port (verified on the
# gsm7252ps: 10000 on down 1/0/52), which is NOT an operational
# speed. Report None unless the link is up, matching the web
# UI's "Unknown"->None so both backends agree field-for-field.
# (ifHighSpeed 0 also maps to None.)
speed_mbps=mbps if (mbps and link_up) else None,
description=alias_map.get(p) or None,
)
)
return result
[docs]
def parse_port_stats(
*,
in_octets: Sequence[SnmpRow],
out_octets: Sequence[SnmpRow],
in_ucast: Sequence[SnmpRow],
out_ucast: Sequence[SnmpRow],
in_errors: Sequence[SnmpRow],
out_errors: Sequence[SnmpRow],
if_types: Sequence[SnmpRow] = (),
) -> list[PortStats]:
from . import oids
rx_b = index_int_column(in_octets, oids.IF_HC_IN_OCTETS)
tx_b = index_int_column(out_octets, oids.IF_HC_OUT_OCTETS)
rx_p = index_int_column(in_ucast, oids.IF_HC_IN_UCAST)
tx_p = index_int_column(out_ucast, oids.IF_HC_OUT_UCAST)
rx_e = index_int_column(in_errors, oids.IF_IN_ERRORS)
tx_e = index_int_column(out_errors, oids.IF_OUT_ERRORS)
# ifHC* counters are ifIndex-keyed (same space as ifType), so filter to the
# physical ports for the same reason get_ports does -- otherwise SNMP
# get_stats emits the M4300's 130 phantom LAG/CPU/VLAN interfaces that the
# web portStatistics page never lists, breaking HTTP<->SNMP stats parity.
physical = physical_ports(if_types)
ports = sorted(
set(rx_b) | set(tx_b) | set(rx_p) | set(tx_p) | set(rx_e) | set(tx_e)
)
if physical is not None:
ports = [p for p in ports if p in physical]
return [
PortStats(
port=p,
rx_bytes=rx_b.get(p),
tx_bytes=tx_b.get(p),
rx_packets=rx_p.get(p),
tx_packets=tx_p.get(p),
rx_errors=rx_e.get(p),
tx_errors=tx_e.get(p),
)
for p in ports
]
[docs]
def decode_port_bitmap(bitmap: bytes | str) -> frozenset[int]:
"""Decode an SNMP VLAN port bitmap. Bit 7 of byte 0 = port 1.
An empty value is a legitimately absent bitmap -> no ports. Both transports
normalize the non-printable OCTET STRING onto the wire to ``bytes``, so
that is the expected form and is used directly (MSB-first). If a bitmap
ever arrives as a printable ``str`` it is latin-1 encoded first; a str that
cannot round-trip through latin-1 is malformed and raises SnmpError naming
the value.
"""
if not bitmap:
return frozenset()
if isinstance(bitmap, bytes):
data = bitmap
else:
try:
data = bitmap.encode("latin-1")
except UnicodeEncodeError as exc:
raise SnmpError(f"malformed VLAN port bitmap {bitmap!r}") from exc
ports: set[int] = set()
for byte_idx, byte_val in enumerate(data):
for bit in range(8):
if byte_val & (0x80 >> bit):
ports.add(byte_idx * 8 + bit + 1)
return frozenset(ports)
def _vlan_bitmap_map(rows: Sequence[SnmpRow], base_oid: str) -> dict[int, bytes | str]:
"""{vlan_id: bitmap_value} for a VLAN bitmap column.
A row absent from the column is skipped; a row present under base_oid
whose VLAN index is non-numeric, or whose value is neither bytes nor str
(wrong SNMP type on the wire), is drift and raises SnmpError naming the
offending OID rather than silently dropping present-but-malformed data.
"""
out: dict[int, bytes | str] = {}
for row in rows:
s = _suffix(row, base_oid)
if s is None:
continue
if not s.isdigit():
raise SnmpError(f"malformed VLAN index {s!r} at {row.oid}")
if not isinstance(row.value, (bytes, str)):
raise SnmpError(f"malformed VLAN port bitmap type at {row.oid}")
out[int(s)] = row.value
return out
def _current_vlan_bitmap_map(
rows: Sequence[SnmpRow], base_oid: str
) -> dict[int, bytes | str]:
"""{vlan_id: bitmap} for a ``dot1qVlanCurrentTable`` bitmap column.
Separate from ``_vlan_bitmap_map`` because the two tables are indexed
differently: the STATIC table is indexed by ``dot1qVlanIndex`` alone, while
the CURRENT table is indexed by ``dot1qVlanTimeMark.dot1qVlanIndex`` (RFC
2674). Live on the GS728TPP (10.2.5.10, firmware 6.0.1.30) every row is
``...4.2.1.4.0.<vlan>`` -- time mark 0. A suffix that is not exactly two
numeric components is drift and raises, rather than being silently dropped.
"""
out: dict[int, bytes | str] = {}
for row in rows:
s = _suffix(row, base_oid)
if s is None:
continue
parts = s.split(".")
if len(parts) != 2 or not all(p.isdigit() for p in parts):
raise SnmpError(f"malformed dot1qVlanCurrentTable index {s!r} at {row.oid}")
if not isinstance(row.value, (bytes, str)):
# Present but the wrong SNMP type on the wire: drift, not absence.
raise SnmpError(f"malformed VLAN port bitmap type at {row.oid}")
out[int(parts[1])] = row.value
return out
[docs]
def parse_vlans(
names: Sequence[SnmpRow],
egress: Sequence[SnmpRow],
untagged: Sequence[SnmpRow],
if_types: Sequence[SnmpRow] = (),
current_egress: Sequence[SnmpRow] = (),
current_untagged: Sequence[SnmpRow] = (),
) -> list[VLANInfo]:
"""VLANs from the Q-BRIDGE static table, completed by the current table.
``if_types`` filters membership to physical ports, exactly as
``parse_port_status``/``parse_pvids`` already do. Without it a LAG shows up
as a phantom member port: VERIFIED on the GS728TPP (10.2.5.10, firmware
6.0.1.30), whose 126-byte PortList sets bit 1000 -- ``po 1``, ifType
161 (ieee8023adLag), confirmed identity-mapped via dot1dBasePortIfIndex --
in 11 of its 13 VLANs. The switch has 28 ports, so "member port 1000" is
not something a caller can act on, and the HTTP backend never reports it.
``current_egress``/``current_untagged`` add VLANs the STATIC table omits.
That is not a hypothetical: the same GS728TPP publishes only 12 static rows
(ids 2..99) while ``dot1qVlanCurrentTable`` has 13 -- VLAN 1, the default
VLAN, exists ONLY there, with ``dot1qVlanStatus = 1 (other)`` rather than
2 (permanent). Reading the static table alone silently loses the VLAN that
carries this switch's own management ports (24/25/27, untagged), which the
web UI does list. Such a VLAN has no ``dot1qVlanStaticName`` row, so its
name is None -- matching what the HTTP backend reports for it.
The static bitmaps win where both tables have the VLAN: they are the
CONFIGURED membership. On the live GS728TPP the two agreed byte-for-byte
for all 12 shared VLANs, so this ordering was measured, not assumed.
"""
from . import oids
physical = physical_ports(if_types)
def ports_of(bitmap: bytes | str) -> frozenset[int]:
decoded = decode_port_bitmap(bitmap)
return decoded if physical is None else frozenset(decoded & physical)
name_map = index_str_column(names, oids.DOT1Q_VLAN_STATIC_NAME)
egress_map = _vlan_bitmap_map(egress, oids.DOT1Q_VLAN_STATIC_EGRESS)
untag_map = _vlan_bitmap_map(untagged, oids.DOT1Q_VLAN_STATIC_UNTAGGED)
cur_egress_map = _current_vlan_bitmap_map(
current_egress, oids.DOT1Q_VLAN_CURRENT_EGRESS
)
cur_untag_map = _current_vlan_bitmap_map(
current_untagged, oids.DOT1Q_VLAN_CURRENT_UNTAGGED
)
result: list[VLANInfo] = []
for vid in sorted(set(name_map) | set(cur_egress_map)):
static_row = vid in name_map
member = ports_of(
egress_map.get(vid, "") if static_row else cur_egress_map.get(vid, "")
)
untag = ports_of(
untag_map.get(vid, "") if static_row else cur_untag_map.get(vid, "")
)
result.append(
VLANInfo(
vlan_id=vid,
name=name_map.get(vid) or None,
member_ports=member,
tagged_ports=member - untag,
untagged_ports=untag,
)
)
return result
[docs]
def parse_pvids(
rows: Sequence[SnmpRow], if_types: Sequence[SnmpRow] = ()
) -> list[tuple[int, int]]:
from . import oids
pvids = index_int_column(rows, oids.DOT1Q_PVID)
physical = physical_ports(if_types)
if physical is None:
return sorted(pvids.items())
# DOT1Q_PVID is keyed by dot1dBasePort. On every real Netgear switch the
# bridge-port and ifIndex spaces COINCIDE for physical ports (SNMP-verified
# on the M4300: PVIDs matched the ifIndex physical set with no translation),
# so filtering the PVID keys directly against the physical ifIndex set drops
# LAG/CPU/VLAN PVIDs correctly. A dot1dBasePortIfIndex translation was tried
# but is WRONG here: it couples PVIDs to the independently-populated FDB
# base-port map (which can point a physical port's base-port at an unrelated
# ifIndex), silently dropping real physical PVIDs.
return sorted((bp, v) for bp, v in pvids.items() if bp in physical)
def _format_mac_bytes(byte_strs: Sequence[str]) -> str:
return ":".join(f"{int(b):02X}" for b in byte_strs)
def _format_mac_octetstring(value: int | str | bytes) -> str | None:
"""Format a raw 6-byte MAC-shaped OCTET STRING as ``XX:XX:XX:XX:XX:XX``.
The value arrives as ``bytes`` from a Hex-STRING varbind, or (for a
transport that normalizes octet strings to latin-1 text) a 6-character
``str``. Returns ``None`` when ``value`` isn't a 6-byte/6-char octet
string -- the caller decides whether that's absence or malformed drift.
"""
if isinstance(value, bytes) and len(value) == 6:
return ":".join(f"{b:02X}" for b in value)
if isinstance(value, str) and len(value) == 6:
return ":".join(f"{ord(c):02X}" for c in value)
return None
def _mac_from_ascii_text(value: int | str | bytes) -> str | None:
"""Recognize a MAC already rendered as ASCII text ``XX:XX:XX:XX:XX:XX``.
Some firmware (verified: the M4300-24X's dot1dBaseBridgeAddress) returns a
MAC OCTET STRING as a 17-character human-readable colon-hex STRING rather
than the proper 6 raw bytes. Returns the normalized upper-case MAC, or
``None`` when ``value`` isn't such a string (so callers fall through to the
raw-octet path / malformed handling).
"""
if not isinstance(value, str):
return None
parts = value.split(":")
if len(parts) != 6:
return None
try:
octets = [int(p, 16) for p in parts]
except ValueError:
return None
if any(len(p) != 2 for p in parts) or any(not 0 <= o <= 0xFF for o in octets):
return None
return ":".join(f"{o:02X}" for o in octets)
def _format_chassis_id(value: int | str | bytes) -> str:
"""Format an lldpRemChassisId value.
The MAC-address chassis subtype formats as ``XX:XX:XX:XX:XX:XX`` (see
``_format_mac_octetstring``). Any other chassis-id subtype (e.g. a chassis
component name) is returned as plain text.
"""
mac = _format_mac_octetstring(value)
if mac is not None:
return mac
return value if isinstance(value, str) else str(value)
[docs]
def parse_base_mac(rows: Sequence[SnmpRow]) -> str | None:
"""Parse dot1dBaseBridgeAddress (BRIDGE-MIB scalar, standard MIB-II) into
a colon-separated MAC string.
An absent scalar (no row under the OID at all) is honestly ``None`` --
not every device necessarily answers this instance. A row that IS present
but isn't a 6-byte/6-char OCTET STRING is drift, not absence, and raises
SnmpError naming the offending OID, consistent with the other column
parsers in this module.
"""
from . import oids
prefix = oids.DOT1D_BASE_BRIDGE_ADDRESS + "."
for row in rows:
if not row.oid.startswith(prefix):
continue
mac = _format_mac_octetstring(row.value) or _mac_from_ascii_text(row.value)
if mac is None:
raise SnmpError(f"malformed base MAC {row.value!r} at {row.oid}")
return mac
return None
def _column_text(value: int | str | bytes) -> str:
"""Render a non-chassis LLDP column (portDesc/sysName) as text."""
if isinstance(value, bytes):
return value.decode("utf-8", errors="replace")
return value if isinstance(value, str) else str(value)
def _format_port_id(value: int | str | bytes) -> str:
"""Format an lldpRemPortId value.
Consistent with ``_format_chassis_id``: a MAC-address port-id subtype
(lldpPortIdSubtype 3) is raw binary and formats as
``XX:XX:XX:XX:XX:XX`` via ``_format_mac_octetstring``, instead of being
UTF-8-decoded (with ``errors="replace"``) into garbled U+FFFD text --
that mismatch, versus chassis-id's correct MAC handling, was the bug.
A genuinely binary portId always arrives as ``bytes`` (the transport's
own printable-ASCII heuristic -- see ``index_str_column``'s docstring --
only emits ``str`` for values that decode cleanly as text), so the
``bytes``-and-6-long check below is the reliable MAC signal. A ``str`` is
additionally treated as raw MAC bytes only when it is NOT printable text
(mirroring the latin-1-normalizing-transport case exercised for
``parse_base_mac``): this guards against a real, everyday ASCII
interface-name portId that happens to be exactly 6 characters (e.g.
``"1/xg51"``) being mistaken for a MAC and corrupted into hex -- unlike
chassis-id, port-id routinely carries short human-readable interface
names, so a bare length-6 check on ``str`` is unsafe here. Any other
value (including a printable 6-char ``str`` or any non-MAC-shaped
value) is plain text, per ``_column_text``.
"""
if isinstance(value, bytes) and len(value) == 6:
mac = _format_mac_octetstring(value)
if mac is not None:
return mac
if isinstance(value, str) and len(value) == 6 and not value.isprintable():
mac = _format_mac_octetstring(value)
if mac is not None:
return mac
return _column_text(value)
[docs]
def parse_lldp(rows: Sequence[SnmpRow]) -> list[LLDPNeighbor]:
"""Group lldpRemTable rows by local port into LLDPNeighbor entries.
The instance suffix is ``<column>.<timeMark>.<localPortNum>.<remIndex>``;
the middle component is the local port. A row present under the table
prefix but with fewer than 4 suffix components, or a non-integer column
or local-port component, is drift (not absence) and raises SnmpError
naming the offending OID. A fully-empty neighbour group (every tracked
column absent) carries no data and is skipped.
"""
from . import oids
prefix = oids.LLDP_REM_TABLE + ".1."
grouped: dict[tuple[str, str, str], dict[int, int | str | bytes]] = {}
for row in rows:
if not row.oid.startswith(prefix):
continue
parts = row.oid[len(prefix) :].split(".")
if len(parts) != 4:
raise SnmpError(f"malformed LLDP index at {row.oid}")
try:
column = int(parts[0])
except ValueError as exc:
raise SnmpError(
f"non-integer LLDP column {parts[0]!r} at {row.oid}"
) from exc
key = (parts[1], parts[2], parts[3]) # timeMark, localPort, remIdx
grouped.setdefault(key, {})[column] = row.value
result: list[LLDPNeighbor] = []
for (_tm, local_port, _rem), cols in grouped.items():
chassis = cols.get(5, "")
port_id = cols.get(7, "")
port_desc = cols.get(8, "")
sys_name = cols.get(9, "")
# A neighbour row group with every column empty carries no data
# (absent); skip it. A present-but-non-integer local-port index is
# drift -> raise.
if not (chassis or port_id or port_desc or sys_name):
continue
try:
lp = int(local_port)
except ValueError as exc:
raise SnmpError(
f"non-integer LLDP local port {local_port!r} at {prefix}...{local_port}"
) from exc
result.append(
LLDPNeighbor(
local_port=lp,
remote_sys_name=_column_text(sys_name) or None,
remote_port_desc=_column_text(port_desc) or None,
remote_chassis_id=_format_chassis_id(chassis) or None,
remote_port_id=_format_port_id(port_id) or None,
)
)
return sorted(result, key=lambda n: n.local_port)
[docs]
def parse_macs(
fdb: Sequence[SnmpRow], bridge_ports: Sequence[SnmpRow]
) -> list[MacEntry]:
"""Build the MAC/FDB table from dot1qTpFdbPort + dot1dBasePortIfIndex.
``dot1qTpFdbPort`` gives the bridge PORT number keyed by
``<vlan>.<mac-as-6-oid-octets>``; ``dot1dBasePortIfIndex`` maps that
bridge port to an ifIndex (falling back to the bridge port number itself
when unmapped). A bridge-port value that is present but not an integer is
table drift and raises SnmpError naming the offending OID.
"""
from . import oids
bridge_to_if = index_int_column(bridge_ports, oids.DOT1D_BASE_PORT_IF_INDEX)
prefix = oids.DOT1Q_TP_FDB_PORT + "."
result: list[MacEntry] = []
for row in fdb:
if not row.oid.startswith(prefix):
continue
parts = row.oid[len(prefix) :].split(".")
if len(parts) != 7: # <vlan>.<6 MAC bytes>
raise SnmpError(f"malformed FDB index at {row.oid}")
try:
vlan_id = int(parts[0])
except ValueError as exc:
raise SnmpError(
f"non-integer VLAN index {parts[0]!r} at {row.oid}"
) from exc
try:
bridge_port = int(row.value)
except ValueError as exc:
raise SnmpError(
f"non-integer bridge port {row.value!r} at {row.oid}"
) from exc
port = bridge_to_if.get(bridge_port, bridge_port)
result.append(
MacEntry(mac=_format_mac_bytes(parts[1:7]), port=port, vlan_id=vlan_id)
)
return sorted(result, key=lambda m: (m.port, m.mac))
DETECT_MAP: dict[int, PoEDetect] = {
1: PoEDetect.DISABLED,
2: PoEDetect.SEARCHING,
3: PoEDetect.DELIVERING,
4: PoEDetect.FAULT,
}
[docs]
def parse_poe(
status: Sequence[SnmpRow], power_mw: Sequence[SnmpRow]
) -> list[PoEStatus]:
"""Build PoE port status from RFC3621 pethPsePortTable + vendor mW.
``status`` is a walk of pethPsePortTable; only columns 3 (admin) and 6
(detect) are honoured (the hard-won fix: never column 1). Rows are
grouped by ``(group, port)`` from the ``<col>.<group>.<port>`` instance
suffix. A port present in the walk but missing either tracked column is
drift (not absence) and raises SnmpError naming the offending port.
``power_mw`` is the vendor per-port power walk, matched to a port by the
final OID suffix component; a port without a vendor mW row gets
``power_mw=None``.
"""
from . import oids
prefix = oids.PETH_PSE_PORT_TABLE + "."
cols: dict[tuple[int, int], dict[int, int]] = {}
for row in status:
if not row.oid.startswith(prefix):
continue
parts = row.oid[len(prefix) :].split(".")
if len(parts) != 3:
continue
column = int(parts[0])
if column not in (3, 6):
continue
try:
key = (int(parts[1]), int(parts[2]))
cols.setdefault(key, {})[column] = int(row.value)
except ValueError as exc:
raise SnmpError(
f"non-integer PoE value {row.value!r} at {row.oid}"
) from exc
# vendor mW keyed by port index (2nd suffix component)
mw: dict[int, int] = {}
for row in power_mw:
parts = row.oid.split(".")
try:
mw[int(parts[-1])] = int(row.value)
except ValueError:
continue
result: list[PoEStatus] = []
for (_group, port), c in sorted(cols.items()):
if 3 not in c:
raise SnmpError(f"PoE port {port} missing admin (col 3)")
if 6 not in c:
raise SnmpError(f"PoE port {port} missing detect (col 6)")
result.append(
PoEStatus(
port=port,
admin_enabled=c[3] == 1,
detect=DETECT_MAP.get(c[6], PoEDetect.UNKNOWN),
power_mw=mw.get(port),
)
)
return result
[docs]
def parse_box_sensors(
rows_by_kind: Sequence[tuple[str, str, Sequence[SnmpRow]]],
) -> list[Sensor]:
"""Build box sensors from walk-discovered Netgear vendor columns.
Each tuple is ``(kind, unit, rows)`` for one vendor column walk (e.g.
fan RPM, PSU power, temperature). Sensor indices are walk-discovered
(they differ per model), not hardcoded. The literal string
``"Not Supported"`` is Netgear's placeholder for an unpopulated slot and
is skipped, not an error; any other non-integer value is present-but-
malformed and raises SnmpError naming the offending OID.
"""
result: list[Sensor] = []
for kind, unit, rows in rows_by_kind:
for row in rows:
parts = row.oid.split(".")
instance = parts[-1]
if row.value == "Not Supported":
continue
try:
value = int(row.value)
except ValueError as exc:
raise SnmpError(
f"non-integer {kind} reading {row.value!r} at {row.oid}"
) from exc
result.append(
Sensor(
name=f"{kind}{instance}", kind=kind, value=float(value), unit=unit
)
)
return result
[docs]
def parse_entity_sensors(
class_rows: Sequence[SnmpRow],
name_rows: Sequence[SnmpRow],
descr_rows: Sequence[SnmpRow],
) -> list[Sensor]:
"""Build box sensors from the standard ENTITY-MIB physical inventory.
For a model whose SNMP agent implements NO Netgear vendor OIDs (verified:
the GS728TPP), the fan/PSU components are exposed ONLY as ENTITY-MIB
``entPhysicalTable`` rows: ``entPhysicalClass`` (6=powerSupply, 7=fan)
identifies each, ``entPhysicalName`` (falling back to ``entPhysicalDescr``)
names it. This is INVENTORY ONLY -- the switch exposes NO live sensor
value/status anywhere in SNMP (ENTITY-SENSOR-MIB and the vendor tree both
answer noSuchObject on real hardware), so each Sensor carries
``value=NaN`` and ``unit="inventory"``: the component is honestly reported
as present without a fabricated reading. (HTTP DOES expose a health status
for these same components -- that is a real per-backend difference, not a
parser bug; see the cross-backend test.)
Rows are matched by their shared entPhysicalIndex (the trailing OID
component). Only powerSupply/fan classes become sensors; chassis/slot/port
rows are ignored. A non-integer class value present under the class column
is drift and raises SnmpError naming the offending OID.
"""
from . import oids
names = index_str_column(name_rows, oids.ENT_PHYSICAL_NAME)
descrs = index_str_column(descr_rows, oids.ENT_PHYSICAL_DESCR)
classes = index_int_column(class_rows, oids.ENT_PHYSICAL_CLASS)
kind_of = {
oids.ENT_CLASS_POWER_SUPPLY: "power",
oids.ENT_CLASS_FAN: "fan",
}
result: list[Sensor] = []
for idx in sorted(classes):
kind = kind_of.get(classes[idx])
if kind is None:
continue
name = names.get(idx) or descrs.get(idx) or f"{kind}{idx}"
result.append(
Sensor(
name=_canon_sensor_name(name),
kind=kind,
value=float("nan"),
unit="inventory",
)
)
return result
def _canon_sensor_name(name: str) -> str:
"""Canonicalize an ENTITY-MIB component name to the box-sensor label the
HTTP DiagnosticsUnitList uses, so the two backends' sensor NAMES are
identical (only the value/unit differ -- SNMP has no live reading).
entPhysicalName renders a PSU as ``"Main PowerSupply"`` /
``"Redundant PowerSupply"``; the web UI labels the same component
``"Main PS"`` / ``"Redundant PS"``. Abbreviating ``PowerSupply`` -> ``PS``
(with or without an internal space) unifies them; a fan name (``"Fan1"``)
already matches and is returned unchanged."""
return name.replace("Power Supply", "PS").replace("PowerSupply", "PS")
def _ip_str(row: SnmpRow) -> str:
"""Return an IP-valued row's value as ``str``.
Both transports normalize IpAddress varbinds to ``str`` (see SnmpRow's
docstring); a row present under an address/netmask/gateway column whose
value is NOT a str is table drift / a malformed reply, not absence, and
raises SnmpError naming the offending OID.
"""
if not isinstance(row.value, str):
raise SnmpError(f"non-IP value {row.value!r} at {row.oid}")
return row.value
def _ipv4_from_rfc4293_index(rows: Sequence[SnmpRow]) -> str | None:
"""The management IPv4 from an RFC-4293 ipAddressTable walk (the address is
in the ROW INDEX: ``<base>.<type>.<len>.<b1>.<b2>.<b3>.<b4>``, type 1=ipv4,
len 4). Skips loopback and non-IPv4 (IPv6) rows. Returns ``None`` when the
walk is empty (older firmware that populates the RFC-1213 ipAddrTable
instead). Used only as a FALLBACK -- see ``parse_mgmt_ip``."""
from . import oids
prefix = oids.IP_ADDRESS_IFINDEX + "."
for row in rows:
if not row.oid.startswith(prefix):
continue
parts = row.oid[len(prefix) :].split(".")
# ipv4 (type 1) with a 4-byte address: type, len=4, then 4 octets.
if len(parts) < 6 or parts[0] != "1" or parts[1] != "4":
continue
ip = ".".join(parts[2:6])
if ip == "127.0.0.1":
continue
return ip
return None
#: The vendor admin-mode enum shared by every logging destination column.
#: 1 = enabled, 2 = disabled -- confirmed twice on m4300-24x against its own
#: ``show logging``: syslog reads 1 under "Syslog Logging : enabled" and the
#: console column reads 2 under "Console Logging : disabled".
_ADMIN_ENABLED = 1
#: Row status in the syslog host table. 1 is what the CLI prints as "Active".
_HOST_STATUS_ACTIVE = 1
def _first_int(rows: Sequence[SnmpRow]) -> int | None:
"""The value of a single-varbind scalar GET, when it is an integer."""
for row in rows:
if isinstance(row.value, int):
return row.value
return None
[docs]
def parse_syslog(
admin_mode: Sequence[SnmpRow],
local_port: Sequence[SnmpRow],
host_addr: Sequence[SnmpRow],
host_port: Sequence[SnmpRow],
host_severity: Sequence[SnmpRow],
host_status: Sequence[SnmpRow],
*,
addr_base: str,
port_base: str,
severity_base: str,
status_base: str,
) -> SyslogConfig:
"""Vendor logging columns -> ``SyslogConfig``.
The host table is indexed by an integer row id, and every per-host column is
matched to the address column by that index rather than by position -- so a
table with a gap in its indices (a deleted row) cannot silently shift one
row's port onto another row's address.
A row whose address is empty is skipped. The address is the only field that
makes a row meaningful, and reporting one collector fewer is far better than
inventing where logs are being sent.
"""
addresses = index_str_column(host_addr, addr_base)
ports = index_int_column(host_port, port_base)
severities = index_int_column(host_severity, severity_base)
statuses = index_int_column(host_status, status_base)
servers = tuple(
SyslogServer(
host=address,
port=ports.get(index, 0),
severity=severities.get(index, 0),
active=statuses.get(index) == _HOST_STATUS_ACTIVE,
# The OID instance IS the table's row index, and it is the handle a
# RowStatus destroy addresses -- so it is surfaced rather than
# dropped. Sparse, exactly as the CLI table shows it.
index=index,
)
for index, address in sorted(addresses.items())
if address.strip()
)
return SyslogConfig(
enabled=_first_int(admin_mode) == _ADMIN_ENABLED,
local_port=_first_int(local_port) or 0,
servers=servers,
)
[docs]
def parse_mgmt_ip(
addr: Sequence[SnmpRow],
netmask: Sequence[SnmpRow],
route_dest: Sequence[SnmpRow],
route_nexthop: Sequence[SnmpRow],
dhcp_mode: Sequence[SnmpRow],
base_mac: Sequence[SnmpRow],
addr_rfc4293: Sequence[SnmpRow] = (),
) -> MgmtIpConfig:
"""Build the management-IP config from ipAddrTable/ipRouteTable + vendor mode.
Address/netmask/gateway come from the standard MIBs (ipAddrTable,
ipRouteTable) and are trustworthy. The DHCP-vs-static mode is UNVERIFIED
(see oids.VendorOids.dhcp_mode_unverified): it is read best-effort and
``IpMode.UNKNOWN`` is returned whenever the mode OID is absent/unset —
never a guessed dhcp/static. The mode OID is INTEGER-typed, so both
transports normalize its value to a Python ``int`` (see ``SnmpRow``'s
docstring); only a recognized present value (``1``/``2``) maps to
DHCP/STATIC, any other present value -- including one that cannot be
coerced to ``int`` at all -- also yields UNKNOWN rather than raising,
since this OID is explicitly best-effort. ``base_mac`` is the standard
(non-UNVERIFIED) dot1dBaseBridgeAddress scalar walk -- see
``parse_base_mac``; an empty walk (OID absent) yields ``base_mac=None``.
"""
from . import oids
ip: str | None = None
ip_index: str | None = None
aprefix = oids.IP_ADENT_ADDR + "."
for row in addr:
if not row.oid.startswith(aprefix):
continue
if row.value == "127.0.0.1":
continue
ip = _ip_str(row)
ip_index = row.oid[len(aprefix) :]
break
# RFC-4293 fallback: firmware that leaves ipAddrTable empty (M4300) carries
# the address in the ipAddressTable index instead. Netmask there is only a
# pointer into the prefix table (unusable on this firmware), so the mask
# stays None -- honest absence, not a fabricated value.
if ip is None:
ip = _ipv4_from_rfc4293_index(addr_rfc4293)
mask: str | None = None
if ip_index is not None:
want = oids.IP_ADENT_NETMASK + "." + ip_index
for r in netmask:
if r.oid == want:
mask = _ip_str(r)
break
dest_rows = {r.oid[len(oids.IP_ROUTE_DEST) + 1 :]: r.value for r in route_dest}
gateway: str | None = None
nprefix = oids.IP_ROUTE_NEXTHOP + "."
for row in route_nexthop:
if not row.oid.startswith(nprefix):
continue
idx = row.oid[len(nprefix) :]
if dest_rows.get(idx) == "0.0.0.0":
gateway = _ip_str(row)
break
mode = IpMode.UNKNOWN
for row in dhcp_mode:
try:
raw_mode = int(row.value)
except (TypeError, ValueError):
break
if raw_mode == 1:
mode = IpMode.DHCP
elif raw_mode == 2:
mode = IpMode.STATIC
break
return MgmtIpConfig(
mode=mode,
address=ip,
netmask=mask,
gateway=gateway,
base_mac=parse_base_mac(base_mac),
)
def _scalar_text(rows: Sequence[SnmpRow], oid: str) -> str | None:
"""Extract one scalar exact-OID GET result's value as text, or None.
Unlike the walk-based column parsers above (matched by base-OID
*prefix*), ``sysDescr``/``sysObjectID`` are fetched with a plain exact-OID
GET (see ``snmp_read.read_system_info``), so ``rows`` is the combined
result of one ``client.get([...])`` call and this matches by exact OID
equality. An absent scalar (no row with this exact OID at all) is
honestly ``None`` -- not every device necessarily answers, and a caller
must never fabricate a value. A row that IS present but isn't decodable
to text is drift, not absence, and raises SnmpError naming the offending
OID, consistent with every other parser in this module.
"""
for row in rows:
if row.oid != oid:
continue
value = row.value
if isinstance(value, bytes):
return value.decode("utf-8", "replace")
if isinstance(value, str):
return value
raise SnmpError(f"non-string value {value!r} at {row.oid}")
return None
[docs]
def parse_hostname(rows: Sequence[SnmpRow]) -> str:
"""Extract ``sysName`` from one exact-OID GET.
Raises rather than returning a placeholder when the scalar is absent. Every
switch in this fleet answers ``sysName`` -- it is a mandatory MIB-II scalar
-- so an absent one is a real failure to report, not an empty hostname to
invent. An empty *string* is a different thing and is passed through: a
switch with no name configured genuinely has one.
"""
from . import oids
value = _scalar_text(rows, oids.SYS_NAME)
if value is None:
raise SnmpError(
f"switch did not answer sysName ({oids.SYS_NAME}); it is a mandatory "
"MIB-II scalar, so this is an agent or transport failure rather "
"than an absent hostname"
)
return value
[docs]
def parse_system_info(rows: Sequence[SnmpRow]) -> tuple[str | None, str | None]:
"""Extract the raw sysDescr/sysObjectID scalar text from one combined GET.
Pure row -> ``(sys_descr, sys_object_id)`` extraction ONLY -- no model
matching happens here. Kept strictly separate from
``detect_model_from_sysdescr`` so the matching heuristic is unit-testable
against plain strings, with no SnmpRow/client machinery involved at all.
"""
from . import oids
sys_descr = _scalar_text(rows, oids.SYS_DESCR)
sys_object_id = _scalar_text(rows, oids.SYS_OBJECT_ID)
return sys_descr, sys_object_id
def _model_match_tokens(model: SwitchModel) -> tuple[str, ...]:
"""Name tokens to search for (uppercased) in a sysDescr string.
Built ONLY from the registry's own ``key``/``display_name`` -- there is
NO hand-invented per-model sysDescr/sysObjectID table anywhere (no MIBs,
no captures, no prior-art map exist for one; see
``detect_model_from_sysdescr``'s docstring). ``display_name`` sometimes
carries a parenthesized alias (e.g. ``"GSM7228PS (S3300)"`` or
``"M4300-24X (XSM4324CS)"``); both the main name and the alias are valid
tokens, since a real switch's sysDescr text could plausibly use either.
"""
tokens = [model.key.upper()]
name = model.display_name
if "(" in name and name.endswith(")"):
main, _, alias = name.partition("(")
tokens.append(main.strip())
tokens.append(alias[:-1].strip())
else:
tokens.append(name.strip())
return tuple(t for t in tokens if t)
# Punctuation stripped from the edges of a whitespace-delimited sysDescr
# word before comparing it to a registered token. Hyphens are deliberately
# EXCLUDED: they are meaningful inside a model identifier itself (e.g.
# "M4300-24X"), so stripping them would merge distinct SKUs together.
_WORD_STRIP_CHARS = string.punctuation.replace("-", "")
def _candidate_tokens(sys_descr: str) -> frozenset[str]:
"""Whitespace-delimited "words" of a sysDescr string, as whole-token
candidates for exact (uppercased) comparison against a registered
model's match tokens.
Only edge punctuation is stripped (e.g. the trailing comma in
``"M4300-24X,"``) -- internal structure, in particular hyphens, is left
intact. This is what makes the comparison a WHOLE-IDENTIFIER match: a
registered token must equal an entire sysDescr word, not merely appear
as a prefix/substring of it. That is the crux of the fix for the
false-positive bug this function exists to prevent (see
``detect_model_from_sysdescr``'s docstring) -- e.g. the single sysDescr
word ``"GS305EPP"`` never equals the registered token ``"GS305EP"``, and
the single word ``"S3300-28X"`` never equals the registered alias
token ``"S3300"``, no matter what non-alphanumeric character (or none)
immediately follows the registered token's text.
"""
return frozenset(
word.strip(_WORD_STRIP_CHARS).upper() for word in sys_descr.split()
)
# Authoritative sysObjectID -> registry-key map. HONESTY CONSTRAINT: every
# entry is confirmed from a REAL hardware capture. sysObjectID is the
# manufacturer's stable product identifier -- unlike free-form sysDescr text
# it is unambiguous, so it is the PREFERRED detector when present. Entries are
# added only when a live capture proves the mapping, NEVER guessed from a spec
# sheet (that is why this map is small: most registered models have no
# committed sysObjectID capture yet).
#
# * 1.3.6.1.4.1.4526.100.10.19 -> gsm7228ps: the S3300-52X-PoE+
# (sw-netgear-s3300-1 @ 10.1.5.11, captured 2026-07-30 --
# tests/fixtures/captures/gsm7228ps.json). Its sysDescr "S3300-52X-PoE+
# ..." is DELIBERATELY unmatchable by ``detect_model_from_sysdescr`` (same
# textual shape as the unregistered S3300-28X SKU -- see that function's
# docstring), so this OID map is the ONLY safe way to auto-detect it.
SYSOBJECTID_MODELS: Mapping[str, str] = MappingProxyType(
{
"1.3.6.1.4.1.4526.100.10.19": "gsm7228ps",
}
)
[docs]
def detect_model_from_sysobjectid(
sys_object_id: str | None, models: Mapping[str, SwitchModel]
) -> str | None:
"""Identify a model from its sysObjectID via ``SYSOBJECTID_MODELS``.
Returns the registry key ONLY when the OID is in the real-capture-confirmed
map AND that key is present in ``models``; otherwise ``None`` (never a
guess). This is the AUTHORITATIVE detector -- sysObjectID is a stable
manufacturer product identifier, so unlike ``detect_model_from_sysdescr``'s
text heuristic it can safely distinguish SKUs whose sysDescr strings are
textually indistinguishable (the S3300-52X vs the unregistered S3300-28X).
``read_system_info`` tries this first and only falls back to sysDescr
matching when it returns ``None``.
"""
if not sys_object_id:
return None
key = SYSOBJECTID_MODELS.get(sys_object_id)
if key is not None and key in models:
return key
return None
[docs]
def detect_model_from_sysdescr(
sys_descr: str | None, models: Mapping[str, SwitchModel]
) -> str | None:
"""Match a switch's sysDescr text against registered models' names.
HONESTY CONSTRAINT: there is no ground-truth sysObjectID -> model table
(see ``oids.SYS_OBJECT_ID`` -- it is read as a raw signal but never used
here). Matching is EXACT (case-insensitive) whole-word matching: the
sysDescr string is split into whitespace-delimited candidate tokens
(``_candidate_tokens``) and a registered model matches only when one of
its own key/display_name/alias tokens (``_model_match_tokens``) equals
one of those candidates in full -- NEVER a bare substring/prefix check,
and NEVER a guess:
* A sysDescr containing an unregistered Netgear model name (e.g.
``"GS752TP"``, not in ``models``) matches no token and correctly
returns ``None`` -- it is NEVER coerced onto some other, wrong,
registered model just because it looks Netgear-ish.
* A non-Netgear/garbage string matches nothing and also returns ``None``.
* CRITICAL (regression that motivated the switch away from substring
matching): a real, unregistered Netgear model whose name EXTENDS a
registered token must also return ``None``, never the shorter
registered model. Bare substring matching used to fail this both when
the extension has no separator (``"GS305EPP"`` used to wrongly match
the registered ``"GS305EP"``, a distinct 123W model vs. the registered
63W one) AND when it has one (``"S3300-28X"`` / ``"S3300-28X-PoE+"``
used to wrongly match the registered alias ``"S3300"`` for
``gsm7228ps``, a distinct S3300 SKU). Whole-word equality rejects both:
neither ``"GS305EPP"`` nor ``"S3300-28X"`` is ever *equal* to the
shorter registered token, regardless of what character (alphanumeric
or not) follows it in the original text.
* A sysDescr matching MORE THAN ONE registered model's tokens (meaning
two registered models' names collide and can't be disambiguated by
this heuristic) ALSO returns ``None`` rather than guessing between
them. This never happens for the current registry (verified: no
model's match tokens equal another's -- e.g. "M4300-24X" vs
"M4300-16X", "GSM7252PS" vs "GSM7228PS"/"S3300" are all mutually
exclusive), but the fallback is kept as a permanent safety net against
a future registry addition introducing a collision.
"""
if not sys_descr:
return None
candidates = _candidate_tokens(sys_descr)
matches = {
model.key
for model in models.values()
if any(token.upper() in candidates for token in _model_match_tokens(model))
}
if len(matches) == 1:
return next(iter(matches))
return None