From b8fa2c6725762e08808d47e1f2e7a2b5f8ae2ab3 Mon Sep 17 00:00:00 2001 From: hansjone Date: Thu, 30 Jul 2026 02:34:25 +0000 Subject: [PATCH] fix(webcrt): skip Netmiko terminal-length on interactive login. Keep disable_paging for collection/MCP; WebCRT opens sessions with interactive=True so the device CLI is not pre-seeded. Co-authored-by: Cursor --- netx_api/ne_session_factory.py | 71 ++++++++++++++++++++++++---- netx_api/webcrt_service.py | 1 + tests/test_ne_session_interactive.py | 53 +++++++++++++++++++++ 3 files changed, 117 insertions(+), 8 deletions(-) create mode 100644 tests/test_ne_session_interactive.py diff --git a/netx_api/ne_session_factory.py b/netx_api/ne_session_factory.py index 726c10d..4fe4647 100644 --- a/netx_api/ne_session_factory.py +++ b/netx_api/ne_session_factory.py @@ -242,6 +242,32 @@ def _netmiko_driver_class(device_type: str) -> type: return cls +def _interactive_driver_class(base_cls: type) -> type: + """Subclass that skips Netmiko auto terminal-length / terminal-width commands. + + Collection and MCP exec need paging disabled; WebCRT must not inject those. + """ + + class _InteractiveSession(base_cls): # type: ignore[misc,valid-type] + def disable_paging(self, *args: Any, **kwargs: Any) -> str: # noqa: ANN401 + return "" + + def set_terminal_width(self, *args: Any, **kwargs: Any) -> str: # noqa: ANN401 + return "" + + _InteractiveSession.__name__ = f"Interactive{getattr(base_cls, '__name__', 'Netmiko')}" + return _InteractiveSession + + +def _build_netmiko_connection(dev: dict[str, Any], *, interactive: bool = False) -> ConnectHandler: + """Instantiate Netmiko from connect kwargs; optional interactive skips paging cmds.""" + if not interactive: + return ConnectHandler(**dev) + device_type = str(dev.get("device_type") or "").strip() + base_cls = _netmiko_driver_class(device_type) + return _interactive_driver_class(base_cls)(**dev) + + def _netmiko_over_ssh_client( ssh_client: paramiko.SSHClient, *, @@ -253,11 +279,14 @@ def _netmiko_over_ssh_client( enable_secret: str, session_timeout: int | None, session_log: Any = None, + interactive: bool = False, ) -> ConnectHandler: """Netmiko session over an already-authenticated SSH client (bastion protocol proxy).""" base_cls = _netmiko_driver_class(device_type) + if interactive: + base_cls = _interactive_driver_class(base_cls) - class _PreauthSession(base_cls): + class _PreauthSession(base_cls): # type: ignore[misc,valid-type] def establish_connection(self, width: int = 511, height: int = 1000) -> None: from netmiko.channel import SSHChannel @@ -321,6 +350,7 @@ def _connect_direct( *, session_timeout: int | None = None, session_log: Any = None, + interactive: bool = False, ) -> ConnectHandler: device_type = normalize_netmiko_device_type(creds["device_type"], creds["protocol"]) dev = _base_connect_kwargs( @@ -333,7 +363,7 @@ def _connect_direct( session_timeout=session_timeout, session_log=session_log, ) - return ConnectHandler(**dev) + return _build_netmiko_connection(dev, interactive=interactive) def _read_channel(conn: ConnectHandler, wait: float = 0.5, max_loops: int = 40) -> str: @@ -523,6 +553,7 @@ def _connect_via_cli_hop( session_log: Any = None, cols: int | None = None, rows: int | None = None, + interactive: bool = False, ) -> ConnectHandler: """Login to ZTE/Huawei/Cisco hop NE, run CLI jump command, then target secondary auth.""" hop_host = str(creds.get("hop_host") or "").strip() @@ -543,7 +574,7 @@ def _connect_via_cli_hop( session_timeout=session_timeout or 180, session_log=session_log, ) - conn = ConnectHandler(**hop_dev) + conn = _build_netmiko_connection(hop_dev, interactive=interactive) try: # MUST resize before stelnet/telnet — nested session captures hop TTY size at start # and often ignores later WINCH. Wrong width → mid-line edit redraw wraps in WebCRT. @@ -594,6 +625,7 @@ def _connect_via_bastion( *, session_timeout: int | None = None, session_log: Any = None, + interactive: bool = False, ) -> ConnectHandler: """SSH to bastion with composite username; bastion proxies to target (protocol proxy).""" hop_host = str(creds.get("hop_host") or "").strip() @@ -626,6 +658,7 @@ def _connect_via_bastion( enable_secret=str(creds.get("enable_secret") or ""), session_timeout=session_timeout or 180, session_log=session_log, + interactive=interactive, ) except Exception: if ssh_client is not None: @@ -656,6 +689,7 @@ def _connect_via_linux_hop( *, session_timeout: int | None = None, session_log: Any = None, + interactive: bool = False, ) -> ConnectHandler: """SSH to Linux bastion, then direct-tcpip tunnel to target (classic ProxyJump-style).""" hop_host = str(creds.get("hop_host") or "").strip() @@ -711,7 +745,7 @@ def _connect_via_linux_hop( session_log=session_log, ) dev["sock"] = channel - conn = ConnectHandler(**dev) + conn = _build_netmiko_connection(dev, interactive=interactive) conn._netx_jump_client = jump # type: ignore[attr-defined] return conn @@ -739,19 +773,40 @@ def open_netmiko_connection( session_log: Any = None, cols: int | None = None, rows: int | None = None, + interactive: bool = False, ) -> ConnectHandler: - """Open a Netmiko connection to the target NE (direct or via configured hop).""" + """Open a Netmiko connection to the target NE (direct or via configured hop). + + ``interactive=True`` (WebCRT) skips Netmiko's automatic ``terminal length`` / + ``terminal width`` (and vendor equivalents). Collection / MCP keep the default. + """ if creds.get("hop_enabled"): vendor = _hop_vendor(creds) if vendor == "linux": - return _connect_via_linux_hop(creds, session_timeout=session_timeout, session_log=session_log) + return _connect_via_linux_hop( + creds, + session_timeout=session_timeout, + session_log=session_log, + interactive=interactive, + ) if vendor == "bastion": - return _connect_via_bastion(creds, session_timeout=session_timeout, session_log=session_log) + return _connect_via_bastion( + creds, + session_timeout=session_timeout, + session_log=session_log, + interactive=interactive, + ) return _connect_via_cli_hop( creds, session_timeout=session_timeout, session_log=session_log, cols=cols, rows=rows, + interactive=interactive, ) - return _connect_direct(creds, session_timeout=session_timeout, session_log=session_log) + return _connect_direct( + creds, + session_timeout=session_timeout, + session_log=session_log, + interactive=interactive, + ) diff --git a/netx_api/webcrt_service.py b/netx_api/webcrt_service.py index f1db64c..c1e0f0f 100644 --- a/netx_api/webcrt_service.py +++ b/netx_api/webcrt_service.py @@ -598,6 +598,7 @@ def create_session( session_log=log_buf, cols=c, rows=r, + interactive=True, ) except Exception as exc: partial = _session_log_text(log_buf).strip() diff --git a/tests/test_ne_session_interactive.py b/tests/test_ne_session_interactive.py new file mode 100644 index 0000000..dcc4a2f --- /dev/null +++ b/tests/test_ne_session_interactive.py @@ -0,0 +1,53 @@ +from __future__ import annotations + +import unittest +from unittest.mock import MagicMock, patch + +from netx_api.ne_session_factory import ( + _build_netmiko_connection, + _interactive_driver_class, + _netmiko_driver_class, + open_netmiko_connection, +) + + +class InteractiveNetmikoTests(unittest.TestCase): + def test_interactive_driver_skips_paging_commands(self) -> None: + base = _netmiko_driver_class("cisco_ios") + cls = _interactive_driver_class(base) + self.assertEqual(cls.disable_paging(None), "") + self.assertEqual(cls.set_terminal_width(None), "") + + def test_build_non_interactive_uses_connect_handler(self) -> None: + fake = MagicMock(name="conn") + with patch("netx_api.ne_session_factory.ConnectHandler", return_value=fake) as ch: + out = _build_netmiko_connection( + { + "device_type": "cisco_ios", + "host": "1.1.1.1", + "username": "u", + "password": "p", + }, + interactive=False, + ) + self.assertIs(out, fake) + ch.assert_called_once() + + def test_open_webcrt_passes_interactive_flag(self) -> None: + creds = { + "device_type": "cisco_ios", + "protocol": "ssh", + "ip_address": "10.0.0.1", + "port": 22, + "username": "u", + "password": "p", + } + with patch("netx_api.ne_session_factory._connect_direct") as direct: + direct.return_value = MagicMock() + open_netmiko_connection(creds, interactive=True) + direct.assert_called_once() + self.assertTrue(direct.call_args.kwargs.get("interactive")) + + +if __name__ == "__main__": + unittest.main()