diff --git a/source/topology_loader.py b/source/topology_loader.py index 456b571..5fcc751 100644 --- a/source/topology_loader.py +++ b/source/topology_loader.py @@ -10,6 +10,8 @@ The loading stage has two public entry points: model by calling topology_parser.parse_topology(). No Mininet code is imported or executed in this module. + +Author: Christos Choutouridis """ import json diff --git a/source/topology_plan.py b/source/topology_plan.py new file mode 100644 index 0000000..d334149 --- /dev/null +++ b/source/topology_plan.py @@ -0,0 +1,812 @@ +"""Lower a validated topology into a fully resolved execution plan. + +This module is the planning stage of the topology pipeline. + +The parser produces a typed description of the user's intent and the validator +proves that the description is semantically coherent. The planner then makes +all implementation choices that must be resolved before executable Mininet +Python can be generated. + +In particular, the planner resolves logical and automatic interface references +into concrete Linux/Mininet interface names. It does not import Mininet, +execute commands, modify the network, or render Python source. + +Author: Christos Choutouridis +""" + +from dataclasses import dataclass +from typing import Dict, List, Optional, Set, Tuple, Union + +from topology_parser import ( + AddressingMode, + DHCPServerSettings, + Interface, + LinkInterfaceSelector, + Node, + NodeType, + RouterSettings, + Service, + ServiceInterfaceSelector, + ServiceType, + Topology, +) + + +class TopologyPlanError(ValueError): + """Raised when a validated topology cannot be lowered into a plan.""" + + +@dataclass(frozen=True) +class PlannedNode: + """Describe one concrete node creation operation. + + Args: + name: Logical node name used by Mininet. + type: Concrete semantic node type to create. + """ + + name: str + type: NodeType + + +@dataclass(frozen=True) +class PlannedLinkEndpoint: + """Describe one fully resolved endpoint of a planned link. + + Args: + node: Logical node name that owns the endpoint. + logical_interface: Resolved logical interface identifier. For a node + with a dynamic interface set, the planner creates a deterministic + synthetic identifier for an automatically allocated interface. + interface_name: Concrete Linux/Mininet interface name. + """ + + node: str + logical_interface: str + interface_name: str + + +@dataclass(frozen=True) +class PlannedLinkSettings: + """Contain resolved provisioning settings for one link. + + Args: + bandwidth_mbps: Optional bandwidth limit in megabits per second. + delay_ms: Optional one-way delay in milliseconds. + loss_percent: Optional packet-loss percentage. + jitter_ms: Optional delay variation in milliseconds. + """ + + bandwidth_mbps: Optional[float] = None + delay_ms: Optional[float] = None + loss_percent: Optional[float] = None + jitter_ms: Optional[float] = None + + +@dataclass(frozen=True) +class PlannedLink: + """Describe one fully resolved Mininet link creation operation. + + Args: + endpoints: Ordered pair of fully resolved link endpoints. + name: Optional logical link name preserved for diagnostics. + settings: Concrete link provisioning settings. + """ + + endpoints: Tuple[PlannedLinkEndpoint, PlannedLinkEndpoint] + name: Optional[str] + settings: PlannedLinkSettings + + +@dataclass(frozen=True) +class PlannedInterface: + """Describe post-link configuration for one concrete interface. + + Args: + node: Logical node name that owns the interface. + logical_interface: Resolved logical interface identifier. + interface_name: Concrete Linux/Mininet interface name. + addressing: Layer-3 addressing strategy. + address: Optional static CIDR address. + gateway: Optional static default gateway. + mac: Optional explicit MAC address. + mtu: Optional MTU value. + """ + + node: str + logical_interface: str + interface_name: str + addressing: AddressingMode + address: Optional[str] + gateway: Optional[str] + mac: Optional[str] + mtu: Optional[int] + + +@dataclass(frozen=True) +class PlannedSysctl: + """Describe one namespace-scoped sysctl operation. + + Args: + node: Logical node on which the sysctl is applied. + key: Fully resolved sysctl key. + value: Integer value written to the key. + """ + + node: str + key: str + value: int + + +@dataclass(frozen=True) +class PlannedDHCPServerSettings: + """Contain fully resolved DHCP server settings. + + Args: + range_start: First address offered to DHCP clients. + range_end: Last address offered to DHCP clients. + gateway: Optional default gateway advertised to DHCP clients. + """ + + range_start: str + range_end: str + gateway: Optional[str] + + +PlannedServiceSettings = PlannedDHCPServerSettings + + +@dataclass(frozen=True) +class PlannedService: + """Describe one fully resolved service startup operation. + + Args: + type: Service implementation type. + node: Logical node on which the service runs. + logical_interface: Resolved logical interface identifier. + interface_name: Concrete Linux/Mininet interface name to bind. + settings: Service-type-specific resolved settings. + """ + + type: ServiceType + node: str + logical_interface: str + interface_name: str + settings: PlannedServiceSettings + + +@dataclass(frozen=True) +class ExecutionPlan: + """Contain the complete, fully resolved execution plan. + + Args: + nodes: Node creation operations in declaration order. + links: Link creation operations in declaration order. + interfaces: Concrete interface configuration operations. + node_settings: Concrete namespace-level configuration operations. + services: Service startup operations in declaration order. + + Notes: + The plan contains no AUTO or ALL selectors. All such decisions are + resolved before an ExecutionPlan is returned. + """ + + nodes: Tuple[PlannedNode, ...] + links: Tuple[PlannedLink, ...] + interfaces: Tuple[PlannedInterface, ...] + node_settings: Tuple[PlannedSysctl, ...] + services: Tuple[PlannedService, ...] + + +@dataclass +class _PlanningContext: + """Hold mutable state used while lowering a topology. + + Args: + nodes: Nodes indexed by logical name. + reserved_interfaces: Explicitly referenced logical interfaces per node. + free_interfaces: Remaining declared interfaces available to AUTO. + dynamic_interface_counters: Synthetic logical-name counters for nodes + whose interface sets were omitted. + physical_interface_counters: Concrete eth-index counters per node. + interface_names: Mapping from resolved logical interfaces to concrete + Linux/Mininet interface names. + interface_models: Mapping from resolved logical interfaces to their + source Interface objects. + """ + + nodes: Dict[str, Node] + reserved_interfaces: Dict[str, Set[str]] + free_interfaces: Dict[str, List[str]] + dynamic_interface_counters: Dict[str, int] + physical_interface_counters: Dict[str, int] + interface_names: Dict[Tuple[str, str], str] + interface_models: Dict[Tuple[str, str], Interface] + + +def _build_context(topology: Topology) -> _PlanningContext: + """Create planning state and reserve all explicit link interfaces. + + Args: + topology: Validated topology to lower. + + Returns: + Initialized mutable planning context. + """ + + nodes = {node.name: node for node in topology.nodes} + reserved: Dict[str, Set[str]] = {node.name: set() for node in topology.nodes} + + for link in topology.links: + for endpoint in link.endpoints: + if isinstance(endpoint.interface, str): + reserved[endpoint.node].add(endpoint.interface) + + free_interfaces: Dict[str, List[str]] = {} + for node in topology.nodes: + if node.interfaces is None: + continue + + free_interfaces[node.name] = [ + name + for name in node.interfaces.keys() + if name not in reserved[node.name] + ] + + return _PlanningContext( + nodes=nodes, + reserved_interfaces=reserved, + free_interfaces=free_interfaces, + dynamic_interface_counters={node.name: 0 for node in topology.nodes}, + physical_interface_counters={node.name: 0 for node in topology.nodes}, + interface_names={}, + interface_models={}, + ) + + +def _plan_nodes(topology: Topology) -> Tuple[PlannedNode, ...]: + """Lower topology nodes into concrete node creation operations. + + Args: + topology: Validated topology to lower. + + Returns: + Planned nodes in source declaration order. + """ + + return tuple( + PlannedNode(name=node.name, type=node.type) + for node in topology.nodes + ) + + +def _allocate_logical_interface( + node: Node, + selector: Union[str, LinkInterfaceSelector], + context: _PlanningContext, +) -> Tuple[str, Interface]: + """Resolve one link endpoint to a concrete logical interface. + + Args: + node: Node that owns the endpoint. + selector: Explicit interface tag or AUTO selector. + context: Mutable planning context. + + Returns: + Pair containing the resolved logical interface name and its interface + configuration model. + + Raises: + TopologyPlanError: If the topology violates a planner precondition. + """ + + if isinstance(selector, str): + if node.interfaces is None or selector not in node.interfaces: + raise TopologyPlanError( + "explicit interface %r cannot be resolved on node %r" + % (selector, node.name) + ) + + return selector, node.interfaces[selector] + + if selector is not LinkInterfaceSelector.AUTO: + raise TopologyPlanError( + "unsupported link interface selector %r" % selector + ) + + if node.interfaces is not None: + free = context.free_interfaces[node.name] + if not free: + raise TopologyPlanError( + "node %r has no declared interface available for AUTO" + % node.name + ) + + logical_name = free.pop(0) + return logical_name, node.interfaces[logical_name] + + counter = context.dynamic_interface_counters[node.name] + context.dynamic_interface_counters[node.name] = counter + 1 + + logical_name = "auto%d" % counter + return logical_name, Interface(name=logical_name) + + +def _allocate_physical_interface( + node_name: str, + logical_name: str, + interface: Interface, + context: _PlanningContext, +) -> str: + """Assign one deterministic concrete interface name. + + Args: + node_name: Logical node that owns the interface. + logical_name: Resolved logical interface identifier. + interface: Interface configuration model. + context: Mutable planning context. + + Returns: + Concrete Linux/Mininet interface name. + + Raises: + TopologyPlanError: If the same logical interface is allocated twice. + """ + + key = (node_name, logical_name) + if key in context.interface_names: + raise TopologyPlanError( + "interface %s:%s was allocated more than once" + % (node_name, logical_name) + ) + + index = context.physical_interface_counters[node_name] + context.physical_interface_counters[node_name] = index + 1 + + interface_name = "%s-eth%d" % (node_name, index) + context.interface_names[key] = interface_name + context.interface_models[key] = interface + + return interface_name + + +def _plan_links( + topology: Topology, + context: _PlanningContext, +) -> Tuple[PlannedLink, ...]: + """Resolve all link endpoints and concrete interface names. + + Args: + topology: Validated topology to lower. + context: Mutable planning context. + + Returns: + Planned links in source declaration order. + """ + + planned_links: List[PlannedLink] = [] + + for link in topology.links: + planned_endpoints: List[PlannedLinkEndpoint] = [] + + for endpoint in link.endpoints: + node = context.nodes[endpoint.node] + logical_name, interface = _allocate_logical_interface( + node, + endpoint.interface, + context, + ) + interface_name = _allocate_physical_interface( + node.name, + logical_name, + interface, + context, + ) + + planned_endpoints.append( + PlannedLinkEndpoint( + node=node.name, + logical_interface=logical_name, + interface_name=interface_name, + ) + ) + + planned_links.append( + PlannedLink( + endpoints=(planned_endpoints[0], planned_endpoints[1]), + name=link.name, + settings=PlannedLinkSettings( + bandwidth_mbps=link.settings.bandwidth_mbps, + delay_ms=link.settings.delay_ms, + loss_percent=link.settings.loss_percent, + jitter_ms=link.settings.jitter_ms, + ), + ) + ) + + return tuple(planned_links) + + +def _interface_requires_materialization(interface: Interface) -> bool: + """Return whether an unused declared interface carries configuration. + + Args: + interface: Declared interface to inspect. + + Returns: + True when silently leaving the interface unmaterialized would discard + requested configuration. + """ + + return ( + interface.addressing is not AddressingMode.NONE + or interface.address is not None + or interface.gateway is not None + or interface.mac is not None + or interface.mtu is not None + ) + + +def _plan_interfaces( + topology: Topology, + context: _PlanningContext, +) -> Tuple[PlannedInterface, ...]: + """Create interface-configuration operations for materialized interfaces. + + Args: + topology: Validated topology to lower. + context: Planning context containing resolved interface mappings. + + Returns: + Planned interface configuration operations. + + Raises: + TopologyPlanError: If a configured declared interface is not attached to + any link and therefore has no concrete Mininet interface to + configure. + """ + + for node in topology.nodes: + if node.interfaces is None: + continue + + for logical_name, interface in node.interfaces.items(): + key = (node.name, logical_name) + if ( + key not in context.interface_names + and _interface_requires_materialization(interface) + ): + raise TopologyPlanError( + "configured interface %s:%s is not attached to any link" + % (node.name, logical_name) + ) + + planned: List[PlannedInterface] = [] + + # The mapping insertion order is the exact allocation order from + # _plan_links(), which is deterministic on supported Python versions. + for key, interface_name in context.interface_names.items(): + node_name, logical_name = key + interface = context.interface_models[key] + + planned.append( + PlannedInterface( + node=node_name, + logical_interface=logical_name, + interface_name=interface_name, + addressing=interface.addressing, + address=interface.address, + gateway=interface.gateway, + mac=interface.mac, + mtu=interface.mtu, + ) + ) + + return tuple(planned) + + +def _plan_node_settings( + topology: Topology, + context: _PlanningContext, +) -> Tuple[PlannedSysctl, ...]: + """Lower node settings into concrete namespace sysctl operations. + + Args: + topology: Validated topology to lower. + context: Planning context containing concrete interface names. + + Returns: + Planned sysctl operations in deterministic order. + """ + + planned: List[PlannedSysctl] = [] + + for node in topology.nodes: + if node.type is not NodeType.ROUTER: + continue + + if not isinstance(node.settings, RouterSettings): + raise TopologyPlanError( + "router %r does not carry RouterSettings" % node.name + ) + + if node.settings.ip_forward is not None: + planned.append( + PlannedSysctl( + node=node.name, + key="net.ipv4.ip_forward", + value=1 if node.settings.ip_forward else 0, + ) + ) + + if node.settings.rp_filter is not None: + value = 1 if node.settings.rp_filter else 0 + + planned.append( + PlannedSysctl( + node=node.name, + key="net.ipv4.conf.all.rp_filter", + value=value, + ) + ) + planned.append( + PlannedSysctl( + node=node.name, + key="net.ipv4.conf.default.rp_filter", + value=value, + ) + ) + + for (owner, _), interface_name in context.interface_names.items(): + if owner != node.name: + continue + + planned.append( + PlannedSysctl( + node=node.name, + key="net.ipv4.conf.%s.rp_filter" % interface_name, + value=value, + ) + ) + + return tuple(planned) + + +def _resolve_service_logical_interface( + service: Service, + node: Node, +) -> str: + """Resolve one validated service interface selector. + + Args: + service: Service whose interface binding is resolved. + node: Node on which the service runs. + + Returns: + Concrete logical interface tag. + + Raises: + TopologyPlanError: If the selector cannot be lowered uniquely. + """ + + if isinstance(service.interface, str): + return service.interface + + if service.interface is ServiceInterfaceSelector.AUTO: + if node.interfaces is None: + raise TopologyPlanError( + "automatic service binding on node %r requires declared " + "interfaces" % node.name + ) + + candidates = [ + name + for name, interface in node.interfaces.items() + if interface.addressing is AddressingMode.STATIC + ] + + if len(candidates) != 1: + raise TopologyPlanError( + "automatic service binding on node %r is not uniquely " + "resolvable" % node.name + ) + + return candidates[0] + + if service.interface is ServiceInterfaceSelector.ALL: + raise TopologyPlanError( + "service type %r cannot be lowered from the ALL selector" + % service.type.value + ) + + raise TopologyPlanError( + "unsupported service interface selector %r" % service.interface + ) + + +def _plan_services( + topology: Topology, + context: _PlanningContext, +) -> Tuple[PlannedService, ...]: + """Resolve service bindings and lower service-specific settings. + + Args: + topology: Validated topology to lower. + context: Planning context containing concrete interface names. + + Returns: + Planned service startup operations in declaration order. + + Raises: + TopologyPlanError: If a service cannot be mapped to a materialized + interface or has unsupported settings. + """ + + planned: List[PlannedService] = [] + + for service in topology.services: + node = context.nodes[service.node] + logical_name = _resolve_service_logical_interface(service, node) + key = (node.name, logical_name) + + if key not in context.interface_names: + raise TopologyPlanError( + "service %r binds to interface %s:%s, but that interface is " + "not attached to any link" + % (service.type.value, node.name, logical_name) + ) + + if service.type is ServiceType.DHCP_SERVER: + if not isinstance(service.settings, DHCPServerSettings): + raise TopologyPlanError( + "DHCP service on %r has incompatible settings" % node.name + ) + + settings = PlannedDHCPServerSettings( + range_start=service.settings.address_range.start, + range_end=service.settings.address_range.end, + gateway=service.settings.gateway, + ) + else: + raise TopologyPlanError( + "unsupported service type %r" % service.type.value + ) + + planned.append( + PlannedService( + type=service.type, + node=node.name, + logical_interface=logical_name, + interface_name=context.interface_names[key], + settings=settings, + ) + ) + + return tuple(planned) + + +def plan(topology: Topology) -> ExecutionPlan: + """Lower a validated topology into a fully resolved execution plan. + + Args: + topology: Topology that has already passed semantic validation. + + Returns: + Fully resolved execution plan suitable for source-code rendering. + + Raises: + TopologyPlanError: If a planner precondition is violated or if a + validated construct cannot be lowered by the current backend. + + Notes: + This function deliberately does not call validate(). Validation and + planning remain separate pipeline stages, and callers are expected to + validate the topology before invoking the planner. + """ + + context = _build_context(topology) + + nodes = _plan_nodes(topology) + links = _plan_links(topology, context) + interfaces = _plan_interfaces(topology, context) + node_settings = _plan_node_settings(topology, context) + services = _plan_services(topology, context) + + return ExecutionPlan( + nodes=nodes, + links=links, + interfaces=interfaces, + node_settings=node_settings, + services=services, + ) + + +def _self_test() -> None: + """Run a smoke test against the public plan() interface.""" + + from topology_parser import parse_topology + from topology_validator import validate + + raw = { + "nodes": [ + { + "name": "h1", + "type": "host", + "interfaces": { + "net0": { + "addressing": "dhcp", + } + }, + }, + { + "name": "s1", + "type": "switch", + "interfaces": { + "client": {}, + "uplink": {}, + }, + }, + { + "name": "r0", + "type": "router", + "interfaces": { + "lan": { + "addressing": "static", + "address": "192.168.1.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": "uplink"}, + ] + }, + ], + "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", + }, + } + ], + } + + topology = parse_topology(raw) + validate(topology) + execution_plan = plan(topology) + + assert len(execution_plan.nodes) == 3 + assert len(execution_plan.links) == 2 + assert execution_plan.links[0].endpoints[1].logical_interface == "client" + assert execution_plan.links[0].endpoints[0].interface_name == "h1-eth0" + assert execution_plan.links[1].endpoints[0].interface_name == "r0-eth0" + assert execution_plan.services[0].interface_name == "r0-eth0" + assert any( + setting.key == "net.ipv4.ip_forward" and setting.value == 1 + for setting in execution_plan.node_settings + ) + + +if __name__ == "__main__": + _self_test() + print("topology_plan: self-test passed") diff --git a/source/topology_renderer.py b/source/topology_renderer.py new file mode 100644 index 0000000..ed78b04 --- /dev/null +++ b/source/topology_renderer.py @@ -0,0 +1,482 @@ +"""Render a resolved execution plan as executable Mininet Python source. + +The renderer is intentionally a serialization stage. It does not perform +semantic validation, allocate AUTO interfaces, import Mininet itself, or make +networking decisions. All topology decisions must already be represented in +the ExecutionPlan produced by topology_plan.plan(). + +The returned Python source is both an inspectable intermediate artifact and the +canonical executable form of the planned topology. + +Author: Christos Choutouridis +""" + +import ast +from typing import List + +from topology_parser import AddressingMode, NodeType, ServiceType +from topology_plan import ( + ExecutionPlan, + PlannedDHCPServerSettings, + PlannedInterface, + PlannedLink, + PlannedService, +) + + +def _python_string(value: str) -> str: + """Return a safe Python string literal. + + Args: + value: String to serialize into generated Python source. + + Returns: + Python source literal representing the string exactly. + """ + + return repr(value) + + +def _render_link_call(link: PlannedLink) -> List[str]: + """Render one Mininet addLink() call. + + Args: + link: Fully resolved link to render. + + Returns: + Source lines implementing the link creation operation. + """ + + first, second = link.endpoints + settings = link.settings + + arguments = [ + "nodes[%s]" % _python_string(first.node), + "nodes[%s]" % _python_string(second.node), + "intfName1=%s" % _python_string(first.interface_name), + "intfName2=%s" % _python_string(second.interface_name), + ] + + has_tc_settings = any( + value is not None + for value in ( + settings.bandwidth_mbps, + settings.delay_ms, + settings.loss_percent, + settings.jitter_ms, + ) + ) + + if has_tc_settings: + arguments.append("cls=TCLink") + + if settings.bandwidth_mbps is not None: + arguments.append("bw=%r" % settings.bandwidth_mbps) + + if settings.delay_ms is not None: + arguments.append("delay=%s" % _python_string("%gms" % settings.delay_ms)) + elif settings.jitter_ms is not None: + # Linux netem expresses jitter as variation around a base delay. A + # zero base delay preserves the declarative meaning when only jitter + # was provided in the topology. + arguments.append("delay=%s" % _python_string("0ms")) + + if settings.loss_percent is not None: + arguments.append("loss=%r" % settings.loss_percent) + + if settings.jitter_ms is not None: + arguments.append("jitter=%s" % _python_string("%gms" % settings.jitter_ms)) + + lines = [" net.addLink("] + for argument in arguments: + lines.append(" %s," % argument) + lines.append(" )") + + return lines + + +def _render_interface_configuration(interface: PlannedInterface) -> List[str]: + """Render configuration commands for one concrete interface. + + Args: + interface: Fully resolved interface configuration operation. + + Returns: + Generated Python source lines for the interface. + """ + + node = "nodes[%s]" % _python_string(interface.node) + name = _python_string(interface.interface_name) + lines: List[str] = [] + + if interface.mac is not None: + lines.append( + " _node_cmd(%s, 'ip', 'link', 'set', 'dev', %s, " + "'address', %s)" + % (node, name, _python_string(interface.mac)) + ) + + if interface.mtu is not None: + lines.append( + " _node_cmd(%s, 'ip', 'link', 'set', 'dev', %s, " + "'mtu', %s)" + % (node, name, _python_string(str(interface.mtu))) + ) + + if interface.addressing in { + AddressingMode.STATIC, + AddressingMode.DHCP, + }: + # addHost(ip=None) prevents Mininet from assigning its default address, + # and this flush makes the generated artifact explicit about the + # desired IPv4 state before static or manual-DHCP configuration. + lines.append( + " _node_cmd(%s, 'ip', '-4', 'addr', 'flush', 'dev', %s)" + % (node, name) + ) + + if interface.addressing is AddressingMode.STATIC: + if interface.address is None: + raise ValueError( + "planned static interface %s:%s has no address" + % (interface.node, interface.logical_interface) + ) + + lines.append( + " _node_cmd(%s, 'ip', 'addr', 'add', %s, 'dev', %s)" + % ( + node, + _python_string(interface.address), + name, + ) + ) + + if ( + interface.addressing in {AddressingMode.STATIC, AddressingMode.DHCP} + or interface.mac is not None + or interface.mtu is not None + ): + lines.append( + " _node_cmd(%s, 'ip', 'link', 'set', 'dev', %s, 'up')" + % (node, name) + ) + + return lines + + +def _render_gateway_configuration(interface: PlannedInterface) -> List[str]: + """Render an optional static default route for one interface. + + Args: + interface: Fully resolved interface configuration operation. + + Returns: + Generated Python source lines, possibly empty. + """ + + if interface.gateway is None: + return [] + + node = "nodes[%s]" % _python_string(interface.node) + + return [ + " _node_cmd(%s, 'ip', 'route', 'replace', 'default', 'via', " + "%s, 'dev', %s)" + % ( + node, + _python_string(interface.gateway), + _python_string(interface.interface_name), + ) + ] + + +def _render_service_start(service: PlannedService, index: int) -> List[str]: + """Render startup code for one planned service. + + Args: + service: Fully resolved service operation. + index: Stable service index used for temporary runtime files. + + Returns: + Generated Python source lines that start and track the service. + + Raises: + ValueError: If the plan contains an unsupported service type or + incompatible settings object. + """ + + if service.type is not ServiceType.DHCP_SERVER: + raise ValueError( + "unsupported planned service type %r" % service.type.value + ) + + if not isinstance(service.settings, PlannedDHCPServerSettings): + raise ValueError("DHCP service has incompatible planned settings") + + node = "nodes[%s]" % _python_string(service.node) + lease_file = "/tmp/mininet-dnsmasq-%d.leases" % index + pid_file = "/tmp/mininet-dnsmasq-%d.pid" % index + log_file = "/tmp/mininet-dnsmasq-%d.log" % index + + args = [ + "dnsmasq", + "--no-daemon", + "--conf-file=", + "--port=0", + "--bind-interfaces", + "--dhcp-authoritative", + "--interface=%s" % service.interface_name, + "--dhcp-range=%s,%s,12h" + % (service.settings.range_start, service.settings.range_end), + "--dhcp-leasefile=%s" % lease_file, + "--pid-file=%s" % pid_file, + ] + + if service.settings.gateway is not None: + args.append( + "--dhcp-option=3,%s" % service.settings.gateway + ) + + rendered_args = ", ".join(_python_string(arg) for arg in args) + + return [ + " pid = _start_background(", + " %s," % node, + " [%s]," % rendered_args, + " %s," % _python_string(log_file), + " )", + " service_processes.append((%s, pid))" % node, + ] + + +def render_python(plan: ExecutionPlan) -> str: + """Render an execution plan as a complete executable Python program. + + Args: + plan: Fully resolved execution plan returned by topology_plan.plan(). + + Returns: + Python source code that creates the network, applies configuration, + starts services, opens the Mininet CLI, and performs structured cleanup. + + Raises: + ValueError: If the supplied plan contains an operation that this + renderer cannot serialize. + """ + + lines: List[str] = [ + "#!/usr/bin/env python3", + '"""Generated Mininet topology. Do not edit by hand."""', + "", + "import shlex", + "", + "from mininet.cli import CLI", + "from mininet.link import TCLink", + "from mininet.net import Mininet", + "", + "", + "def _shell_join(args):", + ' """Quote command arguments for execution inside a Mininet node."""', + "", + " return ' '.join(shlex.quote(str(arg)) for arg in args)", + "", + "", + "def _node_cmd(node, *args):", + ' """Execute one safely quoted command inside a Mininet node."""', + "", + " return node.cmd(_shell_join(args))", + "", + "", + "def _start_background(node, args, log_path):", + ' """Start one background process and return its shell PID."""', + "", + " command = _shell_join(args)", + " shell_command = (", + " command", + " + ' >'", + " + shlex.quote(log_path)", + " + ' 2>&1 & echo $!'", + " )", + " return node.cmd('sh -c ' + shlex.quote(shell_command)).strip()", + "", + "", + "def main():", + ' """Build, configure, expose, and clean up the planned network."""', + "", + " net = Mininet(controller=None, build=False)", + " nodes = {}", + " service_processes = []", + "", + " try:", + " # Create nodes.", + ] + + for node in plan.nodes: + if node.type in {NodeType.HOST, NodeType.ROUTER}: + lines.append( + " nodes[%s] = net.addHost(%s, ip=None)" + % ( + _python_string(node.name), + _python_string(node.name), + ) + ) + elif node.type is NodeType.SWITCH: + lines.append( + " nodes[%s] = net.addSwitch(%s)" + % ( + _python_string(node.name), + _python_string(node.name), + ) + ) + else: + raise ValueError("unsupported planned node type %r" % node.type) + + lines.extend(["", " # Create links."]) + for link in plan.links: + lines.extend(_render_link_call(link)) + + lines.extend( + [ + "", + " # Materialize the Mininet topology.", + " net.build()", + " net.start()", + "", + " # Configure concrete interfaces.", + ] + ) + + for interface in plan.interfaces: + lines.extend(_render_interface_configuration(interface)) + + lines.extend(["", " # Configure static default routes."]) + for interface in plan.interfaces: + lines.extend(_render_gateway_configuration(interface)) + + lines.extend(["", " # Apply node-level networking settings."]) + for setting in plan.node_settings: + lines.append( + " _node_cmd(nodes[%s], 'sysctl', '-w', %s)" + % ( + _python_string(setting.node), + _python_string("%s=%d" % (setting.key, setting.value)), + ) + ) + + lines.extend(["", " # Start planned services."]) + for index, service in enumerate(plan.services): + lines.extend(_render_service_start(service, index)) + + lines.extend( + [ + "", + " # DHCP clients are intentionally not started here.", + " # The assignment captures DHCP before dhclient is run manually.", + " CLI(net)", + "", + " finally:", + " # Stop only the service processes started by this script.", + " for node, pid in reversed(service_processes):", + " if pid:", + " _node_cmd(node, 'kill', pid)", + "", + " net.stop()", + "", + "", + "if __name__ == '__main__':", + " main()", + "", + ] + ) + + return "\n".join(lines) + + +def _self_test() -> None: + """Run a smoke test against the public render_python() interface.""" + + from topology_parser import parse_topology + from topology_plan import plan + from topology_validator import validate + + raw = { + "nodes": [ + { + "name": "h1", + "type": "host", + "interfaces": { + "net0": { + "addressing": "dhcp", + } + }, + }, + { + "name": "s1", + "type": "switch", + }, + { + "name": "r0", + "type": "router", + "interfaces": { + "lan": { + "addressing": "static", + "address": "192.168.1.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"}, + ], + "settings": { + "delay_ms": 1.5, + }, + }, + ], + "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", + }, + } + ], + } + + topology = parse_topology(raw) + validate(topology) + source = render_python(plan(topology)) + + ast.parse(source) + + assert "net.addHost('h1', ip=None)" in source + assert "net.addSwitch('s1')" in source + assert "net.addLink(" in source + assert "cls=TCLink" in source + assert "dnsmasq" in source + assert "CLI(net)" in source + assert "dhclient" in source + + +if __name__ == "__main__": + _self_test() + print("topology_renderer: self-test passed")