mirror of
https://github.com/hansjone/netx.git
synced 2026-10-08 23:33:21 +08:00
Fix ZTE hop false auth rejects by treating CLI prompt as success.
ZTE post-login banners can line-wrap login-failure stats and were misclassified before the target prompt was recognized. Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
parent
fbf73cbaaf
commit
f8d4a396f1
4 changed files with 114 additions and 22 deletions
|
|
@ -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):
|
||||
|
|
|
|||
|
|
@ -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:
|
||||
|
|
|
|||
|
|
@ -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()
|
||||
|
|
|
|||
|
|
@ -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()
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue