From 7b6acb84e4fbbf9193007250f7a7bc156f11c440 Mon Sep 17 00:00:00 2001 From: Christos Choutouridis Date: Sat, 26 Sep 2026 18:53:34 +0300 Subject: [PATCH] Add validation in structural level --- source/topology_validator.py | 1152 ++++++++++++++++++++++++++++++++++ 1 file changed, 1152 insertions(+) create mode 100644 source/topology_validator.py diff --git a/source/topology_validator.py b/source/topology_validator.py new file mode 100644 index 0000000..91b7391 --- /dev/null +++ b/source/topology_validator.py @@ -0,0 +1,1152 @@ +"""Validate the semantic consistency of a parsed topology. + +This module implements the validation stage of the topology pipeline. + +The parser answers: + + "Can this JSON be represented by the topology grammar?" + +The validator answers: + + "Does the parsed topology describe a coherent network?" + +Validation is intentionally independent of Mininet. It does not create +interfaces, execute commands, print diagnostics, or modify network state. +Instead, it collects semantic issues and raises a TopologyValidationError when +validation fails. + +Author: Christos Choutouridis +""" + +from dataclasses import dataclass +import ipaddress +import math +import re +from typing import Dict, List, Optional, Sequence, Set, Tuple + +from topology_parser import ( + AddressingMode, + DHCPServerSettings, + HostSettings, + Interface, + Link, + LinkInterfaceSelector, + LinkSettings, + Node, + NodeType, + RouterSettings, + Service, + ServiceInterfaceSelector, + ServiceType, + SwitchSettings, + Topology, +) + + +_RESERVED_INTERFACE_TAGS = { + LinkInterfaceSelector.AUTO.value, + ServiceInterfaceSelector.ALL.value, +} + +_MAC_ADDRESS_RE = re.compile( + r"^(?:[0-9A-Fa-f]{2}:){5}[0-9A-Fa-f]{2}$" +) + + +@dataclass(frozen=True) +class ValidationIssue: + """Describe one semantic topology validation failure. + + Args: + location: Logical location of the invalid declaration. + message: Human-readable description of the semantic failure. + """ + + location: str + message: str + + +class TopologyValidationError(ValueError): + """Report one or more semantic errors discovered during validation. + + Args: + issues: Sequence of semantic validation failures. + + Attributes: + issues: Immutable tuple containing all collected validation issues. + """ + + def __init__(self, issues: Sequence[ValidationIssue]) -> None: + """Initialize the aggregated validation error. + + Args: + issues: Sequence of semantic validation failures. + """ + + self.issues: Tuple[ValidationIssue, ...] = tuple(issues) + super().__init__(self._format_message()) + + def _format_message(self) -> str: + """Format all validation issues into one exception message. + + Returns: + Multi-line human-readable validation summary. + """ + + count = len(self.issues) + header = "%d topology validation error%s" % ( + count, + "" if count == 1 else "s", + ) + + lines = [header] + for issue in self.issues: + lines.append("- %s: %s" % (issue.location, issue.message)) + + return "\n".join(lines) + + +@dataclass +class _NodeIndex: + """Hold node lookup information prepared for validation. + + Args: + nodes: Unique node declarations indexed by name. + duplicate_names: Node names that were declared more than once. + """ + + nodes: Dict[str, Node] + duplicate_names: Set[str] + + +def _add_issue( + issues: List[ValidationIssue], + location: str, + message: str, +) -> None: + """Append one validation issue. + + Args: + issues: Mutable list receiving validation failures. + location: Logical location associated with the failure. + message: Human-readable description of the failure. + """ + + issues.append(ValidationIssue(location=location, message=message)) + + +def _build_node_index( + topology: Topology, + issues: List[ValidationIssue], +) -> _NodeIndex: + """Build a node lookup table while reporting duplicate names. + + Args: + topology: Parsed topology being validated. + issues: Mutable list receiving validation failures. + + Returns: + Node lookup information for later validation passes. + """ + + nodes: Dict[str, Node] = {} + first_index: Dict[str, int] = {} + duplicate_names: Set[str] = set() + + for index, node in enumerate(topology.nodes): + location = "nodes[%d].name" % index + + if not node.name: + _add_issue(issues, location, "node name must not be empty") + continue + + if node.name in nodes: + duplicate_names.add(node.name) + _add_issue( + issues, + location, + "duplicate node name %r; first declared at nodes[%d]" + % (node.name, first_index[node.name]), + ) + continue + + nodes[node.name] = node + first_index[node.name] = index + + return _NodeIndex( + nodes=nodes, + duplicate_names=duplicate_names, + ) + + +def _validate_node_settings( + node: Node, + location: str, + issues: List[ValidationIssue], +) -> None: + """Validate that a node carries settings matching its semantic type. + + Args: + node: Node whose settings object is checked. + location: Logical location of the node. + issues: Mutable list receiving validation failures. + """ + + expected_type = { + NodeType.HOST: HostSettings, + NodeType.SWITCH: SwitchSettings, + NodeType.ROUTER: RouterSettings, + }[node.type] + + if not isinstance(node.settings, expected_type): + _add_issue( + issues, + location + ".settings", + "settings object does not match node type %r" % node.type.value, + ) + + +def _parse_ipv4_interface( + value: str, + location: str, + issues: List[ValidationIssue], +) -> Optional[ipaddress.IPv4Interface]: + """Parse and validate one IPv4 CIDR interface address. + + Args: + value: Address string expected to contain an IPv4 address and prefix. + location: Logical location associated with the address. + issues: Mutable list receiving validation failures. + + Returns: + Parsed IPv4Interface on success, otherwise None. + """ + + if "/" not in value: + _add_issue( + issues, + location, + "static address must use CIDR notation", + ) + return None + + try: + parsed = ipaddress.ip_interface(value) + except ValueError: + _add_issue(issues, location, "invalid IP interface address %r" % value) + return None + + if not isinstance(parsed, ipaddress.IPv4Interface): + _add_issue(issues, location, "only IPv4 addresses are supported") + return None + + return parsed + + +def _parse_ipv4_address( + value: str, + location: str, + issues: List[ValidationIssue], +) -> Optional[ipaddress.IPv4Address]: + """Parse and validate one IPv4 address without a prefix. + + Args: + value: Address string expected to contain one IPv4 address. + location: Logical location associated with the address. + issues: Mutable list receiving validation failures. + + Returns: + Parsed IPv4Address on success, otherwise None. + """ + + try: + parsed = ipaddress.ip_address(value) + except ValueError: + _add_issue(issues, location, "invalid IPv4 address %r" % value) + return None + + if not isinstance(parsed, ipaddress.IPv4Address): + _add_issue(issues, location, "only IPv4 addresses are supported") + return None + + return parsed + + +def _validate_interface( + node: Node, + interface: Interface, + location: str, + issues: List[ValidationIssue], +) -> None: + """Validate one declared logical interface. + + Args: + node: Node that owns the interface. + interface: Interface being validated. + location: Logical location of the interface. + issues: Mutable list receiving validation failures. + """ + + if not interface.name: + _add_issue(issues, location, "interface tag must not be empty") + + if interface.name in _RESERVED_INTERFACE_TAGS: + _add_issue( + issues, + location, + "interface tag %r is reserved by the grammar" % interface.name, + ) + + if node.type is NodeType.SWITCH and interface.addressing is not AddressingMode.NONE: + _add_issue( + issues, + location + ".addressing", + "Layer-2 switch interfaces must not request L3 addressing", + ) + + parsed_address: Optional[ipaddress.IPv4Interface] = None + + if interface.addressing is AddressingMode.STATIC: + if interface.address is None: + _add_issue( + issues, + location + ".address", + "static addressing requires an address", + ) + else: + parsed_address = _parse_ipv4_interface( + interface.address, + location + ".address", + issues, + ) + + if parsed_address is not None: + network = parsed_address.network + if ( + network.num_addresses > 2 + and parsed_address.ip in {network.network_address, network.broadcast_address} + ): + _add_issue( + issues, + location + ".address", + "static address must not be the network or broadcast address", + ) + + if interface.gateway is not None: + parsed_gateway = _parse_ipv4_address( + interface.gateway, + location + ".gateway", + issues, + ) + + if parsed_address is not None and parsed_gateway is not None: + if parsed_gateway not in parsed_address.network: + _add_issue( + issues, + location + ".gateway", + "gateway must belong to the interface subnet", + ) + elif parsed_gateway == parsed_address.ip: + _add_issue( + issues, + location + ".gateway", + "gateway must not be the interface's own address", + ) + elif ( + parsed_address.network.num_addresses > 2 + and parsed_gateway + in { + parsed_address.network.network_address, + parsed_address.network.broadcast_address, + } + ): + _add_issue( + issues, + location + ".gateway", + "gateway must not be the network or broadcast address", + ) + + elif interface.addressing in {AddressingMode.DHCP, AddressingMode.NONE}: + if interface.address is not None: + _add_issue( + issues, + location + ".address", + "address is only valid with static addressing", + ) + + if interface.gateway is not None: + _add_issue( + issues, + location + ".gateway", + "gateway is only valid with static addressing", + ) + + if interface.mac is not None and _MAC_ADDRESS_RE.fullmatch(interface.mac) is None: + _add_issue( + issues, + location + ".mac", + "MAC address must contain six hexadecimal octets separated by ':'", + ) + + if interface.mtu is not None and interface.mtu <= 0: + _add_issue( + issues, + location + ".mtu", + "MTU must be greater than zero", + ) + + +def _validate_nodes( + topology: Topology, + issues: List[ValidationIssue], +) -> _NodeIndex: + """Validate node declarations and their local interface configuration. + + Args: + topology: Parsed topology being validated. + issues: Mutable list receiving validation failures. + + Returns: + Node index prepared for later cross-reference checks. + """ + + node_index = _build_node_index(topology, issues) + + for node_number, node in enumerate(topology.nodes): + node_location = "nodes[%d]" % node_number + + _validate_node_settings(node, node_location, issues) + + if node.interfaces is None: + continue + + gateway_locations: List[str] = [] + + for interface_name, interface in node.interfaces.items(): + interface_location = "%s.interfaces.%s" % ( + node_location, + interface_name, + ) + + if interface.name != interface_name: + _add_issue( + issues, + interface_location, + "interface object name %r does not match its declaration tag %r" + % (interface.name, interface_name), + ) + + _validate_interface( + node, + interface, + interface_location, + issues, + ) + + if interface.gateway is not None: + gateway_locations.append(interface_location + ".gateway") + + if len(gateway_locations) > 1: + for gateway_location in gateway_locations: + _add_issue( + issues, + gateway_location, + "only one static default gateway is supported per node", + ) + + return node_index + + +def _validate_link_settings( + settings: LinkSettings, + location: str, + issues: List[ValidationIssue], +) -> None: + """Validate optional link provisioning settings. + + Args: + settings: Link settings being validated. + location: Logical location of the settings object. + issues: Mutable list receiving validation failures. + """ + + numeric_settings = { + "bandwidth_mbps": settings.bandwidth_mbps, + "delay_ms": settings.delay_ms, + "loss_percent": settings.loss_percent, + "jitter_ms": settings.jitter_ms, + } + + for name, value in numeric_settings.items(): + if value is not None and not math.isfinite(value): + _add_issue( + issues, + "%s.%s" % (location, name), + "value must be finite", + ) + + if settings.bandwidth_mbps is not None and settings.bandwidth_mbps <= 0: + _add_issue( + issues, + location + ".bandwidth_mbps", + "bandwidth must be greater than zero", + ) + + if settings.delay_ms is not None and settings.delay_ms < 0: + _add_issue( + issues, + location + ".delay_ms", + "delay must not be negative", + ) + + if settings.jitter_ms is not None and settings.jitter_ms < 0: + _add_issue( + issues, + location + ".jitter_ms", + "jitter must not be negative", + ) + + if ( + settings.loss_percent is not None + and not 0 <= settings.loss_percent <= 100 + ): + _add_issue( + issues, + location + ".loss_percent", + "loss percentage must be between 0 and 100", + ) + + +def _validate_link_names( + topology: Topology, + issues: List[ValidationIssue], +) -> None: + """Validate optional logical link names. + + Args: + topology: Parsed topology being validated. + issues: Mutable list receiving validation failures. + """ + + first_index: Dict[str, int] = {} + + for index, link in enumerate(topology.links): + if link.name is None: + continue + + location = "links[%d].name" % index + + if not link.name: + _add_issue(issues, location, "link name must not be empty") + continue + + if link.name in first_index: + _add_issue( + issues, + location, + "duplicate link name %r; first declared at links[%d]" + % (link.name, first_index[link.name]), + ) + continue + + first_index[link.name] = index + + +def _validate_links( + topology: Topology, + node_index: _NodeIndex, + issues: List[ValidationIssue], +) -> None: + """Validate links, endpoint references, and interface allocation capacity. + + Args: + topology: Parsed topology being validated. + node_index: Node lookup information prepared by node validation. + issues: Mutable list receiving validation failures. + """ + + _validate_link_names(topology, issues) + + used_explicit: Dict[Tuple[str, str], str] = {} + auto_requests: Dict[str, List[str]] = {} + + for link_index, link in enumerate(topology.links): + _validate_link_settings( + link.settings, + "links[%d].settings" % link_index, + issues, + ) + + for endpoint_index, endpoint in enumerate(link.endpoints): + endpoint_location = "links[%d].endpoints[%d]" % ( + link_index, + endpoint_index, + ) + + if endpoint.node not in node_index.nodes: + _add_issue( + issues, + endpoint_location + ".node", + "unknown node %r" % endpoint.node, + ) + continue + + if endpoint.node in node_index.duplicate_names: + # The duplicate-name error has already been reported. + # Cross-reference resolution would be ambiguous, so avoid + # generating secondary diagnostics from an arbitrary instance. + continue + + node = node_index.nodes[endpoint.node] + + if endpoint.interface is LinkInterfaceSelector.AUTO: + if node.interfaces is not None: + auto_requests.setdefault(endpoint.node, []).append( + endpoint_location + ".interface" + ) + continue + + interface_name = endpoint.interface + + if not interface_name: + _add_issue( + issues, + endpoint_location + ".interface", + "interface tag must not be empty", + ) + continue + + if node.interfaces is None: + _add_issue( + issues, + endpoint_location + ".interface", + "explicit interface %r cannot be referenced because node %r " + "does not declare an interface set" + % (interface_name, endpoint.node), + ) + continue + + if interface_name not in node.interfaces: + _add_issue( + issues, + endpoint_location + ".interface", + "node %r has no declared interface %r" + % (endpoint.node, interface_name), + ) + continue + + key = (endpoint.node, interface_name) + if key in used_explicit: + _add_issue( + issues, + endpoint_location + ".interface", + "interface %s:%s is already used by %s" + % ( + endpoint.node, + interface_name, + used_explicit[key], + ), + ) + continue + + used_explicit[key] = endpoint_location + + for node_name, request_locations in auto_requests.items(): + node = node_index.nodes[node_name] + + if node.interfaces is None: + continue + + used_count = sum( + 1 + for used_node, _ in used_explicit + if used_node == node_name + ) + free_count = len(node.interfaces) - used_count + + if len(request_locations) > free_count: + shortage = len(request_locations) - free_count + _add_issue( + issues, + "links", + "node %r has %d automatic interface request%s but only %d " + "unused declared interface%s remain%s" + % ( + node_name, + len(request_locations), + "" if len(request_locations) == 1 else "s", + free_count, + "" if free_count == 1 else "s", + "; %d request%s cannot be satisfied" + % ( + shortage, + "" if shortage == 1 else "s", + ), + ), + ) + + +def _resolve_service_interface( + service: Service, + service_location: str, + node: Node, + issues: List[ValidationIssue], +) -> Optional[Interface]: + """Resolve a service binding to one declared interface when possible. + + Args: + service: Service whose interface binding is resolved. + service_location: Logical location of the service. + node: Node on which the service runs. + issues: Mutable list receiving validation failures. + + Returns: + Resolved Interface when the binding identifies exactly one valid + interface, otherwise None. + """ + + selector = service.interface + + if isinstance(selector, str): + if node.interfaces is None: + _add_issue( + issues, + service_location + ".interface", + "explicit interface %r cannot be referenced because node %r " + "does not declare an interface set" + % (selector, node.name), + ) + return None + + if selector not in node.interfaces: + _add_issue( + issues, + service_location + ".interface", + "node %r has no declared interface %r" + % (node.name, selector), + ) + return None + + return node.interfaces[selector] + + if selector is ServiceInterfaceSelector.ALL: + return None + + if selector is ServiceInterfaceSelector.AUTO: + if node.interfaces is None: + _add_issue( + issues, + service_location + ".interface", + "automatic service binding requires declared interfaces", + ) + return None + + candidates = [ + interface + for interface in node.interfaces.values() + if interface.addressing is AddressingMode.STATIC + ] + + if not candidates: + _add_issue( + issues, + service_location + ".interface", + "automatic service binding found no statically addressed interface", + ) + return None + + if len(candidates) > 1: + _add_issue( + issues, + service_location + ".interface", + "automatic service binding is ambiguous across %d statically " + "addressed interfaces" % len(candidates), + ) + return None + + return candidates[0] + + return None + + +def _validate_dhcp_server( + service: Service, + service_location: str, + node: Node, + interface: Optional[Interface], + issues: List[ValidationIssue], +) -> None: + """Validate DHCP-server-specific semantics. + + Args: + service: DHCP server service being validated. + service_location: Logical location of the service. + node: Node on which the service runs. + interface: Resolved service interface, if available. + issues: Mutable list receiving validation failures. + """ + + if service.interface is ServiceInterfaceSelector.ALL: + _add_issue( + issues, + service_location + ".interface", + "DHCP server does not support the 'all' interface selector", + ) + + if not isinstance(service.settings, DHCPServerSettings): + _add_issue( + issues, + service_location + ".settings", + "DHCP server has an incompatible settings object", + ) + return + + start = _parse_ipv4_address( + service.settings.address_range.start, + service_location + ".settings.range.start", + issues, + ) + end = _parse_ipv4_address( + service.settings.address_range.end, + service_location + ".settings.range.end", + issues, + ) + + if start is not None and end is not None and int(start) > int(end): + _add_issue( + issues, + service_location + ".settings.range", + "DHCP range start must not be greater than range end", + ) + + gateway: Optional[ipaddress.IPv4Address] = None + if service.settings.gateway is not None: + gateway = _parse_ipv4_address( + service.settings.gateway, + service_location + ".settings.gateway", + issues, + ) + + if interface is None: + return + + if interface.addressing is not AddressingMode.STATIC: + _add_issue( + issues, + service_location + ".interface", + "DHCP server must bind to a statically addressed interface", + ) + return + + if interface.address is None: + # The interface validator reports this as well. + return + + parsed_interface = _parse_ipv4_interface( + interface.address, + service_location + ".interface", + issues, + ) + if parsed_interface is None: + return + + network = parsed_interface.network + + for value, location, label in ( + (start, service_location + ".settings.range.start", "range start"), + (end, service_location + ".settings.range.end", "range end"), + ): + if value is None: + continue + + if value not in network: + _add_issue( + issues, + location, + "%s must belong to DHCP interface subnet %s" + % (label, network), + ) + continue + + if ( + network.num_addresses > 2 + and value in {network.network_address, network.broadcast_address} + ): + _add_issue( + issues, + location, + "%s must not be the network or broadcast address" % label, + ) + + if gateway is not None: + if gateway not in network: + _add_issue( + issues, + service_location + ".settings.gateway", + "DHCP gateway must belong to DHCP interface subnet %s" + % network, + ) + elif ( + network.num_addresses > 2 + and gateway in {network.network_address, network.broadcast_address} + ): + _add_issue( + issues, + service_location + ".settings.gateway", + "DHCP gateway must not be the network or broadcast address", + ) + + if ( + start is not None + and end is not None + and int(start) <= int(end) + ): + server_address = parsed_interface.ip + + if int(start) <= int(server_address) <= int(end): + _add_issue( + issues, + service_location + ".settings.range", + "DHCP range must not include the server interface address %s" + % server_address, + ) + + if ( + gateway is not None + and int(start) <= int(gateway) <= int(end) + ): + _add_issue( + issues, + service_location + ".settings.range", + "DHCP range must not include the advertised gateway %s" + % gateway, + ) + + +def _validate_services( + topology: Topology, + node_index: _NodeIndex, + issues: List[ValidationIssue], +) -> None: + """Validate service references and service-specific semantics. + + Args: + topology: Parsed topology being validated. + node_index: Node lookup information prepared by node validation. + issues: Mutable list receiving validation failures. + """ + + for index, service in enumerate(topology.services): + location = "services[%d]" % index + + if service.node not in node_index.nodes: + _add_issue( + issues, + location + ".node", + "unknown node %r" % service.node, + ) + continue + + if service.node in node_index.duplicate_names: + continue + + node = node_index.nodes[service.node] + interface = _resolve_service_interface( + service, + location, + node, + issues, + ) + + if service.type is ServiceType.DHCP_SERVER: + _validate_dhcp_server( + service, + location, + node, + interface, + issues, + ) + + +def validate(topology: Topology) -> None: + """Validate the semantic consistency of a parsed topology. + + Args: + topology: Parsed topology produced by topology_parser.parse_topology(). + + Raises: + TopologyValidationError: If one or more semantic consistency errors are + discovered. + + Notes: + Validation is non-destructive and does not invoke Mininet or execute + Linux networking commands. All discovered semantic issues are + collected before the exception is raised. + """ + + issues: List[ValidationIssue] = [] + + node_index = _validate_nodes(topology, issues) + _validate_links(topology, node_index, issues) + _validate_services(topology, node_index, issues) + + if issues: + raise TopologyValidationError(issues) + + +def _self_test() -> None: + """Run smoke tests against the public validate() interface.""" + + from topology_parser import parse_topology + + valid_raw = { + "nodes": [ + { + "name": "h1", + "type": "host", + "interfaces": { + "net0": { + "addressing": "dhcp", + } + }, + }, + { + "name": "h3", + "type": "host", + "interfaces": { + "net0": { + "addressing": "static", + "address": "192.168.2.10/24", + "gateway": "192.168.2.1", + } + }, + }, + { + "name": "s1", + "type": "switch", + }, + { + "name": "s2", + "type": "switch", + }, + { + "name": "r0", + "type": "router", + "interfaces": { + "lan": { + "addressing": "static", + "address": "192.168.1.1/24", + }, + "wan": { + "addressing": "static", + "address": "192.168.2.1/24", + }, + }, + "settings": { + "ip_forward": True, + "rp_filter": False, + }, + }, + ], + "links": [ + { + "endpoints": [ + {"node": "h1", "interface": "net0"}, + {"node": "s1", "interface": "auto"}, + ] + }, + { + "endpoints": [ + {"node": "r0", "interface": "lan"}, + {"node": "s1", "interface": "auto"}, + ] + }, + { + "endpoints": [ + {"node": "r0", "interface": "wan"}, + {"node": "s2", "interface": "auto"}, + ] + }, + { + "endpoints": [ + {"node": "h3", "interface": "net0"}, + {"node": "s2", "interface": "auto"}, + ] + }, + ], + "services": [ + { + "type": "dhcp-server", + "node": "r0", + "interface": "lan", + "settings": { + "range": { + "start": "192.168.1.100", + "end": "192.168.1.200", + }, + "gateway": "192.168.1.1", + }, + } + ], + } + + validate(parse_topology(valid_raw)) + + invalid_raw = { + "nodes": [ + { + "name": "h1", + "type": "host", + "interfaces": { + "net0": { + "addressing": "dhcp", + "gateway": "192.168.1.1", + } + }, + }, + { + "name": "h1", + "type": "host", + }, + { + "name": "s1", + "type": "switch", + "interfaces": { + "p1": {}, + }, + }, + ], + "links": [ + { + "endpoints": [ + {"node": "missing", "interface": "auto"}, + {"node": "s1", "interface": "p1"}, + ] + }, + { + "endpoints": [ + {"node": "s1", "interface": "p1"}, + {"node": "s1", "interface": "auto"}, + ] + }, + ], + "services": [], + } + + try: + validate(parse_topology(invalid_raw)) + except TopologyValidationError as exc: + assert len(exc.issues) >= 4 + assert any("duplicate node name" in issue.message for issue in exc.issues) + assert any("unknown node" in issue.message for issue in exc.issues) + assert any("already used" in issue.message for issue in exc.issues) + assert any("only valid with static addressing" in issue.message for issue in exc.issues) + else: + raise AssertionError( + "invalid topology must raise TopologyValidationError" + ) + + +if __name__ == "__main__": + _self_test() + print("topology_validator: self-test passed")