From 25a1e52a6b489104e072334d9340d39ad8d23554 Mon Sep 17 00:00:00 2001 From: dv Date: Wed, 29 Apr 2026 17:29:10 +0300 Subject: [PATCH] =?UTF-8?q?=D1=89=D0=B0=D1=81=20=D0=B2=D1=81=D0=B5=20?= =?UTF-8?q?=D1=81=D0=BB=D0=BE=D0=BC=D0=B0=D1=8E?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/app/plugins/builtin/common/ping.py | 2 +- backend/app/services/discovery/db_resolver.py | 12 +- .../app/services/discovery/orchestrator.py | 53 +++++- backend/app/services/discovery/ssh_client.py | 157 ++++++++++++++++-- frontend/src/components/DiscoveryDrawer.tsx | 49 ++++++ 5 files changed, 246 insertions(+), 27 deletions(-) diff --git a/backend/app/plugins/builtin/common/ping.py b/backend/app/plugins/builtin/common/ping.py index dee4f35..bef9be1 100644 --- a/backend/app/plugins/builtin/common/ping.py +++ b/backend/app/plugins/builtin/common/ping.py @@ -21,7 +21,7 @@ class PingAction(BaseAction): proc = None try: proc = await asyncio.create_subprocess_exec( - "ping", "-c", "1", "-W", "2", device.ip, + "ping", "-c", "4", "-W", "2", device.ip, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, ) diff --git a/backend/app/services/discovery/db_resolver.py b/backend/app/services/discovery/db_resolver.py index 76e4009..3f49a1a 100644 --- a/backend/app/services/discovery/db_resolver.py +++ b/backend/app/services/discovery/db_resolver.py @@ -19,6 +19,7 @@ async def autodetect_db_from_server( ssh_port: int, ssh_options: list, server_config_path: str = "/etc/caps/config.ini", + progress_cb=None, ) -> dict | None: """SSH into server, read its caps config, extract DB connection params. @@ -28,7 +29,7 @@ async def autodetect_db_from_server( from configparser import RawConfigParser config_text = await _read_server_config_with_retry( - server_ip, ssh_port, ssh_options, server_config_path + server_ip, ssh_port, ssh_options, server_config_path, progress_cb=progress_cb ) parser = RawConfigParser(strict=False) @@ -55,13 +56,15 @@ async def _read_server_config_with_retry( ssh_options: list, server_config_path: str, attempts: int = 2, + progress_cb=None, ) -> str: """Read server config, retrying once for transient command timeouts.""" last_exc: Exception | None = None for attempt in range(1, attempts + 1): try: return await ssh_run_multi( - server_ip, ssh_port, ssh_options, f"cat {server_config_path}", timeout=30 + server_ip, ssh_port, ssh_options, f"cat {server_config_path}", timeout=30, + progress_cb=progress_cb, ) except TimeoutError as exc: last_exc = exc @@ -88,13 +91,16 @@ async def get_rack_hosts( tunnel_host: str | None = None, tunnel_ssh_port: int = 22, ssh_options: list | None = None, + progress_cb=None, ) -> list[tuple[str, str, str, str | None]]: """Return list of (external_id, host_ip, rack_name, rack_type) from the racks table. Supports an optional SSH tunnel when the DB is not directly reachable. """ if via_tunnel and tunnel_host and ssh_options: - conn = await open_ssh_tunnel(tunnel_host, tunnel_ssh_port, ssh_options, host, port) + conn = await open_ssh_tunnel( + tunnel_host, tunnel_ssh_port, ssh_options, host, port, progress_cb=progress_cb + ) async with conn: listener = await conn.forward_local_port("127.0.0.1", 0, host, port) local_port = listener.get_port() diff --git a/backend/app/services/discovery/orchestrator.py b/backend/app/services/discovery/orchestrator.py index 3fdd4f5..3db1811 100644 --- a/backend/app/services/discovery/orchestrator.py +++ b/backend/app/services/discovery/orchestrator.py @@ -33,6 +33,7 @@ from .ssh_client import ( get_ip_neighbors_multi, ssh_run_multi, verify_is_mikrotik, + emit_ping_status, ) from .types import DeviceAction, DiscoveredDevice, DiscoveryResult, ProgressCallback @@ -92,6 +93,7 @@ async def discover_via_ssh( via_tunnel=use_tunnel, tunnel_host=obj.server_ip, tunnel_ssh_port=ssh_port, ssh_options=ssh_options, + progress_cb=_emit, ) if db_device_obj is not None: db_device_obj.status = "online" @@ -153,7 +155,9 @@ async def discover_via_ssh( ) if rack_ssh_ok: - neigh_ips = await get_ip_neighbors_multi(rack_host, rack_ssh_port or embedded_ssh_ports[0], ssh_options) + neigh_ips = await get_ip_neighbors_multi( + rack_host, rack_ssh_port or embedded_ssh_ports[0], ssh_options, progress_cb=_emit + ) neighbor_candidates.update(neigh_ips) discovered_device_ips.add(rack_host) @@ -163,8 +167,14 @@ async def discover_via_ssh( _rack_found = len(devices_found) for found in devices_found: + ping_online, ping_ms = await emit_ping_status( + found.ip, _emit, phase="discovered_device" + ) await _upsert_plugin_device( - db, obj, found, external_id, rack_id_mapping, result + db, obj, found, external_id, rack_id_mapping, result, + ping_online=ping_online, + ping_ms=ping_ms if ping_online else None, + checked_at=now, ) discovered_device_ips.add(found.ip) @@ -270,6 +280,7 @@ async def _resolve_db_connection( try: detected = await autodetect_db_from_server( server_ip=obj.server_ip, ssh_port=ssh_port, ssh_options=ssh_options, + progress_cb=_emit, ) except Exception as exc: result.errors.append( @@ -301,7 +312,8 @@ async def _resolve_db_connection( async def _emit_db_autodetect_error(obj, ssh_port, ssh_options, result: DiscoveryResult) -> None: try: raw = await ssh_run_multi( - obj.server_ip, ssh_port, ssh_options, "cat /etc/caps/config.ini", timeout=30 + obj.server_ip, ssh_port, ssh_options, "cat /etc/caps/config.ini", timeout=30, + progress_cb=None, ) diag_parser = RawConfigParser(strict=False) diag_parser.read_string(raw) @@ -502,7 +514,8 @@ async def _ssh_fetch_rack_config( for attempt in range(1, 3): try: config_text = await ssh_run_multi( - rack_host, ssh_port, ssh_options, f"cat {config_path}", timeout=30 + rack_host, ssh_port, ssh_options, f"cat {config_path}", timeout=30, + progress_cb=_emit, ) return True, config_text, ssh_port except TimeoutError as exc: @@ -616,6 +629,9 @@ async def _upsert_plugin_device( result: DiscoveryResult, slave_rack_id: int | None = None, slave_identifier: str | None = None, + ping_online: bool | None = None, + ping_ms: int | None = None, + checked_at: datetime | None = None, ) -> None: """Upsert a single device discovered from a plugin section. @@ -684,6 +700,17 @@ async def _upsert_plugin_device( existing.connection_params = cp changes.append(f"connection_params.credentials updated") + if ping_online is not None: + new_status = "online" if ping_online else "offline" + if existing.status != new_status: + existing.status = new_status + changes.append(f"status: ->{new_status}") + if checked_at is not None: + existing.last_ping_at = checked_at + if ping_online: + existing.last_seen = checked_at + existing.last_ping_ms = ping_ms if ping_online else None + if changes: logger.info("[discovery] %s (%s) - UPDATED: %s", found.ip, found.plugin_name, ", ".join(changes)) result.updated += 1 @@ -713,6 +740,10 @@ async def _upsert_plugin_device( rack_id=rack_id, location=device_location, device_meta={"plugin": found.plugin_name, "cls": found.cls}, connection_params=conn_params, + status="online" if ping_online else "offline" if ping_online is not None else "unknown", + last_seen=checked_at if ping_online else None, + last_ping_at=checked_at if ping_online is not None else None, + last_ping_ms=ping_ms if ping_online else None, )) cred_note = f", http_user={found.http_username}" if found.http_username else "" logger.info( @@ -752,7 +783,9 @@ async def _discover_slave_racks( await _emit("disc_action", {"message": f"Проверяем слейв-кандидата {candidate_ip}..."}) try: - slave_config, slave_ssh_port = await fetch_slave_config(candidate_ip, ssh_ports, ssh_options, config_path) + slave_config, slave_ssh_port = await fetch_slave_config( + candidate_ip, ssh_ports, ssh_options, config_path, progress_cb=_emit + ) is_slave = False slave_identifier = "" master_ip_for_slave = "" @@ -794,6 +827,9 @@ async def _discover_slave_racks( }) for found in slave_devices: + ping_online, ping_ms = await emit_ping_status( + found.ip, _emit, phase="discovered_device" + ) await _upsert_plugin_device( db, obj, found, external_id="", @@ -801,6 +837,9 @@ async def _discover_slave_racks( result=result, slave_rack_id=slave_rack_id, slave_identifier=slave_identifier, + ping_online=ping_online, + ping_ms=ping_ms if ping_online else None, + checked_at=now, ) try: @@ -906,7 +945,7 @@ async def _discover_mikrotik( ) -> None: target_host = obj.server_ip or rack_hosts[0][1] logger.info("[discovery] Phase 3: MikroTik via SSH_CLIENT on %s", target_host) - mikrotik_ip = await discover_mikrotik_multi(target_host, ssh_port, ssh_options) + mikrotik_ip = await discover_mikrotik_multi(target_host, ssh_port, ssh_options, progress_cb=_emit) if not mikrotik_ip: logger.info("[discovery] MikroTik: no IP found via SSH_CLIENT") @@ -914,7 +953,7 @@ async def _discover_mikrotik( logger.info("[discovery] MikroTik candidate: %s — verifying...", mikrotik_ip) await _emit("disc_action", {"message": f"Проверка MikroTik {mikrotik_ip} (WinBox/SSH)..."}) - is_mikrotik = await verify_is_mikrotik(mikrotik_ip, ssh_port, ssh_options) + is_mikrotik = await verify_is_mikrotik(mikrotik_ip, ssh_port, ssh_options, progress_cb=_emit) if not is_mikrotik: logger.info("[discovery] %s — NOT a MikroTik (WinBox and SSH banner check failed), skipping", mikrotik_ip) await _emit("disc_action", {"message": f"{mikrotik_ip} — не MikroTik, пропускаем"}) diff --git a/backend/app/services/discovery/ssh_client.py b/backend/app/services/discovery/ssh_client.py index b0ec2e3..a110769 100644 --- a/backend/app/services/discovery/ssh_client.py +++ b/backend/app/services/discovery/ssh_client.py @@ -6,11 +6,59 @@ No SQLAlchemy, no business logic — pure network I/O. import asyncio import logging import re +import socket +import time +from collections.abc import Awaitable, Callable import asyncssh logger = logging.getLogger(__name__) +ProgressCallback = Callable[[str, dict], Awaitable[None]] + +PING_TIMEOUT = 3 +SSH_CONNECT_TIMEOUT = 15 + +DISCOVERY_SSH_KWARGS: dict = { + "kex_algs": ["curve25519-sha256"], + "encryption_algs": ["aes128-ctr"], + "mac_algs": ["hmac-sha2-256"], +} + + +async def _connect_with_sock_compat(**kwargs) -> asyncssh.SSHClientConnection: + """Connect with IPQoS disabled and conservative SSH algorithms. + + AsyncSSH doesn't expose OpenSSH's `IPQoS=none` option directly, so discovery + opens the TCP socket itself, clears IP_TOS, and passes the connected socket + to AsyncSSH. + """ + host = kwargs.pop("host") + port = kwargs.pop("port", 22) + connect_timeout = kwargs.pop("connect_timeout", 10) + + sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + sock.setblocking(False) + try: + try: + sock.setsockopt(socket.IPPROTO_IP, socket.IP_TOS, 0) + except OSError: + logger.debug("[discovery] Could not clear IP_TOS for %s:%s", host, port) + + loop = asyncio.get_running_loop() + await asyncio.wait_for(loop.sock_connect(sock, (host, port)), timeout=connect_timeout) + return await asyncssh.connect( + host=host, + port=port, + sock=sock, + connect_timeout=connect_timeout, + **DISCOVERY_SSH_KWARGS, + **kwargs, + ) + except BaseException: + sock.close() + raise + # ── Single connection helpers (used internally and for legacy paths) ────────── @@ -24,8 +72,8 @@ async def read_remote_file( """SSH into host and return the contents of remote_path.""" if not await is_port_open(host, port, timeout=3.0): raise ConnectionRefusedError(f"{host}:{port} — порт недоступен") - async with asyncssh.connect( - host, port=port, username=username, password=password, + async with await _connect_with_sock_compat( + host=host, port=port, username=username, password=password, known_hosts=None, connect_timeout=10, ) as conn: result = await conn.run(f"cat {remote_path}", check=True) @@ -40,8 +88,8 @@ async def get_ip_neighbors( if not await is_port_open(host, port, timeout=3.0): logger.info("[discovery] %s:%d — port closed, skipping ip neigh", host, port) return [] - async with asyncssh.connect( - host, port=port, username=username, password=password, + async with await _connect_with_sock_compat( + host=host, port=port, username=username, password=password, known_hosts=None, connect_timeout=10, ) as conn: result = await conn.run("ip neigh show", check=False) @@ -69,12 +117,59 @@ async def is_port_open(host: str, port: int, timeout: float = 3.0) -> bool: return False +async def ping_host(host: str, timeout: int = PING_TIMEOUT) -> tuple[bool, int]: + """Ping a host once. Returns (success, latency_ms).""" + start = time.monotonic() + try: + proc = await asyncio.create_subprocess_exec( + "ping", "-c", "1", "-W", str(timeout), host, + stdout=asyncio.subprocess.DEVNULL, + stderr=asyncio.subprocess.DEVNULL, + ) + await asyncio.wait_for(proc.wait(), timeout=timeout + 2) + ms = int((time.monotonic() - start) * 1000) + return proc.returncode == 0, ms + except Exception: + return False, 0 + + +async def emit_ping_status( + host: str, + progress_cb: ProgressCallback | None = None, + *, + phase: str = "ssh", +) -> tuple[bool, int]: + online, ping_ms = await ping_host(host) + if progress_cb is not None: + await progress_cb("disc_device_ping", { + "ip": host, + "status": "online" if online else "offline", + "ping_ms": ping_ms if online else None, + "phase": phase, + }) + logger.info( + "[discovery] ping %s before %s: %s%s", + host, + phase, + "online" if online else "offline", + f" ({ping_ms} ms)" if online else "", + ) + return online, ping_ms + + # ── Multi-auth helpers ──────────────────────────────────────────────────────── async def ssh_run_multi( - host: str, port: int, options: list, command: str, timeout: int = 15, + host: str, + port: int, + options: list, + command: str, + timeout: int = 15, + progress_cb: ProgressCallback | None = None, ) -> str: """Try each SSH auth option in order; return stdout on first success or raise.""" + await emit_ping_status(host, progress_cb) + # Fast pre-check: skip SSH entirely if port is not open if not await is_port_open(host, port, timeout=3.0): raise ConnectionRefusedError(f"{host}:{port} — порт недоступен (TCP check failed)") @@ -84,7 +179,10 @@ async def ssh_run_multi( _log_attempt(i, len(options), host, opt) try: connect_kwargs = _build_connect_kwargs(host, port, opt) - conn = await asyncssh.connect(**connect_kwargs) + conn = await asyncio.wait_for( + _connect_with_sock_compat(**connect_kwargs, connect_timeout=SSH_CONNECT_TIMEOUT), + timeout=SSH_CONNECT_TIMEOUT, + ) except Exception as exc: last_exc = exc logger.info( @@ -116,20 +214,34 @@ async def ssh_run_multi( raise last_exc -async def get_ip_neighbors_multi(host: str, port: int, options: list) -> list[str]: +async def get_ip_neighbors_multi( + host: str, + port: int, + options: list, + progress_cb: ProgressCallback | None = None, +) -> list[str]: """Multi-auth version of get_ip_neighbors.""" try: - stdout = await ssh_run_multi(host, port, options, "ip neigh show", timeout=15) + stdout = await ssh_run_multi( + host, port, options, "ip neigh show", timeout=15, progress_cb=progress_cb + ) return _parse_neigh_output(stdout) except Exception as exc: logger.warning("[discovery] ip neigh failed on %s: %s", host, exc) return [] -async def discover_mikrotik_multi(host: str, port: int, options: list) -> str | None: +async def discover_mikrotik_multi( + host: str, + port: int, + options: list, + progress_cb: ProgressCallback | None = None, +) -> str | None: """Discover MikroTik IP via SSH_CLIENT environment variable (multi-auth).""" try: - stdout = await ssh_run_multi(host, port, options, "echo $SSH_CLIENT", timeout=15) + stdout = await ssh_run_multi( + host, port, options, "echo $SSH_CLIENT", timeout=15, progress_cb=progress_cb + ) parts = stdout.strip().split() return parts[0] if parts else None except Exception as exc: @@ -138,13 +250,19 @@ async def discover_mikrotik_multi(host: str, port: int, options: list) -> str | async def fetch_slave_config( - host: str, ports: list[int], options: list, config_path: str, + host: str, + ports: list[int], + options: list, + config_path: str, + progress_cb: ProgressCallback | None = None, ) -> tuple[str | None, int | None]: """Read a remote config file from a slave candidate. Returns (config, port).""" for port in ports: for attempt in range(1, 3): try: - config_text = await ssh_run_multi(host, port, options, f"cat {config_path}", timeout=30) + config_text = await ssh_run_multi( + host, port, options, f"cat {config_path}", timeout=30, progress_cb=progress_cb + ) return config_text, port except TimeoutError as exc: if attempt < 2: @@ -170,8 +288,8 @@ async def discover_proxmox_via_port_scan( Migrate to multi-auth when needed. """ try: - async with asyncssh.connect( - host, port=port, username=username, password=password, + async with await _connect_with_sock_compat( + host=host, port=port, username=username, password=password, known_hosts=None, connect_timeout=10, ) as conn: scan_script = ( @@ -196,6 +314,7 @@ async def open_ssh_tunnel( ssh_options: list, target_host: str, target_port: int, + progress_cb: ProgressCallback | None = None, ): """Open an asyncssh connection, trying each option in order. @@ -203,6 +322,7 @@ async def open_ssh_tunnel( The caller is responsible for closing the connection. """ last_exc: Exception = Exception("No SSH options provided") + await emit_ping_status(tunnel_host, progress_cb, phase="ssh_tunnel") for opt in ssh_options: try: connect_kwargs = _build_connect_kwargs(tunnel_host, tunnel_port, opt) @@ -217,7 +337,10 @@ async def open_ssh_tunnel( "[discovery] Tunnel attempt for %s — password: user=%s password=%s", tunnel_host, opt.username, masked, ) - conn = await asyncssh.connect(**connect_kwargs) + conn = await asyncio.wait_for( + _connect_with_sock_compat(**connect_kwargs), + timeout=SSH_CONNECT_TIMEOUT, + ) logger.info( "[discovery] Tunnel auth succeeded for %s with %s(%s)", tunnel_host, opt.type, opt.username, @@ -239,6 +362,7 @@ async def verify_is_mikrotik( ssh_port: int = 22, ssh_options: list | None = None, timeout: float = 3.0, + progress_cb: ProgressCallback | None = None, ) -> bool: """Check whether host is actually a MikroTik RouterOS device. @@ -265,11 +389,12 @@ async def verify_is_mikrotik( # Probe 2: SSH banner if ssh_options: + await emit_ping_status(host, progress_cb, phase="mikrotik_verify") for opt in ssh_options: try: kwargs = _build_connect_kwargs(host, ssh_port, opt) kwargs["connect_timeout"] = int(timeout) - async with asyncssh.connect(**kwargs) as conn: + async with await _connect_with_sock_compat(**kwargs) as conn: banner = (conn.get_extra_info("server_version") or "").lower() if "rosssh" in banner or "mikrotik" in banner: logger.info( diff --git a/frontend/src/components/DiscoveryDrawer.tsx b/frontend/src/components/DiscoveryDrawer.tsx index b82fb5f..c4404ed 100644 --- a/frontend/src/components/DiscoveryDrawer.tsx +++ b/frontend/src/components/DiscoveryDrawer.tsx @@ -28,6 +28,13 @@ interface InfraItem { ip: string } +interface DevicePingItem { + ip: string + status: 'online' | 'offline' + ping_ms?: number | null + phase?: string +} + type PhaseState = 'idle' | 'running' | 'done' interface Props { @@ -50,6 +57,7 @@ export function DiscoveryDrawer({ open, onClose, objectName, taskId, onComplete const [mikrotikPhase, setMikrotikPhase] = useState('idle') const [proxmoxPhase, setProxmoxPhase] = useState('idle') const [infraFound, setInfraFound] = useState([]) + const [devicePings, setDevicePings] = useState([]) const [result, setResult] = useState<{ created: number; updated: number; skipped: number; errors: string[] } | null>(null) const ctrlRef = useRef(null) @@ -66,6 +74,7 @@ export function DiscoveryDrawer({ open, onClose, objectName, taskId, onComplete setMikrotikPhase('idle') setProxmoxPhase('idle') setInfraFound([]) + setDevicePings([]) setResult(null) const ctrl = new AbortController() @@ -129,6 +138,18 @@ export function DiscoveryDrawer({ open, onClose, objectName, taskId, onComplete } else if (ev.event === 'disc_infra_found') { setInfraFound((prev) => [...prev, { type: data.type, ip: data.ip }]) + } else if (ev.event === 'disc_device_ping') { + setDevicePings((prev) => { + const item = { + ip: data.ip, + status: data.status === 'online' ? 'online' : 'offline', + ping_ms: data.ping_ms, + phase: data.phase, + } satisfies DevicePingItem + const withoutCurrent = prev.filter((p) => p.ip !== item.ip) + return [item, ...withoutCurrent].slice(0, 12) + }) + } else if (ev.event === 'disc_done') { setResult({ created: data.created ?? 0, @@ -231,6 +252,34 @@ export function DiscoveryDrawer({ open, onClose, objectName, taskId, onComplete )} + {/* Ping status */} + {devicePings.length > 0 && ( +
+ + Ping + + {devicePings.map((p) => ( +
+ {p.status === 'online' ? ( + + ) : ( + + )} + {p.ip} + + {p.status} + + {p.status === 'online' && p.ping_ms != null && ( + {p.ping_ms} ms + )} +
+ ))} +
+ )} + {/* Slave phase */} {slavesPhase !== 'idle' && (