Christian Manivong
c868d41322
CI / test (3.10) (push) Failing after 15s
CI / test (3.11) (push) Failing after 15s
CI / test (3.12) (push) Successful in 30s
CI / test (3.10) (pull_request) Successful in 30s
CI / test (3.11) (pull_request) Successful in 28s
CI / test (3.12) (pull_request) Successful in 29s
carp_virtual_ip() and its state table moved here from napalm-device-types (4.10.1, NAPALM/napalm-device-types#39): under the agreed structure the base package carries only the shape of the reading, and what a CARP state means is protocol knowledge. Tests moved along (test_carp_states.py).
125 lines
4.7 KiB
Python
125 lines
4.7 KiB
Python
"""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, Optional
|
|
|
|
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) -> Optional[tuple[str, Optional[int]]]:
|
|
"""``"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: Optional[int],
|
|
vhid: object,
|
|
state: Optional[str],
|
|
instance: str,
|
|
interface: Optional[str],
|
|
) -> Optional[VirtualIpDict]:
|
|
"""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)
|