1023 lines
43 KiB
Python
1023 lines
43 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 logging
|
|
import re
|
|
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
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# ── URI extraction ────────────────────────────────────────────────────────────
|
|
|
|
_SERIAL_SCHEMES = {"serial", "com"}
|
|
_SERVICE_HOSTS = {"localhost", "127.0.0.1"}
|
|
|
|
def _extract_ip(uri: str) -> str | None:
|
|
"""Return IP from a URI string, or None if not a network device."""
|
|
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
|
|
except Exception:
|
|
pass
|
|
return 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
|
|
|
|
|
|
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}'")
|
|
ip = _extract_ip(raw)
|
|
if not ip:
|
|
logger.info(f"[ssh_discovery] → IGNORED (not a valid network IP)")
|
|
continue
|
|
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)
|
|
logger.info(f"[ssh_discovery] → EXTRACTED {ip} (category={_cls_to_category(cls)}, role={_cls_to_role(cls, plugin_name)})")
|
|
discovered.append(DiscoveredDevice(
|
|
ip=ip,
|
|
category=_cls_to_category(cls),
|
|
role=_cls_to_role(cls, plugin_name),
|
|
plugin_name=plugin_name,
|
|
cls=cls,
|
|
))
|
|
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 _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 _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]]:
|
|
"""Return list of (external_id, host_ip, rack_name) 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]]:
|
|
"""Return list of (external_id, host_ip, rack_name) 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 FROM \"{table}\" WHERE data->>'host' IS NOT NULL"
|
|
)
|
|
rows = await cursor.fetchall()
|
|
return [(str(r[0]), str(r[1]), str(r[2] or r[0])) for r in rows if r[1]]
|
|
|
|
|
|
# ── 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)
|
|
|
|
|
|
async def discover_via_ssh(
|
|
db: AsyncSession,
|
|
obj: Object,
|
|
config_path: str = "/etc/caps/config.ini",
|
|
) -> DiscoveryResult:
|
|
result = DiscoveryResult()
|
|
|
|
if not obj.ssh_user or not obj.ssh_pass_enc:
|
|
result.errors.append("SSH credentials not configured on this object (ssh_user / ssh_password)")
|
|
return result
|
|
|
|
try:
|
|
ssh_password = decrypt(obj.ssh_pass_enc)
|
|
except Exception:
|
|
result.errors.append("Failed to decrypt SSH password")
|
|
return result
|
|
|
|
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
|
|
detected = await _autodetect_db_from_server(
|
|
server_ip=obj.server_ip,
|
|
ssh_port=ssh_port,
|
|
ssh_user=obj.ssh_user,
|
|
ssh_password=ssh_password,
|
|
)
|
|
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 _read_remote_config(
|
|
host=obj.server_ip,
|
|
port=ssh_port,
|
|
username=obj.ssh_user,
|
|
password=ssh_password,
|
|
config_path="/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 1: get rack hosts from object's DB
|
|
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=obj.db_via_tunnel,
|
|
tunnel_host=obj.server_ip,
|
|
tunnel_ssh_port=ssh_port,
|
|
tunnel_ssh_user=obj.ssh_user,
|
|
tunnel_ssh_password=ssh_password,
|
|
)
|
|
except Exception as exc:
|
|
result.errors.append(f"Cannot fetch rack hosts from object DB: {exc}")
|
|
return result
|
|
|
|
if not rack_hosts:
|
|
result.errors.append("No racks with host IPs found in object DB")
|
|
return result
|
|
|
|
# Step 2: SSH into each rack, parse config, upsert devices
|
|
# 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, _, _ in rack_hosts:
|
|
existing_rack_row = (await db.execute(
|
|
select(Rack).where(Rack.object_id == obj.id, Rack.name == ext_id)
|
|
)).scalar_one_or_none()
|
|
if existing_rack_row:
|
|
rack_id_mapping[ext_id] = existing_rack_row.id
|
|
else:
|
|
new_rack = Rack(object_id=obj.id, name=ext_id)
|
|
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 '{ext_id}' (id={new_rack.id})")
|
|
|
|
# Step 1b: Add main CAPS server and DB server as devices
|
|
# 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))
|
|
if db_host and db_host != obj.server_ip:
|
|
infra_devices.append((db_host, "vm", "db_server", True)) # DB query succeeded = online
|
|
|
|
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 external_id, rack_host, rack_name in 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
|
|
try:
|
|
config_text = await _read_remote_config(
|
|
host=rack_host,
|
|
port=obj.ssh_port or 22,
|
|
username=obj.ssh_user,
|
|
password=ssh_password,
|
|
config_path=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}")
|
|
continue # no config to parse
|
|
|
|
# Get all rack IPs to exclude them from device discovery
|
|
all_rack_ips = [host for _, host, _ in rack_hosts]
|
|
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:
|
|
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},
|
|
))
|
|
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")
|
|
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,
|
|
))
|
|
|
|
# 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}")
|
|
|
|
# Step 3: Discover MikroTik via 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_via_ssh_client(
|
|
host=obj.server_ip or rack_hosts[0][1],
|
|
port=obj.ssh_port or 22,
|
|
username=obj.ssh_user,
|
|
password=ssh_password,
|
|
)
|
|
if mikrotik_ip:
|
|
logger.info(f"[ssh_discovery] MikroTik found via SSH_CLIENT at {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
|
|
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 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
|