From 513513147980502d1dab0a69cd8f5a5fbb22317d Mon Sep 17 00:00:00 2001 From: dv Date: Mon, 13 Apr 2026 16:17:09 +0300 Subject: [PATCH] =?UTF-8?q?=D1=80=D0=B0=D0=B7=D0=B1=D0=B8=D0=BB=20ssh=20di?= =?UTF-8?q?scovery?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .claude/settings.local.json | 3 +- backend/app/services/database.py | 19 +- backend/app/services/device_discovery.py | 1523 +---------------- backend/app/services/discovery/__init__.py | 20 + .../app/services/discovery/config_parser.py | 178 ++ backend/app/services/discovery/db_resolver.py | 103 ++ .../app/services/discovery/orchestrator.py | 927 ++++++++++ backend/app/services/discovery/ssh_client.py | 222 +++ backend/app/services/discovery/types.py | 42 + backend/app/services/discovery/uri.py | 68 + 10 files changed, 1586 insertions(+), 1519 deletions(-) create mode 100644 backend/app/services/discovery/__init__.py create mode 100644 backend/app/services/discovery/config_parser.py create mode 100644 backend/app/services/discovery/db_resolver.py create mode 100644 backend/app/services/discovery/orchestrator.py create mode 100644 backend/app/services/discovery/ssh_client.py create mode 100644 backend/app/services/discovery/types.py create mode 100644 backend/app/services/discovery/uri.py diff --git a/.claude/settings.local.json b/.claude/settings.local.json index a28bede..096c26a 100644 --- a/.claude/settings.local.json +++ b/.claude/settings.local.json @@ -16,7 +16,8 @@ "Bash(docker compose:*)", "Bash(grep -r 8000 /c/Users/Professional/Documents/VSCode/utp_service/backend/ --include=*.py --include=*.sh)", "Bash(grep -n \"rack_type\" /c/Users/Professional/Documents/VSCode/utp_service/backend/alembic/versions/*.py)", - "Bash(python3 -c \":*)" + "Bash(python3 -c \":*)", + "Bash(python -c ':*)" ] } } diff --git a/backend/app/services/database.py b/backend/app/services/database.py index 28d54a1..aa27977 100644 --- a/backend/app/services/database.py +++ b/backend/app/services/database.py @@ -6,13 +6,26 @@ from app.config import settings engine = create_async_engine( settings.DATABASE_URL, - echo=settings.DEBUG, + #echo=settings.DEBUG, + echo=False pool_pre_ping=True, ) + +#if not settings.SQLALCHEMY_LOG: +# for _log_name in ("sqlalchemy.engine", "sqlalchemy.pool", "sqlalchemy.dialects", "sqlalchemy.orm"): +# logging.getLogger(_log_name).setLevel(logging.WARNING) + if not settings.SQLALCHEMY_LOG: - for _log_name in ("sqlalchemy.engine", "sqlalchemy.pool", "sqlalchemy.dialects", "sqlalchemy.orm"): - logging.getLogger(_log_name).setLevel(logging.WARNING) + for name in ( + "sqlalchemy.engine", + "sqlalchemy.pool", + "sqlalchemy.dialects", + "sqlalchemy.orm", + ): + logger = logging.getLogger(name) + logger.setLevel(logging.ERROR) + logger.propagate = False AsyncSessionLocal = async_sessionmaker( bind=engine, diff --git a/backend/app/services/device_discovery.py b/backend/app/services/device_discovery.py index f4de001..50aed9a 100644 --- a/backend/app/services/device_discovery.py +++ b/backend/app/services/device_discovery.py @@ -1,1515 +1,8 @@ -""" -SSH-based device discovery: connects to each rack PC, reads its caps config, -extracts plugin URIs and classifies discovered network devices. - -Discovery chain: - Object DB → racks table (host IPs) → SSH per rack → parse config → upsert devices - -Limitations (devices not discoverable automatically): - - Serial/USB devices (no network URI) - - RODOS-8 relays, Moxa switches (not represented as plugins) - - Physical MikroTik router (not in rack configs) -""" - -import asyncio -import logging -import re -from collections.abc import Callable, Awaitable -from datetime import datetime, timezone -from configparser import RawConfigParser -from dataclasses import dataclass, field -from io import StringIO -from urllib.parse import urlparse - -import asyncssh -import psycopg -from sqlalchemy import select -from sqlalchemy.ext.asyncio import AsyncSession - -from app.models.device import Device -from app.models.object import Object -from app.services.crypto import decrypt, encrypt - -logger = logging.getLogger(__name__) - -# ── URI extraction ──────────────────────────────────────────────────────────── - -_SERIAL_SCHEMES = {"serial", "com"} -_SERVICE_HOSTS = {"localhost", "127.0.0.1"} - -def _extract_device_info(uri: str) -> tuple[str, str, str] | None: - """Return (ip, username, password) from a URI, or None if not a network device. - - Username and password are extracted from URI credentials (http://user:pass@ip). - They may be empty strings if not present in the URI. - """ - if not uri: - return None - try: - parsed = urlparse(uri) - scheme = (parsed.scheme or "").lower() - if scheme in _SERIAL_SCHEMES: - return None - hostname = parsed.hostname - if not hostname or hostname in _SERVICE_HOSTS: - return None - # Validate it looks like an IP (simple check) - if re.match(r"^\d{1,3}(\.\d{1,3}){3}$", hostname): - return hostname, parsed.username or "", parsed.password or "" - except Exception: - pass - return None - - -def _extract_ip(uri: str) -> str | None: - """Return IP from a URI string, or None if not a network device.""" - info = _extract_device_info(uri) - return info[0] if info else None - - -# ── Plugin class → device category ─────────────────────────────────────────── - -def _cls_to_category(cls: str) -> str: - cls_lower = cls.lower() - if "camera" in cls_lower or "facedetect" in cls_lower: - return "camera" - if "controller" in cls_lower or "ovenmk" in cls_lower or "rodos" in cls_lower: - return "io_board" - if "payment" in cls_lower or "sberpilot" in cls_lower or "payx" in cls_lower: - return "bank_terminal" - if "payonline" in cls_lower or "kkt" in cls_lower or "fiscal" in cls_lower: - return "cash_register" - return "other" - - -def _cls_to_role(cls: str, plugin_name: str) -> str: - cls_lower = cls.lower() - name_lower = plugin_name.lower() - if "camera" in cls_lower or "camera" in name_lower: - if "grz" in name_lower or "plate" in name_lower or "lpr" in name_lower or "frame" in name_lower: - return "plate" - if "face" in name_lower or "lico" in name_lower: - return "face" - return "overview" - return "other" - - -# Plugins whose URIs point at the CAPS server itself, not at separate devices -_SKIP_PLUGIN_PREFIXES = ("mnemo", "processor", "recognition", "photo", "voice") - - -# ── Config parsing ──────────────────────────────────────────────────────────── - -@dataclass -class DiscoveredDevice: - ip: str - category: str - role: str - plugin_name: str - cls: str - http_username: str = "" # from URI credentials (e.g. http://user:pass@ip) - http_password: str = "" - - -def _parse_config(config_text: str, rack_host: str, all_rack_hosts: list[str] | None = None) -> list[DiscoveredDevice]: - """Extract discovered network devices from a caps config file. - - Args: - config_text: Config file content - rack_host: IP of current rack (will be excluded from results) - all_rack_hosts: All rack IPs (will be excluded to avoid confusing rack servers with devices) - """ - if all_rack_hosts is None: - all_rack_hosts = [rack_host] - - parser = RawConfigParser(strict=False) - parser.read_string(config_text) - - discovered: list[DiscoveredDevice] = [] - seen_ips: set[str] = set() - all_sections = parser.sections() - plugin_sections = [s for s in all_sections if s.startswith("plugin:")] - - logger.info(f"[ssh_discovery] Config parsing: found {len(all_sections)} sections total, {len(plugin_sections)} are plugins") - - for section in all_sections: - if not section.startswith("plugin:"): - continue - - plugin_name = section[len("plugin:"):] - if any(plugin_name.startswith(prefix) for prefix in _SKIP_PLUGIN_PREFIXES): - logger.info(f"[ssh_discovery] Plugin '{plugin_name}' - IGNORED (in skip list)") - continue - - cls = parser.get(section, "cls", fallback="") - if not cls: - logger.info(f"[ssh_discovery] Plugin '{plugin_name}' - IGNORED (no 'cls' field)") - continue - - logger.info(f"[ssh_discovery] Plugin '{plugin_name}' (cls={cls})") - - # uri/frame_uri/host → actual device (camera, io_board, etc.) - device_uri_fields = ("uri", "frame_uri", "host") - found_ip_in_plugin = False - for field_name in device_uri_fields: - raw = parser.get(section, field_name, fallback=None) - if not raw: - logger.info(f"[ssh_discovery] {field_name}: (not set)") - continue - logger.info(f"[ssh_discovery] {field_name}='{raw}'") - info = _extract_device_info(raw) - if not info: - logger.info(f"[ssh_discovery] → IGNORED (not a valid network IP)") - continue - ip, http_user, http_pass = info - if ip in all_rack_hosts: - logger.info(f"[ssh_discovery] → IGNORED (is a rack server IP: {ip})") - continue - if ip in seen_ips: - logger.info(f"[ssh_discovery] → IGNORED (duplicate, already seen)") - continue - - seen_ips.add(ip) - cred_str = f", http_user={http_user}" if http_user else "" - logger.info(f"[ssh_discovery] → EXTRACTED {ip} (category={_cls_to_category(cls)}, role={_cls_to_role(cls, plugin_name)}{cred_str})") - discovered.append(DiscoveredDevice( - ip=ip, - category=_cls_to_category(cls), - role=_cls_to_role(cls, plugin_name), - plugin_name=plugin_name, - cls=cls, - http_username=http_user, - http_password=http_pass, - )) - found_ip_in_plugin = True - - # stream_uri → RTSP server (separate VM, not the camera itself) - stream_raw = parser.get(section, "stream_uri", fallback=None) - if stream_raw: - logger.info(f"[ssh_discovery] stream_uri='{stream_raw}'") - stream_ip = _extract_ip(stream_raw) - if stream_ip and stream_ip not in all_rack_hosts and stream_ip not in seen_ips: - seen_ips.add(stream_ip) - logger.info(f"[ssh_discovery] → EXTRACTED {stream_ip} as RTSP server (category=vm)") - discovered.append(DiscoveredDevice( - ip=stream_ip, - category="vm", - role="other", - plugin_name=f"{plugin_name}_rtsp", - cls="rtsp_server", - )) - found_ip_in_plugin = True - - if not found_ip_in_plugin: - logger.info(f"[ssh_discovery] → No network devices found in this plugin") - - # Parse [caps.voice] section for Asterisk server - voice_addr = parser.get("caps.voice", "server_addr", fallback=None) - if voice_addr: - logger.info(f"[ssh_discovery] [caps.voice] server_addr='{voice_addr}'") - voice_ip = _extract_ip(f"sip://{voice_addr}") or (voice_addr.strip() if re.match(r"^\d{1,3}(\.\d{1,3}){3}$", voice_addr.strip()) else None) - if voice_ip and voice_ip not in all_rack_hosts and voice_ip not in seen_ips: - seen_ips.add(voice_ip) - logger.info(f"[ssh_discovery] → EXTRACTED {voice_ip} as Asterisk (category=vm)") - discovered.append(DiscoveredDevice( - ip=voice_ip, - category="vm", - role="other", - plugin_name="asterisk", - cls="asterisk", - )) - - return discovered - - -# ── SSH operations ──────────────────────────────────────────────────────────── - -async def _read_remote_config( - host: str, - port: int, - username: str, - password: str, - config_path: str, -) -> str: - async with asyncssh.connect( - host, - port=port, - username=username, - password=password, - known_hosts=None, - connect_timeout=10, - ) as conn: - result = await conn.run(f"cat {config_path}", check=True) - return result.stdout - - -async def _get_ip_neighbors( - host: str, - port: int, - username: str, - password: str, -) -> list[str]: - """Run `ip neigh show` and return REACHABLE neighbour IPs.""" - try: - async with asyncssh.connect( - host, - port=port, - username=username, - password=password, - known_hosts=None, - connect_timeout=10, - ) as conn: - result = await conn.run("ip neigh show", check=False) - ips = [] - for line in result.stdout.strip().splitlines(): - parts = line.split() - if not parts: - continue - ip = parts[0] - state = parts[-1] if parts else "" - # Accept REACHABLE and DELAY; skip STALE/FAILED/INCOMPLETE - if re.match(r"^\d{1,3}(\.\d{1,3}){3}$", ip) and state in ("REACHABLE", "DELAY"): - ips.append(ip) - return ips - except Exception as e: - logger.warning(f"[ssh_discovery] ip neigh failed on {host}: {e}") - return [] - - -async def _check_if_slave( - host: str, - port: int, - username: str, - password: str, - master_ip: str, - config_path: str, -) -> tuple[bool, str, str]: - """ - SSH into candidate host and check if it is a slave rack pointing at master_ip. - Returns (is_slave, identifier, config_text). - - is_slave: True if [plugin:linking] has secondary cls and uri contains master_ip - - identifier: value of [caps.rack] identifier field - - config_text: raw config (for device parsing if slave confirmed) - """ - try: - config_text = await _read_remote_config(host, port, username, password, config_path) - parser = RawConfigParser(strict=False) - parser.read_string(config_text) - linking_cls = parser.get("plugin:linking", "cls", fallback="") - linking_uri = parser.get("plugin:linking", "uri", fallback="") - identifier = parser.get("caps.rack", "identifier", fallback="").strip() - if "secondary" in linking_cls.lower() and master_ip in linking_uri: - return True, identifier or host, config_text - except Exception: - pass - return False, "", "" - - -async def _discover_mikrotik_via_ssh_client( - host: str, - port: int, - username: str, - password: str, -) -> str | None: - """ - Discover MikroTik router IP via SSH_CLIENT environment variable. - When connected via L2TP → MikroTik → main_server, the SSH_CLIENT env var - shows the client IP connecting to the server (MikroTik's L2TP interface IP). - """ - try: - async with asyncssh.connect( - host, - port=port, - username=username, - password=password, - known_hosts=None, - connect_timeout=10, - ) as conn: - result = await conn.run("echo $SSH_CLIENT", check=False) - ssh_client_output = result.stdout.strip() - if ssh_client_output: - # SSH_CLIENT format: "client_ip client_port server_port" - parts = ssh_client_output.split() - if parts: - return parts[0] - except Exception as e: - logger.warning(f"[ssh_discovery] Failed to discover MikroTik via SSH_CLIENT: {e}") - return None - - -async def _discover_proxmox_via_port_scan( - host: str, - port: int, - username: str, - password: str, -) -> list[str]: - """ - Discover Proxmox servers on the local subnet by scanning port 8006. - Runs a subnet scanning script that checks port 8006 on all IPs in the subnet. - """ - try: - async with asyncssh.connect( - host, - port=port, - username=username, - password=password, - known_hosts=None, - connect_timeout=10, - ) as conn: - # Scan script: determine subnet and scan all IPs for port 8006 - scan_script = ( - "subnet=$(ip -o -f inet addr show | awk '!/ lo |docker|br-|wg/ {print $4; exit}' | " - "cut -d/ -f1 | awk -F. '{print $1\".\"$2\".\"$3}') && " - "(set +m; for ip in $subnet.{1..254}; do " - "timeout 0.3 bash -c \"echo > /dev/tcp/$ip/8006\" 2>/dev/null && echo \"$ip\" & " - "done; wait)" - ) - result = await conn.run(scan_script, check=False) - ips = [line.strip() for line in result.stdout.strip().split('\n') if line.strip()] - return ips - except Exception as e: - logger.warning(f"[ssh_discovery] Failed to discover Proxmox servers via port scan: {e}") - return [] - - -def _parse_db_uri(uri: str) -> dict | None: - """Parse postgresql+psycopg2://user:pass@host:port/dbname → dict.""" - try: - parsed = urlparse(uri) - if not parsed.hostname: - return None - return { - "host": parsed.hostname, - "port": parsed.port or 5432, - "user": parsed.username or "", - "password": parsed.password or "", - "dbname": (parsed.path or "").lstrip("/"), - } - except Exception: - return None - - -async def _autodetect_db_from_server( - server_ip: str, - ssh_port: int, - ssh_user: str, - ssh_password: str, - server_config_path: str = "/etc/caps/config.ini", -) -> dict | None: - """SSH into server, read its caps.conf, extract DB connection params.""" - try: - config_text = await _read_remote_config( - host=server_ip, - port=ssh_port, - username=ssh_user, - password=ssh_password, - config_path=server_config_path, - ) - except Exception: - return None - - parser = RawConfigParser(strict=False) - parser.read_string(config_text) - - # Search all sections + DEFAULT (keys before any section header) for db_uri - sections_to_check = list(parser.sections()) + ["DEFAULT"] - for section in sections_to_check: - try: - db_uri = parser.get(section, "db_uri", fallback=None) - if db_uri and db_uri.startswith("postgresql"): - return _parse_db_uri(db_uri) - except Exception: - continue - - # Also check raw defaults (keys before any section) - db_uri = parser.defaults().get("db_uri") - if db_uri and db_uri.startswith("postgresql"): - return _parse_db_uri(db_uri) - - return None - - -async def _autodetect_db_from_server_multi( - server_ip: str, - ssh_port: int, - ssh_options: list, - server_config_path: str = "/etc/caps/config.ini", -) -> dict | None: - """Multi-auth version of _autodetect_db_from_server.""" - try: - config_text = await _ssh_run_multi(server_ip, ssh_port, ssh_options, f"cat {server_config_path}") - except Exception: - return None - - parser = RawConfigParser(strict=False) - parser.read_string(config_text) - - sections_to_check = list(parser.sections()) + ["DEFAULT"] - for section in sections_to_check: - try: - db_uri = parser.get(section, "db_uri", fallback=None) - if db_uri and db_uri.startswith("postgresql"): - return _parse_db_uri(db_uri) - except Exception: - continue - - db_uri = parser.defaults().get("db_uri") - if db_uri and db_uri.startswith("postgresql"): - return _parse_db_uri(db_uri) - return None - - -async def _ssh_run_multi(host: str, port: int, options: list, command: str, timeout: int = 15) -> str: - """Try multiple SSH auth options to run a command; return stdout on success or raise.""" - last_exc: Exception = Exception("No auth options provided") - for i, opt in enumerate(options): - if opt.type == "key": - logger.info( - "[ssh_discovery] SSH attempt %d/%d for %s — key: user=%s key_path=%s", - i + 1, len(options), host, opt.username, opt.key_path, - ) - else: - masked = (opt.password[:1] + "***" + opt.password[-1:]) if len(opt.password) > 2 else "***" - logger.info( - "[ssh_discovery] SSH attempt %d/%d for %s — password: user=%s password=%s", - i + 1, len(options), host, opt.username, masked, - ) - try: - connect_kwargs: dict = dict( - host=host, port=port, username=opt.username, - known_hosts=None, connect_timeout=10, - ) - if opt.type == "key": - connect_kwargs["client_keys"] = [opt.key_path] - connect_kwargs["passphrase"] = None - else: - connect_kwargs["password"] = opt.password - async with asyncssh.connect(**connect_kwargs) as conn: - result = await asyncio.wait_for(conn.run(command, check=True), timeout=timeout) - logger.info("[ssh_discovery] SSH auth succeeded for %s with %s(%s)", host, opt.type, opt.username) - return result.stdout - except Exception as exc: - last_exc = exc - logger.info("[ssh_discovery] SSH auth failed for %s with %s(%s): %s", host, opt.type, opt.username, exc) - raise last_exc - - -async def _get_ip_neighbors_multi(host: str, port: int, options: list) -> list[str]: - """Multi-auth version of _get_ip_neighbors.""" - try: - stdout = await _ssh_run_multi(host, port, options, "ip neigh show", timeout=15) - ips = [] - for line in stdout.strip().splitlines(): - parts = line.split() - if not parts: - continue - ip = parts[0] - state = parts[-1] if parts else "" - if re.match(r"^\d{1,3}(\.\d{1,3}){3}$", ip) and state in ("REACHABLE", "DELAY"): - ips.append(ip) - return ips - except Exception as e: - logger.warning(f"[ssh_discovery] ip neigh failed on {host}: {e}") - return [] - - -async def _discover_mikrotik_multi(host: str, port: int, options: list) -> str | None: - """Multi-auth version of _discover_mikrotik_via_ssh_client.""" - try: - stdout = await _ssh_run_multi(host, port, options, "echo $SSH_CLIENT", timeout=15) - ssh_client_output = stdout.strip() - if ssh_client_output: - parts = ssh_client_output.split() - if parts: - return parts[0] - except Exception as e: - logger.warning(f"[ssh_discovery] Failed to discover MikroTik via SSH_CLIENT: {e}") - return None - - -async def _fetch_slave_config( - host: str, port: int, options: list, config_path: str -) -> str | None: - """Read config from a slave candidate via SSH (one connection). Returns None on failure.""" - try: - return await _ssh_run_multi(host, port, options, f"cat {config_path}", timeout=15) - except Exception: - return None - - -def _parse_slave_config(config_text: str, master_ip: str, host: str) -> tuple[bool, str, str]: - """Check parsed config against one master_ip. No network I/O.""" - parser = RawConfigParser(strict=False) - parser.read_string(config_text) - linking_cls = parser.get("plugin:linking", "cls", fallback="") - linking_uri = parser.get("plugin:linking", "uri", fallback="") - identifier = parser.get("caps.rack", "identifier", fallback="").strip() - if "secondary" in linking_cls.lower() and master_ip in linking_uri: - return True, identifier or host, config_text - return False, "", "" - - - -async def _open_tunnel(tunnel_host: str, tunnel_port: int, ssh_options: list, target_host: str, target_port: int): - """Try each ssh_option in order; return open asyncssh connection on first success or raise.""" - last_exc: Exception = Exception("No SSH options provided") - for opt in ssh_options: - try: - connect_kwargs: dict = dict( - host=tunnel_host, port=tunnel_port, username=opt.username, - known_hosts=None, connect_timeout=10, - ) - if opt.type == "key": - connect_kwargs["client_keys"] = [opt.key_path] - connect_kwargs["passphrase"] = None - logger.info("[ssh_discovery] Tunnel attempt for %s — key: user=%s key_path=%s", tunnel_host, opt.username, opt.key_path) - else: - connect_kwargs["password"] = opt.password - masked = (opt.password[:1] + "***" + opt.password[-1:]) if len(opt.password) > 2 else "***" - logger.info("[ssh_discovery] Tunnel attempt for %s — password: user=%s password=%s", tunnel_host, opt.username, masked) - conn = await asyncssh.connect(**connect_kwargs) - logger.info("[ssh_discovery] Tunnel auth succeeded for %s with %s(%s)", tunnel_host, opt.type, opt.username) - return conn - except Exception as exc: - last_exc = exc - logger.info("[ssh_discovery] Tunnel auth failed for %s with %s(%s): %s", tunnel_host, opt.type, opt.username, exc) - raise last_exc - - -async def _get_rack_hosts_direct( - host: str, - port: int, - user: str, - password: str, - dbname: str, - table: str, - via_tunnel: bool = False, - tunnel_host: str | None = None, - tunnel_ssh_port: int = 22, - ssh_options: list | None = None, -) -> list[tuple[str, str, str, str | None]]: - """Return list of (external_id, host_ip, rack_name, rack_type) from the racks table.""" - if via_tunnel and tunnel_host and ssh_options: - conn = await _open_tunnel(tunnel_host, tunnel_ssh_port, ssh_options, host, port) - async with conn: - listener = await conn.forward_local_port("127.0.0.1", 0, host, port) - local_port = listener.get_port() - dsn = ( - f"host=127.0.0.1 port={local_port} dbname={dbname} " - f"user={user} password={password} connect_timeout=10" - ) - return await _fetch_rack_hosts(dsn, table) - else: - dsn = f"host={host} port={port} dbname={dbname} user={user} password={password} connect_timeout=10" - return await _fetch_rack_hosts(dsn, table) - - -async def _fetch_rack_hosts(dsn: str, table: str) -> list[tuple[str, str, str, str | None]]: - """Return list of (external_id, host_ip, rack_name, rack_type) from the racks table.""" - async with await psycopg.AsyncConnection.connect(dsn) as conn: - cursor = await conn.execute( - f"SELECT id, data->>'host' AS host, name, data->>'rack_type' AS rack_type " - f"FROM \"{table}\" WHERE data->>'host' IS NOT NULL" - ) - rows = await cursor.fetchall() - result = [] - for r in rows: - if not r[1]: - continue - ext_id = str(r[0]) - short_name = str(r[2]) if r[2] else None - # Combine CAPS id with its short name: "lane-1-1 (01)" - rack_name = f"{ext_id} ({short_name})" if short_name and short_name != ext_id else ext_id - result.append((ext_id, str(r[1]), rack_name, r[3] or None)) - return result - - -# ── Main entry point ────────────────────────────────────────────────────────── - -@dataclass -class DeviceAction: - """Track each device discovery action for summary reporting.""" - ip: str - hostname: str - action: str # "ADDED", "UPDATED", "SKIPPED", "IGNORED", "ERROR" - reason: str - category: str = "" - role: str = "" - - -@dataclass -class DiscoveryResult: - created: int = 0 - updated: int = 0 - skipped: int = 0 - errors: list[str] = field(default_factory=list) - actions: list[DeviceAction] = field(default_factory=list) - - -ProgressCallback = Callable[[str, dict], Awaitable[None]] - - -async def discover_via_ssh( - db: AsyncSession, - obj: Object, - config_path: str = "/etc/caps/config.ini", - progress_cb: ProgressCallback | None = None, -) -> DiscoveryResult: - result = DiscoveryResult() - - async def _emit(event_type: str, data: dict) -> None: - if progress_cb is not None: - await progress_cb(event_type, data) - - # Build ordered list of SSH auth options for this object - from app.services.ssh import build_object_auth_options, _AuthOption - ssh_options = await build_object_auth_options(obj, db) - - if not ssh_options: - result.errors.append("SSH credentials not configured on this object") - return result - - # Legacy single-cred vars for calls that still use them (tunnels etc.) - # Use the first option as the primary; fall back to legacy fields - _first = ssh_options[0] - ssh_password = _first.password if _first.type == "password" else "" - ssh_port = obj.ssh_port or 22 - - # ── Resolve DB connection ───────────────────────────────────────────────── - # Priority 1: fully configured DB on the object - # Priority 2: autodetect from server config via SSH (only server_ip needed) - - db_host: str | None = obj.db_host - db_port: int = obj.db_port or 5432 - db_name: str | None = obj.db_name - db_user: str | None = obj.db_user - db_password: str | None = None - db_table: str | None = obj.db_table - server_ip_ssh_verified = False - - if db_host and db_name and db_user and obj.db_pass_enc and db_table: - try: - db_password = decrypt(obj.db_pass_enc) - except Exception: - result.errors.append("Failed to decrypt DB password") - return result - elif obj.server_ip: - # Autodetect: SSH into server, read its caps.conf to get DB URI - await _emit("disc_action", {"message": f"Чтение конфигурации сервера {obj.server_ip}..."}) - detected = await _autodetect_db_from_server_multi( - server_ip=obj.server_ip, - ssh_port=ssh_port, - ssh_options=ssh_options, - ) - if not detected or not all([detected.get("host"), detected.get("user"), detected.get("dbname")]): - # Diagnostic: try to read the config and list what sections/keys are there - try: - raw = await _ssh_run_multi(obj.server_ip, ssh_port, ssh_options, "cat /etc/caps/config.ini") - diag_parser = RawConfigParser(strict=False) - diag_parser.read_string(raw) - found_sections = diag_parser.sections() - found_keys = {s: list(diag_parser.options(s)) for s in found_sections} - result.errors.append( - f"db_uri not found in server config. " - f"Sections found: {found_sections}. " - f"Keys per section: {found_keys}. " - f"Fill in DB fields manually on the object." - ) - except Exception as diag_exc: - result.errors.append( - f"Could not read server config for auto-detection: {diag_exc}. " - f"Fill in DB fields manually on the object." - ) - return result - db_host = detected["host"] - db_port = detected["port"] - db_user = detected["user"] - db_password = detected["password"] - db_name = detected["dbname"] - db_table = db_table or obj.db_table or "racks" - server_ip_ssh_verified = True - else: - result.errors.append( - "DB connection not configured. Either fill in the DB fields on the object " - "or set server_ip so the DB can be auto-detected via SSH." - ) - return result - - # Step 0.5: upsert DB server as a device now (before connection attempt), - # so the user can see it and configure tunnel even if connection fails. - now_pre = datetime.now(timezone.utc) - db_device_obj: Device | None = None - if db_host and db_host != obj.server_ip: - try: - db_device_obj = (await db.execute( - select(Device).where(Device.object_id == obj.id, Device.ip == db_host) - )).scalar_one_or_none() - if db_device_obj is None: - db_device_obj = Device( - object_id=obj.id, - ip=db_host, - hostname="db_server", - category="vm", - role="other", - source="auto_ssh", - location="server_room", - status="unknown", - device_meta={"is_caps_db": True}, - ) - db.add(db_device_obj) - await db.flush() - logger.info("[ssh_discovery] %s (db_server) - PRE-ADDED before connection attempt", db_host) - result.created += 1 - result.actions.append(DeviceAction( - ip=db_host, hostname="db_server", action="ADDED", - reason="object infrastructure device (vm/db_server)", category="vm", role="other", - )) - except Exception as exc: - logger.warning("[ssh_discovery] Failed to pre-upsert db_server device %s: %s", db_host, exc) - - # Read tunnel flag from the device (user may have set it via UI), fall back to object setting - device_via_tunnel = bool( - (db_device_obj.connection_params or {}).get("db_via_tunnel") - ) if db_device_obj else False - use_tunnel = device_via_tunnel or obj.db_via_tunnel - - # Step 1: get rack hosts from object's DB - await _emit("disc_action", {"message": "Получение списка стоек из БД..."}) - logger.info( - "[ssh_discovery] Connecting to object DB: host=%s port=%s dbname=%s user=%s table=%s via_tunnel=%s", - db_host, db_port, db_name, db_user, db_table, use_tunnel, - ) - try: - rack_hosts = await _get_rack_hosts_direct( - host=db_host, - port=db_port, - user=db_user, - password=db_password or "", - dbname=db_name, - table=db_table, - via_tunnel=use_tunnel, - tunnel_host=obj.server_ip, - tunnel_ssh_port=ssh_port, - ssh_options=ssh_options, - ) - # Connection succeeded — mark DB device online - if db_device_obj is not None: - db_device_obj.status = "online" - db_device_obj.last_seen = now_pre - except Exception as exc: - # Mark DB device offline so the user sees it's unreachable - if db_device_obj is not None: - db_device_obj.status = "offline" - err_msg = f"Cannot fetch rack hosts from object DB: {exc}" - if "timeout" in str(exc).lower(): - err_msg += ". Попробуйте настроить SSH-туннель (отредактируйте устройство БД и включите опцию)" - result.errors.append(err_msg) - return result - - if not rack_hosts: - result.errors.append("No racks with host IPs found in object DB") - return result - - await _emit("disc_server", {"rack_count": len(rack_hosts)}) - - # Step 2: SSH into each rack, parse config, upsert devices - # neighbor_candidates: IPs seen in ARP tables of rack PCs; used for slave discovery - # discovered_device_ips: all IPs added as devices during this run - neighbor_candidates: set[str] = set() - discovered_device_ips: set[str] = set() - - # Build mapping of external_id -> Rack.id for device assignment. - # If a rack with that name doesn't exist yet in our racks table, create it now. - from app.models.rack import Rack - rack_id_mapping: dict[str, int] = {} - for ext_id, _, rack_name, rack_type_db in rack_hosts: - # rack_name already has the combined form "lane-1-1 (01)" from _fetch_rack_hosts; - # also search for the legacy ext_id-only name to handle already-created racks. - existing_rack_row = (await db.execute( - select(Rack).where( - Rack.object_id == obj.id, - Rack.name.in_([ext_id, rack_name]), - ) - )).scalar_one_or_none() - if existing_rack_row: - rack_id_mapping[ext_id] = existing_rack_row.id - if existing_rack_row.name != rack_name: - logger.info(f"[ssh_discovery] Renamed rack '{existing_rack_row.name}' → '{rack_name}'") - existing_rack_row.name = rack_name - if rack_type_db and not existing_rack_row.rack_type: - existing_rack_row.rack_type = rack_type_db - logger.info(f"[ssh_discovery] Updated rack '{rack_name}' rack_type={rack_type_db}") - else: - new_rack = Rack(object_id=obj.id, name=rack_name, rack_type=rack_type_db) - db.add(new_rack) - await db.flush() # get the new PK - rack_id_mapping[ext_id] = new_rack.id - logger.info(f"[ssh_discovery] Created rack entry '{rack_name}' rack_type={rack_type_db} (id={new_rack.id})") - - # Step 1b: Add main CAPS server as device (DB server was already upserted in Step 0.5) - # verified=True means we successfully connected to this host during this discovery run - now = datetime.now(timezone.utc) - infra_devices = [] - if obj.server_ip: - infra_devices.append((obj.server_ip, "main_server", "caps_server", server_ip_ssh_verified)) - - for infra_ip, infra_category, infra_hostname, infra_verified in infra_devices: - try: - existing_infra = (await db.execute( - select(Device).where( - Device.object_id == obj.id, - Device.ip == infra_ip, - ) - )).scalar_one_or_none() - if existing_infra is None: - db.add(Device( - object_id=obj.id, - ip=infra_ip, - hostname=infra_hostname, - category=infra_category, - role="other", - source="auto_ssh", - location="server_room", - status="online" if infra_verified else "unknown", - last_seen=now if infra_verified else None, - )) - logger.info(f"[ssh_discovery] {infra_ip} ({infra_hostname}) - ADDED: category={infra_category}, location=server_room, status={'online' if infra_verified else 'unknown'}, source=auto_ssh") - result.created += 1 - result.actions.append(DeviceAction( - ip=infra_ip, - hostname=infra_hostname, - action="ADDED", - reason=f"object infrastructure device ({infra_category})", - category=infra_category, - role="other", - )) - else: - infra_changes = [] - if infra_verified: - if existing_infra.status != "online": - existing_infra.status = "online" - infra_changes.append("status: →online") - existing_infra.last_seen = now - infra_changes.append(f"last_seen: →{now.isoformat()}") - if infra_changes: - logger.info(f"[ssh_discovery] {infra_ip} ({infra_hostname}) - UPDATED: {', '.join(infra_changes)}") - result.updated += 1 - result.actions.append(DeviceAction( - ip=infra_ip, - hostname=existing_infra.hostname or infra_hostname, - action="UPDATED", - reason="; ".join(infra_changes), - category=existing_infra.category, - role=existing_infra.role, - )) - else: - logger.info(f"[ssh_discovery] {infra_ip} ({infra_hostname}) - SKIPPED: already exists (category={existing_infra.category}, hostname={existing_infra.hostname})") - result.skipped += 1 - result.actions.append(DeviceAction( - ip=infra_ip, - hostname=existing_infra.hostname or infra_hostname, - action="SKIPPED", - reason=f"already exists (category={existing_infra.category})", - category=existing_infra.category, - role=existing_infra.role, - )) - except Exception as exc: - logger.error(f"[ssh_discovery] {infra_ip} ({infra_hostname}) - ERROR: {exc}") - result.errors.append(f"Error upserting infra device {infra_ip}: {exc}") - - if infra_devices: - await db.flush() - - for rack_idx, (external_id, rack_host, rack_name, _rack_type_db) in enumerate(rack_hosts): - await _emit("disc_rack_start", { - "rack_name": rack_name, - "rack_host": rack_host, - "index": rack_idx, - "total": len(rack_hosts), - }) - - # Resolve config_path override from the embedded device's connection_params - existing_rack_pc = (await db.execute( - select(Device).where( - Device.object_id == obj.id, - Device.ip == rack_host, - ) - )).scalar_one_or_none() - - effective_config_path = config_path - if existing_rack_pc and existing_rack_pc.connection_params: - effective_config_path = existing_rack_pc.connection_params.get( - "config_path", config_path - ) - - rack_hostname = external_id - rack_ssh_ok = False - config_text = None - await _emit("disc_action", {"message": f"SSH → {rack_host} ({rack_name})"}) - try: - config_text = await _ssh_run_multi(rack_host, ssh_port, ssh_options, f"cat {effective_config_path}") - rack_ssh_ok = True - except Exception as exc: - result.errors.append(f"SSH to {rack_host} failed: {exc}") - - logger.info(f"[ssh_discovery] Processing rack {rack_host} ({rack_hostname}), ssh_ok={rack_ssh_ok}, in_db={existing_rack_pc is not None}") - - try: - if existing_rack_pc is None: - # New device — insert with status based on SSH result - db.add(Device( - object_id=obj.id, - ip=rack_host, - hostname=rack_hostname, - category="embedded", - role="other", - source="auto_ssh", - rack_id=rack_id_mapping.get(external_id), - status="online" if rack_ssh_ok else "unknown", - last_seen=now if rack_ssh_ok else None, - )) - status_str = "online" if rack_ssh_ok else "unknown" - logger.info(f"[ssh_discovery] {rack_host} ({rack_hostname}) - ADDED: status={status_str}") - result.created += 1 - result.actions.append(DeviceAction( - ip=rack_host, - hostname=rack_hostname, - action="ADDED", - reason="rack PC from racks table" + (" (SSH ok)" if rack_ssh_ok else " (SSH failed, added as unknown)"), - category="embedded", - role="other", - )) - else: - rack_changes = [] - if rack_ssh_ok: - if existing_rack_pc.status != "online": - existing_rack_pc.status = "online" - rack_changes.append("status: →online") - existing_rack_pc.last_seen = now - rack_changes.append("last_seen: updated") - if existing_rack_pc.rack_id is None and rack_id_mapping.get(external_id) is not None: - existing_rack_pc.rack_id = rack_id_mapping[external_id] - rack_changes.append(f"rack_id: None→{existing_rack_pc.rack_id}") - if rack_changes: - logger.info(f"[ssh_discovery] {rack_host} ({existing_rack_pc.hostname}) - UPDATED: {', '.join(rack_changes)}") - result.updated += 1 - result.actions.append(DeviceAction( - ip=rack_host, - hostname=existing_rack_pc.hostname, - action="UPDATED", - reason="; ".join(rack_changes), - category=existing_rack_pc.category, - role=existing_rack_pc.role, - )) - else: - reason = "SSH failed, rack already in DB" if not rack_ssh_ok else "rack already up to date" - logger.info(f"[ssh_discovery] {rack_host} ({existing_rack_pc.hostname}) - SKIPPED: {reason}") - result.skipped += 1 - result.actions.append(DeviceAction( - ip=rack_host, - hostname=existing_rack_pc.hostname, - action="SKIPPED", - reason=reason, - category=existing_rack_pc.category, - role=existing_rack_pc.role, - )) - except Exception as exc: - logger.error(f"[ssh_discovery] {rack_host} - ERROR: {exc}") - result.errors.append(f"Error processing rack {rack_host}: {exc}") - result.actions.append(DeviceAction( - ip=rack_host, - hostname=rack_hostname, - action="ERROR", - reason=str(exc), - category="embedded", - role="other", - )) - - if not rack_ssh_ok: - try: - await db.flush() - except Exception as exc: - result.errors.append(f"Flush error for offline rack {rack_host}: {exc}") - await _emit("disc_rack_done", {"rack_name": rack_name, "rack_host": rack_host, "found": 0, "ssh_ok": False}) - continue # no config to parse - - # Collect ARP neighbours for slave rack discovery (best-effort) - neigh_ips = await _get_ip_neighbors_multi(rack_host, ssh_port, ssh_options) - neighbor_candidates.update(neigh_ips) - discovered_device_ips.add(rack_host) # rack PC itself is a known device - - # Get all rack IPs to exclude them from device discovery - all_rack_ips = [host for _, host, _, _ in rack_hosts] - await _emit("disc_action", {"message": f"Парсинг конфигурации {rack_name}..."}) - devices_found = _parse_config(config_text, rack_host, all_rack_ips) - logger.info(f"[ssh_discovery] Found {len(devices_found)} devices from plugins in {rack_host}") - - for found in devices_found: - # vm devices (RTSP, Asterisk, etc.) are shared infrastructure — place in server_room, - # not tied to any specific rack. Other devices belong to the current rack. - is_shared = found.category == "vm" - if is_shared: - hostname = found.plugin_name # e.g. "asterisk", "camera_1_rtsp" - rack_id = None - device_location = "server_room" - else: - hostname = f"{external_id}-{found.plugin_name}" # e.g. "lane-1-1-camera_1" - rack_id = rack_id_mapping.get(external_id) - device_location = None - - try: - existing = (await db.execute( - select(Device).where( - Device.object_id == obj.id, - Device.ip == found.ip, - ) - )).scalar_one_or_none() - - if existing is not None: - changes = [] - - # PROTECT: Don't overwrite embedded (rack) devices with plugin devices! - # If device is currently embedded (a rack), never change it - if existing.category == "embedded": - logger.info(f"[ssh_discovery] {found.ip} ({found.plugin_name}) - SKIPPED: device is rack server (embedded), cannot overwrite with plugin device") - result.skipped += 1 - result.actions.append(DeviceAction( - ip=found.ip, - hostname=existing.hostname, - action="SKIPPED", - reason=f"device is rack server (embedded), cannot overwrite with plugin device", - category=existing.category, - role=existing.role, - )) - continue - - if existing.category != found.category and existing.source == "auto_ssh": - changes.append(f"category: {existing.category}→{found.category}") - existing.category = found.category - if existing.role == "other" and found.role != "other": - changes.append(f"role: other→{found.role}") - existing.role = found.role - if not is_shared: - # Rack-specific device: update hostname if stale - if not existing.hostname or existing.hostname.startswith("rack"): - if existing.hostname != hostname: - changes.append(f"hostname: '{existing.hostname}'→'{hostname}'") - existing.hostname = hostname - # Update rack_id if not set - if existing.rack_id is None and rack_id is not None: - existing.rack_id = rack_id - changes.append(f"rack_id: None→{rack_id}") - # Shared vm devices: never update hostname/rack_id — they're already in server_room - - if changes: - logger.info(f"[ssh_discovery] {found.ip} ({found.plugin_name}) - UPDATED: {', '.join(changes)}") - result.updated += 1 - result.actions.append(DeviceAction( - ip=found.ip, - hostname=existing.hostname, - action="UPDATED", - reason="; ".join(changes), - category=found.category, - role=found.role, - )) - else: - logger.info(f"[ssh_discovery] {found.ip} ({found.plugin_name}) - SKIPPED: already exists with same values (source={existing.source}, category={existing.category}, role={existing.role}, hostname={existing.hostname})") - result.skipped += 1 - result.actions.append(DeviceAction( - ip=found.ip, - hostname=existing.hostname, - action="SKIPPED", - reason=f"already exists with same values (source={existing.source}, category={existing.category}, role={existing.role}, hostname={existing.hostname})", - category=existing.category, - role=existing.role, - )) - else: - # Populate connection_params with HTTP credentials from URI if present - init_connection_params: dict = {} - if found.http_username: - init_connection_params["username"] = found.http_username - init_connection_params["password_enc"] = encrypt(found.http_password) if found.http_password else "" - db.add(Device( - object_id=obj.id, - ip=found.ip, - hostname=hostname, - category=found.category, - role=found.role, - source="auto_ssh", - rack_id=rack_id, - location=device_location, - device_meta={"plugin": found.plugin_name, "cls": found.cls}, - connection_params=init_connection_params, - )) - cred_note = f", http_user={found.http_username}" if found.http_username else "" - logger.info(f"[ssh_discovery] {found.ip} ({found.plugin_name}) - ADDED: category={found.category}, role={found.role}, hostname={hostname}, rack_id={rack_id}, location={device_location}, source=auto_ssh{cred_note}") - result.created += 1 - result.actions.append(DeviceAction( - ip=found.ip, - hostname=hostname, - action="ADDED", - reason=f"new device from plugin {found.plugin_name}", - category=found.category, - role=found.role, - )) - except Exception as exc: - logger.error(f"[ssh_discovery] {found.ip} ({found.plugin_name}) - ERROR: {exc}") - result.errors.append(f"Error upserting {found.ip}: {exc}") - result.actions.append(DeviceAction( - ip=found.ip, - hostname=hostname, - action="ERROR", - reason=str(exc), - category=found.category, - role=found.role, - )) - - # Track all plugin device IPs as known (for slave candidate filtering) - for found in devices_found: - discovered_device_ips.add(found.ip) - - # Flush after each rack so that subsequent SELECTs (for other racks - # that may share the same device IPs) see the just-added rows. - # Without this, autoflush=False means pending db.add() calls are - # invisible to SELECT queries, causing UniqueViolationError on flush. - try: - await db.flush() - except Exception as exc: - result.errors.append(f"Flush error after rack {rack_host}: {exc}") - await _emit("disc_rack_done", {"rack_name": rack_name, "rack_host": rack_host, "found": len(devices_found), "ssh_ok": True}) - - # Step 2.5: Discover slave racks via ARP neighbor tables - # Filter out already-known IPs: rack hosts, discovered devices, infra IPs, common gateways - all_rack_ips_set = {host for _, host, _, _ in rack_hosts} - infra_ips = {ip for ip in [obj.server_ip, db_host] if ip} - common_gateways = {ip for ip in neighbor_candidates if ip.endswith(".1") or ip.endswith(".254")} - slave_candidates = ( - neighbor_candidates - - all_rack_ips_set - - discovered_device_ips - - infra_ips - - common_gateways - ) - await _emit("disc_phase", {"phase": "slaves", "candidate_count": len(slave_candidates)}) - if slave_candidates: - logger.info(f"[ssh_discovery] Step 2.5: Slave discovery — {len(slave_candidates)} ARP candidate(s): {slave_candidates}") - await _emit("disc_action", {"message": f"Поиск слейв-стоек: проверяем {len(slave_candidates)} ARP-кандидатов..."}) - else: - await _emit("disc_action", {"message": "Слейв-стойки: ARP-кандидатов не найдено"}) - for candidate_ip in slave_candidates: - logger.info(f"[ssh_discovery] Checking slave candidate {candidate_ip}...") - await _emit("disc_action", {"message": f"Проверяем слейв-кандидата {candidate_ip}..."}) - - # Fetch config once — then check against all masters without extra SSH connections - slave_config_text = await _fetch_slave_config(candidate_ip, ssh_port, ssh_options, config_path) - is_slave = False - slave_identifier = "" - master_ip_for_slave = "" - - if slave_config_text is None: - logger.info(f"[ssh_discovery] {candidate_ip} — could not read config, skipping") - else: - for _, master_host, _, _ in rack_hosts: - ok, identifier, _ = _parse_slave_config(slave_config_text, master_host, candidate_ip) - if ok: - is_slave = True - master_ip_for_slave = master_host - master_ext_id = next( - (ext_id for ext_id, host, _, _ in rack_hosts if host == master_host), - master_host, - ) - slave_identifier = f"{master_ext_id}-slave" - break - - if not is_slave: - logger.info(f"[ssh_discovery] {candidate_ip} — not a slave rack (no secondary linking pointing to known master)") - continue - - logger.info(f"[ssh_discovery] {candidate_ip} — SLAVE rack confirmed (name={slave_identifier}, master={master_ip_for_slave})") - await _emit("disc_action", {"message": f"Слейв подтверждён: {slave_identifier} ({candidate_ip})"}) - - # Create Rack entry for the slave in our DB - slave_rack_id: int | None = None - try: - existing_slave_rack = (await db.execute( - select(Rack).where(Rack.object_id == obj.id, Rack.name == slave_identifier) - )).scalar_one_or_none() - if existing_slave_rack: - slave_rack_id = existing_slave_rack.id - else: - new_slave_rack = Rack(object_id=obj.id, name=slave_identifier, description="slave (auto-discovered via ARP)") - db.add(new_slave_rack) - await db.flush() - slave_rack_id = new_slave_rack.id - logger.info(f"[ssh_discovery] Created rack entry '{slave_identifier}' (id={slave_rack_id}) for slave") - except Exception as exc: - result.errors.append(f"Error creating rack entry for slave {candidate_ip}: {exc}") - - # Upsert the slave rack PC itself as embedded device - try: - existing_slave_dev = (await db.execute( - select(Device).where(Device.object_id == obj.id, Device.ip == candidate_ip) - )).scalar_one_or_none() - if existing_slave_dev is None: - db.add(Device( - object_id=obj.id, - ip=candidate_ip, - hostname=slave_identifier, - category="embedded", - role="other", - source="auto_ssh", - rack_id=slave_rack_id, - status="online", - last_seen=now, - )) - logger.info(f"[ssh_discovery] {candidate_ip} ({slave_identifier}) - ADDED: slave rack PC") - result.created += 1 - result.actions.append(DeviceAction( - ip=candidate_ip, hostname=slave_identifier, - action="ADDED", reason=f"slave rack PC (master={master_ip_for_slave})", - category="embedded", role="other", - )) - else: - slave_changes = [] - if existing_slave_dev.status != "online": - existing_slave_dev.status = "online" - slave_changes.append("status: →online") - existing_slave_dev.last_seen = now - slave_changes.append("last_seen: updated") - if existing_slave_dev.rack_id is None and slave_rack_id: - existing_slave_dev.rack_id = slave_rack_id - slave_changes.append(f"rack_id: →{slave_rack_id}") - result.updated += 1 - result.actions.append(DeviceAction( - ip=candidate_ip, hostname=existing_slave_dev.hostname or slave_identifier, - action="UPDATED", reason="; ".join(slave_changes), - category="embedded", role="other", - )) - await db.flush() - except Exception as exc: - result.errors.append(f"Error upserting slave device {candidate_ip}: {exc}") - - # Parse slave config and add its devices - all_rack_ips_with_slave = list(all_rack_ips_set) + [candidate_ip] - slave_devices = _parse_config(slave_config_text, candidate_ip, all_rack_ips_with_slave) - logger.info(f"[ssh_discovery] Slave {candidate_ip}: found {len(slave_devices)} plugin devices") - await _emit("disc_slave_found", {"rack_name": slave_identifier, "ip": candidate_ip, "found": len(slave_devices)}) - for found in slave_devices: - is_shared = found.category == "vm" - if is_shared: - hostname = found.plugin_name - s_rack_id = None - device_location = "server_room" - else: - hostname = f"{slave_identifier}-{found.plugin_name}" - s_rack_id = slave_rack_id - device_location = None - try: - existing = (await db.execute( - select(Device).where(Device.object_id == obj.id, Device.ip == found.ip) - )).scalar_one_or_none() - if existing is None: - slave_conn_params: dict = {} - if found.http_username: - slave_conn_params["username"] = found.http_username - slave_conn_params["password_enc"] = encrypt(found.http_password) if found.http_password else "" - db.add(Device( - object_id=obj.id, ip=found.ip, hostname=hostname, - category=found.category, role=found.role, - source="auto_ssh", rack_id=s_rack_id, location=device_location, - device_meta={"plugin": found.plugin_name, "cls": found.cls}, - connection_params=slave_conn_params, - )) - logger.info(f"[ssh_discovery] {found.ip} ({hostname}) - ADDED from slave {slave_identifier}") - result.created += 1 - result.actions.append(DeviceAction( - ip=found.ip, hostname=hostname, action="ADDED", - reason=f"slave {slave_identifier} plugin {found.plugin_name}", - category=found.category, role=found.role, - )) - else: - result.skipped += 1 - result.actions.append(DeviceAction( - ip=found.ip, hostname=existing.hostname or hostname, action="SKIPPED", - reason="already exists", category=existing.category, role=existing.role, - )) - except Exception as exc: - result.errors.append(f"Error upserting slave device {found.ip}: {exc}") - try: - await db.flush() - except Exception as exc: - result.errors.append(f"Flush error after slave {candidate_ip}: {exc}") - - # Step 3: Discover MikroTik via SSH_CLIENT - await _emit("disc_phase", {"phase": "mikrotik"}) - await _emit("disc_action", {"message": "Поиск MikroTik (SSH_CLIENT)..."}) - logger.info(f"[ssh_discovery] Step 3: Attempting to discover MikroTik via SSH_CLIENT on {obj.server_ip or 'main_server'}") - mikrotik_ip = await _discover_mikrotik_multi( - host=obj.server_ip or rack_hosts[0][1], - port=ssh_port, - options=ssh_options, - ) - if mikrotik_ip: - logger.info(f"[ssh_discovery] MikroTik found via SSH_CLIENT at {mikrotik_ip}") - await _emit("disc_infra_found", {"type": "mikrotik", "ip": mikrotik_ip}) - try: - existing = (await db.execute( - select(Device).where( - Device.object_id == obj.id, - Device.ip == mikrotik_ip, - ) - )).scalar_one_or_none() - - if existing is not None: - logger.info(f"[ssh_discovery] {mikrotik_ip} (mikrotik) - SKIPPED: already exists (source={existing.source}, category={existing.category}, hostname={existing.hostname})") - result.skipped += 1 - result.actions.append(DeviceAction( - ip=mikrotik_ip, - hostname="mikrotik", - action="SKIPPED", - reason=f"already exists (source={existing.source}, category={existing.category}, hostname={existing.hostname})", - category=existing.category, - role=existing.role, - )) - else: - db.add(Device( - object_id=obj.id, - ip=mikrotik_ip, - hostname="mikrotik", - category="router", - role="other", - source="auto_ssh", - )) - logger.info(f"[ssh_discovery] {mikrotik_ip} (mikrotik) - ADDED: category=router, role=other, hostname=mikrotik, source=auto_ssh") - result.created += 1 - result.actions.append(DeviceAction( - ip=mikrotik_ip, - hostname="mikrotik", - action="ADDED", - reason="discovered via SSH_CLIENT", - category="router", - role="other", - )) - except Exception as exc: - logger.error(f"[ssh_discovery] {mikrotik_ip} (mikrotik) - ERROR: {exc}") - result.errors.append(f"Error upserting {mikrotik_ip}: {exc}") - result.actions.append(DeviceAction( - ip=mikrotik_ip, - hostname="mikrotik", - action="ERROR", - reason=str(exc), - category="router", - role="other", - )) - - await db.flush() - else: - logger.info(f"[ssh_discovery] MikroTik: No IP found via SSH_CLIENT (reason: SSH_CLIENT env var not available or parsing failed)") - - # Step 4: Discover Proxmox via port 8006 scan - await _emit("disc_phase", {"phase": "proxmox"}) - await _emit("disc_action", {"message": "Сканирование порта 8006 (Proxmox)..."}) - logger.info(f"[ssh_discovery] Step 4: Attempting to discover Proxmox servers via port 8006 scan on {obj.server_ip or 'main_server'}") - proxmox_ips = await _discover_proxmox_via_port_scan( - host=obj.server_ip or rack_hosts[0][1], - port=obj.ssh_port or 22, - username=obj.ssh_user, - password=ssh_password, - ) - if proxmox_ips: - logger.info(f"[ssh_discovery] Proxmox discovery found {len(proxmox_ips)} server(s) on port 8006: {proxmox_ips}") - for proxmox_ip in proxmox_ips: - await _emit("disc_infra_found", {"type": "proxmox", "ip": proxmox_ip}) - for idx, proxmox_ip in enumerate(proxmox_ips, 1): - hostname = f"proxmox_{idx}" - try: - existing = (await db.execute( - select(Device).where( - Device.object_id == obj.id, - Device.ip == proxmox_ip, - ) - )).scalar_one_or_none() - - if existing is not None: - logger.info(f"[ssh_discovery] {proxmox_ip} ({hostname}) - SKIPPED: already exists (source={existing.source}, category={existing.category}, hostname={existing.hostname})") - result.skipped += 1 - result.actions.append(DeviceAction( - ip=proxmox_ip, - hostname=hostname, - action="SKIPPED", - reason=f"already exists (source={existing.source}, category={existing.category}, hostname={existing.hostname})", - category=existing.category, - role=existing.role, - )) - else: - db.add(Device( - object_id=obj.id, - ip=proxmox_ip, - hostname=hostname, - category="vm", - role="other", - source="auto_ssh", - )) - logger.info(f"[ssh_discovery] {proxmox_ip} ({hostname}) - ADDED: category=vm, role=other, hostname={hostname}, source=auto_ssh") - result.created += 1 - result.actions.append(DeviceAction( - ip=proxmox_ip, - hostname=hostname, - action="ADDED", - reason="discovered via port 8006 scan", - category="vm", - role="other", - )) - except Exception as exc: - logger.error(f"[ssh_discovery] {proxmox_ip} ({hostname}) - ERROR: {exc}") - result.errors.append(f"Error upserting {proxmox_ip}: {exc}") - result.actions.append(DeviceAction( - ip=proxmox_ip, - hostname=hostname, - action="ERROR", - reason=str(exc), - category="vm", - role="other", - )) - - await db.flush() - else: - logger.info(f"[ssh_discovery] Proxmox: No servers found via port 8006 scan (reason: no hosts responding on port 8006)") - - # ── Summary Report ───────────────────────────────────────────────────────────── - logger.info("=" * 100) - logger.info(f"[ssh_discovery] SUMMARY: ADDED={result.created}, UPDATED={result.updated}, SKIPPED={result.skipped}, ERRORS={len(result.errors)}") - logger.info(f"[ssh_discovery] Total actions collected: {len(result.actions)}") - logger.info("=" * 100) - - # Group actions by status - added = [a for a in result.actions if a.action == "ADDED"] - updated = [a for a in result.actions if a.action == "UPDATED"] - skipped = [a for a in result.actions if a.action == "SKIPPED"] - errors = [a for a in result.actions if a.action == "ERROR"] - - logger.info(f"[ssh_discovery] Actions by type: ADDED={len(added)}, UPDATED={len(updated)}, SKIPPED={len(skipped)}, ERROR={len(errors)}") - - if added: - logger.info("[ssh_discovery] ─── ADDED devices ───") - for action in added: - logger.info(f" {action.ip:20} ({action.hostname:30}) | {action.category:15} | {action.role:10} | {action.reason}") - - if updated: - logger.info("[ssh_discovery] ─── UPDATED devices ───") - for action in updated: - logger.info(f" {action.ip:20} ({action.hostname:30}) | {action.category:15} | {action.role:10} | {action.reason}") - - if skipped: - logger.info("[ssh_discovery] ─── SKIPPED devices ───") - for action in skipped: - logger.info(f" {action.ip:20} ({action.hostname:30}) | {action.category:15} | {action.role:10} | {action.reason}") - - if errors: - logger.error("[ssh_discovery] ─── ERROR devices ───") - for action in errors: - logger.error(f" {action.ip:20} ({action.hostname:30}) | {action.category:15} | {action.role:10} | {action.reason}") - - logger.info("=" * 100) - - return result +# Backward-compatibility shim — logic lives in app.services.discovery +from app.services.discovery import ( # noqa: F401 + DeviceAction, + DiscoveredDevice, + DiscoveryResult, + ProgressCallback, + discover_via_ssh, +) diff --git a/backend/app/services/discovery/__init__.py b/backend/app/services/discovery/__init__.py new file mode 100644 index 0000000..7060129 --- /dev/null +++ b/backend/app/services/discovery/__init__.py @@ -0,0 +1,20 @@ +""" +SSH-based device discovery package. + +Public API: + discover_via_ssh — run a full discovery pass for one Object + DiscoveryResult — summary of what was created/updated/skipped + DeviceAction — per-device action record + DiscoveredDevice — device parsed from a CAPS config plugin section +""" + +from .orchestrator import discover_via_ssh +from .types import DeviceAction, DiscoveredDevice, DiscoveryResult, ProgressCallback + +__all__ = [ + "discover_via_ssh", + "DiscoveryResult", + "DeviceAction", + "DiscoveredDevice", + "ProgressCallback", +] diff --git a/backend/app/services/discovery/config_parser.py b/backend/app/services/discovery/config_parser.py new file mode 100644 index 0000000..59fc399 --- /dev/null +++ b/backend/app/services/discovery/config_parser.py @@ -0,0 +1,178 @@ +""" +CAPS INI config parsing utilities. +All functions are pure (no network I/O, no DB). +""" + +import logging +import re +from configparser import RawConfigParser +from urllib.parse import urlparse + +from .types import DiscoveredDevice +from .uri import ( + _SKIP_PLUGIN_PREFIXES, + extract_device_info, + extract_ip, + cls_to_category, + cls_to_role, +) + +logger = logging.getLogger(__name__) + + +def parse_config( + config_text: str, + rack_host: str, + all_rack_hosts: list[str] | None = None, +) -> list[DiscoveredDevice]: + """Extract discovered network devices from a CAPS config file. + + Args: + config_text: Config file content. + rack_host: IP of the current rack PC (excluded from results). + all_rack_hosts: All rack IPs (excluded so rack servers aren't mistaken for devices). + """ + if all_rack_hosts is None: + all_rack_hosts = [rack_host] + + parser = RawConfigParser(strict=False) + parser.read_string(config_text) + + discovered: list[DiscoveredDevice] = [] + seen_ips: set[str] = set() + all_sections = parser.sections() + plugin_sections = [s for s in all_sections if s.startswith("plugin:")] + + logger.info( + "[discovery] Config parsing: %d sections total, %d plugins", + len(all_sections), len(plugin_sections), + ) + + for section in all_sections: + if not section.startswith("plugin:"): + continue + + plugin_name = section[len("plugin:"):] + if any(plugin_name.startswith(prefix) for prefix in _SKIP_PLUGIN_PREFIXES): + logger.info("[discovery] Plugin '%s' - IGNORED (skip list)", plugin_name) + continue + + cls = parser.get(section, "cls", fallback="") + if not cls: + logger.info("[discovery] Plugin '%s' - IGNORED (no 'cls' field)", plugin_name) + continue + + logger.info("[discovery] Plugin '%s' (cls=%s)", plugin_name, cls) + + # uri / frame_uri / host → actual device (camera, io_board, etc.) + device_uri_fields = ("uri", "frame_uri", "host") + found_ip_in_plugin = False + for field_name in device_uri_fields: + raw = parser.get(section, field_name, fallback=None) + if not raw: + logger.info("[discovery] %s: (not set)", field_name) + continue + logger.info("[discovery] %s='%s'", field_name, raw) + info = extract_device_info(raw) + if not info: + logger.info("[discovery] → IGNORED (not a valid network IP)") + continue + ip, http_user, http_pass = info + if ip in all_rack_hosts: + logger.info("[discovery] → IGNORED (is a rack server IP: %s)", ip) + continue + if ip in seen_ips: + logger.info("[discovery] → IGNORED (duplicate)") + continue + + seen_ips.add(ip) + cred_str = f", http_user={http_user}" if http_user else "" + logger.info( + "[discovery] → EXTRACTED %s (category=%s, role=%s%s)", + ip, cls_to_category(cls), cls_to_role(cls, plugin_name), cred_str, + ) + discovered.append(DiscoveredDevice( + ip=ip, + category=cls_to_category(cls), + role=cls_to_role(cls, plugin_name), + plugin_name=plugin_name, + cls=cls, + http_username=http_user, + http_password=http_pass, + )) + found_ip_in_plugin = True + + # stream_uri → RTSP server (a separate VM, not the camera itself) + stream_raw = parser.get(section, "stream_uri", fallback=None) + if stream_raw: + logger.info("[discovery] stream_uri='%s'", stream_raw) + stream_ip = extract_ip(stream_raw) + if stream_ip and stream_ip not in all_rack_hosts and stream_ip not in seen_ips: + seen_ips.add(stream_ip) + logger.info("[discovery] → EXTRACTED %s as RTSP server (category=vm)", stream_ip) + discovered.append(DiscoveredDevice( + ip=stream_ip, + category="vm", + role="other", + plugin_name=f"{plugin_name}_rtsp", + cls="rtsp_server", + )) + found_ip_in_plugin = True + + if not found_ip_in_plugin: + logger.info("[discovery] → No network devices found in this plugin") + + # [caps.voice] section → Asterisk server + voice_addr = parser.get("caps.voice", "server_addr", fallback=None) + if voice_addr: + logger.info("[discovery] [caps.voice] server_addr='%s'", voice_addr) + voice_ip = extract_ip(f"sip://{voice_addr}") or ( + voice_addr.strip() + if re.match(r"^\d{1,3}(\.\d{1,3}){3}$", voice_addr.strip()) + else None + ) + if voice_ip and voice_ip not in all_rack_hosts and voice_ip not in seen_ips: + seen_ips.add(voice_ip) + logger.info("[discovery] → EXTRACTED %s as Asterisk (category=vm)", voice_ip) + discovered.append(DiscoveredDevice( + ip=voice_ip, + category="vm", + role="other", + plugin_name="asterisk", + cls="asterisk", + )) + + return discovered + + +def parse_slave_check(config_text: str, master_ip: str, host: str) -> tuple[bool, str, str]: + """Check a parsed config against one master IP. No network I/O. + + Returns: + (is_slave, identifier, config_text) — config_text is passed through unchanged. + """ + parser = RawConfigParser(strict=False) + parser.read_string(config_text) + linking_cls = parser.get("plugin:linking", "cls", fallback="") + linking_uri = parser.get("plugin:linking", "uri", fallback="") + identifier = parser.get("caps.rack", "identifier", fallback="").strip() + if "secondary" in linking_cls.lower() and master_ip in linking_uri: + return True, identifier or host, config_text + return False, "", "" + + +def parse_db_uri(uri: str) -> dict | None: + """Parse postgresql+psycopg2://user:pass@host:port/dbname → dict.""" + try: + parsed = urlparse(uri) + if not parsed.hostname: + return None + return { + "host": parsed.hostname, + "port": parsed.port or 5432, + "user": parsed.username or "", + "password": parsed.password or "", + "dbname": (parsed.path or "").lstrip("/"), + } + except Exception: + return None diff --git a/backend/app/services/discovery/db_resolver.py b/backend/app/services/discovery/db_resolver.py new file mode 100644 index 0000000..d7b470d --- /dev/null +++ b/backend/app/services/discovery/db_resolver.py @@ -0,0 +1,103 @@ +""" +DB-related helpers for device discovery: + - Auto-detect the CAPS DB connection from a remote config file. + - Fetch rack host IPs from the CAPS PostgreSQL database. +""" + +import logging + +import psycopg + +from .config_parser import parse_db_uri +from .ssh_client import ssh_run_multi, open_ssh_tunnel + +logger = logging.getLogger(__name__) + + +async def autodetect_db_from_server( + server_ip: str, + ssh_port: int, + ssh_options: list, + server_config_path: str = "/etc/caps/config.ini", +) -> dict | None: + """SSH into server, read its caps config, extract DB connection params. + + Returns a dict with keys: host, port, user, password, dbname. + Returns None if no db_uri is found. + """ + from configparser import RawConfigParser + + try: + config_text = await ssh_run_multi( + server_ip, ssh_port, ssh_options, f"cat {server_config_path}" + ) + except Exception: + return None + + parser = RawConfigParser(strict=False) + parser.read_string(config_text) + + for section in list(parser.sections()) + ["DEFAULT"]: + try: + db_uri = parser.get(section, "db_uri", fallback=None) + if db_uri and db_uri.startswith("postgresql"): + return parse_db_uri(db_uri) + except Exception: + continue + + db_uri = parser.defaults().get("db_uri") + if db_uri and db_uri.startswith("postgresql"): + return parse_db_uri(db_uri) + + return None + + +async def get_rack_hosts( + host: str, + port: int, + user: str, + password: str, + dbname: str, + table: str, + via_tunnel: bool = False, + tunnel_host: str | None = None, + tunnel_ssh_port: int = 22, + ssh_options: list | None = 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) + async with conn: + listener = await conn.forward_local_port("127.0.0.1", 0, host, port) + local_port = listener.get_port() + dsn = ( + f"host=127.0.0.1 port={local_port} dbname={dbname} " + f"user={user} password={password} connect_timeout=10" + ) + return await _fetch_rack_rows(dsn, table) + else: + dsn = f"host={host} port={port} dbname={dbname} user={user} password={password} connect_timeout=10" + return await _fetch_rack_rows(dsn, table) + + +async def _fetch_rack_rows(dsn: str, table: str) -> list[tuple[str, str, str, str | None]]: + async with await psycopg.AsyncConnection.connect(dsn) as conn: + cursor = await conn.execute( + f"SELECT id, data->>'host' AS host, name, data->>'rack_type' AS rack_type " + f"FROM \"{table}\" WHERE data->>'host' IS NOT NULL" + ) + rows = await cursor.fetchall() + + result = [] + for r in rows: + if not r[1]: + continue + ext_id = str(r[0]) + short_name = str(r[2]) if r[2] else None + # Combine CAPS id with short name: "lane-1-1 (01)" + rack_name = f"{ext_id} ({short_name})" if short_name and short_name != ext_id else ext_id + result.append((ext_id, str(r[1]), rack_name, r[3] or None)) + return result diff --git a/backend/app/services/discovery/orchestrator.py b/backend/app/services/discovery/orchestrator.py new file mode 100644 index 0000000..5ede2e3 --- /dev/null +++ b/backend/app/services/discovery/orchestrator.py @@ -0,0 +1,927 @@ +""" +Main discovery orchestrator. + +Coordinates all phases of SSH-based device discovery: + 0. Resolve DB connection (explicit config or autodetect via SSH). + 0.5 Pre-upsert the DB server device (so users can configure tunnels on failure). + 1. Fetch rack hosts from the CAPS DB; sync Rack table entries. + 1b. Upsert infrastructure devices (main_server). + 2. SSH into each rack → parse CAPS config → upsert plugin devices. + 2.5 Discover slave racks via ARP neighbor tables. + 3. Discover MikroTik via SSH_CLIENT env var. + 4. Discover Proxmox via port 8006 scan. +""" + +import logging +from configparser import RawConfigParser +from datetime import datetime, timezone + +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession + +from app.models.device import Device +from app.models.object import Object +from app.models.rack import Rack +from app.services.crypto import decrypt, encrypt + +from .config_parser import parse_config, parse_slave_check +from .db_resolver import autodetect_db_from_server, get_rack_hosts +from .ssh_client import ( + discover_mikrotik_multi, + discover_proxmox_via_port_scan, + fetch_slave_config, + get_ip_neighbors_multi, + ssh_run_multi, +) +from .types import DeviceAction, DiscoveredDevice, DiscoveryResult, ProgressCallback + +logger = logging.getLogger(__name__) + + +# ── Public entry point ──────────────────────────────────────────────────────── + +async def discover_via_ssh( + db: AsyncSession, + obj: Object, + config_path: str = "/etc/caps/config.ini", + progress_cb: ProgressCallback | None = None, +) -> DiscoveryResult: + result = DiscoveryResult() + + async def _emit(event_type: str, data: dict) -> None: + if progress_cb is not None: + await progress_cb(event_type, data) + + from app.services.ssh import build_object_auth_options + ssh_options = await build_object_auth_options(obj, db) + + if not ssh_options: + result.errors.append("SSH credentials not configured on this object") + return result + + ssh_port = obj.ssh_port or 22 + # First option's password is used by single-auth Proxmox scan (legacy). + _first = ssh_options[0] + ssh_password = _first.password if _first.type == "password" else "" + + # ── Phase 0: Resolve DB connection ──────────────────────────────────────── + db_params = await _resolve_db_connection(obj, ssh_port, ssh_options, result, _emit) + if db_params is None: + return result + + db_host, db_port, db_name, db_user, db_password, db_table, server_ip_ssh_verified = db_params + + # ── Phase 0.5: Pre-upsert DB server device ──────────────────────────────── + now = datetime.now(timezone.utc) + db_device_obj, use_tunnel = await _pre_upsert_db_device( + db, obj, db_host, result, now + ) + + # ── Phase 1: Fetch rack hosts ────────────────────────────────────────────── + await _emit("disc_action", {"message": "Получение списка стоек из БД..."}) + logger.info( + "[discovery] Connecting to object DB: host=%s port=%s dbname=%s user=%s table=%s via_tunnel=%s", + db_host, db_port, db_name, db_user, db_table, use_tunnel, + ) + try: + rack_hosts = await get_rack_hosts( + host=db_host, port=db_port, user=db_user, + password=db_password or "", dbname=db_name, table=db_table, + via_tunnel=use_tunnel, + tunnel_host=obj.server_ip, tunnel_ssh_port=ssh_port, + ssh_options=ssh_options, + ) + if db_device_obj is not None: + db_device_obj.status = "online" + db_device_obj.last_seen = now + except Exception as exc: + if db_device_obj is not None: + db_device_obj.status = "offline" + err_msg = f"Cannot fetch rack hosts from object DB: {exc}" + if "timeout" in str(exc).lower(): + err_msg += ". Попробуйте настроить SSH-туннель (отредактируйте устройство БД и включите опцию)" + result.errors.append(err_msg) + return result + + if not rack_hosts: + result.errors.append("No racks with host IPs found in object DB") + return result + + await _emit("disc_server", {"rack_count": len(rack_hosts)}) + + # ── Phase 1a: Sync Rack table entries ────────────────────────────────────── + rack_id_mapping = await _sync_rack_entries(db, obj, rack_hosts) + + # ── Phase 1b: Upsert infrastructure devices ─────────────────────────────── + infra_devices = [] + if obj.server_ip: + infra_devices.append((obj.server_ip, "main_server", "caps_server", server_ip_ssh_verified)) + await _upsert_infra_devices(db, obj, infra_devices, result, now) + + # ── Phase 2: Process racks ──────────────────────────────────────────────── + neighbor_candidates: set[str] = set() + discovered_device_ips: set[str] = set() + + for rack_idx, (external_id, rack_host, rack_name, _rack_type_db) in enumerate(rack_hosts): + await _emit("disc_rack_start", { + "rack_name": rack_name, "rack_host": rack_host, + "index": rack_idx, "total": len(rack_hosts), + }) + + # Allow per-device config_path override stored in connection_params. + effective_config_path = await _resolve_rack_config_path( + db, obj, rack_host, config_path + ) + + rack_ssh_ok, config_text = await _ssh_fetch_rack_config( + rack_host, ssh_port, ssh_options, effective_config_path, rack_name, result, _emit + ) + + await _upsert_rack_device( + db, obj, external_id, rack_host, rack_name, + rack_id_mapping, rack_ssh_ok, now, result + ) + + if not rack_ssh_ok: + try: + await db.flush() + except Exception as exc: + result.errors.append(f"Flush error for offline rack {rack_host}: {exc}") + await _emit("disc_rack_done", { + "rack_name": rack_name, "rack_host": rack_host, "found": 0, "ssh_ok": False + }) + continue + + neigh_ips = await get_ip_neighbors_multi(rack_host, ssh_port, ssh_options) + neighbor_candidates.update(neigh_ips) + discovered_device_ips.add(rack_host) + + all_rack_ips = [h for _, h, _, _ in rack_hosts] + await _emit("disc_action", {"message": f"Парсинг конфигурации {rack_name}..."}) + devices_found = parse_config(config_text, rack_host, all_rack_ips) + logger.info("[discovery] Found %d devices from plugins in %s", len(devices_found), rack_host) + + for found in devices_found: + await _upsert_plugin_device( + db, obj, found, external_id, rack_id_mapping, result + ) + discovered_device_ips.add(found.ip) + + try: + await db.flush() + except Exception as exc: + result.errors.append(f"Flush error after rack {rack_host}: {exc}") + await _emit("disc_rack_done", { + "rack_name": rack_name, "rack_host": rack_host, + "found": len(devices_found), "ssh_ok": True, + }) + + # ── Phase 2.5: Slave rack discovery via ARP ─────────────────────────────── + all_rack_ips_set = {h for _, h, _, _ in rack_hosts} + infra_ips = {ip for ip in [obj.server_ip, db_host] if ip} + common_gateways = {ip for ip in neighbor_candidates if ip.endswith(".1") or ip.endswith(".254")} + slave_candidates = ( + neighbor_candidates - all_rack_ips_set - discovered_device_ips - infra_ips - common_gateways + ) + + await _emit("disc_phase", {"phase": "slaves", "candidate_count": len(slave_candidates)}) + if slave_candidates: + logger.info("[discovery] Phase 2.5: Slave discovery — %d candidate(s)", len(slave_candidates)) + await _emit("disc_action", { + "message": f"Поиск слейв-стоек: проверяем {len(slave_candidates)} ARP-кандидатов..." + }) + else: + await _emit("disc_action", {"message": "Слейв-стойки: ARP-кандидатов не найдено"}) + + await _discover_slave_racks( + db, obj, slave_candidates, rack_hosts, ssh_port, ssh_options, + config_path, all_rack_ips_set, now, result, _emit + ) + + # ── Phase 3: MikroTik via SSH_CLIENT ────────────────────────────────────── + await _emit("disc_phase", {"phase": "mikrotik"}) + await _emit("disc_action", {"message": "Поиск MikroTik (SSH_CLIENT)..."}) + await _discover_mikrotik(db, obj, rack_hosts, ssh_port, ssh_options, result, _emit) + + # ── Phase 4: Proxmox via port 8006 scan ─────────────────────────────────── + await _emit("disc_phase", {"phase": "proxmox"}) + await _emit("disc_action", {"message": "Сканирование порта 8006 (Proxmox)..."}) + await _discover_proxmox(db, obj, ssh_port, ssh_password, rack_hosts, result, _emit) + + # ── Summary ─────────────────────────────────────────────────────────────── + _log_summary(result) + return result + + +# ── Phase helpers ───────────────────────────────────────────────────────────── + +async def _resolve_db_connection( + obj: Object, + ssh_port: int, + ssh_options: list, + result: DiscoveryResult, + _emit, +) -> tuple | None: + """Determine the DB connection parameters. + + Returns (db_host, db_port, db_name, db_user, db_password, db_table, server_ip_ssh_verified) + or None if resolution failed (errors added to result). + """ + db_host = obj.db_host + db_port = obj.db_port or 5432 + db_name = obj.db_name + db_user = obj.db_user + db_password: str | None = None + db_table = obj.db_table + server_ip_ssh_verified = False + + if db_host and db_name and db_user and obj.db_pass_enc and db_table: + try: + db_password = decrypt(obj.db_pass_enc) + except Exception: + result.errors.append("Failed to decrypt DB password") + return None + elif obj.server_ip: + await _emit("disc_action", {"message": f"Чтение конфигурации сервера {obj.server_ip}..."}) + detected = await autodetect_db_from_server( + server_ip=obj.server_ip, ssh_port=ssh_port, ssh_options=ssh_options, + ) + if not detected or not all([detected.get("host"), detected.get("user"), detected.get("dbname")]): + await _emit_db_autodetect_error(obj, ssh_port, ssh_options, result) + return None + + db_host = detected["host"] + db_port = detected["port"] + db_user = detected["user"] + db_password = detected["password"] + db_name = detected["dbname"] + db_table = db_table or obj.db_table or "racks" + server_ip_ssh_verified = True + else: + result.errors.append( + "DB connection not configured. Either fill in the DB fields on the object " + "or set server_ip so the DB can be auto-detected via SSH." + ) + return None + + return db_host, db_port, db_name, db_user, db_password, db_table, server_ip_ssh_verified + + +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") + diag_parser = RawConfigParser(strict=False) + diag_parser.read_string(raw) + found_sections = diag_parser.sections() + found_keys = {s: list(diag_parser.options(s)) for s in found_sections} + result.errors.append( + f"db_uri not found in server config. " + f"Sections found: {found_sections}. " + f"Keys per section: {found_keys}. " + f"Fill in DB fields manually on the object." + ) + except Exception as diag_exc: + result.errors.append( + f"Could not read server config for auto-detection: {diag_exc}. " + f"Fill in DB fields manually on the object." + ) + + +async def _pre_upsert_db_device( + db: AsyncSession, + obj: Object, + db_host: str | None, + result: DiscoveryResult, + now: datetime, +) -> tuple: + """Upsert the CAPS DB server as a device before the DB connection attempt. + + Returns (db_device_obj, use_tunnel). + """ + db_device_obj: Device | None = None + if db_host and db_host != obj.server_ip: + try: + db_device_obj = (await db.execute( + select(Device).where(Device.object_id == obj.id, Device.ip == db_host) + )).scalar_one_or_none() + if db_device_obj is None: + db_device_obj = Device( + object_id=obj.id, ip=db_host, hostname="db_server", + category="vm", role="other", source="auto_ssh", + location="server_room", status="unknown", + device_meta={"is_caps_db": True}, + ) + db.add(db_device_obj) + await db.flush() + logger.info("[discovery] %s (db_server) - PRE-ADDED before connection attempt", db_host) + result.created += 1 + result.actions.append(DeviceAction( + ip=db_host, hostname="db_server", action="ADDED", + reason="object infrastructure device (vm/db_server)", + category="vm", role="other", + )) + except Exception as exc: + logger.warning("[discovery] Failed to pre-upsert db_server %s: %s", db_host, exc) + + device_via_tunnel = bool( + (db_device_obj.connection_params or {}).get("db_via_tunnel") + ) if db_device_obj else False + use_tunnel = device_via_tunnel or obj.db_via_tunnel + + return db_device_obj, use_tunnel + + +async def _sync_rack_entries( + db: AsyncSession, + obj: Object, + rack_hosts: list[tuple], +) -> dict[str, int]: + """Ensure a Rack row exists for every rack returned from the CAPS DB. + + Returns a mapping of external_id → Rack.id. + """ + rack_id_mapping: dict[str, int] = {} + for ext_id, _, rack_name, rack_type_db in rack_hosts: + existing = (await db.execute( + select(Rack).where( + Rack.object_id == obj.id, + Rack.name.in_([ext_id, rack_name]), + ) + )).scalar_one_or_none() + if existing: + rack_id_mapping[ext_id] = existing.id + if existing.name != rack_name: + logger.info("[discovery] Renamed rack '%s' → '%s'", existing.name, rack_name) + existing.name = rack_name + if rack_type_db and not existing.rack_type: + existing.rack_type = rack_type_db + else: + new_rack = Rack(object_id=obj.id, name=rack_name, rack_type=rack_type_db) + db.add(new_rack) + await db.flush() + rack_id_mapping[ext_id] = new_rack.id + logger.info("[discovery] Created rack '%s' rack_type=%s (id=%d)", rack_name, rack_type_db, new_rack.id) + return rack_id_mapping + + +async def _upsert_infra_devices( + db: AsyncSession, + obj: Object, + infra_devices: list[tuple], + result: DiscoveryResult, + now: datetime, +) -> None: + for infra_ip, infra_category, infra_hostname, infra_verified in infra_devices: + try: + existing = (await db.execute( + select(Device).where(Device.object_id == obj.id, Device.ip == infra_ip) + )).scalar_one_or_none() + + if existing is None: + db.add(Device( + object_id=obj.id, ip=infra_ip, hostname=infra_hostname, + category=infra_category, role="other", source="auto_ssh", + location="server_room", + status="online" if infra_verified else "unknown", + last_seen=now if infra_verified else None, + )) + logger.info( + "[discovery] %s (%s) - ADDED: category=%s, status=%s", + infra_ip, infra_hostname, infra_category, + "online" if infra_verified else "unknown", + ) + result.created += 1 + result.actions.append(DeviceAction( + ip=infra_ip, hostname=infra_hostname, action="ADDED", + reason=f"object infrastructure device ({infra_category})", + category=infra_category, role="other", + )) + else: + changes = [] + if infra_verified: + if existing.status != "online": + existing.status = "online" + changes.append("status: →online") + existing.last_seen = now + changes.append(f"last_seen: →{now.isoformat()}") + if changes: + logger.info("[discovery] %s (%s) - UPDATED: %s", infra_ip, infra_hostname, ", ".join(changes)) + result.updated += 1 + result.actions.append(DeviceAction( + ip=infra_ip, hostname=existing.hostname or infra_hostname, + action="UPDATED", reason="; ".join(changes), + category=existing.category, role=existing.role, + )) + else: + logger.info("[discovery] %s (%s) - SKIPPED: already up to date", infra_ip, infra_hostname) + result.skipped += 1 + result.actions.append(DeviceAction( + ip=infra_ip, hostname=existing.hostname or infra_hostname, + action="SKIPPED", reason=f"already exists (category={existing.category})", + category=existing.category, role=existing.role, + )) + except Exception as exc: + logger.error("[discovery] %s (%s) - ERROR: %s", infra_ip, infra_hostname, exc) + result.errors.append(f"Error upserting infra device {infra_ip}: {exc}") + + if infra_devices: + await db.flush() + + +async def _resolve_rack_config_path( + db: AsyncSession, obj: Object, rack_host: str, default_path: str +) -> str: + """Return the config path to use for this rack, honouring any per-device override.""" + existing = (await db.execute( + select(Device).where(Device.object_id == obj.id, Device.ip == rack_host) + )).scalar_one_or_none() + if existing and existing.connection_params: + return existing.connection_params.get("config_path", default_path) + return default_path + + +async def _ssh_fetch_rack_config( + rack_host: str, + ssh_port: int, + ssh_options: list, + config_path: str, + rack_name: str, + result: DiscoveryResult, + _emit, +) -> tuple[bool, str | None]: + """Attempt to SSH into rack_host and read its config file. + + Returns (ssh_ok, config_text). + """ + await _emit("disc_action", {"message": f"SSH → {rack_host} ({rack_name})"}) + try: + config_text = await ssh_run_multi(rack_host, ssh_port, ssh_options, f"cat {config_path}") + return True, config_text + except Exception as exc: + result.errors.append(f"SSH to {rack_host} failed: {exc}") + return False, None + + +async def _upsert_rack_device( + db: AsyncSession, + obj: Object, + external_id: str, + rack_host: str, + rack_name: str, + rack_id_mapping: dict, + rack_ssh_ok: bool, + now: datetime, + result: DiscoveryResult, +) -> None: + """Upsert the rack PC itself as an embedded device.""" + rack_hostname = external_id + logger.info( + "[discovery] Processing rack %s (%s), ssh_ok=%s", + rack_host, rack_hostname, rack_ssh_ok, + ) + try: + existing = (await db.execute( + select(Device).where(Device.object_id == obj.id, Device.ip == rack_host) + )).scalar_one_or_none() + + if existing is None: + status = "online" if rack_ssh_ok else "unknown" + db.add(Device( + object_id=obj.id, ip=rack_host, hostname=rack_hostname, + category="embedded", role="other", source="auto_ssh", + rack_id=rack_id_mapping.get(external_id), + status=status, last_seen=now if rack_ssh_ok else None, + )) + logger.info("[discovery] %s (%s) - ADDED: status=%s", rack_host, rack_hostname, status) + result.created += 1 + result.actions.append(DeviceAction( + ip=rack_host, hostname=rack_hostname, action="ADDED", + reason="rack PC from racks table" + (" (SSH ok)" if rack_ssh_ok else " (SSH failed, added as unknown)"), + category="embedded", role="other", + )) + else: + changes = [] + if rack_ssh_ok: + if existing.status != "online": + existing.status = "online" + changes.append("status: →online") + existing.last_seen = now + changes.append("last_seen: updated") + if existing.rack_id is None and rack_id_mapping.get(external_id) is not None: + existing.rack_id = rack_id_mapping[external_id] + changes.append(f"rack_id: None→{existing.rack_id}") + if changes: + logger.info("[discovery] %s (%s) - UPDATED: %s", rack_host, existing.hostname, ", ".join(changes)) + result.updated += 1 + result.actions.append(DeviceAction( + ip=rack_host, hostname=existing.hostname, action="UPDATED", + reason="; ".join(changes), category=existing.category, role=existing.role, + )) + else: + reason = "SSH failed, rack already in DB" if not rack_ssh_ok else "rack already up to date" + logger.info("[discovery] %s (%s) - SKIPPED: %s", rack_host, existing.hostname, reason) + result.skipped += 1 + result.actions.append(DeviceAction( + ip=rack_host, hostname=existing.hostname, action="SKIPPED", + reason=reason, category=existing.category, role=existing.role, + )) + except Exception as exc: + logger.error("[discovery] %s - ERROR: %s", rack_host, exc) + result.errors.append(f"Error processing rack {rack_host}: {exc}") + result.actions.append(DeviceAction( + ip=rack_host, hostname=rack_hostname, action="ERROR", reason=str(exc), + category="embedded", role="other", + )) + + +async def _upsert_plugin_device( + db: AsyncSession, + obj: Object, + found: DiscoveredDevice, + external_id: str, + rack_id_mapping: dict, + result: DiscoveryResult, + slave_rack_id: int | None = None, + slave_identifier: str | None = None, +) -> None: + """Upsert a single device discovered from a plugin section. + + Works for both primary rack devices and slave rack devices. + When slave_rack_id / slave_identifier are given, the device is treated as + belonging to that slave rack instead of the primary rack keyed by external_id. + """ + is_slave = slave_identifier is not None + is_shared = found.category == "vm" + + if is_shared: + hostname = found.plugin_name + rack_id: int | None = None + device_location: str | None = "server_room" + elif is_slave: + hostname = f"{slave_identifier}-{found.plugin_name}" + rack_id = slave_rack_id + device_location = None + else: + hostname = f"{external_id}-{found.plugin_name}" + rack_id = rack_id_mapping.get(external_id) + device_location = None + + try: + existing = (await db.execute( + select(Device).where(Device.object_id == obj.id, Device.ip == found.ip) + )).scalar_one_or_none() + + if existing is not None: + # Protect embedded (rack PC) devices from being overwritten by plugin devices. + if existing.category == "embedded": + logger.info( + "[discovery] %s (%s) - SKIPPED: device is rack server (embedded)", + found.ip, found.plugin_name, + ) + result.skipped += 1 + result.actions.append(DeviceAction( + ip=found.ip, hostname=existing.hostname, action="SKIPPED", + reason="device is rack server (embedded), cannot overwrite with plugin device", + category=existing.category, role=existing.role, + )) + return + + changes = [] + if existing.category != found.category and existing.source == "auto_ssh": + changes.append(f"category: {existing.category}→{found.category}") + existing.category = found.category + if existing.role == "other" and found.role != "other": + changes.append(f"role: other→{found.role}") + existing.role = found.role + if not is_shared: + if not existing.hostname or existing.hostname.startswith("rack"): + if existing.hostname != hostname: + changes.append(f"hostname: '{existing.hostname}'→'{hostname}'") + existing.hostname = hostname + if existing.rack_id is None and rack_id is not None: + existing.rack_id = rack_id + changes.append(f"rack_id: None→{rack_id}") + + if changes: + logger.info("[discovery] %s (%s) - UPDATED: %s", found.ip, found.plugin_name, ", ".join(changes)) + result.updated += 1 + result.actions.append(DeviceAction( + ip=found.ip, hostname=existing.hostname, action="UPDATED", + reason="; ".join(changes), category=found.category, role=found.role, + )) + else: + logger.info( + "[discovery] %s (%s) - SKIPPED: already exists with same values", + found.ip, found.plugin_name, + ) + result.skipped += 1 + result.actions.append(DeviceAction( + ip=found.ip, hostname=existing.hostname, action="SKIPPED", + reason=f"already exists (source={existing.source}, category={existing.category})", + category=existing.category, role=existing.role, + )) + else: + conn_params: dict = {} + if found.http_username: + conn_params["username"] = found.http_username + conn_params["password_enc"] = encrypt(found.http_password) if found.http_password else "" + db.add(Device( + object_id=obj.id, ip=found.ip, hostname=hostname, + category=found.category, role=found.role, source="auto_ssh", + rack_id=rack_id, location=device_location, + device_meta={"plugin": found.plugin_name, "cls": found.cls}, + connection_params=conn_params, + )) + cred_note = f", http_user={found.http_username}" if found.http_username else "" + logger.info( + "[discovery] %s (%s) - ADDED: category=%s, role=%s, rack_id=%s%s", + found.ip, found.plugin_name, found.category, found.role, rack_id, cred_note, + ) + result.created += 1 + result.actions.append(DeviceAction( + ip=found.ip, hostname=hostname, action="ADDED", + reason=f"new device from plugin {found.plugin_name}", + category=found.category, role=found.role, + )) + except Exception as exc: + logger.error("[discovery] %s (%s) - ERROR: %s", found.ip, found.plugin_name, exc) + result.errors.append(f"Error upserting {found.ip}: {exc}") + result.actions.append(DeviceAction( + ip=found.ip, hostname=hostname, action="ERROR", reason=str(exc), + category=found.category, role=found.role, + )) + + +async def _discover_slave_racks( + db: AsyncSession, + obj: Object, + slave_candidates: set[str], + rack_hosts: list[tuple], + ssh_port: int, + ssh_options: list, + config_path: str, + all_rack_ips_set: set[str], + now: datetime, + result: DiscoveryResult, + _emit, +) -> None: + for candidate_ip in slave_candidates: + logger.info("[discovery] Checking slave candidate %s...", candidate_ip) + await _emit("disc_action", {"message": f"Проверяем слейв-кандидата {candidate_ip}..."}) + + slave_config = await fetch_slave_config(candidate_ip, ssh_port, ssh_options, config_path) + is_slave = False + slave_identifier = "" + master_ip_for_slave = "" + + if slave_config is None: + logger.info("[discovery] %s — could not read config, skipping", candidate_ip) + else: + for _, master_host, _, _ in rack_hosts: + ok, _, _ = parse_slave_check(slave_config, master_host, candidate_ip) + if ok: + is_slave = True + master_ip_for_slave = master_host + master_ext_id = next( + (ext_id for ext_id, h, _, _ in rack_hosts if h == master_host), + master_host, + ) + slave_identifier = f"{master_ext_id}-slave" + break + + if not is_slave: + logger.info("[discovery] %s — not a slave rack", candidate_ip) + continue + + logger.info("[discovery] %s — SLAVE confirmed (name=%s, master=%s)", candidate_ip, slave_identifier, master_ip_for_slave) + await _emit("disc_action", {"message": f"Слейв подтверждён: {slave_identifier} ({candidate_ip})"}) + + slave_rack_id = await _ensure_slave_rack(db, obj, slave_identifier, result) + + await _upsert_slave_device(db, obj, candidate_ip, slave_identifier, slave_rack_id, master_ip_for_slave, now, result) + + all_rack_ips_with_slave = list(all_rack_ips_set) + [candidate_ip] + slave_devices = parse_config(slave_config, candidate_ip, all_rack_ips_with_slave) + logger.info("[discovery] Slave %s: found %d plugin devices", candidate_ip, len(slave_devices)) + await _emit("disc_slave_found", { + "rack_name": slave_identifier, "ip": candidate_ip, "found": len(slave_devices) + }) + + for found in slave_devices: + await _upsert_plugin_device( + db, obj, found, + external_id="", # not used for slaves (is_slave=True path) + rack_id_mapping={}, + result=result, + slave_rack_id=slave_rack_id, + slave_identifier=slave_identifier, + ) + + try: + await db.flush() + except Exception as exc: + result.errors.append(f"Flush error after slave {candidate_ip}: {exc}") + + +async def _ensure_slave_rack( + db: AsyncSession, obj: Object, slave_identifier: str, result: DiscoveryResult +) -> int | None: + """Ensure a Rack entry exists for the slave and return its ID.""" + try: + existing = (await db.execute( + select(Rack).where(Rack.object_id == obj.id, Rack.name == slave_identifier) + )).scalar_one_or_none() + if existing: + return existing.id + new_rack = Rack( + object_id=obj.id, name=slave_identifier, + description="slave (auto-discovered via ARP)", + ) + db.add(new_rack) + await db.flush() + logger.info("[discovery] Created rack '%s' (id=%d) for slave", slave_identifier, new_rack.id) + return new_rack.id + except Exception as exc: + result.errors.append(f"Error creating rack entry for slave: {exc}") + return None + + +async def _upsert_slave_device( + db: AsyncSession, + obj: Object, + candidate_ip: str, + slave_identifier: str, + slave_rack_id: int | None, + master_ip: str, + now: datetime, + result: DiscoveryResult, +) -> None: + try: + existing = (await db.execute( + select(Device).where(Device.object_id == obj.id, Device.ip == candidate_ip) + )).scalar_one_or_none() + if existing is None: + db.add(Device( + object_id=obj.id, ip=candidate_ip, hostname=slave_identifier, + category="embedded", role="other", source="auto_ssh", + rack_id=slave_rack_id, status="online", last_seen=now, + )) + logger.info("[discovery] %s (%s) - ADDED: slave rack PC", candidate_ip, slave_identifier) + result.created += 1 + result.actions.append(DeviceAction( + ip=candidate_ip, hostname=slave_identifier, action="ADDED", + reason=f"slave rack PC (master={master_ip})", + category="embedded", role="other", + )) + else: + changes = [] + if existing.status != "online": + existing.status = "online" + changes.append("status: →online") + existing.last_seen = now + changes.append("last_seen: updated") + if existing.rack_id is None and slave_rack_id: + existing.rack_id = slave_rack_id + changes.append(f"rack_id: →{slave_rack_id}") + result.updated += 1 + result.actions.append(DeviceAction( + ip=candidate_ip, hostname=existing.hostname or slave_identifier, + action="UPDATED", reason="; ".join(changes), + category="embedded", role="other", + )) + await db.flush() + except Exception as exc: + result.errors.append(f"Error upserting slave device {candidate_ip}: {exc}") + + +async def _discover_mikrotik( + db: AsyncSession, + obj: Object, + rack_hosts: list[tuple], + ssh_port: int, + ssh_options: list, + result: DiscoveryResult, + _emit, +) -> 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) + + if not mikrotik_ip: + logger.info("[discovery] MikroTik: no IP found via SSH_CLIENT") + return + + logger.info("[discovery] MikroTik found at %s", mikrotik_ip) + await _emit("disc_infra_found", {"type": "mikrotik", "ip": mikrotik_ip}) + try: + existing = (await db.execute( + select(Device).where(Device.object_id == obj.id, Device.ip == mikrotik_ip) + )).scalar_one_or_none() + if existing is not None: + logger.info("[discovery] %s (mikrotik) - SKIPPED: already exists", mikrotik_ip) + result.skipped += 1 + result.actions.append(DeviceAction( + ip=mikrotik_ip, hostname="mikrotik", action="SKIPPED", + reason=f"already exists (source={existing.source}, category={existing.category})", + category=existing.category, role=existing.role, + )) + else: + db.add(Device( + object_id=obj.id, ip=mikrotik_ip, hostname="mikrotik", + category="router", role="other", source="auto_ssh", + )) + logger.info("[discovery] %s (mikrotik) - ADDED: category=router", mikrotik_ip) + result.created += 1 + result.actions.append(DeviceAction( + ip=mikrotik_ip, hostname="mikrotik", action="ADDED", + reason="discovered via SSH_CLIENT", + category="router", role="other", + )) + except Exception as exc: + logger.error("[discovery] %s (mikrotik) - ERROR: %s", mikrotik_ip, exc) + result.errors.append(f"Error upserting {mikrotik_ip}: {exc}") + result.actions.append(DeviceAction( + ip=mikrotik_ip, hostname="mikrotik", action="ERROR", reason=str(exc), + category="router", role="other", + )) + await db.flush() + + +async def _discover_proxmox( + db: AsyncSession, + obj: Object, + ssh_port: int, + ssh_password: str, + rack_hosts: list[tuple], + result: DiscoveryResult, + _emit, +) -> None: + target_host = obj.server_ip or rack_hosts[0][1] + logger.info("[discovery] Phase 4: Proxmox port 8006 scan on %s", target_host) + proxmox_ips = await discover_proxmox_via_port_scan( + host=target_host, port=obj.ssh_port or 22, + username=obj.ssh_user, password=ssh_password, + ) + if not proxmox_ips: + logger.info("[discovery] Proxmox: no servers found on port 8006") + return + + logger.info("[discovery] Proxmox: found %d server(s): %s", len(proxmox_ips), proxmox_ips) + for proxmox_ip in proxmox_ips: + await _emit("disc_infra_found", {"type": "proxmox", "ip": proxmox_ip}) + + for idx, proxmox_ip in enumerate(proxmox_ips, 1): + hostname = f"proxmox_{idx}" + try: + existing = (await db.execute( + select(Device).where(Device.object_id == obj.id, Device.ip == proxmox_ip) + )).scalar_one_or_none() + if existing is not None: + logger.info("[discovery] %s (%s) - SKIPPED: already exists", proxmox_ip, hostname) + result.skipped += 1 + result.actions.append(DeviceAction( + ip=proxmox_ip, hostname=hostname, action="SKIPPED", + reason=f"already exists (source={existing.source}, category={existing.category})", + category=existing.category, role=existing.role, + )) + else: + db.add(Device( + object_id=obj.id, ip=proxmox_ip, hostname=hostname, + category="vm", role="other", source="auto_ssh", + )) + logger.info("[discovery] %s (%s) - ADDED: category=vm", proxmox_ip, hostname) + result.created += 1 + result.actions.append(DeviceAction( + ip=proxmox_ip, hostname=hostname, action="ADDED", + reason="discovered via port 8006 scan", + category="vm", role="other", + )) + except Exception as exc: + logger.error("[discovery] %s (%s) - ERROR: %s", proxmox_ip, hostname, exc) + result.errors.append(f"Error upserting {proxmox_ip}: {exc}") + result.actions.append(DeviceAction( + ip=proxmox_ip, hostname=hostname, action="ERROR", reason=str(exc), + category="vm", role="other", + )) + await db.flush() + + +# ── Summary logging ─────────────────────────────────────────────────────────── + +def _log_summary(result: DiscoveryResult) -> None: + sep = "=" * 100 + logger.info(sep) + logger.info( + "[discovery] SUMMARY: ADDED=%d, UPDATED=%d, SKIPPED=%d, ERRORS=%d", + result.created, result.updated, result.skipped, len(result.errors), + ) + by_action: dict[str, list[DeviceAction]] = {} + for action in result.actions: + by_action.setdefault(action.action, []).append(action) + + for label, actions in by_action.items(): + log_fn = logger.error if label == "ERROR" else logger.info + log_fn("[discovery] ─── %s devices ───", label) + for a in actions: + log_fn(" %-20s (%-30s) | %-15s | %-10s | %s", a.ip, a.hostname, a.category, a.role, a.reason) + logger.info(sep) diff --git a/backend/app/services/discovery/ssh_client.py b/backend/app/services/discovery/ssh_client.py new file mode 100644 index 0000000..06d726c --- /dev/null +++ b/backend/app/services/discovery/ssh_client.py @@ -0,0 +1,222 @@ +""" +Low-level SSH operations for device discovery. +No SQLAlchemy, no business logic — pure network I/O. +""" + +import asyncio +import logging +import re + +import asyncssh + +logger = logging.getLogger(__name__) + + +# ── Single connection helpers (used internally and for legacy paths) ────────── + +async def read_remote_file( + host: str, + port: int, + username: str, + password: str, + remote_path: str, +) -> str: + """SSH into host and return the contents of remote_path.""" + async with asyncssh.connect( + 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) + return result.stdout + + +async def get_ip_neighbors( + host: str, port: int, username: str, password: str, +) -> list[str]: + """Run `ip neigh show` and return REACHABLE/DELAY neighbour IPs.""" + try: + async with asyncssh.connect( + host, port=port, username=username, password=password, + known_hosts=None, connect_timeout=10, + ) as conn: + result = await conn.run("ip neigh show", check=False) + return _parse_neigh_output(result.stdout) + except Exception as exc: + logger.warning("[discovery] ip neigh failed on %s: %s", host, exc) + return [] + + +# ── Multi-auth helpers ──────────────────────────────────────────────────────── + +async def ssh_run_multi( + host: str, port: int, options: list, command: str, timeout: int = 15, +) -> str: + """Try each SSH auth option in order; return stdout on first success or raise.""" + last_exc: Exception = Exception("No auth options provided") + for i, opt in enumerate(options): + _log_attempt(i, len(options), host, opt) + try: + connect_kwargs = _build_connect_kwargs(host, port, opt) + async with asyncssh.connect(**connect_kwargs) as conn: + result = await asyncio.wait_for( + conn.run(command, check=True), timeout=timeout + ) + logger.info( + "[discovery] SSH auth succeeded for %s with %s(%s)", + host, opt.type, opt.username, + ) + return result.stdout + except Exception as exc: + last_exc = exc + logger.info( + "[discovery] SSH auth failed for %s with %s(%s): %s", + host, opt.type, opt.username, exc, + ) + raise last_exc + + +async def get_ip_neighbors_multi(host: str, port: int, options: list) -> list[str]: + """Multi-auth version of get_ip_neighbors.""" + try: + stdout = await ssh_run_multi(host, port, options, "ip neigh show", timeout=15) + 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: + """Discover MikroTik IP via SSH_CLIENT environment variable (multi-auth).""" + try: + stdout = await ssh_run_multi(host, port, options, "echo $SSH_CLIENT", timeout=15) + parts = stdout.strip().split() + return parts[0] if parts else None + except Exception as exc: + logger.warning("[discovery] Failed to discover MikroTik via SSH_CLIENT: %s", exc) + return None + + +async def fetch_slave_config( + host: str, port: int, options: list, config_path: str, +) -> str | None: + """Read a remote config file from a slave candidate. Returns None on failure.""" + try: + return await ssh_run_multi(host, port, options, f"cat {config_path}", timeout=15) + except Exception: + return None + + +async def discover_proxmox_via_port_scan( + host: str, port: int, username: str, password: str, +) -> list[str]: + """Discover Proxmox servers on the local subnet by scanning port 8006. + + Note: uses single-auth (password) because it's called from the legacy path. + Migrate to multi-auth when needed. + """ + try: + async with asyncssh.connect( + host, port=port, username=username, password=password, + known_hosts=None, connect_timeout=10, + ) as conn: + scan_script = ( + "subnet=$(ip -o -f inet addr show | awk '!/ lo |docker|br-|wg/ {print $4; exit}' | " + "cut -d/ -f1 | awk -F. '{print $1\".\"$2\".\"$3}') && " + "(set +m; for ip in $subnet.{1..254}; do " + "timeout 0.3 bash -c \"echo > /dev/tcp/$ip/8006\" 2>/dev/null && echo \"$ip\" & " + "done; wait)" + ) + result = await conn.run(scan_script, check=False) + return [ln.strip() for ln in result.stdout.strip().split("\n") if ln.strip()] + except Exception as exc: + logger.warning("[discovery] Proxmox port scan failed on %s: %s", host, exc) + return [] + + +# ── Tunnel ──────────────────────────────────────────────────────────────────── + +async def open_ssh_tunnel( + tunnel_host: str, + tunnel_port: int, + ssh_options: list, + target_host: str, + target_port: int, +): + """Open an asyncssh connection, trying each option in order. + + Returns an open asyncssh connection on first success, or raises. + The caller is responsible for closing the connection. + """ + last_exc: Exception = Exception("No SSH options provided") + for opt in ssh_options: + try: + connect_kwargs = _build_connect_kwargs(tunnel_host, tunnel_port, opt) + if opt.type == "key": + logger.info( + "[discovery] Tunnel attempt for %s — key: user=%s key_path=%s", + tunnel_host, opt.username, opt.key_path, + ) + else: + masked = _mask(opt.password) + logger.info( + "[discovery] Tunnel attempt for %s — password: user=%s password=%s", + tunnel_host, opt.username, masked, + ) + conn = await asyncssh.connect(**connect_kwargs) + logger.info( + "[discovery] Tunnel auth succeeded for %s with %s(%s)", + tunnel_host, opt.type, opt.username, + ) + return conn + except Exception as exc: + last_exc = exc + logger.info( + "[discovery] Tunnel auth failed for %s with %s(%s): %s", + tunnel_host, opt.type, opt.username, exc, + ) + raise last_exc + + +# ── Internal helpers ────────────────────────────────────────────────────────── + +def _parse_neigh_output(stdout: str) -> list[str]: + ips = [] + for line in stdout.strip().splitlines(): + parts = line.split() + if not parts: + continue + ip = parts[0] + state = parts[-1] + if re.match(r"^\d{1,3}(\.\d{1,3}){3}$", ip) and state in ("REACHABLE", "DELAY"): + ips.append(ip) + return ips + + +def _build_connect_kwargs(host: str, port: int, opt) -> dict: + kwargs: dict = dict( + host=host, port=port, username=opt.username, + known_hosts=None, connect_timeout=10, + ) + if opt.type == "key": + kwargs["client_keys"] = [opt.key_path] + kwargs["passphrase"] = None + else: + kwargs["password"] = opt.password + return kwargs + + +def _mask(password: str) -> str: + return (password[:1] + "***" + password[-1:]) if len(password) > 2 else "***" + + +def _log_attempt(idx: int, total: int, host: str, opt) -> None: + if opt.type == "key": + logger.info( + "[discovery] SSH attempt %d/%d for %s — key: user=%s key_path=%s", + idx + 1, total, host, opt.username, opt.key_path, + ) + else: + logger.info( + "[discovery] SSH attempt %d/%d for %s — password: user=%s password=%s", + idx + 1, total, host, opt.username, _mask(opt.password), + ) diff --git a/backend/app/services/discovery/types.py b/backend/app/services/discovery/types.py new file mode 100644 index 0000000..41e273d --- /dev/null +++ b/backend/app/services/discovery/types.py @@ -0,0 +1,42 @@ +""" +Shared dataclasses and type aliases for the discovery subsystem. +""" + +from collections.abc import Callable, Awaitable +from dataclasses import dataclass, field + + +@dataclass +class DiscoveredDevice: + """A network device extracted from a CAPS config plugin section.""" + ip: str + category: str + role: str + plugin_name: str + cls: str + http_username: str = "" # from URI credentials (e.g. http://user:pass@ip) + http_password: str = "" + + +@dataclass +class DeviceAction: + """Records what happened to one device during a discovery run.""" + ip: str + hostname: str + action: str # "ADDED", "UPDATED", "SKIPPED", "IGNORED", "ERROR" + reason: str + category: str = "" + role: str = "" + + +@dataclass +class DiscoveryResult: + created: int = 0 + updated: int = 0 + skipped: int = 0 + errors: list[str] = field(default_factory=list) + actions: list[DeviceAction] = field(default_factory=list) + + +# Async callback: (event_type, data_dict) → None +ProgressCallback = Callable[[str, dict], Awaitable[None]] diff --git a/backend/app/services/discovery/uri.py b/backend/app/services/discovery/uri.py new file mode 100644 index 0000000..8e9e7a5 --- /dev/null +++ b/backend/app/services/discovery/uri.py @@ -0,0 +1,68 @@ +""" +URI parsing and device classification utilities. +All functions are pure (no I/O, no DB). +""" + +import re +from urllib.parse import urlparse + + +_SERIAL_SCHEMES = {"serial", "com"} +_SERVICE_HOSTS = {"localhost", "127.0.0.1"} + +# Plugin name prefixes whose URIs point at the CAPS server itself, not separate devices. +_SKIP_PLUGIN_PREFIXES = ("mnemo", "processor", "recognition", "photo", "voice") + + +def extract_device_info(uri: str) -> tuple[str, str, str] | None: + """Return (ip, username, password) from a URI, or None if not a network device. + + Username and password are extracted from URI credentials (http://user:pass@ip). + They may be empty strings if not present in the URI. + """ + if not uri: + return None + try: + parsed = urlparse(uri) + scheme = (parsed.scheme or "").lower() + if scheme in _SERIAL_SCHEMES: + return None + hostname = parsed.hostname + if not hostname or hostname in _SERVICE_HOSTS: + return None + if re.match(r"^\d{1,3}(\.\d{1,3}){3}$", hostname): + return hostname, parsed.username or "", parsed.password or "" + except Exception: + pass + return None + + +def extract_ip(uri: str) -> str | None: + """Return IP from a URI string, or None if not a network device.""" + info = extract_device_info(uri) + return info[0] if info else None + + +def cls_to_category(cls: str) -> str: + cls_lower = cls.lower() + if "camera" in cls_lower or "facedetect" in cls_lower: + return "camera" + if "controller" in cls_lower or "ovenmk" in cls_lower or "rodos" in cls_lower: + return "io_board" + if "payment" in cls_lower or "sberpilot" in cls_lower or "payx" in cls_lower: + return "bank_terminal" + if "payonline" in cls_lower or "kkt" in cls_lower or "fiscal" in cls_lower: + return "cash_register" + return "other" + + +def cls_to_role(cls: str, plugin_name: str) -> str: + cls_lower = cls.lower() + name_lower = plugin_name.lower() + if "camera" in cls_lower or "camera" in name_lower: + if "grz" in name_lower or "plate" in name_lower or "lpr" in name_lower or "frame" in name_lower: + return "plate" + if "face" in name_lower or "lico" in name_lower: + return "face" + return "overview" + return "other"