diff --git a/netx_api/ne_cli_errors.py b/netx_api/ne_cli_errors.py index 076767f..7158925 100644 --- a/netx_api/ne_cli_errors.py +++ b/netx_api/ne_cli_errors.py @@ -6,6 +6,8 @@ import re from typing import Any # Prefer specific auth signals over generic Netmiko prompt timeouts. +# Note: do NOT match bare ``authentication failure`` — ZTE/Huawei success banners +# say ``0 authentication failure occurred`` (stats, not a reject), often line-wrapped. _AUTH_PATTERNS: tuple[re.Pattern[str], ...] = tuple( re.compile(p, re.I) for p in ( @@ -13,8 +15,7 @@ _AUTH_PATTERNS: tuple[re.Pattern[str], ...] = tuple( r"permission denied\s*\([^)]*publickey[^)]*\)", r"permission denied", r"authentication failed", - r"authentication failure", - r"auth(?:entication)?\s*fail", + r"auth(?:entication)?\s*fail(?!ures?\s+occurred)", r"login\s*(?:invalid|failed|incorrect|rejected)", r"access denied", r"bad (?:secret|password|secrets)", @@ -26,16 +27,20 @@ _AUTH_PATTERNS: tuple[re.Pattern[str], ...] = tuple( ) ) -# Huawei / ZTE post-login security banner (success path via SSH or bastion hop). +# Huawei / ZTE post-login security banner (success path via SSH or CLI hop). # Example: "Afterwards, 0 authentication failure occurred." +# ZTE may wrap mid-word: "... SSH. After" + "wards, 0 authentication failure occurred." +# or worse: "...Afterwa" + "rds, 0 ..." leaving "rwards, 0 authentication failure occurred." _LOGIN_SUCCESS_NOTICE = re.compile( - r"^.*(?:" - r"last\s+successful\s+login\s+was\s+performed" - r"|afterwards,\s*\d+\s+authentication\s+failures?\s+occurred" - r"|上次成功登录" - r"|之后发生了?\s*\d+\s*次认证失败" - r").*$", - re.I | re.M, + r"(?is)(?:" + r"last\s+successful\s+login\s+was\s+performed[^\n]*" + r"|login\s+at\s+[^\n]*through\s+ssh[^\n]*" + r"|(?:after)?wards,\s*\d+\s+authentication\s+failures?\s+occurred[^\n]*" + r"|afterwards,\s*\d+\s+authentication\s+failures?\s+occurred[^\n]*" + r"|\d+\s+authentication\s+failures?\s+occurred[^\n]*" + r"|上次成功登录[^\n]*" + r"|之后发生了?\s*\d+\s*次认证失败[^\n]*" + r")" ) _PROMPT_TIMEOUT = re.compile(r"pattern not detected|readtimeout|read timeout", re.I) @@ -62,6 +67,11 @@ def find_auth_failure_snippet(text: str, *, max_len: int = 220) -> str | None: return None +def saw_zte_login_banner(text: str) -> bool: + """True when ZTE post-login MOTD is present (nested ssh hop success path).""" + return bool(_LOGIN_SUCCESS_NOTICE.search(str(text or ""))) + + def format_cli_failure(exc: BaseException | str, transcript: str = "", *, limit: int = 1020) -> str: """Human/ops-facing failure message; promote auth rejects above Pattern/ReadTimeout.""" if isinstance(exc, BaseException): diff --git a/netx_api/ne_session_connect.py b/netx_api/ne_session_connect.py index 932d164..3813473 100644 --- a/netx_api/ne_session_connect.py +++ b/netx_api/ne_session_connect.py @@ -669,6 +669,20 @@ def _looks_like_target_cli_prompt(text: str) -> bool: return bool(re.search(r"(?:[>#\]])\s*$", tail)) or bool(re.search(r"^<[^>\r\n]+>\s*$", tail)) +def _target_auth_failure_snippet(text: str, *, sent_pass: bool) -> str | None: + """Return auth-reject snippet unless we already reached target CLI prompt. + + Prompt-at-tail is the ground truth for hop login success; banner/stat lines + (e.g. ZTE ``0 authentication failure occurred``) must not override it. + """ + from .ne_cli_errors import find_auth_failure_snippet + + acc = str(text or "") + if sent_pass and _looks_like_target_cli_prompt(acc): + return None + return find_auth_failure_snippet(acc) + + def _looks_like_password_change_prompt(text: str) -> bool: """Huawei/VRP ``Change now? [Y/N]:`` after successful password (not host-key).""" tail = _auth_prompt_tail(text, lines=2) @@ -691,6 +705,12 @@ def _saw_huawei_last_login(text: str) -> bool: ) +def _saw_zte_login_banner(text: str) -> bool: + from .ne_cli_errors import saw_zte_login_banner + + return saw_zte_login_banner(text) + + def _interactive_target_auth( conn: ConnectHandler, username: str, @@ -704,8 +724,6 @@ def _interactive_target_auth( When ``emit_raw`` is False, device bytes are assumed to already reach the UI via ``session_log`` ProgressBytesIO — only emit ``[netx]`` markers (avoids doubled echo). """ - from .ne_cli_errors import find_auth_failure_snippet - # Hop stelnet can show Trying/Connected + two [Y/N] before password; keep budget generous. deadline = time.time() + max(60, int(settings.ne_connect_timeout_sec or 30) + 30) sent_user = False @@ -721,7 +739,9 @@ def _interactive_target_auth( acc += buf if emit_raw: _emit_progress(progress_cb, buf) - denied = find_auth_failure_snippet(acc) + if sent_pass and _looks_like_target_cli_prompt(acc): + return + denied = _target_auth_failure_snippet(acc, sent_pass=sent_pass) if denied: raise paramiko.AuthenticationException(f"target_auth_rejected: {denied}") @@ -761,21 +781,23 @@ def _interactive_target_auth( continue 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}") if _looks_like_target_cli_prompt(acc): return + denied = _target_auth_failure_snippet(acc, sent_pass=True) + if denied: + raise paramiko.AuthenticationException(f"target_auth_rejected: {denied}") # Last-login banner already printed — hand off quickly even if prompt parse lags. - if _saw_huawei_last_login(acc) and empty_after_pass >= 1: + if (_saw_huawei_last_login(acc) or _saw_zte_login_banner(acc)) and empty_after_pass >= 1: return # Do not treat a single empty read right after password as success — - # Huawei still prints last-login banner / prompt. + # Huawei/ZTE still prints last-login banner / prompt. if not buf.strip(): empty_after_pass += 1 # Faster handoff: live echo already shows login; WS cannot accept stdin # until open_netmiko_connection returns. - if empty_after_pass >= (2 if _saw_huawei_last_login(acc) else 3): + if empty_after_pass >= ( + 2 if (_saw_huawei_last_login(acc) or _saw_zte_login_banner(acc)) else 3 + ): return else: empty_after_pass = 0 @@ -788,7 +810,7 @@ def _interactive_target_auth( # 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) + denied = _target_auth_failure_snippet(acc, sent_pass=True) if denied: raise paramiko.AuthenticationException(f"target_auth_rejected: {denied}") return @@ -800,12 +822,12 @@ def _interactive_target_auth( and _looks_like_target_cli_prompt(buf) ): # Passwordless target after username only (no stelnet host-key dance). - denied = find_auth_failure_snippet(acc) + denied = _target_auth_failure_snippet(acc, sent_pass=False) if denied: raise paramiko.AuthenticationException(f"target_auth_rejected: {denied}") return time.sleep(0.15) - denied = find_auth_failure_snippet(acc) + denied = _target_auth_failure_snippet(acc, sent_pass=sent_pass) if denied: raise paramiko.AuthenticationException(f"target_auth_rejected: {denied}") if not sent_pass: diff --git a/tests/test_cli_hop_target_auth.py b/tests/test_cli_hop_target_auth.py index ecf8008..d4eafc2 100644 --- a/tests/test_cli_hop_target_auth.py +++ b/tests/test_cli_hop_target_auth.py @@ -156,5 +156,34 @@ class HuaweiStelnetAuthTests(unittest.TestCase): self.assertTrue(any("password-change" in p for p in seen)) +class ZteHopAuthTests(unittest.TestCase): + @patch("netx_api.ne_session_connect._read_channel") + @patch("netx_api.ne_session_connect._send_line") + def test_zte_banner_with_prompt_not_auth_failure( + self, + mock_send: MagicMock, + mock_read: MagicMock, + ) -> None: + """ZTE login stats contain 'authentication failure' — prompt is ground truth.""" + mock_read.side_effect = [ + "Username:", + "Password:", + ( + "Welcome to ZXR10 ZXCTN 6120H\n" + "Login at 09:43:53 08-31-2026 from 10.229.147.122 through SSH.\n" + "The last successful login was performed at 09:43:44 08-31-2026 " + "from 10.229.147.122 through SSH. Afterwa\n" + "rwards, 0 authentication failure occurred.\n" + "AL5458-ACC-6120HS#" + ), + ] + conn = MagicMock() + _interactive_target_auth(conn, "ipran", "secret") + self.assertEqual( + [c.args[1] for c in mock_send.call_args_list], + ["ipran", "secret"], + ) + + if __name__ == "__main__": unittest.main() diff --git a/tests/test_ne_cli_errors.py b/tests/test_ne_cli_errors.py index 9243032..19c974a 100644 --- a/tests/test_ne_cli_errors.py +++ b/tests/test_ne_cli_errors.py @@ -67,6 +67,37 @@ class CliAuthClassifyTests(unittest.TestCase): ) self.assertIn("Username or password is wrong", find_auth_failure_snippet(text) or "") + def test_zte_post_login_banner_not_auth_failure(self): + """ZTE nested ssh hop: '0 authentication failure occurred' is login stats.""" + text = ( + "Welcome to ZXR10 ZXCTN 6120H Carrier-Class Router of ZTE Corporation\n" + "Login at 09:43:53 08-31-2026 from 10.229.147.122 through SSH.\n" + "The last successful login was performed at 09:43:44 08-31-2026 " + "from 10.229.147.122 through SSH. Afterwards, 0 authentication " + "failure occurred.\n" + "AL5458-ACC-6120HS#" + ) + self.assertIsNone(find_auth_failure_snippet(text)) + + def test_zte_wrapped_afterwards_banner_not_auth_failure(self): + """Narrow PTY wraps 'Afterwards' — must not classify as auth reject.""" + text = ( + "The last successful login was performed at 09:43:44 08-31-2026 " + "from 10.229.147.122 through SSH. After\n" + "wards, 0 authentication failure occurred.\n" + "AL5458-ACC-6120HS#" + ) + self.assertIsNone(find_auth_failure_snippet(text)) + + def test_zte_severely_wrapped_afterwards_banner_not_auth_failure(self): + """Production saw 'rwards' after mid-word wrap — still login stats, not reject.""" + text = ( + "from 10.229.147.122 through SSH. Afterwa\n" + "rwards, 0 authentication failure occurred.\n" + "AL5458-ACC-6120HS#" + ) + self.assertIsNone(find_auth_failure_snippet(text)) + if __name__ == "__main__": unittest.main()