From 559cc14d2c9a3433d2b51fe095faf5ffdea8a3f0 Mon Sep 17 00:00:00 2001 From: oliver Date: Fri, 31 Jul 2026 16:58:47 +0800 Subject: [PATCH] Classify CLI auth rejects instead of reporting them as prompt timeouts. Detect Permission denied in hop login and session logs for config sync and WebCRT. Co-authored-by: Cursor --- netx_api/config_sync_runner.py | 16 ++++-- netx_api/ne_cli_errors.py | 90 ++++++++++++++++++++++++++++++++++ netx_api/ne_session_factory.py | 17 +++++++ netx_api/webcrt_service.py | 5 +- tests/test_ne_cli_errors.py | 38 ++++++++++++++ 5 files changed, 161 insertions(+), 5 deletions(-) create mode 100644 netx_api/ne_cli_errors.py create mode 100644 tests/test_ne_cli_errors.py diff --git a/netx_api/config_sync_runner.py b/netx_api/config_sync_runner.py index 5a0077a..ffdb969 100644 --- a/netx_api/config_sync_runner.py +++ b/netx_api/config_sync_runner.py @@ -2,6 +2,7 @@ from __future__ import annotations +import io import logging import time from concurrent.futures import ThreadPoolExecutor, TimeoutError as FuturesTimeout @@ -19,6 +20,7 @@ from .config_sync_commands import command_list, commands_for_vendor from .config_sync_service import finalize_cycle, sync_cycle_progress from .db import SessionLocal from .models import ConfigSyncCycle, ConfigSyncPolicy, ConfigSyncTask, NeConfigHistory, NeConfigSnapshot +from .ne_cli_errors import format_cli_failure, session_log_text from .ne_session_factory import close_netmiko_connection, open_netmiko_connection _log = logging.getLogger("netx.config_sync.runner") @@ -31,8 +33,7 @@ def _utcnow() -> datetime: def _format_error(exc: BaseException) -> str: - head = f"{type(exc).__name__}: {exc}" - return head[:1020] + return format_cli_failure(exc, limit=1020) def _pool_for_cycle(cycle_id: str, concurrency: int) -> ThreadPoolExecutor: @@ -103,11 +104,18 @@ def _claim_task(cycle_id: str, task_id: str) -> bool: def _collect_commands(creds: dict[str, Any], commands: list[str]) -> list[str]: per_cmd = int(settings.ne_collect_read_timeout_sec or 120) session_timeout = per_cmd * max(1, len(commands)) + 60 - conn = open_netmiko_connection(creds, session_timeout=session_timeout) + log_buf = io.BytesIO() + try: + conn = open_netmiko_connection(creds, session_timeout=session_timeout, session_log=log_buf) + except Exception as exc: + raise RuntimeError(format_cli_failure(exc, session_log_text(log_buf))) from exc try: outputs: list[str] = [] for command in commands: - out = conn.send_command(command_string=command, read_timeout=per_cmd) + try: + out = conn.send_command(command_string=command, read_timeout=per_cmd) + except Exception as exc: + raise RuntimeError(format_cli_failure(exc, session_log_text(log_buf))) from exc outputs.append(str(out or "")) return outputs finally: diff --git a/netx_api/ne_cli_errors.py b/netx_api/ne_cli_errors.py new file mode 100644 index 0000000..d551306 --- /dev/null +++ b/netx_api/ne_cli_errors.py @@ -0,0 +1,90 @@ +"""Classify CLI/Netmiko failures so auth rejects are not mistaken for read timeouts.""" + +from __future__ import annotations + +import re +from typing import Any + +# Prefer specific auth signals over generic Netmiko prompt timeouts. +_AUTH_PATTERNS: tuple[re.Pattern[str], ...] = tuple( + re.compile(p, re.I) + for p in ( + r"permission denied\s*\([^)]*password[^)]*\)", + r"permission denied\s*\([^)]*publickey[^)]*\)", + r"permission denied", + r"authentication failed", + r"authentication failure", + r"auth(?:entication)?\s*fail", + r"login\s*(?:invalid|failed|incorrect|rejected)", + r"access denied", + r"bad (?:secret|password|secrets)", + r"incorrect password", + r"%\s*(?:error|login):\s*authentication", + r"username or password is (?:wrong|incorrect|invalid)", + r"wrong password", + r"password rejected", + ) +) + +_PROMPT_TIMEOUT = re.compile(r"pattern not detected|readtimeout|read timeout", re.I) + + +def find_auth_failure_snippet(text: str, *, max_len: int = 220) -> str | None: + """Return a short matching auth-failure line/snippet, or None.""" + blob = str(text or "") + if not blob.strip(): + return None + for pat in _AUTH_PATTERNS: + m = pat.search(blob) + if not m: + continue + # Prefer the whole line containing the match. + start = blob.rfind("\n", 0, m.start()) + 1 + end = blob.find("\n", m.end()) + if end < 0: + end = len(blob) + line = blob[start:end].strip() + if not line: + line = m.group(0).strip() + return line[:max_len] + return None + + +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): + exc_text = f"{type(exc).__name__}: {exc}" + else: + exc_text = str(exc or "") + combined = f"{exc_text}\n{transcript or ''}" + auth = find_auth_failure_snippet(combined) + if auth: + # Keep enough of the original class for searchability when it was a timeout wrapper. + if _PROMPT_TIMEOUT.search(exc_text): + msg = f"auth_rejected: {auth} (reported_as_prompt_timeout)" + else: + msg = f"auth_rejected: {auth}" + return msg[:limit] + return exc_text[:limit] + + +def session_log_text(session_log: Any) -> str: + """Decode Netmiko session_log file/BytesIO into text.""" + if session_log is None: + return "" + try: + if hasattr(session_log, "getvalue"): + raw = session_log.getvalue() + elif hasattr(session_log, "read"): + try: + session_log.seek(0) + except Exception: + pass + raw = session_log.read() + else: + return "" + except Exception: + return "" + if isinstance(raw, bytes): + return raw.decode("utf-8", errors="replace") + return str(raw or "") diff --git a/netx_api/ne_session_factory.py b/netx_api/ne_session_factory.py index 4fe4647..392d3bb 100644 --- a/netx_api/ne_session_factory.py +++ b/netx_api/ne_session_factory.py @@ -489,11 +489,19 @@ def _prompt_needs_auth(text: str) -> tuple[bool, bool]: def _interactive_target_auth(conn: ConnectHandler, username: str, password: str) -> None: """Respond to username/password prompts after hop command (target credentials).""" + from .ne_cli_errors import find_auth_failure_snippet + deadline = time.time() + int(settings.ne_connect_timeout_sec or 30) sent_user = False sent_pass = False + acc = "" while time.time() < deadline: buf = _read_channel(conn, wait=0.3, max_loops=8) + if buf: + acc += buf + denied = find_auth_failure_snippet(acc) + if denied: + raise paramiko.AuthenticationException(f"target_auth_rejected: {denied}") need_user, need_pass = _prompt_needs_auth(buf) if need_pass and not sent_pass: _send_line(conn, password) @@ -504,14 +512,23 @@ def _interactive_target_auth(conn: ConnectHandler, username: str, password: str) sent_user = True continue if sent_pass and not need_user and not need_pass: + 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): + 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 time.sleep(0.3) + denied = find_auth_failure_snippet(acc) + if denied: + raise paramiko.AuthenticationException(f"target_auth_rejected: {denied}") if not sent_pass: raise TimeoutError("target_auth_timeout") diff --git a/netx_api/webcrt_service.py b/netx_api/webcrt_service.py index c1e0f0f..4f6aa46 100644 --- a/netx_api/webcrt_service.py +++ b/netx_api/webcrt_service.py @@ -612,7 +612,10 @@ def create_session( error=str(exc)[:500], transcript_len=len(partial), ) - detail = f"connect_failed:{exc}" + from .ne_cli_errors import format_cli_failure + + classified = format_cli_failure(exc, partial) + detail = f"connect_failed:{classified}" if partial: # Keep detail bounded; UI surfaces this on open failure. detail = f"{detail}\n--- device transcript ---\n{partial[-4000:]}" diff --git a/tests/test_ne_cli_errors.py b/tests/test_ne_cli_errors.py new file mode 100644 index 0000000..466fd95 --- /dev/null +++ b/tests/test_ne_cli_errors.py @@ -0,0 +1,38 @@ +"""Unit tests for CLI auth-failure classification.""" + +from __future__ import annotations + +import unittest + +from netx_api.ne_cli_errors import find_auth_failure_snippet, format_cli_failure + + +class CliAuthClassifyTests(unittest.TestCase): + def test_permission_denied_password(self): + text = "banner\nca-oper@114.1.105.3: Permission denied (password).\n" + self.assertIn("Permission denied", find_auth_failure_snippet(text) or "") + + def test_prompt_timeout_promoted_to_auth(self): + exc = "ReadTimeout: Pattern not detected: '[>#]' in output." + transcript = "Warning\nca-oper@114.1.105.3: Permission denied (password).\n" + msg = format_cli_failure(exc, transcript) + self.assertTrue(msg.startswith("auth_rejected:")) + self.assertIn("Permission denied", msg) + self.assertIn("prompt_timeout", msg) + + def test_plain_timeout_unchanged(self): + exc = RuntimeError("ReadTimeout: Pattern not detected: '[>#]' in output.") + msg = format_cli_failure(exc, "show running-config\n...still dumping...\n") + self.assertIn("ReadTimeout", msg) + self.assertFalse(msg.startswith("auth_rejected:")) + + def test_authentication_exception(self): + class AuthenticationException(Exception): + pass + + msg = format_cli_failure(AuthenticationException("target_auth_rejected: Permission denied")) + self.assertTrue(msg.startswith("auth_rejected:")) + + +if __name__ == "__main__": + unittest.main()