From b540ea21c102bf3ca51440fb35060df7ccfaeb95 Mon Sep 17 00:00:00 2001 From: hansjone Date: Thu, 30 Jul 2026 02:34:25 +0000 Subject: [PATCH] fix(webcrt): resize hop PTY before nested CLI jump Nested stelnet/telnet captures hop TTY width at start; Netmiko's default 511 cols made mid-line edit redraws wrap and garble in ~80-col WebCRT. Co-authored-by: Cursor --- netx_api/ne_session_factory.py | 34 ++++++++++++++++++++++++++- netx_api/webcrt_router.py | 1 + netx_api/webcrt_service.py | 7 +++++- tests/test_cli_hop_return.py | 5 +++- tests/test_webcrt.py | 42 ++++++++++++++++++++++++++++++++++ web/src/services/api.ts | 2 ++ 6 files changed, 88 insertions(+), 3 deletions(-) diff --git a/netx_api/ne_session_factory.py b/netx_api/ne_session_factory.py index 97f7693..726c10d 100644 --- a/netx_api/ne_session_factory.py +++ b/netx_api/ne_session_factory.py @@ -497,11 +497,32 @@ def _hop_netmiko_device_type(vendor: str, hop_protocol: str) -> str: return normalize_netmiko_device_type(base, hop_protocol) +def _resize_pty(conn: ConnectHandler, cols: int | None = None, rows: int | None = None) -> None: + """Set SSH PTY size so nested telnet/stelnet inherits the interactive terminal geometry. + + Netmiko defaults to 511x1000. If a CLI hop jump runs at that size and WebCRT is ~80 + columns, mid-line edit redraws (spaces / clear-to-EOL) wrap and garble the display. + """ + if cols is None and rows is None: + return + channel = getattr(conn, "remote_conn", None) + if channel is None or not hasattr(channel, "resize_pty"): + return + c = max(20, min(500, int(cols if cols is not None else 80))) + r = max(5, min(200, int(rows if rows is not None else 24))) + try: + channel.resize_pty(width=c, height=r) + except Exception: + _log.debug("resize_pty failed cols=%s rows=%s", c, r, exc_info=True) + + def _connect_via_cli_hop( creds: dict[str, Any], *, session_timeout: int | None = None, session_log: Any = None, + cols: int | None = None, + rows: int | None = None, ) -> 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() @@ -524,6 +545,9 @@ def _connect_via_cli_hop( ) conn = ConnectHandler(**hop_dev) 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. + _resize_pty(conn, cols, rows) pre = _read_channel(conn, wait=0.5) hop_prompt = extract_cli_prompt_marker(pre) if not hop_prompt: @@ -713,6 +737,8 @@ def open_netmiko_connection( *, session_timeout: int | None = None, session_log: Any = None, + cols: int | None = None, + rows: int | None = None, ) -> ConnectHandler: """Open a Netmiko connection to the target NE (direct or via configured hop).""" if creds.get("hop_enabled"): @@ -721,5 +747,11 @@ def open_netmiko_connection( return _connect_via_linux_hop(creds, session_timeout=session_timeout, session_log=session_log) if vendor == "bastion": return _connect_via_bastion(creds, session_timeout=session_timeout, session_log=session_log) - return _connect_via_cli_hop(creds, session_timeout=session_timeout, session_log=session_log) + return _connect_via_cli_hop( + creds, + session_timeout=session_timeout, + session_log=session_log, + cols=cols, + rows=rows, + ) return _connect_direct(creds, session_timeout=session_timeout, session_log=session_log) diff --git a/netx_api/webcrt_router.py b/netx_api/webcrt_router.py index e815751..5e15a35 100644 --- a/netx_api/webcrt_router.py +++ b/netx_api/webcrt_router.py @@ -115,6 +115,7 @@ async def websocket_session(websocket: WebSocket, session_id: str) -> None: "rows": sess.rows, "device_type": sess.device_type, "vendor": sess.vendor, + "cli_hop": bool(sess.cli_hop_guard), } ) diff --git a/netx_api/webcrt_service.py b/netx_api/webcrt_service.py index fbc53d6..f1db64c 100644 --- a/netx_api/webcrt_service.py +++ b/netx_api/webcrt_service.py @@ -593,7 +593,11 @@ def create_session( log_buf = io.BytesIO() try: conn = open_netmiko_connection( - creds, session_timeout=connect_timeout, session_log=log_buf + creds, + session_timeout=connect_timeout, + session_log=log_buf, + cols=c, + rows=r, ) except Exception as exc: partial = _session_log_text(log_buf).strip() @@ -686,6 +690,7 @@ def create_session( "cols": sess.cols, "rows": sess.rows, "ws_path": f"/v1/webcrt/sessions/{session_id}/ws", + "cli_hop": bool(hop_guard), } diff --git a/tests/test_cli_hop_return.py b/tests/test_cli_hop_return.py index 307f2d9..27a81a1 100644 --- a/tests/test_cli_hop_return.py +++ b/tests/test_cli_hop_return.py @@ -62,6 +62,7 @@ class CliHopReturnDetectionTests(unittest.TestCase): from netx_api.ne_session_factory import _connect_via_cli_hop conn = MagicMock() + conn.remote_conn = MagicMock() mock_ch.return_value = conn mock_read.side_effect = ["\n", ""] mock_auth.return_value = None @@ -79,8 +80,10 @@ class CliHopReturnDetectionTests(unittest.TestCase): "ip_address": "10.0.0.2", "port": 22, } - out = _connect_via_cli_hop(creds) + out = _connect_via_cli_hop(creds, cols=120, rows=40) self.assertIs(out, conn) + # Nested stelnet must see WebCRT geometry, not Netmiko's default 511x1000. + conn.remote_conn.resize_pty.assert_called_with(width=120, height=40) guard = get_cli_hop_guard(conn) self.assertIsNotNone(guard) assert guard is not None diff --git a/tests/test_webcrt.py b/tests/test_webcrt.py index 59bd8eb..bd93fac 100644 --- a/tests/test_webcrt.py +++ b/tests/test_webcrt.py @@ -122,6 +122,7 @@ class WebcrtServiceTests(unittest.TestCase): with patch.object(svc.settings, "webcrt_max_sessions", 1): out = svc.create_session(db, ne_id="ne-a", cols=80, rows=24, client="test") self.assertIn("session_id", out) + self.assertFalse(out.get("cli_hop")) with self.assertRaises(HTTPException) as ctx: svc.create_session(db, ne_id="ne-a", cols=80, rows=24, client="test") self.assertEqual(ctx.exception.status_code, 429) @@ -179,6 +180,7 @@ class WebcrtServiceTests(unittest.TestCase): self.assertIn("session_log", mock_open.call_args.kwargs) fake.remote_conn.resize_pty.assert_called() self.assertEqual(out["ne_id"], "ne-hop") + self.assertFalse(out.get("cli_hop")) # bastion hop is not vendor CLI hop guard sess = svc.get_session(out["session_id"]) assert sess is not None boot = sess.bootstrap_output.decode("utf-8", errors="replace") @@ -190,6 +192,46 @@ class WebcrtServiceTests(unittest.TestCase): self.assertEqual(fake.written[len(before) :], ["\n"]) svc.close_session(out["session_id"], reason="test") + @patch.object(svc, "_audit") + @patch.object(svc, "get_cli_hop_guard") + @patch.object(svc, "open_netmiko_connection") + @patch("netx_api.cli_resolve.resolve_cli_target") + def test_create_session_reports_cli_hop( + self, + mock_resolve: MagicMock, + mock_open: MagicMock, + mock_guard: MagicMock, + _mock_audit: MagicMock, + ) -> None: + mock_resolve.return_value = ( + { + "username": "u", + "password": "p", + "hop_enabled": True, + "hop_vendor": "huawei", + "ip_address": "10.0.0.3", + "protocol": "ssh", + }, + { + "id": "ne-cli-hop", + "name": "C", + "ip_address": "10.0.0.3", + "protocol": "ssh", + "source": "managed", + }, + ) + mock_open.side_effect = lambda *a, **k: _FakeConn() + mock_guard.return_value = {"hop_prompt": "", "hop_vendor": "huawei", "hop_host": "1.1.1.1"} + out = svc.create_session(MagicMock(), ne_id="ne-cli-hop", cols=100, rows=30, client="test") + self.assertTrue(out.get("cli_hop")) + self.assertEqual(mock_open.call_args.kwargs.get("cols"), 100) + self.assertEqual(mock_open.call_args.kwargs.get("rows"), 30) + sess = svc.get_session(out["session_id"]) + assert sess is not None + self.assertTrue(sess.cli_hop_guard) + self.assertEqual(sess.cli_hop_prompt, "") + svc.close_session(out["session_id"], reason="test") + @patch.object(svc, "_audit") @patch.object(svc, "open_netmiko_connection") @patch("netx_api.cli_resolve.resolve_cli_target") diff --git a/web/src/services/api.ts b/web/src/services/api.ts index ae8590f..5f022c5 100644 --- a/web/src/services/api.ts +++ b/web/src/services/api.ts @@ -410,6 +410,8 @@ export type WebcrtSessionCreateResult = { cols: number; rows: number; ws_path: string; + /** True when session used a vendor CLI hop (nested stelnet/telnet). */ + cli_hop?: boolean; }; export const createWebcrtSession = (body: {