feat: merge IP-scanned and Proxmox-imported devices by MAC

Reconcile the same physical device discovered by both the nmap IP scan and
the Proxmox importer into a single inventory row, keyed on MAC. Previously
each path only deduped by IP, and the importer captured no MAC (and no IP for
stopped guests), so most guests double-listed.

Backend:
- mac_utils.normalize_mac: canonical MAC (lowercase, ':'-separated), the
  cross-source join key. Normalized on write and on compare.
- proxmox_service: capture the guest NIC MAC agent-free from the net0 config
  (qemu virtio=<MAC>, lxc hwaddr=<MAC>); works for stopped guests. Resolver
  now returns (ip, mac).
- proxmox persist: match existing Node/PendingDevice by ieee OR ip OR MAC;
  fill mac, keep the vm/lxc type, union sources.
- scanner persist: match PendingDevice by ip OR MAC; fill the IP a Proxmox
  import lacked, keep a pve row's type, union the scan source. Stamp query
  matches raw + normalized MAC (legacy-safe).
- Multi-source tags: new PendingDevice.discovery_sources JSON column so a
  merged device shows under both the IP and Proxmox filters. Idempotent
  migration backfills from discovery_source (legacy NULL-scalar rows with an
  IP become ["arp"]). _sources_after_merge preserves a scanned row's IP origin
  through the merge without tagging a pure Proxmox guest.
- Import now broadcasts a scan update on completion so an open inventory
  reloads without a manual refresh.

Frontend:
- pendingSources: sourceBuckets/orderedSources map discovery_sources to filter
  buckets; a device with ["arp","proxmox"] matches both filters and renders
  both badges. PendingDevicesModal filter + badges use them.

Tests: MAC normalization, config MAC capture, cross-source merge both
directions, legacy-row IP-tag preservation, no-false-IP-tag guard, refresh
broadcast, and the frontend bucket mapping.

