diff --git a/napalm_proxmox/driver.py b/napalm_proxmox/driver.py index 8cc8bbe..a1c5a6d 100644 --- a/napalm_proxmox/driver.py +++ b/napalm_proxmox/driver.py @@ -705,45 +705,83 @@ class ProxmoxDriver(HypervisorDriver): vlan_entry["untagged"] = list(access_by_vlan.get(vid, [])) return result - # --- Linux bridge fallback (bridge vlan show) --- + # --- Linux bridge fallback --- # Convert result entries to use tagged/untagged keys for entry in result.values(): entry.setdefault("tagged", []) entry.setdefault("untagged", []) - raw = self._exec_ssh_command("bridge vlan show 2>/dev/null || true") - current_iface = "" - for line in raw.splitlines(): - line = line.strip() - if not line: - continue - m = re.match(r"^(\S+)\s+(\d+)", line) - if m: - current_iface = m.group(1) - vid = m.group(2) - else: - m2 = re.match(r"^\s*(\d+)", line) - if m2: - vid = m2.group(1) - else: - continue - if current_iface and vid: - entry = result.setdefault(vid, {"name": "", "tagged": [], "untagged": []}) - if "Untagged" in line or "PVID" in line: - if current_iface not in entry["untagged"]: - entry["untagged"].append(current_iface) - else: - if current_iface not in entry["tagged"]: - entry["tagged"].append(current_iface) + # `bridge vlan show` reports every VID (1-4094) as trunked on each port + # of a VLAN-aware bridge by default — that's Proxmox's standard + # configuration and not meaningful as a VLAN table. Instead, derive the + # VLANs actually in use from the per-VM/container net config (tag=N on + # a bridge), which reflects real assignments. + for tag, bridges in self._get_vm_vlan_tags().items(): + entry = result.setdefault(tag, {"name": "", "tagged": [], "untagged": []}) + for bridge in bridges: + if bridge not in entry["untagged"]: + entry["untagged"].append(bridge) - # Only return VLANs that are actually assigned to at least one interface. - # Linux bridge vlan show reports all 4094 possible VIDs per port — - # filtering here avoids polluting the VLAN table with phantom entries. return { vid: entry for vid, entry in result.items() if entry.get("tagged") or entry.get("untagged") } + def _get_vm_vlan_tags(self) -> dict[str, set[str]]: + """Return ``{vlan_tag: {bridge_names}}`` derived from VM/container net configs. + + Scans every QEMU VM and LXC container on this node for ``netN`` config + entries of the form ``bridge=vmbrX,tag=N,...`` and groups the bridges + each VLAN tag is used on. + """ + tags: dict[str, set[str]] = {} + net_re = re.compile(r"^net\d+$") + + def _collect(vmid: int, config: _JsonDict) -> None: + for key, val in config.items(): + if not net_re.match(key): + continue + bridge = "" + tag: int | None = None + for part in str(val).split(","): + if "=" not in part: + continue + k, v = part.split("=", 1) + k = k.strip().lower() + if k == "bridge": + bridge = v.strip() + elif k == "tag": + try: + tag = int(v.strip()) + except ValueError: + pass + if tag is not None and bridge: + tags.setdefault(str(tag), set()).add(bridge) + + try: + for vm in (self._node_api().qemu.get() or []): + vmid = int(vm.get("vmid", 0)) + try: + config = self._node_api().qemu(vmid).config.get() or {} + _collect(vmid, config) + except Exception as exc: + logger.debug("_get_vm_vlan_tags: QEMU %s config failed: %s", vmid, exc) + except Exception as exc: + logger.warning("_get_vm_vlan_tags: failed to list QEMU VMs: %s", exc) + + try: + for ct in (self._node_api().lxc.get() or []): + vmid = int(ct.get("vmid", 0)) + try: + config = self._node_api().lxc(vmid).config.get() or {} + _collect(vmid, config) + except Exception as exc: + logger.debug("_get_vm_vlan_tags: LXC %s config failed: %s", vmid, exc) + except Exception as exc: + logger.warning("_get_vm_vlan_tags: failed to list LXC containers: %s", exc) + + return tags + # ------------------------------------------------------------------ # # get_network_instances (SDN Zones as VRF-like instances) # ------------------------------------------------------------------ # @@ -1783,15 +1821,20 @@ class ProxmoxDriver(HypervisorDriver): # 1. Available package updates try: - updates = self._api.nodes(self._node_name).apt.update.get() + updates = self.get_available_updates() if updates: warnings.append({ "code": "updates_available", - "severity": "info", + "severity": "warning", + "title": ( + f"{len(updates)} package update" + f"{'s' if len(updates) != 1 else ''} available" + ), + "message": None, "action": None, "meta": { "count": len(updates), - "packages": [u.get("Package", "") for u in updates[:10]], + "packages": [u["name"] for u in updates], }, }) except Exception: