From b4514c13aa54bd521664a45a302e184178b7faf0 Mon Sep 17 00:00:00 2001 From: oliver Date: Tue, 11 Aug 2026 17:32:03 +0800 Subject: [PATCH] Support bastion hop hosts as FQDN and pasted OpenSSH destinations. Use placeholder examples (example.com / RFC5737) in docs and tests. EOF Co-authored-by: Cursor --- netx_api/cli_service.py | 29 ++++-- netx_api/ne_hop_templates.py | 109 ++++++++++++++++++++-- netx_api/ne_service.py | 2 + netx_api/ne_service_common.py | 38 ++++++-- netx_api/ne_service_crud.py | 10 ++- netx_api/ne_session_connect.py | 12 ++- netx_api/ne_session_factory.py | 6 ++ tests/test_bastion_hop.py | 124 ++++++++++++++++++++++++-- web/src/components/HopProxyFields.tsx | 20 ++++- web/src/i18n/en.ts | 11 ++- web/src/i18n/zh.ts | 7 +- web/src/utils/hopProxy.ts | 82 +++++++++++++++++ 12 files changed, 414 insertions(+), 36 deletions(-) diff --git a/netx_api/cli_service.py b/netx_api/cli_service.py index 704e95c..ad42887 100644 --- a/netx_api/cli_service.py +++ b/netx_api/cli_service.py @@ -22,6 +22,7 @@ from .ne_service import ( _normalize_hop_target_auth_mode, _normalize_hop_vendor, _normalize_protocol, + _normalize_saved_hop_endpoint, _require_crypto, ) @@ -33,8 +34,12 @@ def _now() -> datetime: def _validate_profile_hop(body: CliConnectProfileCreate | CliConnectProfileUpdate, *, hop_enabled: bool) -> None: if not hop_enabled: return - host = str(getattr(body, "hop_host", None) or "").strip() - user = str(getattr(body, "hop_username", None) or "").strip() + vendor = _normalize_hop_vendor(getattr(body, "hop_vendor", None) or "zte") + host, user = _normalize_saved_hop_endpoint( + hop_vendor=vendor, + hop_host=str(getattr(body, "hop_host", None) or ""), + hop_username=str(getattr(body, "hop_username", None) or ""), + ) if not host: raise HTTPException(status_code=400, detail="hop_host_required") if not user: @@ -103,6 +108,12 @@ def create_cli_profile(db: Session, body: CliConnectProfileCreate) -> CliConnect _validate_profile_hop(body, hop_enabled=bool(body.hop_enabled)) if not str(body.username or "").strip(): raise HTTPException(status_code=400, detail="username_required") + hop_vendor = _normalize_hop_vendor(body.hop_vendor) + hop_host, hop_username = _normalize_saved_hop_endpoint( + hop_vendor=hop_vendor, + hop_host=str(body.hop_host or ""), + hop_username=str(body.hop_username or ""), + ) row = CliConnectProfile( name=str(body.name or "").strip() or "default", username=str(body.username).strip(), @@ -113,11 +124,11 @@ def create_cli_profile(db: Session, body: CliConnectProfileCreate) -> CliConnect vendor_default=str(body.vendor_default), ne_type_rules=str(body.ne_type_rules or ""), hop_enabled=bool(body.hop_enabled), - hop_vendor=_normalize_hop_vendor(body.hop_vendor), - hop_host=str(body.hop_host or "").strip(), + hop_vendor=hop_vendor, + hop_host=hop_host, hop_port=int(body.hop_port or 22), hop_protocol=_normalize_protocol(body.hop_protocol), - hop_username=str(body.hop_username or "").strip(), + hop_username=hop_username, hop_password_enc=encrypt_secret(body.hop_password) if body.hop_enabled and body.hop_password else "", hop_command_template=str(body.hop_command_template or "").strip(), hop_vrf=str(body.hop_vrf or "").strip(), @@ -182,6 +193,14 @@ def update_cli_profile(db: Session, profile_id: str, body: CliConnectProfileUpda if "hop_password" in data and data["hop_password"]: _require_crypto() row.hop_password_enc = encrypt_secret(str(data["hop_password"])) + if "hop_host" in data or "hop_username" in data or "hop_vendor" in data: + hop_host, hop_username = _normalize_saved_hop_endpoint( + hop_vendor=str(row.hop_vendor or ""), + hop_host=str(row.hop_host or ""), + hop_username=str(row.hop_username or ""), + ) + row.hop_host = hop_host + row.hop_username = hop_username if body.is_default is True: db.query(CliConnectProfile).filter(CliConnectProfile.id != row.id).update({CliConnectProfile.is_default: False}) row.is_default = True diff --git a/netx_api/ne_hop_templates.py b/netx_api/ne_hop_templates.py index e4f0399..4fdfa53 100644 --- a/netx_api/ne_hop_templates.py +++ b/netx_api/ne_hop_templates.py @@ -1,6 +1,7 @@ """Vendor hop / bastion username templates and rendering.""" from __future__ import annotations +import re from typing import Any _HOP_PLACEHOLDERS = ("target_ip", "target_port", "target_user", "target_password", "vrf", "hop_user", "hop_host") @@ -8,6 +9,96 @@ _HOP_PLACEHOLDERS = ("target_ip", "target_port", "target_user", "target_password # ZTE CLI jump: ssh/telnet [vrf ] — target user/password via secondary auth. _LEGACY_HOP_TEMPLATES = frozenset({"ssh {target_user}@{target_ip}", "ssh {target_ip}", "telnet {target_ip}"}) +_SSH_PREFIX_RE = re.compile(r"^(?:ssh(?:\s+-p\s+\d+)?\s+)", re.IGNORECASE) + + +def normalize_hop_host(value: str) -> str: + """Normalize bastion/jump host: IP or FQDN (strip ssh://, port suffix, trailing /).""" + host = str(value or "").strip() + if not host: + return "" + host = _SSH_PREFIX_RE.sub("", host).strip() + if "://" in host: + # ssh://user@host:port/ → keep right-hand host-ish fragment for further parse + host = host.split("://", 1)[1] + host = host.strip().rstrip("/") + # Bracketed IPv6: [2001:db8::1]:22 + if host.startswith("[") and "]" in host: + inside, _, rest = host[1:].partition("]") + if rest in ("",) or rest.startswith(":"): + return inside.strip() + return host + # host:port (not IPv6) — keep host only when port is numeric + if host.count(":") == 1: + left, right = host.rsplit(":", 1) + if right.isdigit() and left and "@" not in left: + return left.strip() + return host + + +def parse_bastion_ssh_destination(value: str) -> dict[str, str]: + """Parse OpenSSH-style bastion destination (IP or domain bastion host). + + Examples:: + + ssh-bastion.example.com + 192.0.2.10 + bastion-user@target-user@198.51.100.20@ssh-bastion.example.com + ssh bastion-user@target-user@198.51.100.20@ssh-bastion.example.com + + OpenSSH splits ``user@host`` at the **last** ``@``, so the bastion address + (IP or FQDN) is the final segment; preceding segments form the SSH username. + """ + raw = str(value or "").strip() + raw = _SSH_PREFIX_RE.sub("", raw).strip().strip('"').strip("'") + if not raw: + return {"hop_host": "", "hop_username": "", "target_user": "", "target_ip": "", "ssh_username": ""} + + if "@" not in raw: + host = normalize_hop_host(raw) + return { + "hop_host": host, + "hop_username": "", + "target_user": "", + "target_ip": "", + "ssh_username": "", + } + + user_part, host_part = raw.rsplit("@", 1) + hop_host = normalize_hop_host(host_part) + ssh_username = str(user_part or "").strip() + parts = [p for p in ssh_username.split("@") if p != ""] + hop_username = parts[0] if parts else "" + target_user = parts[1] if len(parts) >= 2 else "" + target_ip = parts[2] if len(parts) >= 3 else "" + return { + "hop_host": hop_host, + "hop_username": hop_username, + "target_user": target_user, + "target_ip": target_ip, + "ssh_username": ssh_username, + } + + +def expand_bastion_hop_fields( + *, + hop_host: str, + hop_username: str = "", +) -> tuple[str, str]: + """If hop_host is a pasted ``user@…@bastion`` string, split into (host, username). + + Hostname-only / IP values are returned unchanged. Existing hop_username wins + unless the paste clearly includes a composite username. + """ + raw_host = str(hop_host or "").strip() + cur_user = str(hop_username or "").strip() + if "@" not in raw_host: + return normalize_hop_host(raw_host), cur_user + parsed = parse_bastion_ssh_destination(raw_host) + host = str(parsed.get("hop_host") or "") + pasted_user = str(parsed.get("hop_username") or "") + return host, (pasted_user or cur_user) + def default_zte_hop_template(protocol: str, vrf: str = "") -> str: cmd = "telnet" if str(protocol or "ssh").strip().lower() == "telnet" else "ssh" v = str(vrf or "").strip() @@ -52,22 +143,27 @@ def resolve_bastion_ssh_username(rendered: str, hop_host: str) -> str: """Map template output to the SSH username Paramiko must send. CLI ``ssh hop@target@ip@bastion`` is parsed by OpenSSH as user ``hop@target@ip`` - and host ``bastion``. Legacy templates that included ``{hop_host}`` duplicated the - bastion address inside the username and break authentication. + and host ``bastion`` (IP or FQDN). Legacy templates that included ``{hop_host}`` + duplicated the bastion address inside the username and break authentication. """ user = str(rendered or "").strip() - host = str(hop_host or "").strip() + host = normalize_hop_host(hop_host) if not user or not host: return user + # Prefer exact suffix strip; also accept case-insensitive FQDN match. suffix = f"@{host}" if user.endswith(suffix): - return user[:-len(suffix)] + return user[: -len(suffix)] + lower_user = user.lower() + lower_suffix = suffix.lower() + if lower_user.endswith(lower_suffix): + return user[: -len(suffix)] return user def bastion_ssh_cli(username: str, hop_host: str, hop_port: int = 22) -> str: """Human-readable ssh command equivalent (for logs/UI).""" - host = str(hop_host or "").strip() + host = normalize_hop_host(hop_host) user = str(username or "").strip() target = f"{user}@{host}" if user else host port = int(hop_port or 22) @@ -107,7 +203,7 @@ def render_hop_command(template: str, creds: dict[str, Any]) -> str: "target_password": str(creds.get("password") or ""), "vrf": str(creds.get("hop_vrf") or "").strip(), "hop_user": str(creds.get("hop_username") or ""), - "hop_host": str(creds.get("hop_host") or "").strip(), + "hop_host": normalize_hop_host(str(creds.get("hop_host") or "")), } out = tpl for key in _HOP_PLACEHOLDERS: @@ -116,4 +212,3 @@ def render_hop_command(template: str, creds: dict[str, Any]) -> str: raise ValueError("hop_command_template_invalid_placeholder") return out - diff --git a/netx_api/ne_service.py b/netx_api/ne_service.py index 3c0ccbf..a3950f4 100644 --- a/netx_api/ne_service.py +++ b/netx_api/ne_service.py @@ -9,6 +9,7 @@ from .ne_service_common import ( _normalize_hop_target_auth_mode, _normalize_hop_vendor, _normalize_protocol, + _normalize_saved_hop_endpoint, _require_crypto, get_device_credentials, row_to_out, @@ -44,6 +45,7 @@ __all__ = [ "_normalize_hop_target_auth_mode", "_normalize_hop_vendor", "_normalize_protocol", + "_normalize_saved_hop_endpoint", "_require_crypto", "batch_apply_account", "batch_apply_hop_proxy", diff --git a/netx_api/ne_service_common.py b/netx_api/ne_service_common.py index 5b94704..baa4c83 100644 --- a/netx_api/ne_service_common.py +++ b/netx_api/ne_service_common.py @@ -18,6 +18,7 @@ from .device_types import ( from .models import ManagedNE from .ne_crypto import CredentialCryptoError, credentials_configured, decrypt_secret, encrypt_secret from .ne_schemas import ManagedNeCreate, ManagedNeOut, ManagedNeUpdate +from .ne_hop_templates import expand_bastion_hop_fields, normalize_hop_host from .ne_session_factory import default_bastion_username_template, default_hop_command_template from .timeutil import utcnow_naive @@ -118,13 +119,18 @@ def _infer_managed_ne_type_vendor(ne_type: str, vendor: str) -> tuple[str, str]: def _validate_hop_on_create(body: ManagedNeCreate) -> None: if not body.hop_enabled: return - if not str(body.hop_host or "").strip(): + hop_vendor = _normalize_hop_vendor(body.hop_vendor) + hop_host, hop_username = _normalize_saved_hop_endpoint( + hop_vendor=hop_vendor, + hop_host=str(body.hop_host or ""), + hop_username=str(body.hop_username or ""), + ) + if not hop_host: raise HTTPException(status_code=400, detail="hop_host_required") - if not str(body.hop_username or "").strip(): + if not hop_username: raise HTTPException(status_code=400, detail="hop_username_required") if not str(body.hop_password or "").strip(): raise HTTPException(status_code=400, detail="hop_password_required") - hop_vendor = _normalize_hop_vendor(body.hop_vendor) if hop_vendor == "bastion" and _normalize_hop_target_auth_mode(body.hop_target_auth_mode) == "manual": if not str(body.password or "").strip(): raise HTTPException(status_code=400, detail="password_required") @@ -141,13 +147,27 @@ def _import_cell_str(value: Any) -> str: return "" if text.lower() == "nan" else text +def _normalize_saved_hop_endpoint(*, hop_vendor: str, hop_host: str, hop_username: str) -> tuple[str, str]: + """Accept IP or FQDN; split pasted OpenSSH ``user@target@ip@bastion`` into fields.""" + host = str(hop_host or "").strip() + user = str(hop_username or "").strip() + if str(hop_vendor or "").strip().lower() == "bastion": + return expand_bastion_hop_fields(hop_host=host, hop_username=user) + return normalize_hop_host(host), user + + def _apply_hop_create(row: ManagedNE, body: ManagedNeCreate) -> None: row.hop_enabled = bool(body.hop_enabled) row.hop_vendor = _normalize_hop_vendor(body.hop_vendor) - row.hop_host = str(body.hop_host or "").strip() + hop_host, hop_username = _normalize_saved_hop_endpoint( + hop_vendor=row.hop_vendor, + hop_host=str(body.hop_host or ""), + hop_username=str(body.hop_username or ""), + ) + row.hop_host = hop_host row.hop_port = int(body.hop_port or 22) row.hop_protocol = _normalize_protocol(body.hop_protocol) - row.hop_username = str(body.hop_username or "").strip() + row.hop_username = hop_username row.hop_password_enc = encrypt_secret(body.hop_password) if body.hop_enabled else "" row.hop_command_template = str(body.hop_command_template or "").strip() row.hop_vrf = str(body.hop_vrf or "").strip() @@ -176,6 +196,14 @@ def _apply_hop_update(row: ManagedNE, data: dict[str, Any]) -> None: row.hop_vrf = str(data["hop_vrf"]).strip() if "hop_target_auth_mode" in data and data["hop_target_auth_mode"] is not None: row.hop_target_auth_mode = _normalize_hop_target_auth_mode(data["hop_target_auth_mode"]) + if "hop_host" in data or "hop_username" in data or "hop_vendor" in data: + hop_host, hop_username = _normalize_saved_hop_endpoint( + hop_vendor=str(row.hop_vendor or ""), + hop_host=str(row.hop_host or ""), + hop_username=str(row.hop_username or ""), + ) + row.hop_host = hop_host + row.hop_username = hop_username if row.hop_enabled: if not str(row.hop_host or "").strip(): raise HTTPException(status_code=400, detail="hop_host_required") diff --git a/netx_api/ne_service_crud.py b/netx_api/ne_service_crud.py index 44ec6d6..e84d5f4 100644 --- a/netx_api/ne_service_crud.py +++ b/netx_api/ne_service_crud.py @@ -26,6 +26,7 @@ from .ne_service_common import ( _normalize_hop_vendor, _normalize_ip, _normalize_protocol, + _normalize_saved_hop_endpoint, _normalize_vendor, _now, _require_crypto, @@ -192,8 +193,12 @@ def update_managed_ne(db: Session, ne_id: str, body: ManagedNeUpdate) -> Managed def batch_apply_hop_proxy(db: Session, ids: list[str], hop: HopProxyConfig) -> dict[str, Any]: """Apply the same jump-host (proxy) settings to multiple managed NEs.""" - hop_host = str(hop.hop_host or "").strip() - hop_user = str(hop.hop_username or "").strip() + hop_vendor = _normalize_hop_vendor(hop.hop_vendor) + hop_host, hop_user = _normalize_saved_hop_endpoint( + hop_vendor=hop_vendor, + hop_host=str(hop.hop_host or ""), + hop_username=str(hop.hop_username or ""), + ) hop_pass = str(hop.hop_password or "").strip() if hop_pass: _require_crypto() @@ -205,7 +210,6 @@ def batch_apply_hop_proxy(db: Session, ids: list[str], hop: HopProxyConfig) -> d if not hop_pass and hop_auth_mode != "bastion_managed": raise HTTPException(status_code=400, detail="hop_password_required") - hop_vendor = _normalize_hop_vendor(hop.hop_vendor) template = str(hop.hop_command_template or "").strip() if hop_vendor == "bastion" and not template: template = default_bastion_username_template() diff --git a/netx_api/ne_session_connect.py b/netx_api/ne_session_connect.py index fac2c7e..eebda88 100644 --- a/netx_api/ne_session_connect.py +++ b/netx_api/ne_session_connect.py @@ -16,6 +16,8 @@ from .ne_cli_hop import ( ) from .ne_hop_templates import ( _hop_vendor, + expand_bastion_hop_fields, + normalize_hop_host, render_hop_command, resolve_bastion_ssh_username, ) @@ -497,11 +499,15 @@ def _connect_via_bastion( keepalive: int | None = None, ) -> ConnectHandler: """SSH to bastion with composite username; bastion proxies to target (protocol proxy).""" - hop_host = str(creds.get("hop_host") or "").strip() - hop_user = str(creds.get("hop_username") or "").strip() + hop_host, hop_user = expand_bastion_hop_fields( + hop_host=str(creds.get("hop_host") or ""), + hop_username=str(creds.get("hop_username") or ""), + ) hop_pass = str(creds.get("hop_password") or "") if not hop_host or not hop_user or not hop_pass: raise ValueError("hop_credentials_incomplete") + # Keep render/logs aligned when hop_host was a pasted user@…@fqdn string. + creds = {**creds, "hop_host": hop_host, "hop_username": hop_user} composite_rendered = render_hop_command(str(creds.get("hop_command_template") or ""), creds) ssh_username = resolve_bastion_ssh_username(composite_rendered, hop_host) @@ -563,7 +569,7 @@ def _connect_via_linux_hop( keepalive: int | None = None, ) -> ConnectHandler: """SSH to Linux bastion, then direct-tcpip tunnel to target (classic ProxyJump-style).""" - hop_host = str(creds.get("hop_host") or "").strip() + hop_host = normalize_hop_host(str(creds.get("hop_host") or "")) 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: diff --git a/netx_api/ne_session_factory.py b/netx_api/ne_session_factory.py index 6daf365..d8caefc 100644 --- a/netx_api/ne_session_factory.py +++ b/netx_api/ne_session_factory.py @@ -17,6 +17,9 @@ from .ne_hop_templates import ( default_hop_command_template, default_huawei_hop_template, default_zte_hop_template, + expand_bastion_hop_fields, + normalize_hop_host, + parse_bastion_ssh_destination, render_hop_command, resolve_bastion_ssh_username, ) @@ -60,9 +63,12 @@ __all__ = [ "default_hop_command_template", "default_huawei_hop_template", "default_zte_hop_template", + "expand_bastion_hop_fields", "extract_cli_prompt_marker", "get_cli_hop_guard", + "normalize_hop_host", "open_netmiko_connection", + "parse_bastion_ssh_destination", "render_hop_command", "resolve_bastion_ssh_username", "should_close_cli_hop_session", diff --git a/tests/test_bastion_hop.py b/tests/test_bastion_hop.py index 845fbd0..7e76c14 100644 --- a/tests/test_bastion_hop.py +++ b/tests/test_bastion_hop.py @@ -48,20 +48,70 @@ class BastionTemplateTests(unittest.TestCase): def test_resolve_strips_legacy_hop_host_suffix(self) -> None: self.assertEqual( - resolve_bastion_ssh_username("ZTE-FIVIE@ca-admin@114.1.198.1@10.34.145.25", "10.34.145.25"), - "ZTE-FIVIE@ca-admin@114.1.198.1", + resolve_bastion_ssh_username("bastion-user@target-user@198.51.100.10@192.0.2.10", "192.0.2.10"), + "bastion-user@target-user@198.51.100.10", + ) + + def test_resolve_strips_domain_bastion_suffix(self) -> None: + self.assertEqual( + resolve_bastion_ssh_username( + "bastion-user@target-user@198.51.100.20@ssh-bastion.example.com", + "ssh-bastion.example.com", + ), + "bastion-user@target-user@198.51.100.20", ) def test_resolve_keeps_username_without_hop_host_suffix(self) -> None: self.assertEqual( - resolve_bastion_ssh_username("ZTE-FIVIE@ca-admin@114.1.198.1", "10.34.145.25"), - "ZTE-FIVIE@ca-admin@114.1.198.1", + resolve_bastion_ssh_username("bastion-user@target-user@198.51.100.10", "192.0.2.10"), + "bastion-user@target-user@198.51.100.10", ) def test_bastion_ssh_cli(self) -> None: self.assertEqual( - bastion_ssh_cli("ZTE-FIVIE@ca-admin@114.1.198.1", "10.34.145.25"), - "ssh ZTE-FIVIE@ca-admin@114.1.198.1@10.34.145.25", + bastion_ssh_cli("bastion-user@target-user@198.51.100.10", "192.0.2.10"), + "ssh bastion-user@target-user@198.51.100.10@192.0.2.10", + ) + + def test_bastion_ssh_cli_domain(self) -> None: + self.assertEqual( + bastion_ssh_cli("bastion-user@target-user@198.51.100.20", "ssh-bastion.example.com"), + "ssh bastion-user@target-user@198.51.100.20@ssh-bastion.example.com", + ) + + def test_parse_domain_bastion_destination(self) -> None: + from netx_api.ne_hop_templates import expand_bastion_hop_fields, parse_bastion_ssh_destination + + parsed = parse_bastion_ssh_destination( + "ssh bastion-user@target-user@198.51.100.20@ssh-bastion.example.com" + ) + self.assertEqual(parsed["hop_host"], "ssh-bastion.example.com") + self.assertEqual(parsed["hop_username"], "bastion-user") + self.assertEqual(parsed["target_user"], "target-user") + self.assertEqual(parsed["target_ip"], "198.51.100.20") + self.assertEqual(parsed["ssh_username"], "bastion-user@target-user@198.51.100.20") + host, user = expand_bastion_hop_fields( + hop_host="bastion-user@target-user@198.51.100.20@ssh-bastion.example.com", + hop_username="", + ) + self.assertEqual(host, "ssh-bastion.example.com") + self.assertEqual(user, "bastion-user") + + def test_render_bastion_with_domain_hop_host(self) -> None: + creds = { + "hop_vendor": "bastion", + "hop_username": "bastion-user", + "hop_host": "ssh-bastion.example.com", + "username": "target-user", + "ip_address": "198.51.100.20", + "hop_protocol": "ssh", + "hop_vrf": "", + } + out = render_hop_command("", creds) + self.assertEqual(out, "bastion-user@target-user@198.51.100.20") + self.assertEqual( + bastion_ssh_cli(out, creds["hop_host"]), + "ssh bastion-user@target-user@198.51.100.20@ssh-bastion.example.com", ) def test_netmiko_driver_class_resolves_zte(self) -> None: @@ -132,6 +182,68 @@ class BastionConnectImplTests(unittest.TestCase): self.assertEqual(wrap_kwargs["password"], "vault-pass") conn.disconnect.assert_not_called() + @patch("netx_api.ne_session_connect._netmiko_over_ssh_client") + @patch("netx_api.ne_session_connect._bastion_ssh_connect") + def test_bastion_domain_host_connect(self, bastion_ssh, netmiko_wrap) -> None: + from netx_api.ne_session_factory import _connect_via_bastion + + bastion_ssh.return_value = MagicMock() + netmiko_wrap.return_value = MagicMock() + creds = { + "hop_host": "ssh-bastion.example.com", + "hop_username": "bastion-user", + "hop_password": "vault-pass", + "hop_port": 22, + "device_type": "zte_zxros", + "protocol": "ssh", + "username": "target-user", + "ip_address": "198.51.100.20", + "password": "", + "hop_target_auth_mode": "bastion_managed", + "hop_vendor": "bastion", + "hop_protocol": "ssh", + "hop_vrf": "", + } + _connect_via_bastion(creds) + bastion_ssh.assert_called_once_with( + host="ssh-bastion.example.com", + port=22, + username="bastion-user@target-user@198.51.100.20", + password="vault-pass", + timeout=unittest.mock.ANY, + ) + + @patch("netx_api.ne_session_connect._netmiko_over_ssh_client") + @patch("netx_api.ne_session_connect._bastion_ssh_connect") + def test_bastion_pasted_destination_expands_on_connect(self, bastion_ssh, netmiko_wrap) -> None: + from netx_api.ne_session_factory import _connect_via_bastion + + bastion_ssh.return_value = MagicMock() + netmiko_wrap.return_value = MagicMock() + creds = { + "hop_host": "bastion-user@target-user@198.51.100.20@ssh-bastion.example.com", + "hop_username": "", + "hop_password": "vault-pass", + "hop_port": 22, + "device_type": "zte_zxros", + "protocol": "ssh", + "username": "target-user", + "ip_address": "198.51.100.20", + "password": "", + "hop_target_auth_mode": "bastion_managed", + "hop_vendor": "bastion", + "hop_protocol": "ssh", + "hop_vrf": "", + } + _connect_via_bastion(creds) + bastion_ssh.assert_called_once_with( + host="ssh-bastion.example.com", + port=22, + username="bastion-user@target-user@198.51.100.20", + password="vault-pass", + timeout=unittest.mock.ANY, + ) + @patch("netx_api.ne_session_connect._interactive_target_auth") @patch("netx_api.ne_session_connect._read_channel") @patch("netx_api.ne_session_connect._netmiko_over_ssh_client") diff --git a/web/src/components/HopProxyFields.tsx b/web/src/components/HopProxyFields.tsx index 073370f..950553c 100644 --- a/web/src/components/HopProxyFields.tsx +++ b/web/src/components/HopProxyFields.tsx @@ -3,6 +3,7 @@ import { useI18n } from "../i18n"; import { HOP_VENDORS, defaultHopTemplate, + expandBastionHopFields, isAutoHopTemplate, isBastionHopVendor, isLinuxHopVendor, @@ -127,7 +128,24 @@ export function HopProxyFields({