1153 lines
33 KiB
Python
1153 lines
33 KiB
Python
"""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 <cchoutou@ece.auth.gr>
|
|
"""
|
|
|
|
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")
|