from typing import Any, List, Mapping, Optional from .net_cidr import cidr_info, is_cidr from .net_device_validate_types import DeviceValidation from .net_hostname_validate import validate_hostname from .net_ipv4 import is_ipv4 from .net_ipv6 import format_ipv6, is_ipv6, parse_ipv6 from .net_mac_address import validate_mac_address from .net_port import is_port 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, )