Transports

Getting bytes to and from a device. Each transport satisfies a client protocol defined in the protocol layer, so a reader or writer works unchanged across them — and equally against a mock.

These are also the classes you construct by hand to point the library at a virtual switch.

Transport implementations (sync net-snmp CLI, async pysnmp) for SNMP I/O.

Synchronous

Synchronous SNMP transport implementations.

Synchronous SNMP v2c client over the net-snmp CLI tools (subprocess).

No Python SNMP package is used. The net-snmp binaries (snmpget/snmpbulkwalk) are a system requirement — install the OS snmp package (apt-get install -y snmp). Args are passed as a list; shell is never used.

netgear_switch.transport.sync.snmp_netsnmp_cli.parse_netsnmp_lines(text, *, empty_subtree_ok=False)

Parse snmpget/snmpbulkwalk output (-On -Oe -OU -Ln) into SnmpRows.

For a GET (empty_subtree_ok=False, the default) a “No Such Object/Instance” line raises SnmpError: the caller asked for a specific scalar and its absence must be surfaced, never fabricated as empty.

For a WALK (empty_subtree_ok=True) a “No Such Object/Instance” line is the NORMAL response a real agent gives when the walked subtree has no entries at all (e.g. the RFC3621 PoE MIB on a non-PoE switch, an unimplemented sensor table, or an unpopulated ipAddrTable). Verified against live hardware: snmpbulkwalk of an absent subtree emits exactly one <base> = No Such Object available on this agent at this OID line. In walk mode that line is treated like the benign end-of-MIB terminator — skipped, returning the rows collected so far ([] for an empty subtree) instead of raising. This is what lets optional reads degrade to an empty result rather than crashing the whole call.

The benign “No more variables left in this MIB View …/ past the end of the MIB tree” terminator that snmpbulkwalk appends at the end of a successful walk is always skipped. Multi-line Hex-STRING continuations are joined into one bytes value.

Return type:

list[SnmpRow]

class netgear_switch.transport.sync.snmp_netsnmp_cli.NetsnmpCliClient(host, community, *, timeout=10, retries=1, runner=subprocess.run)

Bases: object

Read-only sync SNMP client shelling out to net-snmp CLI tools.

get(oids)

Return type:

list[SnmpRow]

walk(base_oid)

Return type:

list[SnmpRow]

set(varbind)

set_many(varbinds)

Synchronous NSDP UDP transport (stdlib sockets only).

Binds a UDP client port and exchanges one request/response datagram with the switch over unicast (the query_ip pattern — preferred over broadcast discovery for a known host). client_port defaults to the real NSDP client port 63321, but the virtual face lets tests pass client_port=0 to bind an unprivileged ephemeral port on loopback (so no root/CAP_NET_BIND_SERVICE and no SO_BINDTODEVICE are needed under test). Errors (timeout / malformed / bad password) surface as NsdpError, never silently.

class netgear_switch.transport.sync.nsdp_udp.UdpNsdpClient(

host,

*,

interface=None,

client_mac=None,

client_port=63321,

server_port=63322,

timeout=2.0,

auth_scheme='auto',

sock_factory=socket.socket,

)

Bases: object

Sync NSDP read+write client over UDP for a single switch.

read(tags)

Return type:

NSDPPacket

write(tlvs, *, password)

Return type:

NSDPPacket

Asynchronous

Asynchronous SNMP transport implementations.

Asynchronous SNMP v2c client on pysnmp v7. pysnmp is imported lazily.

Value parity: each pysnmp SMI value is normalized to the SAME plain Python type the net-snmp CLI client (Task 10) produces — int for integer-family, str for text/OID/IP, bytes for non-printable octet strings (Hex-STRING). Task 16’s sync/async equivalence test compares these values, so they must match.

