napalm-proxmox/napalm_proxmox/routing_mixin.py

# Copyright 2025 The NetOrk Project Authors. All rights reserved.
#
# 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.

"""Routing-related NAPALM getters for Proxmox VE nodes."""

from __future__ import annotations

import re
from typing import Any

import napalm.base.constants as C


_JsonDict = dict[str, Any]


class ProxmoxRoutingMixin:
    """Mixin providing routing NAPALM methods (get_route_to, ping, traceroute)."""

    def ping(
        self,
        destination: str,
        source: str = C.PING_SOURCE,
        ttl: int = C.PING_TTL,
        timeout: int = C.PING_TIMEOUT,
        size: int = C.PING_SIZE,
        count: int = C.PING_COUNT,
        vrf: str = C.PING_VRF,
        source_interface: str = C.PING_SOURCE_INTERFACE,
    ) -> _JsonDict:
        """Execute ping on the Proxmox node and return results."""
        cmd_parts = [
            f"ping -c {count}",
            f"-W {timeout}",
            f"-s {size}",
            f"-t {ttl}",
        ]
        if source:
            cmd_parts.append(f"-I {source}")
        elif source_interface:
            cmd_parts.append(f"-I {source_interface}")
        cmd_parts.append(destination)
        cmd = " ".join(cmd_parts)

        raw = self._exec_ssh_command(f"{cmd} 2>&1 || true")
        if not raw:
            return {"error": "Ping command not available via exec endpoint"}

        # Detect common failure strings before parsing statistics
        _error_patterns = (
            "Name or service not known",
            "Network is unreachable",
            "connect: No route to host",
            "unknown host",
        )
        for _pat in _error_patterns:
            if _pat.lower() in raw.lower():
                return {"error": raw.strip()}

        # Parse statistics line: "5 packets transmitted, 5 received, 0% packet loss"
        m_stat = re.search(
            r"(\d+) packets transmitted,\s+(\d+) received.*?([\d.]+)% packet loss",
            raw,
        )
        if not m_stat:
            return {"error": raw.strip()}

        sent = int(m_stat.group(1))
        received = int(m_stat.group(2))
        loss = sent - received

        # RTT line: "rtt min/avg/max/mdev = 0.123/0.456/0.789/0.100 ms"
        m_rtt = re.search(
            r"rtt min/avg/max/mdev = ([\d.]+)/([\d.]+)/([\d.]+)/([\d.]+)",
            raw,
        )
        rtt_min = float(m_rtt.group(1)) if m_rtt else 0.0
        rtt_avg = float(m_rtt.group(2)) if m_rtt else 0.0
        rtt_max = float(m_rtt.group(3)) if m_rtt else 0.0
        rtt_std = float(m_rtt.group(4)) if m_rtt else 0.0

        # Individual probe lines
        probes = []
        for m_probe in re.finditer(
            r"icmp_seq=\d+.*?time=([\d.]+) ms.*?from ([\d.a-fA-F:]+)", raw
        ):
            probes.append(
                {"ip_address": m_probe.group(2), "rtt": float(m_probe.group(1))}
            )

        return {
            "success": {
                "probes_sent": sent,
                "packet_loss": loss,
                "rtt_min": rtt_min,
                "rtt_max": rtt_max,
                "rtt_avg": rtt_avg,
                "rtt_stddev": rtt_std,
                "results": probes,
            }
        }

    def traceroute(
        self,
        destination: str,
        source: str = "",
        ttl: int = 255,
        timeout: int = 2,
        vrf: str = "",
    ) -> _JsonDict:
        """Execute traceroute on the Proxmox node and return results."""
        cmd_parts = [f"traceroute -m {ttl}", f"-w {timeout}", "-n"]
        if source:
            cmd_parts.append(f"-s {source}")
        cmd_parts.append(destination)
        raw = self._exec_ssh_command(" ".join(cmd_parts) + " 2>&1 || true")

        if not raw:
            return {"error": "traceroute not available via exec endpoint"}

        hops: _JsonDict = {}
        for line in raw.splitlines():
            m = re.match(
                r"^\s*(\d+)\s+([\d.a-fA-F:]+|\*)\s+([\d.]+|[\d.]+\s+ms|\*)",
                line,
            )
            if not m:
                continue
            hop_id = int(m.group(1))
            ip_addr = m.group(2)
            if ip_addr == "*":
                continue
            # Parse RTT probes: each hop can have up to 3
            rtts = re.findall(r"([\d.]+)\s+ms", line)
            probes_dict = {}
            for idx, rtt in enumerate(rtts, start=1):
                probes_dict[idx] = {
                    "rtt": float(rtt),
                    "ip_address": ip_addr,
                    "host_name": ip_addr,
                }
            if probes_dict:
                hops[hop_id] = {"probes": probes_dict}

        if not hops:
            return {"error": raw.strip()}
        return {"success": hops}

    def get_bgp_config(self, group: str = "", neighbor: str = "") -> _JsonDict:
        raise NotImplementedError("BGP configuration is not managed via Proxmox API")

    def get_bgp_neighbors(self) -> _JsonDict:
        raise NotImplementedError("BGP is not managed via Proxmox API")

    def get_bgp_neighbors_detail(self, neighbor_address: str = "") -> _JsonDict:
        raise NotImplementedError("BGP is not managed via Proxmox API")

    def get_route_to(
        self, destination: str = "", protocol: str = "", longer: bool = False
    ) -> _JsonDict:
        """Return routing table entries for the given destination."""
        cmd = f"ip route show {destination} 2>/dev/null || true"
        raw = self._exec_ssh_command(cmd)
        routes: _JsonDict = {}
        for line in raw.splitlines():
            line = line.strip()
            if not line:
                continue
            parts = line.split()
            if not parts:
                continue
            prefix = parts[0]
            next_hop = ""
            out_iface = ""
            proto = "static"
            for i, tok in enumerate(parts):
                if tok == "via" and i + 1 < len(parts):
                    next_hop = parts[i + 1]
                if tok == "dev" and i + 1 < len(parts):
                    out_iface = parts[i + 1]
                if tok == "proto" and i + 1 < len(parts):
                    proto = parts[i + 1]

            if protocol and protocol.lower() not in proto.lower():
                continue

            routes.setdefault(prefix, []).append(
                {
                    "protocol": proto,
                    "current_active": True,
                    "last_active": True,
                    "age": -1,
                    "next_hop": next_hop,
                    "outgoing_interface": out_iface,
                    "selected_next_hop": True,
                    "preference": 1,
                    "inactive_reason": "",
                    "routing_table": "default",
                    "protocol_attributes": {},
                }
            )
        return routes

    def get_optics(self) -> _JsonDict:
        raise NotImplementedError("Optics not available via Proxmox API")

    def get_probes_config(self) -> _JsonDict:
        raise NotImplementedError

    def get_probes_results(self) -> _JsonDict:
        raise NotImplementedError

    def get_firewall_policies(self) -> _JsonDict:
        raise NotImplementedError("Use the Proxmox firewall API directly")