455 lines
16 KiB
Python
455 lines
16 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 re
|
|
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
|
|
|
|
# ── 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) -> list[DiscoveredDevice]:
|
|
"""Extract discovered network devices from a caps config file."""
|
|
parser = RawConfigParser(strict=False)
|
|
parser.read_string(config_text)
|
|
|
|
discovered: list[DiscoveredDevice] = []
|
|
seen_ips: set[str] = set()
|
|
|
|
for section in parser.sections():
|
|
if not section.startswith("plugin:"):
|
|
continue
|
|
|
|
plugin_name = section[len("plugin:"):]
|
|
if any(plugin_name.startswith(prefix) for prefix in _SKIP_PLUGIN_PREFIXES):
|
|
continue
|
|
|
|
cls = parser.get(section, "cls", fallback="")
|
|
if not cls:
|
|
continue
|
|
|
|
# Collect all URI-like values in this section
|
|
uri_fields = ("uri", "frame_uri", "host")
|
|
for field_name in uri_fields:
|
|
raw = parser.get(section, field_name, fallback=None)
|
|
if not raw:
|
|
continue
|
|
ip = _extract_ip(raw)
|
|
if not ip or ip == rack_host or ip in seen_ips:
|
|
continue
|
|
seen_ips.add(ip)
|
|
discovered.append(DiscoveredDevice(
|
|
ip=ip,
|
|
category=_cls_to_category(cls),
|
|
role=_cls_to_role(cls, plugin_name),
|
|
plugin_name=plugin_name,
|
|
cls=cls,
|
|
))
|
|
|
|
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
|
|
|
|
|
|
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 DiscoveryResult:
|
|
created: int = 0
|
|
updated: int = 0
|
|
skipped: int = 0
|
|
errors: list[str] = 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
|
|
|
|
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"
|
|
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
|
|
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
|
|
)
|
|
|
|
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,
|
|
)
|
|
except Exception as exc:
|
|
result.errors.append(f"SSH to {rack_host} failed: {exc}")
|
|
continue
|
|
|
|
devices_found = _parse_config(config_text, rack_host)
|
|
|
|
for found in devices_found:
|
|
# Hostname: "rack{rack_name}-{plugin_name}", e.g. "rack01-camera_grz"
|
|
hostname = f"rack{rack_name}-{found.plugin_name}"
|
|
|
|
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:
|
|
changed = False
|
|
if existing.category != found.category and existing.source == "auto_ssh":
|
|
existing.category = found.category
|
|
changed = True
|
|
if existing.role == "other" and found.role != "other":
|
|
existing.role = found.role
|
|
changed = True
|
|
# Update hostname if it was never set or still auto-generated
|
|
if not existing.hostname or existing.hostname.startswith("rack"):
|
|
if existing.hostname != hostname:
|
|
existing.hostname = hostname
|
|
changed = True
|
|
result.updated += 1 if changed else 0
|
|
result.skipped += 0 if changed else 1
|
|
else:
|
|
db.add(Device(
|
|
object_id=obj.id,
|
|
ip=found.ip,
|
|
hostname=hostname,
|
|
category=found.category,
|
|
role=found.role,
|
|
source="auto_ssh",
|
|
device_meta={"plugin": found.plugin_name, "cls": found.cls},
|
|
))
|
|
result.created += 1
|
|
except Exception as exc:
|
|
result.errors.append(f"Error upserting {found.ip}: {exc}")
|
|
|
|
# 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}")
|
|
|
|
return result
|