net.device-validate

Imports name this capability’s declared dependencies, which fune builds next to it in your project; each one links to its page.

from typing import Any, List, Mapping, Optional

from .net_cidr import cidr_info, is_cidr  ← from net.cidr ^1.0.0 · built alongside by fune
from .net_device_validate_types import DeviceValidation
from .net_hostname_validate import validate_hostname  ← from net.hostname-validate ^1.0.0 · built alongside by fune
from .net_ipv4 import is_ipv4  ← from net.ipv4 ^1.0.0 · built alongside by fune
from .net_ipv6 import format_ipv6, is_ipv6, parse_ipv6  ← from net.ipv6 ^1.0.0 · built alongside by fune
from .net_mac_address import validate_mac_address  ← from net.mac-address ^1.0.0 · built alongside by fune
from .net_port import is_port  ← from net.port ^1.0.0 · built alongside by fune

KNOWN = ("name", "host", "ports", "mac", "subnet")


def _present(entry: Mapping[str, Any], key: str) -> bool:
    return key in entry and entry[key] is not None


def _whole_number(value: Any) -> Optional[int]:
    """A whole number, whichever way the JSON reader stored it (22 and 22.0 are the same port)."""
    if isinstance(value, bool):
        return None
    if isinstance(value, int):
        return value
    if isinstance(value, float) and value == value and value not in (float("inf"), float("-inf")) and value.is_integer():
        return int(value)
    return None


def _canonical_ipv6_block(text: str) -> str:
    """An IPv6 block with its host bits cleared and its address in RFC 5952 form."""
    address, _, prefix_text = text.partition("/")
    prefix = int(prefix_text)
    groups = []
    for i, g in enumerate(parse_ipv6(address)):
        bits = min(16, max(0, prefix - 16 * i))
        groups.append(0 if bits == 0 else g & ((0xFFFF << (16 - bits)) & 0xFFFF))
    return "%s/%d" % (format_ipv6(groups), prefix)


def validate_device(entry: Mapping[str, Any]) -> DeviceValidation:
    """Check one inventory entry and say everything that is wrong with it at once,
    so a person fixing a file does not play whack-a-mole one error per run.
    Never raises: a bad entry is a result, not a crash.
    """
    errors: List[str] = []

    name: Optional[str] = None
    if not _present(entry, "name"):
        errors.append("name is required")
    elif not isinstance(entry["name"], str):
        errors.append("name must be text")
    elif entry["name"].strip(" \t\n\r\f\v") == "":
        errors.append("name must not be empty")
    else:
        name = entry["name"].strip(" \t\n\r\f\v")

    host: Optional[str] = None
    host_kind: Optional[str] = None
    if not _present(entry, "host"):
        errors.append("host is required")
    elif not isinstance(entry["host"], str):
        errors.append("host must be text")
    elif is_ipv4(entry["host"]):
        host = entry["host"]
        host_kind = "ipv4"
    elif is_ipv6(entry["host"]):
        host = format_ipv6(parse_ipv6(entry["host"]))
        host_kind = "ipv6"
    else:
        check = validate_hostname(entry["host"])
        if check.valid:
            host = check.hostname
            host_kind = "hostname"
        else:
            errors.append('host "%s" is not an IPv4 address, IPv6 address or host name: %s' % (entry["host"], check.reason))

    ports: List[int] = []
    if not _present(entry, "ports"):
        errors.append("ports is required")
    elif not isinstance(entry["ports"], (list, tuple)):
        errors.append("ports must be a list")
    elif len(entry["ports"]) == 0:
        errors.append("ports must not be empty")
    else:
        for i, item in enumerate(entry["ports"]):
            port = _whole_number(item)
            if port is None:
                errors.append("ports[%d] must be a whole number" % i)
            elif not is_port(port):
                errors.append("ports[%d] %d is not a port (1-65535)" % (i, port))
            elif port in ports:
                errors.append("ports[%d] %d is listed twice" % (i, port))
            else:
                ports.append(port)

    mac: Optional[str] = None
    if _present(entry, "mac"):
        if not isinstance(entry["mac"], str):
            errors.append("mac must be text")
        else:
            check = validate_mac_address(entry["mac"])
            if check.valid:
                mac = check.canonical
            else:
                errors.append('mac "%s" is not valid: %s' % (entry["mac"], check.reason))

    subnet: Optional[str] = None
    if _present(entry, "subnet"):
        if not isinstance(entry["subnet"], str):
            errors.append("subnet must be text")
        elif not is_cidr(entry["subnet"]):
            errors.append('subnet "%s" is not a CIDR block' % entry["subnet"])
        elif ":" in entry["subnet"]:
            subnet = _canonical_ipv6_block(entry["subnet"])
        else:
            subnet = cidr_info(entry["subnet"]).cidr

    # Sorted, not in file order: JavaScript lists integer-like keys first, so
    # "file order" would differ between languages.
    for key in sorted(k for k in entry.keys() if k not in KNOWN):
        errors.append('unknown field "%s"' % key)

    return DeviceValidation(
        valid=len(errors) == 0,
        errors=errors,
        name=name,
        host=host,
        host_kind=host_kind,
        ports=ports,
        mac=mac,
        subnet=subnet,
    )