Merged PR 243: #2139 - Implemented routing
## Summary - Integrated the RouteTable into the Routers frame processing. - Frames are now dropped when their TTL reaches 0 ## Test process Added five tests that check routing passes and fails with correct/incorrect route tables. ## Checklist - [ ] PR is linked to a **work item** - [ ] **acceptance criteria** of linked ticket are met - [ ] performed **self-review** of the code - [ ] written **tests** for any new functionality added with this PR - [ ] updated the **documentation** if this PR changes or adds functionality - [ ] written/updated **design docs** if this PR implements new functionality - [ ] updated the **change log** - [ ] ran **pre-commit** checks for code style - [ ] attended to any **TO-DOs** left in the code #2139 - Implemented routing Related work items: #2139
This commit is contained in:
Christopher McCarthy
180
tests/integration_tests/network/test_broadcast.py
Normal file
180
tests/integration_tests/network/test_broadcast.py
Normal file
@@ -0,0 +1,180 @@
|
||||
from ipaddress import IPv4Address, IPv4Network
|
||||
from typing import Any, Dict, List, Tuple
|
||||
|
||||
import pytest
|
||||
|
||||
from primaite.simulator.network.container import Network
|
||||
from primaite.simulator.network.hardware.nodes.computer import Computer
|
||||
from primaite.simulator.network.hardware.nodes.server import Server
|
||||
from primaite.simulator.network.hardware.nodes.switch import Switch
|
||||
from primaite.simulator.network.transmission.network_layer import IPProtocol
|
||||
from primaite.simulator.network.transmission.transport_layer import Port
|
||||
from primaite.simulator.system.applications.application import Application
|
||||
from primaite.simulator.system.services.service import Service
|
||||
|
||||
|
||||
class BroadcastService(Service):
|
||||
"""A service for sending broadcast and unicast messages over a network."""
|
||||
|
||||
def __init__(self, **kwargs):
|
||||
# Set default service properties for broadcasting
|
||||
kwargs["name"] = "BroadcastService"
|
||||
kwargs["port"] = Port.HTTP
|
||||
kwargs["protocol"] = IPProtocol.TCP
|
||||
super().__init__(**kwargs)
|
||||
|
||||
def describe_state(self) -> Dict:
|
||||
# Implement state description for the service
|
||||
pass
|
||||
|
||||
def unicast(self, ip_address: IPv4Address):
|
||||
# Send a unicast payload to a specific IP address
|
||||
super().send(
|
||||
payload="unicast",
|
||||
dest_ip_address=ip_address,
|
||||
dest_port=Port.HTTP,
|
||||
)
|
||||
|
||||
def broadcast(self, ip_network: IPv4Network):
|
||||
# Send a broadcast payload to an entire IP network
|
||||
super().send(
|
||||
payload="broadcast",
|
||||
dest_ip_address=ip_network,
|
||||
dest_port=Port.HTTP,
|
||||
)
|
||||
|
||||
|
||||
class BroadcastClient(Application):
|
||||
"""A client application to receive broadcast and unicast messages."""
|
||||
|
||||
payloads_received: List = []
|
||||
|
||||
def __init__(self, **kwargs):
|
||||
# Set default client properties
|
||||
kwargs["name"] = "BroadcastClient"
|
||||
kwargs["port"] = Port.HTTP
|
||||
kwargs["protocol"] = IPProtocol.TCP
|
||||
super().__init__(**kwargs)
|
||||
|
||||
def describe_state(self) -> Dict:
|
||||
# Implement state description for the application
|
||||
pass
|
||||
|
||||
def receive(self, payload: Any, session_id: str, **kwargs) -> bool:
|
||||
# Append received payloads to the list and print a message
|
||||
self.payloads_received.append(payload)
|
||||
print(f"Payload: {payload} received on node {self.sys_log.hostname}")
|
||||
|
||||
|
||||
@pytest.fixture(scope="function")
|
||||
def broadcast_network() -> Network:
|
||||
network = Network()
|
||||
|
||||
client_1 = Computer(
|
||||
hostname="client_1",
|
||||
ip_address="192.168.1.2",
|
||||
subnet_mask="255.255.255.0",
|
||||
default_gateway="192.168.1.1",
|
||||
start_up_duration=0,
|
||||
)
|
||||
client_1.power_on()
|
||||
client_1.software_manager.install(BroadcastClient)
|
||||
application_1 = client_1.software_manager.software["BroadcastClient"]
|
||||
application_1.run()
|
||||
|
||||
client_2 = Computer(
|
||||
hostname="client_2",
|
||||
ip_address="192.168.1.3",
|
||||
subnet_mask="255.255.255.0",
|
||||
default_gateway="192.168.1.1",
|
||||
start_up_duration=0,
|
||||
)
|
||||
client_2.power_on()
|
||||
client_2.software_manager.install(BroadcastClient)
|
||||
application_2 = client_2.software_manager.software["BroadcastClient"]
|
||||
application_2.run()
|
||||
|
||||
server_1 = Server(
|
||||
hostname="server_1",
|
||||
ip_address="192.168.1.1",
|
||||
subnet_mask="255.255.255.0",
|
||||
default_gateway="192.168.1.1",
|
||||
start_up_duration=0,
|
||||
)
|
||||
server_1.power_on()
|
||||
|
||||
server_1.software_manager.install(BroadcastService)
|
||||
service: BroadcastService = server_1.software_manager.software["BroadcastService"]
|
||||
service.start()
|
||||
|
||||
switch_1 = Switch(hostname="switch_1", num_ports=6, start_up_duration=0)
|
||||
switch_1.power_on()
|
||||
|
||||
network.connect(endpoint_a=client_1.ethernet_port[1], endpoint_b=switch_1.switch_ports[1])
|
||||
network.connect(endpoint_a=client_2.ethernet_port[1], endpoint_b=switch_1.switch_ports[2])
|
||||
network.connect(endpoint_a=server_1.ethernet_port[1], endpoint_b=switch_1.switch_ports[3])
|
||||
|
||||
return network
|
||||
|
||||
|
||||
@pytest.fixture(scope="function")
|
||||
def broadcast_service_and_clients(broadcast_network) -> Tuple[BroadcastService, BroadcastClient, BroadcastClient]:
|
||||
client_1: BroadcastClient = broadcast_network.get_node_by_hostname("client_1").software_manager.software[
|
||||
"BroadcastClient"
|
||||
]
|
||||
client_2: BroadcastClient = broadcast_network.get_node_by_hostname("client_2").software_manager.software[
|
||||
"BroadcastClient"
|
||||
]
|
||||
service: BroadcastService = broadcast_network.get_node_by_hostname("server_1").software_manager.software[
|
||||
"BroadcastService"
|
||||
]
|
||||
|
||||
return service, client_1, client_2
|
||||
|
||||
|
||||
def test_broadcast_correct_subnet(broadcast_service_and_clients):
|
||||
service, client_1, client_2 = broadcast_service_and_clients
|
||||
|
||||
assert not client_1.payloads_received
|
||||
assert not client_2.payloads_received
|
||||
|
||||
service.broadcast(IPv4Network("192.168.1.0/24"))
|
||||
|
||||
assert client_1.payloads_received == ["broadcast"]
|
||||
assert client_2.payloads_received == ["broadcast"]
|
||||
|
||||
|
||||
def test_broadcast_incorrect_subnet(broadcast_service_and_clients):
|
||||
service, client_1, client_2 = broadcast_service_and_clients
|
||||
|
||||
assert not client_1.payloads_received
|
||||
assert not client_2.payloads_received
|
||||
|
||||
service.broadcast(IPv4Network("192.168.2.0/24"))
|
||||
|
||||
assert not client_1.payloads_received
|
||||
assert not client_2.payloads_received
|
||||
|
||||
|
||||
def test_unicast_correct_address(broadcast_service_and_clients):
|
||||
service, client_1, client_2 = broadcast_service_and_clients
|
||||
|
||||
assert not client_1.payloads_received
|
||||
assert not client_2.payloads_received
|
||||
|
||||
service.unicast(IPv4Address("192.168.1.2"))
|
||||
|
||||
assert client_1.payloads_received == ["unicast"]
|
||||
assert not client_2.payloads_received
|
||||
|
||||
|
||||
def test_unicast_incorrect_address(broadcast_service_and_clients):
|
||||
service, client_1, client_2 = broadcast_service_and_clients
|
||||
|
||||
assert not client_1.payloads_received
|
||||
assert not client_2.payloads_received
|
||||
|
||||
service.unicast(IPv4Address("192.168.2.2"))
|
||||
|
||||
assert not client_1.payloads_received
|
||||
assert not client_2.payloads_received
|
||||
@@ -1,11 +1,16 @@
|
||||
from ipaddress import IPv4Address
|
||||
from typing import Tuple
|
||||
|
||||
import pytest
|
||||
|
||||
from primaite.simulator.network.container import Network
|
||||
from primaite.simulator.network.hardware.base import Link, NIC, Node, NodeOperatingState
|
||||
from primaite.simulator.network.hardware.nodes.computer import Computer
|
||||
from primaite.simulator.network.hardware.nodes.router import ACLAction, Router
|
||||
from primaite.simulator.network.transmission.network_layer import IPProtocol
|
||||
from primaite.simulator.network.transmission.transport_layer import Port
|
||||
from primaite.simulator.system.services.ntp.ntp_client import NTPClient
|
||||
from primaite.simulator.system.services.ntp.ntp_server import NTPServer
|
||||
|
||||
|
||||
@pytest.fixture(scope="function")
|
||||
@@ -34,6 +39,69 @@ def pc_a_pc_b_router_1() -> Tuple[Node, Node, Router]:
|
||||
return pc_a, pc_b, router_1
|
||||
|
||||
|
||||
@pytest.fixture(scope="function")
|
||||
def multi_hop_network() -> Network:
|
||||
network = Network()
|
||||
|
||||
# Configure PC A
|
||||
pc_a = Computer(
|
||||
hostname="pc_a",
|
||||
ip_address="192.168.0.2",
|
||||
subnet_mask="255.255.255.0",
|
||||
default_gateway="192.168.0.1",
|
||||
start_up_duration=0,
|
||||
)
|
||||
pc_a.power_on()
|
||||
network.add_node(pc_a)
|
||||
|
||||
# Configure Router 1
|
||||
router_1 = Router(hostname="router_1", start_up_duration=0)
|
||||
router_1.power_on()
|
||||
network.add_node(router_1)
|
||||
|
||||
# Configure the connection between PC A and Router 1 port 2
|
||||
router_1.configure_port(2, "192.168.0.1", "255.255.255.0")
|
||||
network.connect(pc_a.ethernet_port[1], router_1.ethernet_ports[2])
|
||||
router_1.enable_port(2)
|
||||
|
||||
# Configure Router 1 ACLs
|
||||
router_1.acl.add_rule(action=ACLAction.PERMIT, src_port=Port.ARP, dst_port=Port.ARP, position=22)
|
||||
router_1.acl.add_rule(action=ACLAction.PERMIT, protocol=IPProtocol.ICMP, position=23)
|
||||
|
||||
# Configure PC B
|
||||
pc_b = Computer(
|
||||
hostname="pc_b",
|
||||
ip_address="192.168.2.2",
|
||||
subnet_mask="255.255.255.0",
|
||||
default_gateway="192.168.2.1",
|
||||
start_up_duration=0,
|
||||
)
|
||||
pc_b.power_on()
|
||||
network.add_node(pc_b)
|
||||
|
||||
# Configure Router 2
|
||||
router_2 = Router(hostname="router_2", start_up_duration=0)
|
||||
router_2.power_on()
|
||||
network.add_node(router_2)
|
||||
|
||||
# Configure the connection between PC B and Router 2 port 2
|
||||
router_2.configure_port(2, "192.168.2.1", "255.255.255.0")
|
||||
network.connect(pc_b.ethernet_port[1], router_2.ethernet_ports[2])
|
||||
router_2.enable_port(2)
|
||||
|
||||
# Configure Router 2 ACLs
|
||||
router_2.acl.add_rule(action=ACLAction.PERMIT, src_port=Port.ARP, dst_port=Port.ARP, position=22)
|
||||
router_2.acl.add_rule(action=ACLAction.PERMIT, protocol=IPProtocol.ICMP, position=23)
|
||||
|
||||
# Configure the connection between Router 1 port 1 and Router 2 port 1
|
||||
router_2.configure_port(1, "192.168.1.2", "255.255.255.252")
|
||||
router_1.configure_port(1, "192.168.1.1", "255.255.255.252")
|
||||
network.connect(router_1.ethernet_ports[1], router_2.ethernet_ports[1])
|
||||
router_1.enable_port(1)
|
||||
router_2.enable_port(1)
|
||||
return network
|
||||
|
||||
|
||||
def test_ping_default_gateway(pc_a_pc_b_router_1):
|
||||
pc_a, pc_b, router_1 = pc_a_pc_b_router_1
|
||||
|
||||
@@ -50,3 +118,68 @@ def test_host_on_other_subnet(pc_a_pc_b_router_1):
|
||||
pc_a, pc_b, router_1 = pc_a_pc_b_router_1
|
||||
|
||||
assert pc_a.ping("192.168.1.10")
|
||||
|
||||
|
||||
def test_no_route_no_ping(multi_hop_network):
|
||||
pc_a = multi_hop_network.get_node_by_hostname("pc_a")
|
||||
pc_b = multi_hop_network.get_node_by_hostname("pc_b")
|
||||
|
||||
assert not pc_a.ping(pc_b.ethernet_port[1].ip_address)
|
||||
|
||||
|
||||
def test_with_routes_can_ping(multi_hop_network):
|
||||
pc_a = multi_hop_network.get_node_by_hostname("pc_a")
|
||||
pc_b = multi_hop_network.get_node_by_hostname("pc_b")
|
||||
|
||||
router_1: Router = multi_hop_network.get_node_by_hostname("router_1") # noqa
|
||||
router_2: Router = multi_hop_network.get_node_by_hostname("router_2") # noqa
|
||||
|
||||
# Configure Route from Router 1 to PC B subnet
|
||||
router_1.route_table.add_route(
|
||||
address="192.168.2.0", subnet_mask="255.255.255.0", next_hop_ip_address="192.168.1.2"
|
||||
)
|
||||
|
||||
# Configure Route from Router 2 to PC A subnet
|
||||
router_2.route_table.add_route(
|
||||
address="192.168.0.2", subnet_mask="255.255.255.0", next_hop_ip_address="192.168.1.1"
|
||||
)
|
||||
|
||||
assert pc_a.ping(pc_b.ethernet_port[1].ip_address)
|
||||
|
||||
|
||||
def test_routing_services(multi_hop_network):
|
||||
pc_a = multi_hop_network.get_node_by_hostname("pc_a")
|
||||
|
||||
pc_b = multi_hop_network.get_node_by_hostname("pc_b")
|
||||
|
||||
pc_a.software_manager.install(NTPClient)
|
||||
ntp_client = pc_a.software_manager.software["NTPClient"]
|
||||
ntp_client.start()
|
||||
|
||||
pc_b.software_manager.install(NTPServer)
|
||||
pc_b.software_manager.software["NTPServer"].start()
|
||||
|
||||
ntp_client.configure(ntp_server_ip_address=pc_b.ethernet_port[1].ip_address)
|
||||
|
||||
router_1: Router = multi_hop_network.get_node_by_hostname("router_1") # noqa
|
||||
router_2: Router = multi_hop_network.get_node_by_hostname("router_2") # noqa
|
||||
|
||||
router_1.acl.add_rule(action=ACLAction.PERMIT, src_port=Port.NTP, dst_port=Port.NTP, position=21)
|
||||
router_2.acl.add_rule(action=ACLAction.PERMIT, src_port=Port.NTP, dst_port=Port.NTP, position=21)
|
||||
|
||||
assert ntp_client.time is None
|
||||
ntp_client.request_time()
|
||||
assert ntp_client.time is None
|
||||
|
||||
# Configure Route from Router 1 to PC B subnet
|
||||
router_1.route_table.add_route(
|
||||
address="192.168.2.0", subnet_mask="255.255.255.0", next_hop_ip_address="192.168.1.2"
|
||||
)
|
||||
|
||||
# Configure Route from Router 2 to PC A subnet
|
||||
router_2.route_table.add_route(
|
||||
address="192.168.0.2", subnet_mask="255.255.255.0", next_hop_ip_address="192.168.1.1"
|
||||
)
|
||||
|
||||
ntp_client.request_time()
|
||||
assert ntp_client.time is not None
|
||||
|
||||
Reference in New Issue
Block a user