napalm-openwrt/napalm_openwrt/lldp_mixin.py

# -*- coding: utf-8 -*-
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

from __future__ import annotations

from typing import Any


class OpenWrtLLDPMixin:
    """Mixin providing LLDP NAPALM getters."""

    def get_lldp_neighbors(self) -> dict[str, list[dict[str, Any]]]:
        """Return LLDP neighbors (requires ``lldpd`` to be installed on the device).

        Uses ``lldpctl -f keyvalue`` output::

            lldp.eth0.port.ifname=eth1
            lldp.eth0.chassis.name=router-core
        """
        neighbors: dict[str, list[dict[str, Any]]] = {}
        for row in self._get_lldp_table():
            neighbors.setdefault(row["local_port"], []).append(
                {"hostname": row["system_name"], "port": row["port_id"]}
            )
        return neighbors

    def _lldpd_fix_interface(self) -> None:
        """Ensure lldpd UCI config uses the actual management interface.

        Detects the interface carrying the default route (e.g. br-lan.10 on a
        VLAN trunk) and replaces any stale/wrong interface entry in the lldpd
        UCI config.  Idempotent — only writes if the config needs updating.
        """
        mgmt_iface = self._send_command(
            "ip route show default 2>/dev/null | head -1 | grep -oE 'dev [^ ]+' | awk '{print $2}'"
        ).strip()
        if not mgmt_iface:
            mgmt_iface = "br-lan"
        # LLDP is a L2 protocol — it must run on the bridge, not a VLAN subinterface.
        # Strip .VID suffix so br-ap.10 → br-ap.
        if "." in mgmt_iface:
            mgmt_iface = mgmt_iface.rsplit(".", 1)[0]

        current = self._send_command("uci get lldpd.config.interface 2>/dev/null").strip()
        if mgmt_iface in current:
            return  # already correct

        # Reset the interface list to just the management interface
        self._send_command(
            "uci delete lldpd.config.interface 2>/dev/null; "
            f"uci add_list lldpd.config.interface='{mgmt_iface}' 2>/dev/null; "
            "uci commit lldpd 2>/dev/null; "
            "/etc/init.d/lldpd restart 2>/dev/null"
        )

    def _get_lldp_table(self) -> list[dict[str, Any]]:
        """Parse ``lldpctl -f keyvalue`` into a list of row dicts.

        Ensures ``lldpd`` is enabled and running before querying; if it was
        not already running the daemon needs time to discover neighbors so
        the first call after a fresh install will return an empty list.
        """
        # Ensure lldpd is running on the correct management interface
        self._send_command(
            "pgrep lldpd >/dev/null 2>&1 || "
            "(/etc/init.d/lldpd enable 2>/dev/null; /etc/init.d/lldpd start 2>/dev/null)"
        )
        self._lldpd_fix_interface()
        output = self._send_command("lldpctl -f keyvalue")
        rows: list[dict[str, Any]] = []

        # Group by local interface prefix: lldp.<iface>.*
        entries: dict[str, dict[str, str]] = {}
        for line in output.splitlines():
            line_s = line.strip()
            if "=" not in line_s:
                continue
            key, _, value = line_s.partition("=")
            parts = key.split(".")
            # parts: ['lldp', '<iface>', <category>, <field>, ...]
            if len(parts) < 3 or parts[0] != "lldp":
                continue
            iface = parts[1]
            subkey = ".".join(parts[2:])
            entries.setdefault(iface, {})[subkey] = value

        for iface, data in entries.items():
            rows.append(
                {
                    "local_port": iface,
                    "remote_chassis_id": data.get("chassis.mac", data.get("chassis.id.value", "")),
                    "port_id": data.get("port.ifname", data.get("port.id.value", "")),
                    "mgmt_address": data.get("chassis.mgmt-ip", ""),
                    "port_description": data.get("port.descr", ""),
                    "system_name": data.get("chassis.name", ""),
                }
            )

        return rows

    def get_lldp_neighbors_detail(self, interface: str = "") -> dict[str, list[dict[str, Any]]]:
        """Return detailed LLDP neighbor info."""
        details: dict[str, list[dict[str, Any]]] = {}

        for row in self._get_lldp_table():
            if interface and row["local_port"] != interface:
                continue
            details.setdefault(row["local_port"], []).append(
                {
                    "parent_interface": "",
                    "remote_port": row["port_id"],
                    "remote_port_description": row["port_description"],
                    "remote_chassis_id": row["remote_chassis_id"],
                    "remote_system_name": row["system_name"],
                    "remote_system_description": "",
                    "remote_system_capab": [],
                    "remote_system_enable_capab": [],
                }
            )

        return details