pysnmp ships with no type stubs and is untyped under mypy –strict. Rather than a blanket ignore_missing_imports for the whole package, _pysnmp_asyncio() is the single lazy-import seam. It resolves the module dynamically via importlib.import_module (a plain str -> ModuleType call mypy can’t follow into pysnmp’s untyped internals), so no type: ignore is needed at all; everything downstream of this one seam is deliberately treated as Any, and the rest of the module is fully typed.

class netgear_switch.transport.aio.snmp_pysnmp.PysnmpClient(host, community, *, port=161, timeout=2.0, retries=1)

Bases: object

Async SNMP v2c read/write client for a single switch.

async get(oids)

Return type:

list[SnmpRow]

async walk(base_oid)

Return type:

list[SnmpRow]

async set(varbind)

async set_many(varbinds)

Asynchronous NSDP UDP transport (stdlib asyncio datagram endpoint).

Mirrors the sync UdpNsdpClient but over loop.create_datagram_endpoint. The datagram exchange is factored into an injectable transceive coroutine so read/write are unit-testable with a fake exchange (no real UDP), the async analogue of the sync client’s sock_factory seam. As with the sync client, client_port=0 binds an unprivileged ephemeral port for the virtual face.

class netgear_switch.transport.aio.nsdp_udp.AsyncUdpNsdpClient(

host,

*,

interface=None,

client_mac=None,

client_port=63321,

server_port=63322,

timeout=2.0,

auth_scheme='auto',

transceive=_udp_transceive,

)

Bases: object

Async NSDP read+write client over UDP for a single switch.

async read(tags)

Return type:

NSDPPacket

async write(tlvs, *, password)

Return type:

NSDPPacket

HTTP

httpx-backed HTTP web-UI transport (sync + async).

httpx-backed web-UI clients implementing the session Protocols.

One codebase: all URL/crypto/parse logic lives in the pure protocols.http package; only the actual GET/POST differ between the sync httpx.Client and async httpx.AsyncClient. Legacy Plus switches are HTTP-only, so base_url defaults to http://; the secure flag flips it to https:// (and the Referer scheme with it) for a model whose real UI is HTTPS – the M4300-16X Cheetah UI on :49152. TLS verification defaults off for the switches’ self-signed certs.

httpx is an optional dependency ([http] extra); it is imported at module top-level because this module lives under transport/http and is only ever imported lazily by _dispatch (function-local imports), exactly like the SNMP transports — import netgear_switch never reaches here.

class netgear_switch.transport.http.client.HttpClient(host, password, spec, *, secure=False, verify_tls=False, transport=None)

Bases: object

Synchronous httpx web-UI session (implements HttpSession).

login()

get_page(path)

Return type:

str

post_form(path, data)

Return type:

str

post_multipart(path, data, file)

Return type:

str

post_xml(path, body)

Return type:

str

close()

class netgear_switch.transport.http.client.AsyncHttpClient(host, password, spec, *, secure=False, verify_tls=False, transport=None)

Bases: object

Asynchronous httpx web-UI session (implements AsyncHttpSession).

async login()

async get_page(path)

Return type:

str

async post_form(path, data)

Return type:

str

async post_multipart(path, data, file)

Return type:

str

async post_xml(path, body)

Return type:

str

async aclose()

CLI

Three transports for one command surface. All three satisfy the CliSession protocol — and so does the mock’s in-process CLI face.

CLI transports (SSH/telnet/console) implementing the CliSession seam.

Transport-agnostic CLI session seam + a shared interactive-shell driver.

CliSession is the single seam cli_read.CliReader depends on – the CLI analogue of protocols.http.session.HttpSession. The real SSH/telnet/console transports implement it, and so does the in-process mock face (virtual.faces.cli.VirtualCliFace), so ONE reader codebase runs against both real hardware and the virtual switch.

ShellDriver holds the byte-level interactive-shell logic (send a command, read back until the FASTPATH prompt reappears, strip the command echo and the trailing prompt) so all three real transports share it – they differ only in how a channel’s send/recv bytes are wired. The parsers (protocols.cli.parse) are shared too: the transports carry bytes, the driver frames them into per-command text, and the parsers turn that text into models.

