diff --git a/netx_api/cli_creds.py b/netx_api/cli_creds.py new file mode 100644 index 0000000..29d90de --- /dev/null +++ b/netx_api/cli_creds.py @@ -0,0 +1,72 @@ +"""Shared CLI credential readiness checks (non-interactive exec vs WebCRT).""" + +from __future__ import annotations + +from typing import Any + +from fastapi import HTTPException + +# Reasons surfaced in job rows / execManagedNe responses (grep-friendly). +REASON_IP_REQUIRED = "ip_address_required" +REASON_USERNAME_REQUIRED = "cli_username_required" +REASON_NO_PASSWORD = "no_password" +REASON_HOP_INCOMPLETE = "hop_credentials_incomplete" +REASON_INCOMPLETE = "credentials_incomplete" + + +def cli_creds_ready(creds: dict[str, Any], *, interactive: bool = False) -> tuple[bool, str]: + """Return (ready, reason). ``reason`` is empty when ready. + + Non-interactive paths (LLDP, config sync, collection, MCP exec) require + enough credentials to authenticate without a human at the terminal. + + Interactive WebCRT allows telnet without saved username/password so the user + can type credentials in the terminal (SecureCRT-style). + """ + ip = str(creds.get("ip_address") or "").strip() + if not ip: + return False, REASON_IP_REQUIRED + + hop_enabled = bool(creds.get("hop_enabled")) + hop_vendor = str(creds.get("hop_vendor") or "").strip().lower() + auth_mode = str(creds.get("hop_target_auth_mode") or "bastion_managed").strip().lower() + protocol = str(creds.get("protocol") or "ssh").strip().lower() + username = str(creds.get("username") or "").strip() + password = str(creds.get("password") or "") + + if hop_enabled: + hop_host = str(creds.get("hop_host") or "").strip() + hop_user = str(creds.get("hop_username") or "").strip() + hop_pass = str(creds.get("hop_password") or "") + if not hop_host or not hop_user or not hop_pass: + return False, REASON_HOP_INCOMPLETE + # Bastion-managed: target password may live on the bastion side. + if hop_vendor == "bastion" and auth_mode == "bastion_managed": + return True, "" + if not username: + return False, REASON_USERNAME_REQUIRED + if not password: + return False, REASON_NO_PASSWORD + return True, "" + + if interactive and protocol == "telnet": + return True, "" + + if not username: + return False, REASON_USERNAME_REQUIRED + if not password: + return False, REASON_NO_PASSWORD + return True, "" + + +def cli_creds_skip_reason(creds: dict[str, Any], *, interactive: bool = False) -> str | None: + """Return a skip/fail reason, or None when CLI may proceed.""" + ready, reason = cli_creds_ready(creds, interactive=interactive) + return None if ready else reason + + +def require_cli_creds_ready(creds: dict[str, Any], *, interactive: bool = False) -> None: + """Raise HTTP 400 when credentials cannot support the requested CLI mode.""" + ready, reason = cli_creds_ready(creds, interactive=interactive) + if not ready: + raise HTTPException(status_code=400, detail=reason or REASON_INCOMPLETE) diff --git a/netx_api/config_sync_runner.py b/netx_api/config_sync_runner.py index 1349186..8e91ad6 100644 --- a/netx_api/config_sync_runner.py +++ b/netx_api/config_sync_runner.py @@ -13,6 +13,7 @@ from uuid import uuid4 from fastapi import HTTPException +from .cli_creds import cli_creds_skip_reason from .cli_resolve import resolve_cli_target from .config import settings from .config_sync_codec import compress_text @@ -318,6 +319,19 @@ def _run_single(cycle_id: str, task_id: str) -> None: _update_task(task_id, status="fail", message=_format_error(exc), ended_at=_utcnow()) return + skip = cli_creds_skip_reason(creds, interactive=False) + if skip: + _update_task( + task_id, + status="fail", + message=skip[:1020], + vendor=str(device.get("vendor") or vendor_hint or ""), + ne_name=str(device.get("name") or task.ne_name or ""), + ne_ip=str(device.get("ip_address") or task.ne_ip or ""), + ended_at=_utcnow(), + ) + return + vendor = str(device.get("vendor") or vendor_hint or "") device_type = str(device.get("device_type") or "") ne_name = str(device.get("name") or task.ne_name or "") diff --git a/netx_api/ne_collect_runner.py b/netx_api/ne_collect_runner.py index f4f8624..908455a 100644 --- a/netx_api/ne_collect_runner.py +++ b/netx_api/ne_collect_runner.py @@ -14,6 +14,7 @@ from .config import settings from .db import SessionLocal from fastapi import HTTPException +from .cli_creds import cli_creds_skip_reason from .cli_budget import clamp_cli_workers from .cli_resolve import resolve_cli_target from .models import NeCollectionJob, NeCollectionRun @@ -205,6 +206,10 @@ def _run_single(job_id: str, run_id: str, commands: list[str]) -> None: creds, _device = resolve_cli_target(db, ume_ne_id=tid) else: creds, _device = resolve_cli_target(db, managed_ne_id=tid) + skip = cli_creds_skip_reason(creds, interactive=False) + if skip: + _update_run(run_id, status="fail", message=skip[:1020], ended_at=datetime.now()) + return output = _collect_with_timeout(creds, commands) finished_at = datetime.now() name_part = _safe_filename_part(str(run.ne_name or creds.get("name") or "ne")) diff --git a/netx_api/ne_connect.py b/netx_api/ne_connect.py index 4b4f682..7ee5af2 100644 --- a/netx_api/ne_connect.py +++ b/netx_api/ne_connect.py @@ -11,6 +11,7 @@ from .config import settings from .db import SessionLocal from .models import ManagedNE, UmeCliOverride, UmeInventoryNE from .ne_crypto import CredentialCryptoError +from .cli_creds import cli_creds_skip_reason from .cli_resolve import resolve_cli_target from .ne_service import get_device_credentials from .ne_netmiko import send_show_command @@ -361,6 +362,13 @@ def _run_single(ne_id: str) -> None: ) _update_row(ne_id, "fail", str(exc), detail=detail) return + skip = cli_creds_skip_reason(creds, interactive=False) + if skip: + detail = _truncate_detail( + "\n".join(_connect_context_lines(creds)) + f"\nresult=fail\nerror={skip}" + ) + _update_row(ne_id, "fail", skip, detail=detail) + return status, message, discovered, detail = _probe_device(creds) _update_row(ne_id, status, message, discovered, detail=detail) except Exception as exc: @@ -418,6 +426,13 @@ def _run_single_ume(ume_ne_id: str) -> None: detail = _truncate_detail(traceback.format_exc()) _update_ume_override_row(uid, "fail", str(exc)[:480], detail=detail) return + skip = cli_creds_skip_reason(creds, interactive=False) + if skip: + detail = _truncate_detail( + "\n".join(_connect_context_lines(creds)) + f"\nresult=fail\nerror={skip}" + ) + _update_ume_override_row(uid, "fail", skip, detail=detail) + return status, message, discovered, detail = _probe_device(creds) _update_ume_override_row(uid, status, message, discovered, detail=detail) except Exception as exc: diff --git a/netx_api/ne_exec.py b/netx_api/ne_exec.py index cde728b..e1b5667 100644 --- a/netx_api/ne_exec.py +++ b/netx_api/ne_exec.py @@ -8,6 +8,7 @@ from typing import Any from fastapi import HTTPException from sqlalchemy.orm import Session +from .cli_creds import cli_creds_skip_reason from .cli_resolve import resolve_cli_target from .config import settings from .db import SessionLocal @@ -65,6 +66,16 @@ def execute_managed_ne_commands( validate_ne_exec_command(c) creds, device = resolve_cli_target(db, managed_ne_id=mid or None, ume_ne_id=uid or None) + skip = cli_creds_skip_reason(creds, interactive=False) + if skip: + return { + "ok": False, + "device": device, + "commands": cmds, + "read_timeout_sec": _normalize_read_timeout(read_timeout_sec), + "error": skip, + "detail": skip, + } read_timeout = _normalize_read_timeout(read_timeout_sec) try: diff --git a/netx_api/port_traffic_devices.py b/netx_api/port_traffic_devices.py index 7310f3a..8982b50 100644 --- a/netx_api/port_traffic_devices.py +++ b/netx_api/port_traffic_devices.py @@ -9,6 +9,7 @@ from uuid import uuid4 from fastapi import HTTPException from sqlalchemy.orm import Session +from .cli_creds import require_cli_creds_ready from .cli_resolve import resolve_cli_target from .config import settings from .models import ( @@ -470,6 +471,8 @@ def discover_ports(db: Session, body: DiscoverPortsRequest) -> DiscoverPortsResp except Exception as exc: raise HTTPException(status_code=400, detail=f"resolve_failed: {exc}") from exc + require_cli_creds_ready(creds, interactive=False) + vendor = str(device.get("vendor") or creds.get("vendor") or "") device_type = str(device.get("device_type") or creds.get("device_type") or "") ne_name = str(device.get("name") or creds.get("host") or "") diff --git a/netx_api/port_traffic_runner.py b/netx_api/port_traffic_runner.py index 4e5416a..16e1d06 100644 --- a/netx_api/port_traffic_runner.py +++ b/netx_api/port_traffic_runner.py @@ -10,6 +10,7 @@ from uuid import uuid4 from fastapi import HTTPException +from .cli_creds import cli_creds_skip_reason from .cli_resolve import resolve_cli_target from .cli_timeout import run_cli_with_timeout from .config import settings @@ -228,6 +229,12 @@ def _sample_targets_shared_session(device_id: str, target_ids: list[str]) -> tup _set_target_error(tid, msg) return len(target_ids), msg + skip = cli_creds_skip_reason(creds, interactive=False) + if skip: + for tid in target_ids: + _set_target_error(tid, skip) + return len(target_ids), skip + vendor = str(info.get("vendor") or vendor_hint or "") device_type = str(info.get("device_type") or "") cmds = commands_for_vendor(vendor, device_type) diff --git a/netx_api/topology_discover_scan.py b/netx_api/topology_discover_scan.py index db93d71..91926ab 100644 --- a/netx_api/topology_discover_scan.py +++ b/netx_api/topology_discover_scan.py @@ -8,6 +8,8 @@ from sqlalchemy.orm import Session from .db import SessionLocal from .models import ManagedNE, TopoFabricNode, UmeInventoryNE +from .cli_creds import cli_creds_skip_reason +from .cli_resolve import resolve_cli_target from .ne_exec import execute_managed_ne_commands from .topology_common import ( _DISCOVER_DEADLOCK_RETRIES, @@ -96,6 +98,22 @@ def _discover_one_target( exec_kwargs["ume_ne_id"] = target["ume_ne_id"] else: exec_kwargs["ne_id"] = target["ne_id"] + try: + creds, _device = resolve_cli_target( + db, + managed_ne_id=exec_kwargs.get("ne_id"), + ume_ne_id=exec_kwargs.get("ume_ne_id"), + ) + except HTTPException as exc: + return { + **base, + "ok": False, + "command": cmd, + "error": str(exc.detail or "resolve_failed")[:500], + } + skip = cli_creds_skip_reason(creds, interactive=False) + if skip: + return {**base, "ok": False, "command": cmd, "error": skip} try: from .cli_budget import acquire_cli_slot @@ -104,18 +122,20 @@ def _discover_one_target( return {**base, "ok": False, "command": cmd, "error": "cli_budget_unavailable"} exec_out = execute_managed_ne_commands(db, [cmd], **exec_kwargs) except HTTPException as exc: + detail = str(exc.detail or "exec_failed")[:500] return { **base, "ok": False, "command": cmd, - "error": str(exc.detail or "exec_failed")[:500], + "error": detail, } if not exec_out.get("ok"): + err = str(exec_out.get("error") or exec_out.get("detail") or "exec_failed")[:500] return { **base, "ok": False, "command": cmd, - "error": str(exec_out.get("detail") or exec_out.get("error") or "exec_failed")[:500], + "error": err, } raw = str(exec_out.get("output") or "") diff --git a/netx_api/webcrt_session_registry.py b/netx_api/webcrt_session_registry.py index dfc7c10..3a1773d 100644 --- a/netx_api/webcrt_session_registry.py +++ b/netx_api/webcrt_session_registry.py @@ -205,31 +205,13 @@ def wait_session_ready(session_id: str, *, timeout: float = 120.0) -> WebcrtSess raise HTTPException(status_code=504, detail="connect_timeout") +from .cli_creds import cli_creds_ready as _cli_creds_ready + + def _webcrt_creds_ready(creds: dict[str, Any]) -> bool: - """True when WebCRT can open a session with the resolved credentials. - - Bastion-managed hops store the target password on the bastion side, so an empty - NE password is valid (same as connectivity test). Direct / manual / Linux hops - still require a target password for SSH. - - Telnet (no hop) allows empty username/password so the user can authenticate - interactively in the terminal (SecureCRT-style). - """ - hop_enabled = bool(creds.get("hop_enabled")) - hop_vendor = str(creds.get("hop_vendor") or "").strip().lower() - auth_mode = str(creds.get("hop_target_auth_mode") or "bastion_managed").strip().lower() - protocol = str(creds.get("protocol") or "ssh").strip().lower() - if hop_enabled and hop_vendor == "bastion" and auth_mode == "bastion_managed": - return bool( - str(creds.get("hop_host") or "").strip() - and str(creds.get("hop_username") or "").strip() - and str(creds.get("hop_password") or "") - ) - if protocol == "telnet" and not hop_enabled: - return True - if not str(creds.get("username") or "").strip(): - return False - return bool(str(creds.get("password") or "")) + """True when WebCRT can open a session with the resolved credentials.""" + ready, _ = _cli_creds_ready(creds, interactive=True) + return ready def _finish_connect( diff --git a/tests/test_cli_creds.py b/tests/test_cli_creds.py new file mode 100644 index 0000000..b8fa91d --- /dev/null +++ b/tests/test_cli_creds.py @@ -0,0 +1,119 @@ +from __future__ import annotations + +import unittest + +from netx_api.cli_creds import ( + REASON_HOP_INCOMPLETE, + REASON_NO_PASSWORD, + REASON_USERNAME_REQUIRED, + cli_creds_ready, + cli_creds_skip_reason, +) + + +class CliCredsTests(unittest.TestCase): + def test_direct_ssh_requires_password(self) -> None: + creds = { + "ip_address": "10.0.0.1", + "protocol": "ssh", + "username": "admin", + "password": "", + "hop_enabled": False, + } + ready, reason = cli_creds_ready(creds, interactive=False) + self.assertFalse(ready) + self.assertEqual(reason, REASON_NO_PASSWORD) + self.assertEqual(cli_creds_skip_reason(creds), REASON_NO_PASSWORD) + + def test_direct_ssh_ready(self) -> None: + creds = { + "ip_address": "10.0.0.1", + "protocol": "ssh", + "username": "admin", + "password": "secret", + "hop_enabled": False, + } + ready, reason = cli_creds_ready(creds, interactive=False) + self.assertTrue(ready) + self.assertEqual(reason, "") + self.assertIsNone(cli_creds_skip_reason(creds)) + + def test_bastion_managed_allows_empty_target_password(self) -> None: + creds = { + "ip_address": "10.0.0.2", + "protocol": "ssh", + "username": "ca-oper", + "password": "", + "hop_enabled": True, + "hop_vendor": "bastion", + "hop_target_auth_mode": "bastion_managed", + "hop_host": "10.34.145.27", + "hop_username": "jump", + "hop_password": "hop-secret", + } + ready, reason = cli_creds_ready(creds, interactive=False) + self.assertTrue(ready) + self.assertEqual(reason, "") + + def test_cli_hop_requires_target_password(self) -> None: + creds = { + "ip_address": "10.0.0.3", + "protocol": "ssh", + "username": "admin", + "password": "", + "hop_enabled": True, + "hop_vendor": "zte", + "hop_target_auth_mode": "manual", + "hop_host": "10.1.1.1", + "hop_username": "hop", + "hop_password": "hop-secret", + } + ready, reason = cli_creds_ready(creds, interactive=False) + self.assertFalse(ready) + self.assertEqual(reason, REASON_NO_PASSWORD) + + def test_hop_incomplete(self) -> None: + creds = { + "ip_address": "10.0.0.4", + "protocol": "ssh", + "username": "admin", + "password": "secret", + "hop_enabled": True, + "hop_vendor": "zte", + "hop_host": "", + "hop_username": "hop", + "hop_password": "", + } + ready, reason = cli_creds_ready(creds, interactive=False) + self.assertFalse(ready) + self.assertEqual(reason, REASON_HOP_INCOMPLETE) + + def test_interactive_telnet_allows_empty_password(self) -> None: + creds = { + "ip_address": "10.0.0.5", + "protocol": "telnet", + "username": "", + "password": "", + "hop_enabled": False, + } + ready, reason = cli_creds_ready(creds, interactive=True) + self.assertTrue(ready) + ready_exec, reason_exec = cli_creds_ready(creds, interactive=False) + self.assertFalse(ready_exec) + self.assertEqual(reason_exec, REASON_USERNAME_REQUIRED) + + def test_non_interactive_telnet_requires_password(self) -> None: + creds = { + "ip_address": "10.0.0.6", + "protocol": "telnet", + "username": "admin", + "password": "", + "hop_enabled": False, + } + ready, reason = cli_creds_ready(creds, interactive=False) + self.assertFalse(ready) + self.assertEqual(reason, REASON_NO_PASSWORD) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_config_sync.py b/tests/test_config_sync.py index f92c7ee..793d0e5 100644 --- a/tests/test_config_sync.py +++ b/tests/test_config_sync.py @@ -124,7 +124,13 @@ class ConfigSyncSnapshotOverwriteTests(unittest.TestCase): task.ne_ip = "1.1.1.1" db.get.return_value = task resolve.return_value = ( - {"host": "1.1.1.1"}, + { + "ip_address": "1.1.1.1", + "protocol": "ssh", + "username": "admin", + "password": "secret", + "hop_enabled": False, + }, {"vendor": "Cisco", "device_type": "ios", "name": "r1", "ip_address": "1.1.1.1"}, ) @@ -158,7 +164,13 @@ class ConfigSyncSnapshotOverwriteTests(unittest.TestCase): task.ne_ip = "1.1.1.1" db.get.return_value = task resolve.return_value = ( - {"host": "1.1.1.1"}, + { + "ip_address": "1.1.1.1", + "protocol": "ssh", + "username": "admin", + "password": "secret", + "hop_enabled": False, + }, {"vendor": "Juniper", "device_type": "junos", "name": "r1", "ip_address": "1.1.1.1"}, ) diff --git a/tests/test_ne_exec.py b/tests/test_ne_exec.py index ebcef1c..ae3164d 100644 --- a/tests/test_ne_exec.py +++ b/tests/test_ne_exec.py @@ -8,6 +8,18 @@ from fastapi import HTTPException from netx_api.ne_exec import _validate_command, execute_managed_ne_commands +def _ready_creds(**overrides) -> dict: + base = { + "ip_address": "1.1.1.1", + "protocol": "ssh", + "username": "admin", + "password": "secret", + "hop_enabled": False, + } + base.update(overrides) + return base + + class NeExecValidationTests(unittest.TestCase): def test_allows_show(self) -> None: _validate_command("show ip interface brief") @@ -130,7 +142,7 @@ class NeExecRunTests(unittest.TestCase): @patch("netx_api.ne_exec.resolve_cli_target") def test_execute_success(self, resolve, _collect, _configured) -> None: resolve.return_value = ( - {"ip_address": "1.1.1.1"}, + _ready_creds(), { "source": "managed", "id": "ne-1", @@ -152,6 +164,20 @@ class NeExecRunTests(unittest.TestCase): self.assertEqual(out["output"], "ok-output") self.assertEqual(out["commands"], ["show version"]) + @patch("netx_api.ne_exec.credentials_configured", return_value=True) + @patch("netx_api.ne_exec._collect_on_device", return_value="ok-output") + @patch("netx_api.ne_exec.resolve_cli_target") + def test_execute_skips_when_no_password(self, resolve, collect, _configured) -> None: + resolve.return_value = ( + _ready_creds(password=""), + {"source": "managed", "id": "ne-1", "name": "R2", "ip_address": "192.168.0.128"}, + ) + db = MagicMock() + out = execute_managed_ne_commands(db, ["show version"], ne_id="ne-1") + self.assertFalse(out["ok"]) + self.assertEqual(out["error"], "no_password") + collect.assert_not_called() + @patch("netx_api.ne_exec.credentials_configured", return_value=True) @patch("netx_api.ne_exec._collect_on_device", return_value="ok-output") @patch("netx_api.ne_exec.resolve_cli_target") @@ -181,7 +207,7 @@ class NeExecRunTests(unittest.TestCase): collect.assert_not_called() resolve.return_value = ( - {"ip_address": "1.1.1.1"}, + _ready_creds(), { "source": "managed", "id": "ne-1", @@ -221,7 +247,7 @@ class NeExecBatchTests(unittest.TestCase): session_local.return_value = MagicMock() resolve.return_value = ( - {"ip_address": "1.1.1.1"}, + _ready_creds(), { "source": "managed", "id": "ne-1",