ha-relevant: maybe
This commit is contained in:
Pouzor
2026-07-06 20:11:03 +02:00
parent 05ef746f22
commit adf82f8f01
16 changed files with 572 additions and 76 deletions
+59 -12
View File
@@ -32,6 +32,8 @@ from app.schemas.proxmox import (
ProxmoxTestConnectionResponse,
)
from app.schemas.scan import ScanRunResponse
from app.services.discovery_sources import add_source
from app.services.mac_utils import normalize_mac
from app.services.node_dedupe import dedupe_nodes_by_ieee
from app.services.proxmox_service import (
build_proxmox_cluster_links,
@@ -164,6 +166,10 @@ async def _background_proxmox_import(
run.finished_at = datetime.now(timezone.utc)
run.error = _guest_visibility_advisory(nodes_raw)
await db.commit()
# Nudge the frontend to reload the inventory (same signal the IP scan
# emits) so imported/merged devices appear without a manual refresh.
from app.api.routes.status import broadcast_scan_update
await broadcast_scan_update(run_id=run_id, devices_found=result.device_count)
except Exception as exc:
logger.exception("Proxmox import %s failed", run_id)
await db.rollback()
@@ -223,14 +229,19 @@ async def _persist_pending_import(
if not ieee:
continue
ip = n.get("ip")
mac = normalize_mac(n.get("mac"))
props = build_proxmox_properties(n)
# 1) Already on a canvas? Match by ieee OR (ip when known). Refresh in
# place: merge properties, adopt the pve identity onto a scanned node,
# backfill blank specs/hostname. Do NOT stomp user-set type/status.
# 1) Already on a canvas? Match by ieee OR ip OR mac (the cross-source
# dedup key — a stopped VM has no IP but its configured NIC MAC still
# matches an ARP-scanned node). Refresh in place: merge properties, adopt
# the pve identity onto a scanned node, backfill blank specs/hostname/mac.
# Do NOT stomp user-set type/status.
node_filter = [Node.ieee_address == ieee]
if ip:
node_filter.append(Node.ip == ip)
if mac:
node_filter.append(Node.mac == mac)
existing_nodes = (
await db.execute(select(Node).where(or_(*node_filter)).order_by(Node.id))
).scalars().all()
@@ -242,6 +253,8 @@ async def _persist_pending_import(
en.ieee_address = ieee
if ip and not en.ip:
en.ip = ip
if mac and not en.mac:
en.mac = mac
en.hostname = en.hostname or n.get("hostname")
en.cpu_count = en.cpu_count or n.get("cpu_count")
en.ram_gb = en.ram_gb or n.get("ram_gb")
@@ -251,17 +264,17 @@ async def _persist_pending_import(
if ieee in cluster_members:
en.left_handles = max(en.left_handles or 0, 1)
en.right_handles = max(en.right_handles or 0, 1)
await _ensure_inventory_row(db, ieee, ip, n, props, approved=True)
await _ensure_inventory_row(db, ieee, ip, mac, n, props, approved=True)
pending_updated += 1
continue
# 2) Not on canvas — upsert the pending inventory row.
pending = await _find_pending(db, ieee, ip)
pending = await _find_pending(db, ieee, ip, mac)
if pending is None:
db.add(_new_pending(ieee, ip, n, props, status="pending"))
db.add(_new_pending(ieee, ip, mac, n, props, status="pending"))
pending_created += 1
else:
_refresh_pending(pending, ieee, ip, n, props)
_refresh_pending(pending, ieee, ip, mac, n, props)
pending_updated += 1
links_recorded = await _replace_links(db, edges_raw, cluster_pairs)
@@ -276,22 +289,30 @@ async def _persist_pending_import(
async def _find_pending(
db: AsyncSession, ieee: str, ip: str | None
db: AsyncSession, ieee: str, ip: str | None, mac: str | None
) -> PendingDevice | None:
filters = [PendingDevice.ieee_address == ieee]
if ip:
filters.append(PendingDevice.ip == ip)
if mac:
filters.append(PendingDevice.mac == mac)
return (
await db.execute(select(PendingDevice).where(or_(*filters)))
).scalars().first()
def _new_pending(
ieee: str, ip: str | None, n: dict[str, Any], props: list[dict[str, Any]], status: str
ieee: str,
ip: str | None,
mac: str | None,
n: dict[str, Any],
props: list[dict[str, Any]],
status: str,
) -> PendingDevice:
return PendingDevice(
ieee_address=ieee,
ip=ip,
mac=mac,
hostname=n.get("hostname"),
friendly_name=n.get("label"),
suggested_type=n.get("type"),
@@ -299,19 +320,41 @@ def _new_pending(
model=n.get("model"),
properties=props,
status=status,
discovery_source="proxmox",
discovery_source=_PROXMOX_GUEST_SOURCE,
discovery_sources=[_PROXMOX_GUEST_SOURCE],
)
def _sources_after_merge(row: PendingDevice) -> list[str]:
"""Discovery sources for an inventory row after a Proxmox import merges in.
Must run BEFORE the ``pve-`` ieee is adopted onto the row, so it can tell
whether the row was originally a scanned device. Preserves the prior scan
origin — including legacy rows created before ``discovery_sources`` existed
(empty list) and possibly with a NULL ``discovery_source`` — so the IP tag
survives the merge. A row that carries an IP but was not itself a Proxmox
device (no ``pve-`` ieee) was found by a scan; keep an IP-scan source.
"""
sources = add_source(row.discovery_sources, row.discovery_source)
was_scanned = not (row.ieee_address or "").startswith("pve-")
if was_scanned and row.ip and not any(s in ("arp", "mdns") for s in sources):
sources = add_source(sources, "arp")
return add_source(sources, _PROXMOX_GUEST_SOURCE)
def _refresh_pending(
pending: PendingDevice,
ieee: str,
ip: str | None,
mac: str | None,
n: dict[str, Any],
props: list[dict[str, Any]],
) -> None:
# Compute sources before adopting the pve ieee (needs the pre-merge origin).
pending.discovery_sources = _sources_after_merge(pending)
pending.ieee_address = pending.ieee_address or ieee
pending.ip = ip or pending.ip
pending.mac = pending.mac or mac
pending.hostname = n.get("hostname") or pending.hostname
pending.friendly_name = n.get("label") or pending.friendly_name
pending.suggested_type = n.get("type") or pending.suggested_type
@@ -328,18 +371,22 @@ async def _ensure_inventory_row(
db: AsyncSession,
ieee: str,
ip: str | None,
mac: str | None,
n: dict[str, Any],
props: list[dict[str, Any]],
approved: bool,
) -> None:
"""Ensure an inventory row exists for a device already on a canvas, so it
shows in the inventory with an 'In N canvas' badge. Never changes status."""
inv = await _find_pending(db, ieee, ip)
inv = await _find_pending(db, ieee, ip, mac)
if inv is None:
db.add(_new_pending(ieee, ip, n, props, status="approved" if approved else "pending"))
db.add(_new_pending(ieee, ip, mac, n, props, status="approved" if approved else "pending"))
else:
# Compute sources before adopting the pve ieee (needs the pre-merge origin).
inv.discovery_sources = _sources_after_merge(inv)
inv.ieee_address = inv.ieee_address or ieee
inv.ip = ip or inv.ip
inv.mac = inv.mac or mac
inv.hostname = n.get("hostname") or inv.hostname
inv.suggested_type = n.get("type") or inv.suggested_type
inv.properties = merge_proxmox_properties(list(inv.properties or []), props)
+31
View File
@@ -321,6 +321,37 @@ async def init_db() -> None:
with suppress(OperationalError):
sql = "UPDATE edges SET animated = 'none' WHERE animated = '0' OR animated = 0 OR animated IS NULL"
await conn.exec_driver_sql(sql)
# Multi-source discovery tags: a device found by both an IP scan and a
# Proxmox import carries every source. Backfill from the legacy single
# discovery_source so existing rows show under their filter.
with suppress(OperationalError):
await conn.exec_driver_sql("ALTER TABLE pending_devices ADD COLUMN discovery_sources JSON")
with suppress(OperationalError):
await conn.exec_driver_sql(
"UPDATE pending_devices SET discovery_sources = json_array(discovery_source) "
"WHERE discovery_sources IS NULL AND discovery_source IS NOT NULL"
)
# Legacy IP-scanned rows predating discovery_source have a NULL scalar
# but a real IP — treat them as an ARP scan so they keep the IP tag.
with suppress(OperationalError):
await conn.exec_driver_sql(
"UPDATE pending_devices SET discovery_sources = json_array('arp') "
"WHERE discovery_sources IS NULL AND discovery_source IS NULL AND ip IS NOT NULL"
)
with suppress(OperationalError):
await conn.exec_driver_sql(
"UPDATE pending_devices SET discovery_sources = '[]' WHERE discovery_sources IS NULL"
)
# Canonicalize stored MACs (lowercase, ':' separators) so cross-source
# dedup can match a Proxmox NIC MAC against an ARP-scanned one by equality.
with suppress(OperationalError):
await conn.exec_driver_sql(
"UPDATE pending_devices SET mac = lower(replace(mac, '-', ':')) WHERE mac IS NOT NULL"
)
with suppress(OperationalError):
await conn.exec_driver_sql(
"UPDATE nodes SET mac = lower(replace(mac, '-', ':')) WHERE mac IS NOT NULL"
)
async def get_db() -> AsyncGenerator[AsyncSession, None]:
+6
View File
@@ -119,7 +119,13 @@ class PendingDevice(Base):
services: Mapped[list[Any]] = mapped_column(JSON, default=list)
suggested_type: Mapped[str | None] = mapped_column(String)
status: Mapped[str] = mapped_column(String, default="pending")
# Origin/primary source (first discovery): "arp"/"mdns"/"zigbee"/"zwave"/
# "proxmox". Kept for back-compat; `discovery_sources` is the full set.
discovery_source: Mapped[str | None] = mapped_column(String)
# All sources that have observed this device. A device found by both an IP
# scan and a Proxmox import carries e.g. ["arp", "proxmox"] and shows under
# both inventory filters. Source of truth for the frontend source badges.
discovery_sources: Mapped[list[Any]] = mapped_column(JSON, default=list)
ieee_address: Mapped[str | None] = mapped_column(String, index=True, nullable=True, unique=True)
friendly_name: Mapped[str | None] = mapped_column(String, nullable=True)
device_subtype: Mapped[str | None] = mapped_column(String, nullable=True)
+6 -3
View File
@@ -14,6 +14,9 @@ class PendingDeviceResponse(BaseModel):
suggested_type: str | None
status: str
discovery_source: str | None
# All sources that have observed this device (e.g. ["arp", "proxmox"]). Drives
# the inventory source filter + badges; falls back to [discovery_source].
discovery_sources: list[str] = []
ieee_address: str | None = None
friendly_name: str | None = None
device_subtype: str | None = None
@@ -35,10 +38,10 @@ class PendingDeviceResponse(BaseModel):
node_last_modified: datetime | None = None
node_last_seen: datetime | None = None
@field_validator("properties", mode="before")
@field_validator("properties", "discovery_sources", mode="before")
@classmethod
def _coerce_properties(cls, v: Any) -> list[Any]:
# Legacy rows (column added by migration) have properties = NULL.
def _coerce_list(cls, v: Any) -> list[Any]:
# Legacy rows (columns added by migration) have these = NULL.
return v if isinstance(v, list) else []
model_config = {"from_attributes": True}
+19
View File
@@ -0,0 +1,19 @@
"""Helpers for the multi-valued ``PendingDevice.discovery_sources`` set.
A device discovered by more than one path (e.g. an IP scan *and* a Proxmox
import) accumulates every source that has seen it, so it surfaces under each
matching inventory filter. Order is preserved (origin first) and duplicates are
dropped.
"""
from __future__ import annotations
from collections.abc import Iterable
def add_source(sources: Iterable[str] | None, source: str | None) -> list[str]:
"""Return ``sources`` with ``source`` appended if not already present."""
out = [s for s in (sources or []) if s]
if source and source not in out:
out.append(source)
return out
+18
View File
@@ -0,0 +1,18 @@
"""MAC-address normalization, shared by the scan + Proxmox persist paths.
Different discovery sources emit MACs in different casing/separators (ARP is
lowercase ``bc:24:11:..``, Proxmox config is often uppercase ``BC:24:11:..``).
Canonicalizing on write *and* on compare lets cross-source dedup match a device
by MAC with a plain ``==`` — the join key for merging an IP-scanned row with a
Proxmox-imported one.
"""
from __future__ import annotations
def normalize_mac(mac: str | None) -> str | None:
"""Canonical MAC: lowercase, ``-`` → ``:``, stripped. Blank/None → None."""
if not mac:
return None
normalized = mac.strip().lower().replace("-", ":")
return normalized or None
+53 -17
View File
@@ -18,6 +18,7 @@ from typing import Any
import httpx
from app.services.mac_utils import normalize_mac
from app.services.zigbee_service import merge_zigbee_properties
logger = logging.getLogger(__name__)
@@ -32,6 +33,9 @@ _BYTES_PER_GB = 1024 ** 3
# net0 config line: "name=eth0,bridge=vmbr0,ip=192.168.1.5/24,gw=..."
_LXC_IP_RE = re.compile(r"(?:^|,)ip=([0-9]{1,3}(?:\.[0-9]{1,3}){3})(?:/\d+)?")
# NIC MAC inside a net0 string — qemu "virtio=BC:24:11:..,bridge=.." or lxc
# "..,hwaddr=BC:24:11:..,..". A bare 6-octet MAC match works for both forms.
_NET_MAC_RE = re.compile(r"([0-9a-fA-F]{2}(?::[0-9a-fA-F]{2}){5})")
def _sanitize_proxmox_error(exc: BaseException) -> str:
@@ -121,6 +125,21 @@ def _extract_lxc_ip(config_payload: dict[str, Any] | None) -> str | None:
return match.group(1) if match else None
def _extract_net_mac(config_payload: dict[str, Any] | None) -> str | None:
"""Parse the NIC MAC from a qemu/lxc ``net0`` config string (normalized).
Works agent-free for both guest kinds and for stopped guests — the sole
identity we can reliably cross-match against an ARP-scanned device.
"""
if not config_payload:
return None
net0 = config_payload.get("net0")
if not isinstance(net0, str):
return None
match = _NET_MAC_RE.search(net0)
return normalize_mac(match.group(1)) if match else None
def _host_node(raw: dict[str, Any]) -> dict[str, Any] | None:
"""Build a homelable ``proxmox`` host node from a ``/nodes`` entry."""
name = raw.get("node")
@@ -144,7 +163,9 @@ def _host_node(raw: dict[str, Any]) -> dict[str, Any] | None:
}
def _guest_node(raw: dict[str, Any], host_name: str, kind: str, ip: str | None) -> dict[str, Any] | None:
def _guest_node(
raw: dict[str, Any], host_name: str, kind: str, ip: str | None, mac: str | None = None
) -> dict[str, Any] | None:
"""Build a homelable vm/lxc node from a ``/qemu`` or ``/lxc`` list entry."""
vmid = raw.get("vmid")
if vmid is None:
@@ -159,6 +180,7 @@ def _guest_node(raw: dict[str, Any], host_name: str, kind: str, ip: str | None)
"ieee_address": ieee,
"hostname": name,
"ip": ip,
"mac": mac,
"status": "online" if raw.get("status") == "running" else "offline",
"cpu_count": _int_or_none(raw.get("maxcpu") or raw.get("cpus")),
"ram_gb": _gb(raw.get("maxmem")),
@@ -320,34 +342,48 @@ async def _fetch_host_guests(
if not isinstance(entries, list):
continue
for raw in entries:
ip = await _resolve_guest_ip(client, host_name, kind, raw)
node = _guest_node(raw, host_name, kind, ip)
ip, mac = await _resolve_guest_net(client, host_name, kind, raw)
node = _guest_node(raw, host_name, kind, ip, mac)
if node:
guests.append(node)
return guests
async def _resolve_guest_ip(
async def _resolve_guest_net(
client: httpx.AsyncClient, host_name: str, kind: str, raw: dict[str, Any]
) -> str | None:
"""Best-effort guest IP. qemu → guest agent, lxc → net0 config. Never raises."""
) -> tuple[str | None, str | None]:
"""Best-effort (ip, mac) for a guest. Never raises.
- MAC comes from the guest ``/config`` net0 line for both kinds (agent-free,
works for stopped guests) — the cross-source dedup key.
- IP: qemu → guest agent (running only); lxc → static net0 config.
Partial results are returned even if a later call fails (e.g. config MAC is
kept when the qemu agent call errors).
"""
vmid = raw.get("vmid")
if vmid is None:
return None
return None, None
ip: str | None = None
mac: str | None = None
try:
if kind == "qemu":
if raw.get("status") != "running":
return None
data = await _get_json(
client, f"/nodes/{host_name}/qemu/{vmid}/agent/network-get-interfaces"
)
return _extract_qemu_ip(data)
data = await _get_json(client, f"/nodes/{host_name}/lxc/{vmid}/config")
return _extract_lxc_ip(data)
config = await _get_json(client, f"/nodes/{host_name}/qemu/{vmid}/config")
mac = _extract_net_mac(config)
if raw.get("status") == "running":
data = await _get_json(
client, f"/nodes/{host_name}/qemu/{vmid}/agent/network-get-interfaces"
)
ip = _extract_qemu_ip(data)
else:
config = await _get_json(client, f"/nodes/{host_name}/lxc/{vmid}/config")
ip = _extract_lxc_ip(config)
mac = _extract_net_mac(config)
except httpx.HTTPError:
# Guest agent not installed / container stopped / no perms → no IP. Fine.
return None
# Guest agent not installed / container stopped / no perms. Keep whatever
# was resolved before the failure.
pass
return ip, mac
async def test_proxmox_connection(
+29 -13
View File
@@ -15,8 +15,10 @@ from sqlalchemy import or_, select
from sqlalchemy.ext.asyncio import AsyncSession
from app.db.models import Node, PendingDevice, ScanRun
from app.services.discovery_sources import add_source
from app.services.fingerprint import fingerprint_ports, suggest_node_type
from app.services.http_probe import probe_open_ports
from app.services.mac_utils import normalize_mac
logger = logging.getLogger(__name__)
@@ -522,16 +524,21 @@ async def run_scan(
ip, open_ports, verify_tls=deep_scan.verify_tls
)
norm_mac = normalize_mac(host.get("mac"))
services = fingerprint_ports(open_ports)
suggested_type = suggest_node_type(open_ports, host.get("mac"))
suggested_type = suggest_node_type(open_ports, norm_mac)
# One inventory row per device (by IP). Match across pending AND
# approved so a re-scan of an already-approved device refreshes its
# row instead of spawning a fresh "pending" duplicate. Hidden rows
# are already skipped above.
# One inventory row per device. Match by IP OR MAC across pending AND
# approved so a re-scan refreshes the existing row instead of spawning
# a duplicate — and so a device previously imported from Proxmox (which
# may have no IP but a known NIC MAC) reconciles with this scan instead
# of doubling up. Hidden rows are already skipped above.
match_cond = [PendingDevice.ip == ip]
if norm_mac:
match_cond.append(PendingDevice.mac == norm_mac)
existing_rows = (await db.execute(
select(PendingDevice)
.where(PendingDevice.ip == ip, PendingDevice.status != "hidden")
.where(or_(*match_cond), PendingDevice.status != "hidden")
.order_by(PendingDevice.discovered_at)
)).scalars().all()
@@ -543,32 +550,41 @@ async def run_scan(
for dup in existing_rows:
if dup is not keep:
await db.delete(dup)
keep.mac = host.get("mac") or keep.mac
keep.ip = keep.ip or ip # fill an IP a Proxmox import lacked
keep.mac = norm_mac or keep.mac
keep.hostname = host.get("hostname") or keep.hostname
keep.os = host.get("os") or keep.os
keep.services = services
keep.suggested_type = suggested_type
# Don't downgrade a Proxmox-typed guest (vm/lxc) to the generic
# scan guess; the importer knows the true type.
if not (keep.ieee_address or "").startswith("pve-"):
keep.suggested_type = suggested_type
# Merged row carries both sources (e.g. ["proxmox", "arp"]).
keep.discovery_sources = add_source(keep.discovery_sources, discovery_source)
# status preserved — an approved device stays approved.
else:
db.add(PendingDevice(
ip=ip,
mac=host.get("mac"),
mac=norm_mac,
hostname=host.get("hostname"),
os=host.get("os"),
services=services,
suggested_type=suggested_type,
status="pending",
discovery_source=discovery_source,
discovery_sources=[discovery_source],
))
devices_found += 1
# Stamp last_scan on any canvas node that matches this device by IP
# (or MAC, when known) so the inventory shows when the scanner last
# observed it. Matches across designs.
host_mac = host.get("mac")
# observed it. Match both the normalized and raw MAC so a legacy
# canvas node whose mac predates normalization still matches. Across
# designs.
node_match = [Node.ip == ip]
if host_mac:
node_match.append(Node.mac == host_mac)
for m in {norm_mac, host.get("mac")}:
if m:
node_match.append(Node.mac == m)
matching_nodes = (await db.execute(
select(Node).where(or_(*node_match))
)).scalars().all()