1493 lines
66 KiB
Python
1493 lines
66 KiB
Python
"""
|
|
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 _check_if_slave_multi(
|
|
host: str, port: int, options: list, master_ip: str, config_path: str
|
|
) -> tuple[bool, str, str]:
|
|
"""Multi-auth version of _check_if_slave."""
|
|
try:
|
|
config_text = await _ssh_run_multi(host, port, options, f"cat {config_path}", timeout=15)
|
|
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 _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,
|
|
tunnel_ssh_user: str | None = None,
|
|
tunnel_ssh_password: str | 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 tunnel_ssh_user and tunnel_ssh_password:
|
|
async with asyncssh.connect(
|
|
tunnel_host,
|
|
port=tunnel_ssh_port,
|
|
username=tunnel_ssh_user,
|
|
password=tunnel_ssh_password,
|
|
known_hosts=None,
|
|
) as tunnel:
|
|
listener = await tunnel.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",
|
|
)
|
|
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,
|
|
tunnel_ssh_user=obj.ssh_user,
|
|
tunnel_ssh_password=ssh_password,
|
|
)
|
|
# 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}..."})
|
|
# Check each rack host that could be the master — slave's linking.uri must match
|
|
is_slave = False
|
|
slave_identifier = ""
|
|
slave_config_text = ""
|
|
master_ip_for_slave = ""
|
|
for _, master_host, _, _ in rack_hosts:
|
|
ok, identifier, slave_cfg = await _check_if_slave_multi(
|
|
host=candidate_ip,
|
|
port=ssh_port,
|
|
options=ssh_options,
|
|
master_ip=master_host,
|
|
config_path=config_path,
|
|
)
|
|
if ok:
|
|
is_slave = True
|
|
slave_config_text = slave_cfg
|
|
master_ip_for_slave = master_host
|
|
# Derive slave name from master's external_id: "lane-1-1" → "lane-1-1-slave"
|
|
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
|