From d3fd8406378806cdaa4365c371c8641b4c2f7077 Mon Sep 17 00:00:00 2001 From: oliver Date: Thu, 30 Jul 2026 00:08:22 +0800 Subject: [PATCH] fix(webcrt): allow bastion-managed sessions without NE password. Align session credential checks with connectivity test so target password can live only on the bastion. Co-authored-by: Cursor --- netx_api/webcrt_service.py | 23 ++++++++++- tests/test_webcrt.py | 80 ++++++++++++++++++++++++++++++++++++++ 2 files changed, 102 insertions(+), 1 deletion(-) diff --git a/netx_api/webcrt_service.py b/netx_api/webcrt_service.py index b226416..8b498f4 100644 --- a/netx_api/webcrt_service.py +++ b/netx_api/webcrt_service.py @@ -460,6 +460,27 @@ def get_session(session_id: str) -> WebcrtSession | None: return sess +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. + """ + if not str(creds.get("username") or "").strip(): + return False + 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() + 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 "") + ) + return bool(str(creds.get("password") or "")) + + def create_session( db: Session, *, @@ -487,7 +508,7 @@ def create_session( except Exception as exc: raise HTTPException(status_code=400, detail=f"credential_error:{exc}") from exc - if not str(creds.get("username") or "").strip() or not str(creds.get("password") or ""): + if not _webcrt_creds_ready(creds): raise HTTPException(status_code=400, detail="credentials_incomplete") session_id = str(uuid.uuid4()) diff --git a/tests/test_webcrt.py b/tests/test_webcrt.py index 6dde5f7..1cf151e 100644 --- a/tests/test_webcrt.py +++ b/tests/test_webcrt.py @@ -136,6 +136,9 @@ class WebcrtServiceTests(unittest.TestCase): "hop_enabled": True, "hop_vendor": "bastion", "hop_host": "jump.example", + "hop_username": "jumpuser", + "hop_password": "jumppass", + "hop_target_auth_mode": "bastion_managed", "ip_address": "10.0.0.2", "protocol": "ssh", }, @@ -181,6 +184,83 @@ class WebcrtServiceTests(unittest.TestCase): self.assertEqual(fake.written[len(before) :], ["\n"]) svc.close_session(out["session_id"], reason="test") + @patch.object(svc, "_audit") + @patch.object(svc, "open_netmiko_connection") + @patch("netx_api.cli_resolve.resolve_cli_target") + def test_create_session_bastion_managed_without_target_password( + self, + mock_resolve: MagicMock, + mock_open: MagicMock, + _mock_audit: MagicMock, + ) -> None: + mock_resolve.return_value = ( + { + "username": "ca-oper", + "password": "", + "hop_enabled": True, + "hop_vendor": "bastion", + "hop_host": "10.34.145.27", + "hop_username": "ZTE-TSM", + "hop_password": "bastion-secret", + "hop_target_auth_mode": "bastion_managed", + "hop_command_template": "ssh {target_ip}", + "ip_address": "114.0.44.90", + "protocol": "ssh", + "device_type": "zte_zxros", + }, + { + "id": "ne-bastion", + "name": "KND-PUN-EN1-Z20HS", + "ip_address": "114.0.44.90", + "protocol": "ssh", + "source": "managed", + }, + ) + mock_open.return_value = _FakeConn() + out = svc.create_session(MagicMock(), ne_id="ne-bastion", cols=80, rows=24, client="test") + mock_open.assert_called_once() + self.assertEqual(out["ne_id"], "ne-bastion") + svc.close_session(out["session_id"], reason="test") + + def test_webcrt_creds_ready_bastion_managed(self) -> None: + self.assertTrue( + svc._webcrt_creds_ready( + { + "username": "ca-oper", + "password": "", + "hop_enabled": True, + "hop_vendor": "bastion", + "hop_host": "10.34.145.27", + "hop_username": "ZTE-TSM", + "hop_password": "x", + "hop_target_auth_mode": "bastion_managed", + } + ) + ) + self.assertFalse( + svc._webcrt_creds_ready( + { + "username": "ca-oper", + "password": "", + "hop_enabled": True, + "hop_vendor": "bastion", + "hop_host": "10.34.145.27", + "hop_username": "ZTE-TSM", + "hop_password": "", + "hop_target_auth_mode": "bastion_managed", + } + ) + ) + self.assertFalse( + svc._webcrt_creds_ready( + { + "username": "u", + "password": "", + "hop_enabled": False, + } + ) + ) + @patch.object(svc, "_audit") def test_attach_gen_exclusive_stdout_and_stale_detach(self, _mock_audit: MagicMock) -> None: conn = _FakeConn()