diff --git a/backend/app/services/device_discovery.py b/backend/app/services/device_discovery.py index 3a285da..4694fdf 100644 --- a/backend/app/services/device_discovery.py +++ b/backend/app/services/device_discovery.py @@ -225,6 +225,68 @@ async def _read_remote_config( 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, @@ -524,6 +586,11 @@ async def discover_via_ssh( return result # 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 @@ -727,6 +794,11 @@ async def discover_via_ssh( result.errors.append(f"Flush error for offline rack {rack_host}: {exc}") continue # no config to parse + # Collect ARP neighbours for slave rack discovery (best-effort) + neigh_ips = await _get_ip_neighbors(rack_host, ssh_port, obj.ssh_user, ssh_password) + 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] devices_found = _parse_config(config_text, rack_host, all_rack_ips) @@ -845,6 +917,10 @@ async def discover_via_ssh( 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 @@ -854,6 +930,155 @@ async def discover_via_ssh( except Exception as exc: result.errors.append(f"Flush error after rack {rack_host}: {exc}") + # 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 + ) + if slave_candidates: + logger.info(f"[ssh_discovery] Step 2.5: Slave discovery — {len(slave_candidates)} ARP candidate(s): {slave_candidates}") + for candidate_ip in slave_candidates: + logger.info(f"[ssh_discovery] Checking slave candidate {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( + host=candidate_ip, + port=ssh_port, + username=obj.ssh_user, + password=ssh_password, + master_ip=master_host, + config_path=config_path, + ) + if ok: + is_slave = True + slave_identifier = identifier + slave_config_text = slave_cfg + master_ip_for_slave = master_host + 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 (identifier={slave_identifier}, master={master_ip_for_slave})") + + # 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") + 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: + 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}, + )) + 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 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(