From 6901a7c019460052af7d5a0595cd66ef173eca3b Mon Sep 17 00:00:00 2001 From: hansjone Date: Fri, 28 Aug 2026 21:13:29 +0800 Subject: [PATCH] Fix Huawei stelnet hop auth hanging on host-key [Y/N] prompts. Answer Continue/Save-key prompts during CLI-hop secondary auth so WebCRT terminal login no longer times out after connectivity passes. --- netx_api/ne_session_connect.py | 123 ++++++++++++++++++++++++++---- netx_api/webcrt_channel.py | 13 +++- tests/test_cli_hop_target_auth.py | 113 +++++++++++++++++++++++++++ tests/test_webcrt.py | 7 ++ 4 files changed, 240 insertions(+), 16 deletions(-) create mode 100644 tests/test_cli_hop_target_auth.py diff --git a/netx_api/ne_session_connect.py b/netx_api/ne_session_connect.py index 540e0ec..78ed727 100644 --- a/netx_api/ne_session_connect.py +++ b/netx_api/ne_session_connect.py @@ -374,20 +374,75 @@ def _send_line(conn: ConnectHandler, line: str) -> None: conn.write_channel(text) +def _auth_prompt_tail(text: str) -> str: + """Last non-empty line of an auth transcript (prompt detection).""" + s = str(text or "").replace("\r\n", "\n").replace("\r", "\n") + lines = [ln.strip() for ln in s.split("\n") if ln.strip()] + return lines[-1] if lines else "" + + def _prompt_needs_auth(text: str) -> tuple[bool, bool]: - low = text.lower() - need_user = bool(re.search(r"(username|login|user\s*name)\s*[:>]", low)) - need_pass = bool(re.search(r"password\s*[:>]", low)) + """Detect username/password prompts (Huawei stelnet: ``Please input the username:``).""" + tail = _auth_prompt_tail(text).lower() + if not tail: + return False, False + # Prefer last-line match so earlier echoed prompts in ``acc`` do not stick forever. + need_user = bool( + re.search(r"(?:please\s+input\s+the\s+)?(?:user\s*name|username|login)\s*[:>]\s*$", tail) + ) + need_pass = bool(re.search(r"(?:enter\s+)?password\s*[:>]\s*$", tail)) return need_user, need_pass +def _prompt_needs_host_key_confirm(text: str) -> tuple[bool, bool]: + """Huawei VRP stelnet first-connect host-key prompts. + + Returns ``(continue_access, save_public_key)`` for: + - ``The server is not authenticated. Continue to access it? [Y/N]:`` → Y + - ``Save the server's public key? [Y/N]:`` → N + """ + tail = _auth_prompt_tail(text) + if not tail: + return False, False + low = tail.lower() + # Do not treat password-change ``Change now? [Y/N]:`` as host-key trust. + if re.search(r"(?:change\s*now|please\s*choose|password\s+needs\s+to\s+be\s+changed)", low): + return False, False + continue_access = bool( + re.search(r"(?:not\s+authenticated|continue\s+to\s+access)", low) + and re.search(r"\[Y/N\]\s*:\s*$", tail, flags=re.I) + ) + save_key = bool( + re.search(r"save\s+the\s+server'?s?\s+public\s+key", low) + and re.search(r"\[Y/N\]\s*:\s*$", tail, flags=re.I) + ) + return continue_access, save_key + + +def _looks_like_target_cli_prompt(text: str) -> bool: + """True when transcript ends at a device CLI prompt (not ``[Y/N]:`` / login).""" + tail = _auth_prompt_tail(text) + if not tail: + return False + if re.search(r"\[Y/N\]", tail, flags=re.I): + return False + need_user, need_pass = _prompt_needs_auth(tail) + if need_user or need_pass: + return False + # Huawei ```` / ``[sysname]``; Cisco ``R1#`` / ``R1>``. + return bool(re.search(r"(?:[>#\]])\s*$", tail)) or bool(re.search(r"^<[^>\r\n]+>\s*$", tail)) + + def _interactive_target_auth(conn: ConnectHandler, username: str, password: str) -> None: - """Respond to username/password prompts after hop command (target credentials).""" + """Respond to username/password (and Huawei stelnet host-key) prompts after hop command.""" from .ne_cli_errors import find_auth_failure_snippet - deadline = time.time() + int(settings.ne_connect_timeout_sec or 30) + # Hop stelnet can show Trying/Connected + two [Y/N] before password; keep budget generous. + deadline = time.time() + max(45, int(settings.ne_connect_timeout_sec or 30) + 15) sent_user = False sent_pass = False + answered_continue = False + answered_save_key = False acc = "" while time.time() < deadline: buf = _read_channel(conn, wait=0.3, max_loops=8) @@ -396,7 +451,19 @@ def _interactive_target_auth(conn: ConnectHandler, username: str, password: str) denied = find_auth_failure_snippet(acc) if denied: raise paramiko.AuthenticationException(f"target_auth_rejected: {denied}") - need_user, need_pass = _prompt_needs_auth(buf) + + # Match against accumulated tail so prompts split across reads are still seen. + continue_access, save_key = _prompt_needs_host_key_confirm(acc) + if continue_access and not answered_continue: + _send_line(conn, "Y") + answered_continue = True + continue + if save_key and not answered_save_key: + _send_line(conn, "N") + answered_save_key = True + continue + + need_user, need_pass = _prompt_needs_auth(acc) if need_pass and not sent_pass: _send_line(conn, password) sent_pass = True @@ -405,20 +472,40 @@ def _interactive_target_auth(conn: ConnectHandler, username: str, password: str) _send_line(conn, username) sent_user = True continue - if sent_pass and not need_user and not need_pass: + + if sent_pass and not need_user and not need_pass and not continue_access and not save_key: + denied = find_auth_failure_snippet(acc) + if denied: + raise paramiko.AuthenticationException(f"target_auth_rejected: {denied}") + # Prefer a real CLI prompt (Huawei last-login banner may precede it). + if _looks_like_target_cli_prompt(acc) or not buf.strip(): + return + # Banner mid-stream after password — keep reading briefly within deadline. + time.sleep(0.2) + continue + + if not buf.strip(): + time.sleep(0.3) + continue + # Huawei stelnet often reprints the hop ``[sysname]`` after host-key trust and + # *before* ``Enter password:``. Do not treat that as target login success. + if sent_pass and _looks_like_target_cli_prompt(buf): denied = find_auth_failure_snippet(acc) if denied: raise paramiko.AuthenticationException(f"target_auth_rejected: {denied}") return - if not buf.strip(): - time.sleep(0.3) - continue - if re.search(r"[>#]\s*$", buf): + if ( + sent_user + and not sent_pass + and not answered_continue + and not answered_save_key + and _looks_like_target_cli_prompt(buf) + ): + # Passwordless target after username only (no stelnet host-key dance). denied = find_auth_failure_snippet(acc) if denied: raise paramiko.AuthenticationException(f"target_auth_rejected: {denied}") - if sent_pass or (sent_user and not need_pass): - return + return time.sleep(0.3) denied = find_auth_failure_snippet(acc) if denied: @@ -502,6 +589,16 @@ def _connect_via_cli_hop( _send_line(conn, "") pre = pre + _read_channel(conn, wait=0.35, max_loops=10) hop_prompt = extract_cli_prompt_marker(pre) + # WebCRT skips hop session_preparation; wait a bit longer for a settled CLI + # before stelnet/telnet so the jump command is not typed into a half-ready PTY. + if interactive and not hop_prompt: + for _ in range(4): + more = _read_channel(conn, wait=0.4, max_loops=8) + if more: + pre += more + hop_prompt = extract_cli_prompt_marker(pre) + if hop_prompt: + break hop_cmd = render_hop_command(str(creds.get("hop_command_template") or ""), creds) _send_line(conn, hop_cmd) _interactive_target_auth(conn, str(creds["username"]), str(creds["password"])) diff --git a/netx_api/webcrt_channel.py b/netx_api/webcrt_channel.py index b205e5b..496a099 100644 --- a/netx_api/webcrt_channel.py +++ b/netx_api/webcrt_channel.py @@ -150,14 +150,21 @@ def _looks_like_login_prompt(text: str) -> bool: def _looks_like_password_change_prompt(text: str) -> bool: - """Huawei/VRP post-auth ``Change now? [Y/N]:`` (Netmiko already answers N).""" + """Huawei/VRP post-auth ``Change now? [Y/N]:`` (Netmiko already answers N). + + Do not match bare ``[Y/N]:`` — stelnet host-key trust prompts share that suffix + and are answered in ``_interactive_target_auth``, not by skipping Enter here. + """ s = str(text or "").replace("\r\n", "\n").replace("\r", "\n") lines = [ln.strip() for ln in s.split("\n") if ln.strip()] if not lines: return False last = lines[-1] - return bool(re.search(r"(?i)(change\s*now|please\s*choose|password\s+needs\s+to\s+be\s+changed).{0,80}:\s*$", last)) or bool( - re.search(r"\[Y/N\]\s*:\s*$", last, flags=re.I) + return bool( + re.search( + r"(?i)(change\s*now|please\s*choose|password\s+needs\s+to\s+be\s+changed).{0,80}:\s*$", + last, + ) ) diff --git a/tests/test_cli_hop_target_auth.py b/tests/test_cli_hop_target_auth.py new file mode 100644 index 0000000..2059376 --- /dev/null +++ b/tests/test_cli_hop_target_auth.py @@ -0,0 +1,113 @@ +"""Unit tests for CLI-hop secondary auth (Huawei stelnet host-key + login).""" + +from __future__ import annotations + +import unittest +from unittest.mock import MagicMock, patch + +from netx_api.ne_session_connect import ( + _interactive_target_auth, + _looks_like_target_cli_prompt, + _prompt_needs_auth, + _prompt_needs_host_key_confirm, +) + + +class PromptDetectTests(unittest.TestCase): + def test_huawei_username_and_password(self) -> None: + self.assertEqual( + _prompt_needs_auth("Please input the username:"), + (True, False), + ) + self.assertEqual(_prompt_needs_auth("Enter password:"), (False, True)) + self.assertEqual(_prompt_needs_auth("Password:"), (False, True)) + + def test_host_key_prompts(self) -> None: + cont, save = _prompt_needs_host_key_confirm( + "The server is not authenticated. Continue to access it? [Y/N]:" + ) + self.assertTrue(cont) + self.assertFalse(save) + cont, save = _prompt_needs_host_key_confirm("Save the server's public key? [Y/N]:") + self.assertFalse(cont) + self.assertTrue(save) + + def test_password_change_not_host_key(self) -> None: + cont, save = _prompt_needs_host_key_confirm("Change now? [Y/N]:") + self.assertFalse(cont) + self.assertFalse(save) + + def test_cli_prompt_rejects_yn(self) -> None: + self.assertFalse(_looks_like_target_cli_prompt("Continue? [Y/N]:")) + self.assertTrue(_looks_like_target_cli_prompt("")) + self.assertTrue(_looks_like_target_cli_prompt("[HW-VM-AR1000V-1]")) + + +class HuaweiStelnetAuthTests(unittest.TestCase): + @patch("netx_api.ne_session_connect._read_channel") + @patch("netx_api.ne_session_connect._send_line") + def test_answers_host_key_then_login( + self, + mock_send: MagicMock, + mock_read: MagicMock, + ) -> None: + """Transcript from Huawei AR stelnet first connect to untrusted target.""" + mock_read.side_effect = [ + "Please input the username:", + ( + "Trying 1.1.1.2 ...\n" + "Press CTRL+K to abort\n" + "Connected to 1.1.1.2 ...\n" + "The server is not authenticated. Continue to access it? [Y/N]:" + ), + "Save the server's public key? [Y/N]:", + ( + "Jan 1 2017 01:16:39+00:00 HW-VM-AR1000V-1 " + "%%01SSH/4/SAVE_PUBLICKEY(l)[3]:When deciding whether to save " + "the server's public key 1.1.1.2, the user chose N.\n" + "[HW-VM-AR1000V-1]\n" + "Enter password:" + ), + ( + " -----------------------------------------------------------------------------\n" + " User last login information:\n" + " -----------------------------------------------------------------------------\n" + " Access Type: SSH\n" + " IP-Address : 172.16.0.6\n" + " Time : 2017-01-05 00:11:36+00:00\n" + " ----------------\n" + "" + ), + ] + conn = MagicMock() + _interactive_target_auth(conn, "ipran", "secret") + self.assertEqual( + [c.args[1] for c in mock_send.call_args_list], + ["ipran", "Y", "N", "secret"], + ) + + @patch("netx_api.ne_session_connect._read_channel") + @patch("netx_api.ne_session_connect._send_line") + def test_hop_prompt_before_password_not_success( + self, + mock_send: MagicMock, + mock_read: MagicMock, + ) -> None: + """``[hop]`` between host-key and password must not end auth early.""" + mock_read.side_effect = [ + "Please input the username:", + "The server is not authenticated. Continue to access it? [Y/N]:", + "Save the server's public key? [Y/N]:", + "[HW-VM-AR1000V-1]\n", # hop reprint — must keep waiting + "Enter password:", + "\n", + "", + ] + conn = MagicMock() + _interactive_target_auth(conn, "ipran", "secret") + sent = [c.args[1] for c in mock_send.call_args_list] + self.assertEqual(sent, ["ipran", "Y", "N", "secret"]) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_webcrt.py b/tests/test_webcrt.py index 2c364f8..8ac2a35 100644 --- a/tests/test_webcrt.py +++ b/tests/test_webcrt.py @@ -111,6 +111,13 @@ class WebcrtServiceTests(unittest.TestCase): self.assertEqual(svc.prepare_bootstrap_output("banner\n:"), "banner\n") self.assertTrue(svc._looks_like_password_change_prompt("Change now? [Y/N]:")) self.assertFalse(svc._looks_like_password_change_prompt("Change now? [Y/N]:N")) + # Bare / stelnet host-key [Y/N] must not be treated as password-change. + self.assertFalse(svc._looks_like_password_change_prompt("[Y/N]:")) + self.assertFalse( + svc._looks_like_password_change_prompt( + "The server is not authenticated. Continue to access it? [Y/N]:" + ) + ) # WS attach must not send Enter when bootstrap is a login prompt. self.assertFalse( (not svc._looks_like_cli_prompt("Username:") and not svc._looks_like_login_prompt("Username:"))