"""The firewall's CARP addresses, read from its interface overview (netOrk #1121). ``/api/interfaces/overview/export`` lists every address of an interface; one bound to a CARP VHID carries ``vhid`` and this node's CARP ``status``. That covers the CARP address itself and an IP alias bound to its VHID alike -- both float with the VHID's MASTER. An address without a VHID stays on this node and is no virtual IP. ``interfaces/vip_settings/search_item`` would list the same addresses, but with each VHID's CARP password in it. The overview has none, so a secret never reaches the caller. What a state means -- MASTER holds, BACKUP stands by -- is CARP protocol knowledge, so it lives here with the driver that speaks CARP (:func:`carp_virtual_ip`), not in napalm-device-types, which carries only the shape of the reading. """ from __future__ import annotations import ipaddress from typing import Any from napalm_device_types.models import VirtualIpDict #: CARP's state of a VHID on one member -> (running, holds). _CARP_STATES: dict[str, tuple[bool, bool]] = { "MASTER": (True, True), "BACKUP": (True, False), "INIT": (False, False), "DISABLED": (False, False), } def _address(text: str) -> tuple[str, int | None] | None: """``"10.0.0.1/24"`` -> ``("10.0.0.1", 24)``; None for what is no address.""" host, slash, prefix = text.partition("/") try: address = ipaddress.ip_address(host) except ValueError: return None if not slash: return str(address), None if not prefix.isdigit() or int(prefix) > address.max_prefixlen: return None return str(address), int(prefix) def carp_virtual_ip( *, address: str, prefix_length: int | None, vhid: object, state: str | None, instance: str, interface: str | None, ) -> VirtualIpDict | None: """One address bound to a CARP VHID, as a virtual IP; None for what is no address. Who holds it comes from CARP's *state* on this member, not from its interfaces: MASTER holds it, BACKUP stands by, INIT and DISABLED neither run nor hold. A state this does not know says nothing (``running`` and ``holds`` None), and the caller falls back on the interfaces. *address* may carry its prefix (``"10.0.0.1/24"``), which then wins over *prefix_length*. """ parsed = _address(address.strip()) if address else None if parsed is None: return None normalised, written = parsed prefix = written if written is not None else prefix_length if prefix is not None and not 0 <= prefix <= ipaddress.ip_address(normalised).max_prefixlen: prefix = None vhid_text = str(vhid).strip() if vhid is not None else "" running, holds = _CARP_STATES.get((state or "").strip().upper(), (None, None)) return { "mechanism": "carp", "instance": instance or f"vhid {vhid_text}", "address": normalised, "prefix_length": prefix, "interface": interface, "group_key": f"vhid:{vhid_text}" if vhid_text.isdigit() else None, "running": running, "holds": holds, } def _interfaces(overview: Any) -> list[Any]: """The interface list, whether the answer is the list or wraps it.""" if isinstance(overview, dict): return overview.get("interfaces") or [] if isinstance(overview, list): return overview raise ValueError(f"unexpected interface overview: {type(overview).__name__}") def carp_addresses(overview: Any) -> list[VirtualIpDict]: """Every address bound to a CARP VHID in *overview*, sorted by address. :raises ValueError: when the answer is neither the interface list nor wraps it. """ entries: list[VirtualIpDict] = [] for interface in _interfaces(overview): if not isinstance(interface, dict): continue identifier = interface.get("identifier") or "" for family in ("ipv4", "ipv6"): for address in interface.get(family) or []: if not isinstance(address, dict) or not address.get("vhid"): continue entry = carp_virtual_ip( address=str(address.get("ipaddr") or ""), prefix_length=None, vhid=address.get("vhid"), state=address.get("status"), instance=interface.get("description") or identifier, interface=interface.get("device") or identifier or None, ) if entry is not None: entries.append(entry) return sorted(entries, key=_order) def _order(entry: VirtualIpDict) -> tuple[int, int]: address = ipaddress.ip_address(entry["address"]) return address.version, int(address)