dv/utp_service
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

щас все сломаю

Browse source
This commit is contained in:
dv 2026-04-29 17:29:10 +03:00
parent e4b993dbf3
commit 25a1e52a6b
5 changed files with 246 additions and 27 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
backend/app/plugins/builtin/common/ping.py
View file

@ -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,
)

12
backend/app/services/discovery/db_resolver.py
View file

@ -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()

53
backend/app/services/discovery/orchestrator.py
View file

@ -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, пропускаем"})

157
backend/app/services/discovery/ssh_client.py
View file

@ -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(

49
frontend/src/components/DiscoveryDrawer.tsx
View file

@ -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<PhaseState>('idle')
const [proxmoxPhase, setProxmoxPhase] = useState<PhaseState>('idle')
const [infraFound, setInfraFound] = useState<InfraItem[]>([])
const [devicePings, setDevicePings] = useState<DevicePingItem[]>([])
const [result, setResult] = useState<{ created: number; updated: number; skipped: number; errors: string[] } | null>(null)
const ctrlRef = useRef<AbortController | null>(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
</div>
)}
{/* Ping status */}
{devicePings.length > 0 && (
<div style={{ marginBottom: 16, padding: '8px 0', borderTop: '1px solid #f0f0f0' }}>
<Typography.Text type="secondary" style={{ display: 'block', fontSize: 12, marginBottom: 6 }}>
Ping
</Typography.Text>
{devicePings.map((p) => (
<div
key={p.ip}
style={{ display: 'flex', alignItems: 'center', gap: 8, padding: '3px 0' }}
>
{p.status === 'online' ? (
<CheckCircleOutlined style={{ color: '#52c41a', fontSize: 11 }} />
) : (
<CloseCircleOutlined style={{ color: '#ff4d4f', fontSize: 11 }} />
)}
<Typography.Text style={{ flex: 1, fontSize: 12 }}>{p.ip}</Typography.Text>
<Tag color={p.status === 'online' ? 'green' : 'red'} style={{ fontSize: 11 }}>
{p.status}
</Tag>
{p.status === 'online' && p.ping_ms != null && (
<Typography.Text type="secondary" style={{ fontSize: 11 }}>{p.ping_ms} ms</Typography.Text>
)}
</div>
))}
</div>
)}
{/* Slave phase */}
{slavesPhase !== 'idle' && (
<div style={{ marginBottom: 12, padding: '8px 0', borderTop: '1px solid #f0f0f0' }}>