class netgear_switch.transport.cli.session.CliSession(*args, **kwargs)

Bases: Protocol

A ready-to-use authenticated CLI session for one switch.

run issues one command and returns its output text with the echoed command line and the trailing prompt removed. Setup (enable + disable paging) is the transport’s responsibility, done before the first run.

run(command)

Return type:

str

run_scp_copy(command, scp_password)

Issue an interactive copy scp://... and drive its mid-command prompts (host-key TOFU, remote password, (y/n) overwrite), returning the transcript on success and raising CliTransportError on failure. Only the FASTPATH cert-deploy path (cli_write) uses this; a session used purely for reads never calls it.

Return type:

str

run_write_memory(command='write memory', *, prestuff)

Issue write memory and answer its (y/n) save-config confirm.

Return type:

str

close()

exception netgear_switch.transport.cli.session.CliTransportError

Bases: Exception

A CLI transport failed to connect, authenticate, or read a prompt.

class netgear_switch.transport.cli.session.ShellDriver(

send,

recv,

*,

enable_cmd='enable',

paging_off_cmd='terminal length 0',

enable_password=None,

newline='\r\n',

)

Bases: object

Frames an interactive shell (send/recv bytes) into per-command text.

send writes bytes to the channel; recv returns up to n bytes (blocking, may return a partial chunk). This is deliberately transport-free so SSH, telnet and console reuse it unchanged. It cannot be exercised against real hardware from CI (no network), so it is transport-only and covered by a fake-channel unit test rather than a live session.

setup()

Consume the initial banner/prompt, enable, then disable paging.

run(command)

Return type:

str

run_scp_copy(command, scp_password)

Drive a copy scp://<src> <dest> transfer to completion.

The genuinely-new interactive transport bit: unlike a plain EXEC command (run), copy scp:// prompts the operator mid-flight. This sends the command then loops over _recv chunks, answering each prompt with the same _send/_write_line primitives run uses – no new transport, no pexpect:

• host-key TOFU (... continue connecting (yes/no)?) -> yes

• remote Password: -> scp_password

• (y/n) overwrite confirm -> a bare y (no newline, matching the real FASTPATH prompt)

It returns when the shell prompt reappears after the switch reports the transfer, and raises CliTransportError if the switch reports a failed transfer or the stream ends without a prompt. GROUNDED in the working certbot-hook FastpathScpUpdater._send_copy; MOCK-TESTED, not live-verified (a real SCP upload is a production write needing a staging SCP server), so it cannot be exercised from CI – covered by a byte-level fake-shell test, like the rest of ShellDriver.

Return type:

str

run_write_memory(command='write memory', *, prestuff)

Persist the running config, answering the (y/n) save confirm.

prestuff=True (GSM7252PS) pre-stuffs the y in the SAME write as the command, because that image’s confirm has a tiny timeout that a read-then-answer round trip races – exactly the certbot-hook writemem_stuff behaviour. prestuff=False (M4300) waits for the (y/n) prompt then answers y. GROUNDED in prior art, mock-tested.

Return type:

str

Paramiko-backed SSH CLI transport (implements CliSession).

paramiko is an OPTIONAL dependency (the [ssh] extra) and is imported LAZILY inside connect – import netgear_switch never reaches paramiko, exactly like the httpx transport under transport/http.

PARAMIKO VERSION DECISION (documented as the reviewer asked)

Old FASTPATH firmware (the GSM7252PS/M4300 generation) only offers the legacy key exchange diffie-hellman-group14-sha1 and the ssh-rsa (SHA-1) host-key algorithm. paramiko 3.0 dropped both from its DEFAULT preferred lists (and later releases removed some legacy SHA-1 primitives outright), so a stock modern paramiko negotiates NOTHING with these switches and the handshake fails.

Two mitigations, applied together:

