diff --git a/netx_api/ne_schemas.py b/netx_api/ne_schemas.py index 34f3fb5..9bc85e1 100644 --- a/netx_api/ne_schemas.py +++ b/netx_api/ne_schemas.py @@ -18,7 +18,7 @@ class ManagedNeCreate(BaseModel): port: int = 22 protocol: str = "ssh" username: str - password: str + password: str = "" tags: str = "" remark: str = "" hop_enabled: bool = False diff --git a/netx_api/ne_service.py b/netx_api/ne_service.py index 6dcc19b..8c8ba07 100644 --- a/netx_api/ne_service.py +++ b/netx_api/ne_service.py @@ -32,6 +32,17 @@ IMPORT_COLUMNS = ( "vendor", ) +OPTIONAL_IMPORT_HOP_COLUMNS = ( + "hop_enabled", + "hop_vendor", + "hop_host", + "hop_port", + "hop_username", + "hop_password", + "hop_target_auth_mode", + "hop_command_template", +) + def _now() -> datetime: return datetime.utcnow() @@ -76,12 +87,53 @@ def _validate_hop_on_create(body: ManagedNeCreate) -> None: raise HTTPException(status_code=400, detail="password_required") -def _target_password_optional(body: ManagedNeCreate) -> bool: - return ( - bool(body.hop_enabled) - and _normalize_hop_vendor(body.hop_vendor) == "bastion" - and _normalize_hop_target_auth_mode(body.hop_target_auth_mode) == "bastion_managed" +def _parse_import_bool(value: Any) -> bool: + return str(value or "").strip().lower() in ("1", "true", "yes", "y", "on") + + +def _import_cell_str(value: Any) -> str: + if value is None or (isinstance(value, float) and pd.isna(value)): + return "" + text = str(value).strip() + return "" if text.lower() == "nan" else text + + +def _apply_import_hop(row: ManagedNE, data: Any) -> str | None: + """Apply optional hop columns from an import row. Returns failure reason or None.""" + if "hop_enabled" not in data.index: + return None + if not _parse_import_bool(data.get("hop_enabled")): + row.hop_enabled = False + return None + hop_host = _import_cell_str(data.get("hop_host", "")) + hop_user = _import_cell_str(data.get("hop_username", "")) + hop_pass = _import_cell_str(data.get("hop_password", "")) + if not hop_host or not hop_user or not hop_pass: + return "hop_fields_incomplete" + hop_vendor = _normalize_hop_vendor(_import_cell_str(data.get("hop_vendor", "")) or "zte") + hop_port_raw = data.get("hop_port", 22) + try: + hop_port = int(hop_port_raw) + except (TypeError, ValueError): + hop_port = 22 + template = _import_cell_str(data.get("hop_command_template", "")) + if hop_vendor == "bastion" and not template: + template = default_bastion_username_template() + elif hop_vendor not in ("linux", "bastion") and not template: + template = default_hop_command_template(hop_vendor, "ssh", "") + row.hop_enabled = True + row.hop_vendor = hop_vendor + row.hop_host = hop_host + row.hop_port = hop_port + row.hop_protocol = "ssh" + row.hop_username = hop_user + row.hop_password_enc = encrypt_secret(hop_pass) + row.hop_command_template = template + row.hop_vrf = "" + row.hop_target_auth_mode = _normalize_hop_target_auth_mode( + _import_cell_str(data.get("hop_target_auth_mode", "")) or "bastion_managed" ) + return None def _apply_hop_create(row: ManagedNE, body: ManagedNeCreate) -> None: @@ -210,8 +262,6 @@ def get_managed_ne(db: Session, ne_id: str) -> ManagedNeOut: def create_managed_ne(db: Session, body: ManagedNeCreate) -> ManagedNeOut: _require_crypto() _validate_hop_on_create(body) - if not str(body.password or "").strip() and not _target_password_optional(body): - raise HTTPException(status_code=400, detail="password_required") ip = _normalize_ip(body.ip_address) if not ip: raise HTTPException(status_code=400, detail="ip_address_required") @@ -383,19 +433,36 @@ def build_managed_ne_import_template(fmt: str = "xlsx") -> tuple[str, bytes, str "protocol": "ssh", "name": "Core-SW1", "vendor": "Cisco", + "hop_enabled": "", + "hop_vendor": "", + "hop_host": "", + "hop_port": "", + "hop_username": "", + "hop_password": "", + "hop_target_auth_mode": "", + "hop_command_template": "", }, { - "device_type": "huawei", - "ip": "192.168.0.2", - "username": "admin", - "password": "your_password", + "device_type": "zte_zxros", + "ip": "2.2.2.2", + "username": "target-user", + "password": "", "port": 22, "protocol": "ssh", - "name": "AGG-01", - "vendor": "Huawei", + "name": "PE-via-bastion", + "vendor": "ZTE", + "hop_enabled": "true", + "hop_vendor": "bastion", + "hop_host": "1.1.1.1", + "hop_port": 22, + "hop_username": "bastion-user", + "hop_password": "vault_password", + "hop_target_auth_mode": "bastion_managed", + "hop_command_template": "", }, ] - df = pd.DataFrame(rows, columns=list(IMPORT_COLUMNS)) + all_columns = list(IMPORT_COLUMNS) + list(OPTIONAL_IMPORT_HOP_COLUMNS) + df = pd.DataFrame(rows, columns=all_columns) buf = BytesIO() kind = str(fmt or "xlsx").strip().lower() if kind == "csv": @@ -436,18 +503,18 @@ def import_managed_ne(db: Session, content: bytes, filename: str) -> ImportResul for idx, row in df.iterrows(): row_no = int(idx) + 2 try: - ip = _normalize_ip(str(row.get("ip", ""))) + ip = _normalize_ip(_import_cell_str(row.get("ip", ""))) if not ip: failed.append(ImportFailure(row=row_no, reason="ip_required")) continue - device_type = str(row.get("device_type", "")).strip() + device_type = _import_cell_str(row.get("device_type", "")) if device_type not in SUPPORTED_DEVICE_TYPES: failed.append(ImportFailure(row=row_no, reason="unsupported_device_type")) continue - username = str(row.get("username", "")).strip() - password = str(row.get("password", "")).strip() - if not username or not password: - failed.append(ImportFailure(row=row_no, reason="username_password_required")) + username = _import_cell_str(row.get("username", "")) + password = _import_cell_str(row.get("password", "")) + if not username: + failed.append(ImportFailure(row=row_no, reason="username_required")) continue port_raw = row.get("port", 22) try: @@ -455,8 +522,8 @@ def import_managed_ne(db: Session, content: bytes, filename: str) -> ImportResul except (TypeError, ValueError): port = 22 protocol = _normalize_protocol(str(row.get("protocol", "ssh"))) - display_name = str(row.get("name", "") or "").strip() or ip - vendor_raw = str(row.get("vendor", "") or "Other").strip() + display_name = _import_cell_str(row.get("name", "")) or ip + vendor_raw = _import_cell_str(row.get("vendor", "")) or "Other" vendor = "Other" for v in SUPPORTED_VENDORS: if v.lower() == vendor_raw.lower(): @@ -476,7 +543,11 @@ def import_managed_ne(db: Session, content: bytes, filename: str) -> ImportResul existing.port = port existing.protocol = protocol existing.username = username - existing.password_enc = encrypt_secret(password) + existing.password_enc = encrypt_secret(password) if password else "" + hop_err = _apply_import_hop(existing, row) + if hop_err: + failed.append(ImportFailure(row=row_no, reason=hop_err)) + continue existing.updated_at = now except CredentialCryptoError as exc: failed.append(ImportFailure(row=row_no, reason=str(exc))) diff --git a/tests/test_managed_ne.py b/tests/test_managed_ne.py index faddb73..31ddf65 100644 --- a/tests/test_managed_ne.py +++ b/tests/test_managed_ne.py @@ -178,6 +178,67 @@ class ManagedNeServiceImportTests(unittest.TestCase): self.assertEqual(result.updated, 0) self.assertEqual(len(result.failed), 0) + def test_csv_import_without_password(self): + csv = ( + "device_type,ip,username,password,port,protocol,name,vendor\n" + "zte_zxros,10.1.1.2,target-user,,22,ssh,NE-B,ZTE\n" + ).encode("utf-8") + result = import_managed_ne(self.db, csv, "devices.csv") + self.assertEqual(result.inserted, 1) + self.assertEqual(len(result.failed), 0) + row = self.db.query(ManagedNE).filter(ManagedNE.ip_address == "10.1.1.2").one() + self.assertEqual(row.username, "target-user") + self.assertEqual(row.password_enc, "") + + def test_csv_import_with_bastion_hop(self): + csv = ( + "device_type,ip,username,password,port,protocol,name,vendor," + "hop_enabled,hop_vendor,hop_host,hop_port,hop_username,hop_password,hop_target_auth_mode\n" + "zte_zxros,2.2.2.2,target-user,,22,ssh,NE-C,ZTE," + "true,bastion,1.1.1.1,22,bastion-user,vault-pass,bastion_managed\n" + ).encode("utf-8") + result = import_managed_ne(self.db, csv, "devices.csv") + self.assertEqual(result.inserted, 1) + self.assertEqual(len(result.failed), 0) + row = self.db.query(ManagedNE).filter(ManagedNE.ip_address == "2.2.2.2").one() + self.assertTrue(row.hop_enabled) + self.assertEqual(row.hop_vendor, "bastion") + self.assertEqual(row.hop_host, "1.1.1.1") + self.assertEqual(row.hop_username, "bastion-user") + self.assertEqual(decrypt_secret(row.hop_password_enc), "vault-pass") + + +class ManagedNeCreateOptionalPasswordTests(unittest.TestCase): + def setUp(self): + self._orig = settings.credential_secret_key + settings.credential_secret_key = Fernet.generate_key().decode() + self.engine = create_engine( + "sqlite://", + connect_args={"check_same_thread": False}, + poolclass=StaticPool, + ) + ManagedNE.__table__.create(bind=self.engine, checkfirst=True) + self.db = sessionmaker(bind=self.engine)() + + def tearDown(self): + self.db.close() + settings.credential_secret_key = self._orig + + def test_create_without_password(self): + out = create_managed_ne( + self.db, + ManagedNeCreate( + vendor="ZTE", + device_type="zte_zxros", + ip_address="10.9.9.9", + username="target-user", + password="", + ), + ) + self.assertEqual(out.username, "target-user") + row = self.db.query(ManagedNE).filter(ManagedNE.ip_address == "10.9.9.9").one() + self.assertEqual(row.password_enc, "") + if __name__ == "__main__": unittest.main() diff --git a/web/src/i18n/en.ts b/web/src/i18n/en.ts index bc48be5..54771e2 100644 --- a/web/src/i18n/en.ts +++ b/web/src/i18n/en.ts @@ -169,6 +169,8 @@ const en = { connectDetailTitle: "Connectivity test log", connectDetailEmpty: "No log yet. Run a connectivity test first; failures store full errors and hop context (passwords excluded).", + importHint: + "Import template includes optional hop columns (hop_enabled, etc.), or import NEs first then batch-add proxy. Password may be empty for bastion-managed targets.", importResult: { done: "Import done: {{inserted}} inserted, {{updated}} updated, {{failed}} failed row(s)", }, @@ -185,6 +187,8 @@ const en = { deleted: "Network element deleted", passwordRequired: "Password is required", passwordOptional: "leave blank to keep unchanged", + passwordHint: + "Required for direct login; leave empty when bastion manages target credentials. Use batch proxy later.", }, hop: { sectionTitle: "Jump host / proxy", diff --git a/web/src/i18n/zh.ts b/web/src/i18n/zh.ts index a25d477..205927e 100644 --- a/web/src/i18n/zh.ts +++ b/web/src/i18n/zh.ts @@ -167,6 +167,8 @@ const zh = { connectDetail: "详情", connectDetailTitle: "连通性测试日志", connectDetailEmpty: "暂无日志。请先执行连通性测试；失败时会记录完整错误与跳板上下文（不含密码）。", + importHint: + "导入模板含可选跳板列（hop_enabled 等）；也可先导入网元再批量添加代理。密码可留空（堡垒机托管场景）。", importResult: { done: "导入完成：新增 {{inserted}}，更新 {{updated}}，失败 {{failed}} 行", }, @@ -183,6 +185,7 @@ const zh = { deleted: "网元已删除", passwordRequired: "请填写密码", passwordOptional: "留空则不修改", + passwordHint: "直连网元需填写；经堡垒机托管时可留空，后续可用「批量添加代理」统一配置跳板。", }, hop: { sectionTitle: "跳板 / 代理", diff --git a/web/src/pages/NePage.tsx b/web/src/pages/NePage.tsx index e4db25f..343291b 100644 --- a/web/src/pages/NePage.tsx +++ b/web/src/pages/NePage.tsx @@ -185,10 +185,6 @@ export function NePage() { ...(form.password ? { password: form.password } : {}), ...(form.hop_password ? { hop_password: form.hop_password } : {}), }; - const bastionManaged = - form.hop_enabled && - form.hop_vendor === "bastion" && - form.hop_target_auth_mode === "bastion_managed"; if (form.hop_enabled) { if (!form.hop_host.trim()) throw new Error(t("managedNe.hop.hostRequired")); if (!form.hop_username.trim()) throw new Error(t("managedNe.hop.userRequired")); @@ -199,7 +195,6 @@ export function NePage() { if (!form.hop_password) delete (body as { hop_password?: string }).hop_password; return updateManagedNe(editing.id, body); } - if (!form.password && !bastionManaged) throw new Error(t("managedNe.form.passwordRequired")); return createManagedNe({ ...body, password: form.password || "" }); }, onSuccess: async () => { @@ -372,6 +367,7 @@ export function NePage() { > {importMutation.isPending ? t("managedNe.importing") : t("managedNe.importBtn")} + {t("managedNe.importHint")}