1. Pin the dependency to a release that still ships and prefers the legacy algorithms – paramiko>=2.12,<3 (2.12 is CONFIRMED working against a real GSM7252PS). This is the [ssh] extra’s constraint in pyproject.toml.

2. Belt-and-suspenders, ALSO re-insert the legacy algorithms into the Transport’s preferred KEX / host-key lists explicitly here via get_security_options() when the running paramiko still defines them, so the transport keeps working even if a newer paramiko is installed that retains the primitives but merely de-prioritised them.

This transport CANNOT be live-tested from CI (no network); it is transport-only, and the shared ShellDriver it builds on is unit-tested with a fake channel.

class netgear_switch.transport.cli.ssh.SshCliTransport(host, username, password, spec, *, port=22, timeout=_DEFAULT_TIMEOUT)

Bases: CliSession

An SSH interactive-shell CLI session over paramiko.

connect()

run(command)

Return type:

str

run_scp_copy(command, scp_password)

Issue an interactive copy scp://... and drive its mid-command prompts (host-key TOFU, remote password, (y/n) overwrite), returning the transcript on success and raising CliTransportError on failure. Only the FASTPATH cert-deploy path (cli_write) uses this; a session used purely for reads never calls it.

Return type:

str

run_write_memory(command='write memory', *, prestuff)

Issue write memory and answer its (y/n) save-config confirm.

Return type:

str

close()

Thin telnet CLI transport (implements CliSession), transport-only.

Reuses the SAME ShellDriver (and therefore the same parsers) as the SSH transport – it differs only in carrying bytes over telnetlib instead of a paramiko channel. It CANNOT be live-tested from CI (no network) and shares no parser code that isn’t already covered by the SSH/mock paths.

CAVEAT: the stdlib telnetlib module was DEPRECATED in Python 3.11 and REMOVED in Python 3.13. It is imported lazily so import netgear_switch never depends on it; on 3.13+ constructing this transport raises a clear CliTransportError telling the caller to use SSH or the console transport instead. Telnet on a management switch is plaintext and best avoided anyway.

class netgear_switch.transport.cli.telnet.TelnetCliTransport(host, username, password, spec, *, port=23, timeout=_DEFAULT_TIMEOUT)

Bases: CliSession

A telnet interactive-shell CLI session over telnetlib.

connect()

run(command)

Return type:

str

run_scp_copy(command, scp_password)

Issue an interactive copy scp://... and drive its mid-command prompts (host-key TOFU, remote password, (y/n) overwrite), returning the transcript on success and raising CliTransportError on failure. Only the FASTPATH cert-deploy path (cli_write) uses this; a session used purely for reads never calls it.

Return type:

str

run_write_memory(command='write memory', *, prestuff)

Issue write memory and answer its (y/n) save-config confirm.

Return type:

str

close()

Thin serial-console CLI transport (implements CliSession), transport-only.

Reuses the SAME ShellDriver/parsers as the SSH transport, carrying bytes over a local serial line (pyserial, the [ssh] extra) instead of a network socket. Used to reach a switch’s physical console port. Cannot be exercised from CI (no serial hardware); transport-only, parser-shared.

pyserial is imported lazily so import netgear_switch never depends on it.

class netgear_switch.transport.cli.console.ConsoleCliTransport(

device,

username,

password,

spec,

*,

baudrate=_DEFAULT_BAUD,

timeout=_DEFAULT_TIMEOUT,

)

Bases: CliSession

A serial-console interactive-shell CLI session over pyserial.

connect()

run(command)

Return type:

str

run_scp_copy(command, scp_password)

Issue an interactive copy scp://... and drive its mid-command prompts (host-key TOFU, remote password, (y/n) overwrite), returning the transcript on success and raising CliTransportError on failure. Only the FASTPATH cert-deploy path (cli_write) uses this; a session used purely for reads never calls it.

Return type:

str

run_write_memory(command='write memory', *, prestuff)

Issue write memory and answer its (y/n) save-config confirm.

Return type:

str

close()