oxbrbtx/PrimAITE
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'origin/dev' into feature/2317-refactor-reset

Browse Source
This commit is contained in:
Marek Wolan
2024-02-25 16:41:40 +00:00
parent c115095157 8f85555709
commit a34cf08209
124 changed files with 7441 additions and 3680 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1145
src/primaite/simulator/network/hardware/base.py
View File

File diff suppressed because it is too large Load Diff

0
src/primaite/simulator/network/hardware/network_interface/__init__.py Normal file
View File

0
src/primaite/simulator/network/hardware/network_interface/wireless/__init__.py Normal file
View File

86
src/primaite/simulator/network/hardware/network_interface/wireless/wireless_access_point.py Normal file
View File

@@ -0,0 +1,86 @@
from typing import Dict
from primaite.simulator.network.hardware.base import (
IPWirelessNetworkInterface,
Layer3Interface,
WirelessNetworkInterface,
)
from primaite.simulator.network.transmission.data_link_layer import Frame
class WirelessAccessPoint(IPWirelessNetworkInterface):
"""
Represents a Wireless Access Point (AP) in a network.
This class models a Wireless Access Point, a device that allows wireless devices to connect to a wired network
using Wi-Fi or other wireless standards. The Wireless Access Point bridges the wireless and wired segments of
the network, allowing wireless devices to communicate with other devices on the network.
As an integral component of wireless networking, a Wireless Access Point provides functionalities for network
management, signal broadcasting, security enforcement, and connection handling. It also possesses Layer 3
capabilities such as IP addressing and subnetting, allowing for network segmentation and routing.
Inherits from:
- WirelessNetworkInterface: Provides basic properties and methods specific to wireless interfaces.
- Layer3Interface: Provides Layer 3 properties like ip_address and subnet_mask, enabling the device to manage
network traffic and routing.
This class can be further specialised or extended to support specific features or standards related to wireless
networking, such as different Wi-Fi versions, frequency bands, or advanced security protocols.
"""
def describe_state(self) -> Dict:
"""
Produce a dictionary describing the current state of this object.
:return: Current state of this object and child objects.
:rtype: Dict
"""
# Get the state from the WirelessNetworkInterface
state = WirelessNetworkInterface.describe_state(self)
# Update the state with information from Layer3Interface
state.update(Layer3Interface.describe_state(self))
# Update the state with NIC-specific information
state.update(
{
"wake_on_lan": self.wake_on_lan,
}
)
return state
def enable(self):
"""Enable the interface."""
pass
def disable(self):
"""Disable the interface."""
pass
def send_frame(self, frame: Frame) -> bool:
"""
Attempts to send a network frame through the interface.
:param frame: The network frame to be sent.
:return: A boolean indicating whether the frame was successfully sent.
"""
pass
def receive_frame(self, frame: Frame) -> bool:
"""
Receives a network frame on the interface.
:param frame: The network frame being received.
:return: A boolean indicating whether the frame was successfully received.
"""
pass
def __str__(self) -> str:
"""
String representation of the NIC.
:return: A string combining the port number, MAC address and IP address of the NIC.
"""
return f"Port {self.port_num}: {self.mac_address}/{self.ip_address}"

83
src/primaite/simulator/network/hardware/network_interface/wireless/wireless_nic.py Normal file
View File

@@ -0,0 +1,83 @@
from typing import Dict
from primaite.simulator.network.hardware.base import (
IPWirelessNetworkInterface,
Layer3Interface,
WirelessNetworkInterface,
)
from primaite.simulator.network.transmission.data_link_layer import Frame
class WirelessNIC(IPWirelessNetworkInterface):
"""
Represents a Wireless Network Interface Card (Wireless NIC) in a network device.
This class encapsulates the functionalities and attributes of a wireless NIC, combining the characteristics of a
wireless network interface with Layer 3 features. It is capable of connecting to wireless networks, managing
wireless-specific properties such as signal strength and security protocols, and also handling IP-related
functionalities like IP addressing and subnetting.
Inherits from:
- WirelessNetworkInterface: Provides basic properties and methods specific to wireless interfaces.
- Layer3Interface: Provides Layer 3 properties like ip_address and subnet_mask, enabling the device to participate
in IP-based networking.
This class can be extended to include more advanced features or to tailor its behavior for specific types of
wireless networks or protocols.
"""
def describe_state(self) -> Dict:
"""
Produce a dictionary describing the current state of this object.
:return: Current state of this object and child objects.
:rtype: Dict
"""
# Get the state from the WirelessNetworkInterface
state = WirelessNetworkInterface.describe_state(self)
# Update the state with information from Layer3Interface
state.update(Layer3Interface.describe_state(self))
# Update the state with NIC-specific information
state.update(
{
"wake_on_lan": self.wake_on_lan,
}
)
return state
def enable(self):
"""Enable the interface."""
pass
def disable(self):
"""Disable the interface."""
pass
def send_frame(self, frame: Frame) -> bool:
"""
Attempts to send a network frame through the interface.
:param frame: The network frame to be sent.
:return: A boolean indicating whether the frame was successfully sent.
"""
pass
def receive_frame(self, frame: Frame) -> bool:
"""
Receives a network frame on the interface.
:param frame: The network frame being received.
:return: A boolean indicating whether the frame was successfully received.
"""
pass
def __str__(self) -> str:
"""
String representation of the NIC.
:return: A string combining the port number, MAC address and IP address of the NIC.
"""
return f"Port {self.port_num}: {self.mac_address}/{self.ip_address}"

55
src/primaite/simulator/network/hardware/nodes/computer.py
View File

@@ -1,55 +0,0 @@
from primaite.simulator.network.hardware.base import NIC, Node
from primaite.simulator.system.applications.web_browser import WebBrowser
from primaite.simulator.system.services.dns.dns_client import DNSClient
from primaite.simulator.system.services.ftp.ftp_client import FTPClient
class Computer(Node):
"""
A basic Computer class.
Example:
>>> pc_a = Computer(
hostname="pc_a",
ip_address="192.168.1.10",
subnet_mask="255.255.255.0",
default_gateway="192.168.1.1"
)
>>> pc_a.power_on()
Instances of computer come 'pre-packaged' with the following:
* Core Functionality:
* ARP
* ICMP
* Packet Capture
* Sys Log
* Services:
* DNS Client
* FTP Client
* LDAP Client
* NTP Client
* Applications:
* Email Client
* Web Browser
* Processes:
* Placeholder
"""
def __init__(self, **kwargs):
super().__init__(**kwargs)
self.connect_nic(NIC(ip_address=kwargs["ip_address"], subnet_mask=kwargs["subnet_mask"]))
self._install_system_software()
def _install_system_software(self):
"""Install System Software - software that is usually provided with the OS."""
# DNS Client
self.software_manager.install(DNSClient)
# FTP
self.software_manager.install(FTPClient)
# Web Browser
self.software_manager.install(WebBrowser)
super()._install_system_software()

0
src/primaite/simulator/network/hardware/nodes/host/__init__.py Normal file
View File

32
src/primaite/simulator/network/hardware/nodes/host/computer.py Normal file
View File

@@ -0,0 +1,32 @@
from primaite.simulator.network.hardware.nodes.host.host_node import HostNode
class Computer(HostNode):
"""
A basic Computer class.
Example:
>>> pc_a = Computer(
hostname="pc_a",
ip_address="192.168.1.10",
subnet_mask="255.255.255.0",
default_gateway="192.168.1.1"
)
>>> pc_a.power_on()
Instances of computer come 'pre-packaged' with the following:
* Core Functionality:
* Packet Capture
* Sys Log
* Services:
* ARP Service
* ICMP Service
* DNS Client
* FTP Client
* NTP Client
* Applications:
* Web Browser
"""
pass

385
src/primaite/simulator/network/hardware/nodes/host/host_node.py Normal file
View File

@@ -0,0 +1,385 @@
from __future__ import annotations
from ipaddress import IPv4Address
from typing import Any, Dict, Optional
from primaite import getLogger
from primaite.simulator.network.hardware.base import IPWiredNetworkInterface, Link, Node
from primaite.simulator.network.hardware.node_operating_state import NodeOperatingState
from primaite.simulator.network.transmission.data_link_layer import Frame
from primaite.simulator.system.applications.web_browser import WebBrowser
from primaite.simulator.system.services.arp.arp import ARP, ARPPacket
from primaite.simulator.system.services.dns.dns_client import DNSClient
from primaite.simulator.system.services.ftp.ftp_client import FTPClient
from primaite.simulator.system.services.icmp.icmp import ICMP
from primaite.simulator.system.services.ntp.ntp_client import NTPClient
from primaite.utils.validators import IPV4Address
_LOGGER = getLogger(__name__)
class HostARP(ARP):
"""
The Host ARP Service.
Extends the ARP service for host-specific functionalities within a network, focusing on resolving and caching
MAC addresses and network interfaces (NICs) based on IP addresses, especially concerning the default gateway.
This specialized ARP service for hosts facilitates efficient network communication by managing ARP entries
and handling ARP requests and replies with additional logic for default gateway processing.
"""
def get_default_gateway_mac_address(self) -> Optional[str]:
"""
Retrieves the MAC address of the default gateway as known from the ARP cache.
:return: The MAC address of the default gateway if present in the ARP cache; otherwise, None.
"""
if self.software_manager.node.default_gateway:
return self.get_arp_cache_mac_address(self.software_manager.node.default_gateway)
def get_default_gateway_network_interface(self) -> Optional[NIC]:
"""
Obtains the network interface card (NIC) associated with the default gateway from the ARP cache.
:return: The NIC associated with the default gateway if it exists in the ARP cache; otherwise, None.
"""
if self.software_manager.node.default_gateway and self.software_manager.node.has_enabled_network_interface:
return self.get_arp_cache_network_interface(self.software_manager.node.default_gateway)
def _get_arp_cache_mac_address(
self, ip_address: IPV4Address, is_reattempt: bool = False, is_default_gateway_attempt: bool = False
) -> Optional[str]:
"""
Internal method to retrieve the MAC address associated with an IP address from the ARP cache.
:param ip_address: The IP address whose MAC address is to be retrieved.
:param is_reattempt: Indicates if this call is a reattempt after a failed initial attempt.
:param is_default_gateway_attempt: Indicates if this call is an attempt to get the default gateway's MAC
address.
:return: The MAC address associated with the IP address if found, otherwise None.
"""
arp_entry = self.arp.get(ip_address)
if arp_entry:
return arp_entry.mac_address
if ip_address == self.software_manager.node.default_gateway:
is_reattempt = True
if not is_reattempt:
self.send_arp_request(ip_address)
return self._get_arp_cache_mac_address(
ip_address=ip_address, is_reattempt=True, is_default_gateway_attempt=is_default_gateway_attempt
)
else:
if self.software_manager.node.default_gateway:
if not is_default_gateway_attempt:
self.send_arp_request(self.software_manager.node.default_gateway)
return self._get_arp_cache_mac_address(
ip_address=self.software_manager.node.default_gateway,
is_reattempt=True,
is_default_gateway_attempt=True,
)
return None
def get_arp_cache_mac_address(self, ip_address: IPv4Address) -> Optional[str]:
"""
Retrieves the MAC address associated with a given IP address from the ARP cache.
:param ip_address: The IP address for which the MAC address is sought.
:return: The MAC address if available in the ARP cache; otherwise, None.
"""
return self._get_arp_cache_mac_address(ip_address)
def _get_arp_cache_network_interface(
self, ip_address: IPV4Address, is_reattempt: bool = False, is_default_gateway_attempt: bool = False
) -> Optional[NIC]:
"""
Internal method to retrieve the NIC associated with an IP address from the ARP cache.
:param ip_address: The IP address whose NIC is to be retrieved.
:param is_reattempt: Indicates if this call is a reattempt after a failed initial attempt.
:param is_default_gateway_attempt: Indicates if this call is an attempt to get the NIC of the default gateway.
:return: The NIC associated with the IP address if found, otherwise None.
"""
arp_entry = self.arp.get(ip_address)
if arp_entry:
return self.software_manager.node.network_interfaces[arp_entry.network_interface_uuid]
else:
if ip_address == self.software_manager.node.default_gateway:
is_reattempt = True
if not is_reattempt:
self.send_arp_request(ip_address)
return self._get_arp_cache_network_interface(
ip_address=ip_address, is_reattempt=True, is_default_gateway_attempt=is_default_gateway_attempt
)
else:
if self.software_manager.node.default_gateway:
if not is_default_gateway_attempt:
self.send_arp_request(self.software_manager.node.default_gateway)
return self._get_arp_cache_network_interface(
ip_address=self.software_manager.node.default_gateway,
is_reattempt=True,
is_default_gateway_attempt=True,
)
return None
def get_arp_cache_network_interface(self, ip_address: IPv4Address) -> Optional[NIC]:
"""
Retrieves the network interface card (NIC) associated with a given IP address from the ARP cache.
:param ip_address: The IP address for which the associated NIC is sought.
:return: The NIC if available in the ARP cache; otherwise, None.
"""
return self._get_arp_cache_network_interface(ip_address)
def _process_arp_request(self, arp_packet: ARPPacket, from_network_interface: NIC):
"""
Processes an ARP request.
Adds a new entry to the ARP cache if the target IP address matches the NIC's IP address and sends an ARP
reply back.
:param arp_packet: The ARP packet containing the request.
:param from_network_interface: The NIC that received the ARP request.
"""
super()._process_arp_request(arp_packet, from_network_interface)
# Unmatched ARP Request
if arp_packet.target_ip_address != from_network_interface.ip_address:
self.sys_log.info(
f"Ignoring ARP request for {arp_packet.target_ip_address}. Current IP address is "
f"{from_network_interface.ip_address}"
)
return
arp_packet = arp_packet.generate_reply(from_network_interface.mac_address)
self.send_arp_reply(arp_packet)
class NIC(IPWiredNetworkInterface):
"""
Represents a Network Interface Card (NIC) in a Host Node.
A NIC is a hardware component that provides a computer or other network device with the ability to connect to a
network. It operates at both Layer 2 (Data Link Layer) and Layer 3 (Network Layer) of the OSI model, meaning it
can interpret both MAC addresses and IP addresses. This class combines the functionalities of
WiredNetworkInterface and Layer3Interface, allowing the NIC to manage physical connections and network layer
addressing.
Inherits from:
- WiredNetworkInterface: Provides properties and methods specific to wired connections, including methods to connect
and disconnect from network links and to manage the enabled/disabled state of the interface.
- Layer3Interface: Provides properties for Layer 3 network configuration, such as IP address and subnet mask.
"""
_connected_link: Optional[Link] = None
"The network link to which the network interface is connected."
wake_on_lan: bool = False
"Indicates if the NIC supports Wake-on-LAN functionality."
def model_post_init(self, __context: Any) -> None:
"""
Performs post-initialisation checks to ensure the model's IP configuration is valid.
This method is invoked after the initialisation of a network model object to validate its network settings,
particularly to ensure that the assigned IP address is not a network address. This validation is crucial for
maintaining the integrity of network simulations and avoiding configuration errors that could lead to
unrealistic or incorrect behavior.
:param __context: Contextual information or parameters passed to the method, used for further initializing or
validating the model post-creation.
:raises ValueError: If the IP address is the same as the network address, indicating an incorrect configuration.
"""
if self.ip_network.network_address == self.ip_address:
raise ValueError(f"{self.ip_address}/{self.subnet_mask} must not be a network address")
def describe_state(self) -> Dict:
"""
Produce a dictionary describing the current state of this object.
:return: Current state of this object and child objects.
:rtype: Dict
"""
# Get the state from the IPWiredNetworkInterface
state = super().describe_state()
# Update the state with NIC-specific information
state.update(
{
"wake_on_lan": self.wake_on_lan,
}
)
return state
def set_original_state(self):
"""Sets the original state."""
vals_to_include = {"ip_address", "subnet_mask", "mac_address", "speed", "mtu", "wake_on_lan", "enabled"}
self._original_state = self.model_dump(include=vals_to_include)
def receive_frame(self, frame: Frame) -> bool:
"""
Attempt to receive and process a network frame from the connected Link.
This method processes a frame if the NIC is enabled. It checks the frame's destination and TTL, captures the
frame using PCAP, and forwards it to the connected Node if valid. Returns True if the frame is processed,
False otherwise (e.g., if the NIC is disabled, or TTL expired).
:param frame: The network frame being received.
:return: True if the frame is processed and passed to the node, False otherwise.
"""
if self.enabled:
frame.decrement_ttl()
if frame.ip and frame.ip.ttl < 1:
self._connected_node.sys_log.info(f"Frame discarded at {self} as TTL limit reached")
return False
frame.set_received_timestamp()
self.pcap.capture_inbound(frame)
# If this destination or is broadcast
accept_frame = False
# Check if it's a broadcast:
if frame.ethernet.dst_mac_addr == "ff:ff:ff:ff:ff:ff":
if frame.ip.dst_ip_address in {self.ip_address, self.ip_network.broadcast_address}:
accept_frame = True
else:
if frame.ethernet.dst_mac_addr == self.mac_address:
accept_frame = True
if accept_frame:
self._connected_node.receive_frame(frame=frame, from_network_interface=self)
return True
return False
def __str__(self) -> str:
"""
String representation of the NIC.
:return: A string combining the port number, MAC address and IP address of the NIC.
"""
return f"Port {self.port_num}: {self.mac_address}/{self.ip_address}"
class HostNode(Node):
"""
Represents a host node in the network.
An end-user device within the network, such as a computer or server, equipped with the capability to initiate and
respond to network communications.
A `HostNode` extends the base `Node` class by incorporating host-specific services and applications, thereby
simulating the functionalities typically expected from a networked end-user device.
**Example**::
>>> pc_a = HostNode(
... hostname="pc_a",
... ip_address="192.168.1.10",
... subnet_mask="255.255.255.0",
... default_gateway="192.168.1.1"
... )
>>> pc_a.power_on()
The host node comes pre-equipped with a range of core functionalities, services, and applications necessary
for engaging in various network operations and tasks.
Core Functionality:
-------------------
* Packet Capture: Monitors and logs network traffic.
* Sys Log: Logs system events and errors.
Services:
---------
* ARP (Address Resolution Protocol) Service: Resolves IP addresses to MAC addresses.
* ICMP (Internet Control Message Protocol) Service: Handles ICMP operations, such as ping requests.
* DNS (Domain Name System) Client: Resolves domain names to IP addresses.
* FTP (File Transfer Protocol) Client: Enables file transfers between the host and FTP servers.
* NTP (Network Time Protocol) Client: Synchronizes the system clock with NTP servers.
Applications:
------------
* Web Browser: Provides web browsing capabilities.
"""
network_interfaces: Dict[str, NIC] = {}
"The Network Interfaces on the node."
network_interface: Dict[int, NIC] = {}
"The NICs on the node by port id."
def __init__(self, ip_address: IPV4Address, subnet_mask: IPV4Address, **kwargs):
super().__init__(**kwargs)
self.connect_nic(NIC(ip_address=ip_address, subnet_mask=subnet_mask))
def _install_system_software(self):
"""
Installs the system software and network services typically found on an operating system.
This method equips the host with essential network services and applications, preparing it for various
network-related tasks and operations.
"""
# ARP Service
self.software_manager.install(HostARP)
# ICMP Service
self.software_manager.install(ICMP)
# DNS Client
self.software_manager.install(DNSClient)
# FTP Client
self.software_manager.install(FTPClient)
# NTP Client
self.software_manager.install(NTPClient)
# Web Browser
self.software_manager.install(WebBrowser)
super()._install_system_software()
def default_gateway_hello(self):
"""
Sends a hello message to the default gateway to establish connectivity and resolve the gateway's MAC address.
This method is invoked to ensure the host node can communicate with its default gateway, primarily to confirm
network connectivity and populate the ARP cache with the gateway's MAC address.
"""
if self.operating_state == NodeOperatingState.ON and self.default_gateway:
self.software_manager.arp.get_default_gateway_mac_address()
def receive_frame(self, frame: Frame, from_network_interface: NIC):
"""
Receive a Frame from the connected NIC and process it.
Depending on the protocol, the frame is passed to the appropriate handler such as ARP or ICMP, or up to the
SessionManager if no code manager exists.
:param frame: The Frame being received.
:param from_network_interface: The NIC that received the frame.
"""
super().receive_frame(frame, from_network_interface)
# Check if the destination port is open on the Node
dst_port = None
if frame.tcp:
dst_port = frame.tcp.dst_port
elif frame.udp:
dst_port = frame.udp.dst_port
accept_frame = False
if frame.icmp or dst_port in self.software_manager.get_open_ports():
# accept the frame as the port is open or if it's an ICMP frame
accept_frame = True
# TODO: add internal node firewall check here?
if accept_frame:
self.session_manager.receive_frame(frame, from_network_interface)
else:
self.sys_log.info(f"Ignoring frame from {frame.ip.src_ip_address}")
# TODO: do we need to do anything more here?
pass

12
src/primaite/simulator/network/hardware/nodes/server.py → src/primaite/simulator/network/hardware/nodes/host/server.py
View File

@@ -1,7 +1,7 @@
from primaite.simulator.network.hardware.nodes.computer import Computer
from primaite.simulator.network.hardware.nodes.host.host_node import HostNode
class Server(Computer):
class Server(HostNode):
"""
A basic Server class.
@@ -17,18 +17,14 @@ class Server(Computer):
Instances of Server come 'pre-packaged' with the following:
* Core Functionality:
* ARP
* ICMP
* Packet Capture
* Sys Log
* Services:
* ARP Service
* ICMP Service
* DNS Client
* FTP Client
* LDAP Client
* NTP Client
* Applications:
* Email Client
* Web Browser
* Processes:
* Placeholder
"""

0
src/primaite/simulator/network/hardware/nodes/network/__init__.py Normal file
View File

493
src/primaite/simulator/network/hardware/nodes/network/firewall.py Normal file
View File

@@ -0,0 +1,493 @@
from typing import Dict, Final, Optional, Union
from prettytable import MARKDOWN, PrettyTable
from pydantic import validate_call
from primaite.simulator.network.hardware.nodes.network.router import (
AccessControlList,
ACLAction,
Router,
RouterInterface,
)
from primaite.simulator.network.transmission.data_link_layer import Frame
from primaite.simulator.system.core.sys_log import SysLog
from primaite.utils.validators import IPV4Address
EXTERNAL_PORT_ID: Final[int] = 1
"""The Firewall port ID of the external port."""
INTERNAL_PORT_ID: Final[int] = 2
"""The Firewall port ID of the internal port."""
DMZ_PORT_ID: Final[int] = 3
"""The Firewall port ID of the DMZ port."""
class Firewall(Router):
"""
A Firewall class that extends the functionality of a Router.
The Firewall class acts as a network security system that monitors and controls incoming and outgoing
network traffic based on predetermined security rules. It is an intermediary between internal and external
networks (including DMZ - De-Militarized Zone), ensuring that all inbound and outbound traffic complies with
the security policies.
The Firewall employs Access Control Lists (ACLs) to filter traffic. Both the internal and DMZ ports have both
inbound and outbound ACLs that determine what traffic is allowed to pass.
In addition to the security functions, the Firewall can also perform some routing functions similar to a Router,
forwarding packets between its interfaces based on the destination IP address.
Usage:
To utilise the Firewall class, instantiate it with a hostname and optionally specify sys_log for logging.
Configure the internal, external, and DMZ ports with IP addresses and subnet masks. Define ACL rules to
permit or deny traffic based on your security policies. The Firewall will process frames based on these
rules, determining whether to allow or block traffic at each network interface.
Example:
>>> from primaite.simulator.network.transmission.network_layer import IPProtocol
>>> from primaite.simulator.network.transmission.transport_layer import Port
>>> firewall = Firewall(hostname="Firewall1")
>>> firewall.configure_internal_port(ip_address="192.168.1.1", subnet_mask="255.255.255.0")
>>> firewall.configure_external_port(ip_address="10.0.0.1", subnet_mask="255.255.255.0")
>>> firewall.configure_dmz_port(ip_address="172.16.0.1", subnet_mask="255.255.255.0")
>>> # Permit HTTP traffic to the DMZ
>>> firewall.dmz_inbound_acl.add_rule(
... action=ACLAction.PERMIT,
... protocol=IPProtocol.TCP,
... dst_port=Port.HTTP,
... src_ip_address="0.0.0.0",
... src_wildcard_mask="0.0.0.0",
... dst_ip_address="172.16.0.0",
... dst_wildcard_mask="0.0.0.255"
... )
:ivar str hostname: The Firewall hostname.
"""
internal_inbound_acl: Optional[AccessControlList] = None
"""Access Control List for managing entering the internal network."""
internal_outbound_acl: Optional[AccessControlList] = None
"""Access Control List for managing traffic leaving the internal network."""
dmz_inbound_acl: Optional[AccessControlList] = None
"""Access Control List for managing traffic entering the DMZ."""
dmz_outbound_acl: Optional[AccessControlList] = None
"""Access Control List for managing traffic leaving the DMZ."""
external_inbound_acl: Optional[AccessControlList] = None
"""Access Control List for managing traffic entering from an external network."""
external_outbound_acl: Optional[AccessControlList] = None
"""Access Control List for managing traffic leaving towards an external network."""
def __init__(self, hostname: str, **kwargs):
if not kwargs.get("sys_log"):
kwargs["sys_log"] = SysLog(hostname)
super().__init__(hostname=hostname, num_ports=3, **kwargs)
# Initialise ACLs for internal and dmz interfaces with a default DENY policy
self.internal_inbound_acl = AccessControlList(
sys_log=kwargs["sys_log"], implicit_action=ACLAction.DENY, name=f"{hostname} - Internal Inbound"
)
self.internal_outbound_acl = AccessControlList(
sys_log=kwargs["sys_log"], implicit_action=ACLAction.DENY, name=f"{hostname} - Internal Outbound"
)
self.dmz_inbound_acl = AccessControlList(
sys_log=kwargs["sys_log"], implicit_action=ACLAction.DENY, name=f"{hostname} - DMZ Inbound"
)
self.dmz_outbound_acl = AccessControlList(
sys_log=kwargs["sys_log"], implicit_action=ACLAction.DENY, name=f"{hostname} - DMZ Outbound"
)
# external ACLs should have a default PERMIT policy
self.external_inbound_acl = AccessControlList(
sys_log=kwargs["sys_log"], implicit_action=ACLAction.PERMIT, name=f"{hostname} - External Inbound"
)
self.external_outbound_acl = AccessControlList(
sys_log=kwargs["sys_log"], implicit_action=ACLAction.PERMIT, name=f"{hostname} - External Outbound"
)
self.set_original_state()
def set_original_state(self):
"""Set the original state for the Firewall."""
super().set_original_state()
vals_to_include = {
"internal_port",
"external_port",
"dmz_port",
"internal_inbound_acl",
"internal_outbound_acl",
"dmz_inbound_acl",
"dmz_outbound_acl",
"external_inbound_acl",
"external_outbound_acl",
}
self._original_state.update(self.model_dump(include=vals_to_include))
def describe_state(self) -> Dict:
"""
Describes the current state of the Firewall.
:return: A dictionary representing the current state.
"""
state = super().describe_state()
state.update(
{
"internal_port": self.internal_port.describe_state(),
"external_port": self.external_port.describe_state(),
"dmz_port": self.dmz_port.describe_state(),
"internal_inbound_acl": self.internal_inbound_acl.describe_state(),
"internal_outbound_acl": self.internal_outbound_acl.describe_state(),
"dmz_inbound_acl": self.dmz_inbound_acl.describe_state(),
"dmz_outbound_acl": self.dmz_outbound_acl.describe_state(),
"external_inbound_acl": self.external_inbound_acl.describe_state(),
"external_outbound_acl": self.external_outbound_acl.describe_state(),
}
)
return state
def show(self, markdown: bool = False):
"""
Displays the current configuration of the firewall's network interfaces in a table format.
The table includes information about each port (External, Internal, DMZ), their MAC addresses, IP
configurations, link speeds, and operational status. The output can be formatted as Markdown if specified.
:param markdown: If True, formats the output table in Markdown style. Useful for documentation or reporting
purposes within Markdown-compatible platforms.
"""
table = PrettyTable(["Port", "MAC Address", "Address", "Speed", "Status"])
if markdown:
table.set_style(MARKDOWN)
table.align = "l"
table.title = f"{self.hostname} Network Interfaces"
ports = {"External": self.external_port, "Internal": self.internal_port, "DMZ": self.dmz_port}
for port, network_interface in ports.items():
table.add_row(
[
port,
network_interface.mac_address,
f"{network_interface.ip_address}/{network_interface.ip_network.prefixlen}",
network_interface.speed,
"Enabled" if network_interface.enabled else "Disabled",
]
)
print(table)
def show_rules(self, external: bool = True, internal: bool = True, dmz: bool = True, markdown: bool = False):
"""
Prints the configured ACL rules for each specified network zone of the firewall.
This method allows selective viewing of ACL rules applied to external, internal, and DMZ interfaces, providing
a clear overview of the firewall's current traffic filtering policies. Each section can be independently
toggled.
:param external: If True, shows ACL rules for external interfaces.
:param internal: If True, shows ACL rules for internal interfaces.
:param dmz: If True, shows ACL rules for DMZ interfaces.
:param markdown: If True, formats the output in Markdown, enhancing readability in Markdown-compatible viewers.
"""
print(f"{self.hostname} Firewall Rules")
print()
if external:
self.external_inbound_acl.show(markdown)
print()
self.external_outbound_acl.show(markdown)
print()
if internal:
self.internal_inbound_acl.show(markdown)
print()
self.internal_outbound_acl.show(markdown)
print()
if dmz:
self.dmz_inbound_acl.show(markdown)
print()
self.dmz_outbound_acl.show(markdown)
print()
def receive_frame(self, frame: Frame, from_network_interface: RouterInterface):
"""
Receive a frame and process it.
Acts as the primary entry point for all network frames arriving at the Firewall, determining the flow of
traffic based on the source network interface controller (NIC) and applying the appropriate Access Control
List (ACL) rules.
This method categorizes the incoming traffic into three main pathways based on the source NIC: external inbound,
internal outbound, and DMZ (De-Militarized Zone) outbound. It plays a crucial role in enforcing the firewall's
security policies by directing each frame to the corresponding processing method that evaluates it against
specific ACL rules.
Based on the originating NIC:
- Frames from the external port are processed as external inbound traffic, potentially destined for either the
DMZ or the internal network.
- Frames from the internal port are treated as internal outbound traffic, aimed at reaching the external
network or a service within the DMZ.
- Frames from the DMZ port are handled as DMZ outbound traffic, with potential destinations including the
internal network or the external network.
:param frame: The network frame to be processed.
:param from_network_interface: The network interface controller from which the frame is coming. Used to
determine the direction of the traffic (inbound or outbound) and the zone (external, internal,
DMZ) it belongs to.
"""
# If the frame comes from the external port, it's considered as external inbound traffic
if from_network_interface == self.external_port:
self._process_external_inbound_frame(frame, from_network_interface)
return
# If the frame comes from the internal port, it's considered as internal outbound traffic
elif from_network_interface == self.internal_port:
self._process_internal_outbound_frame(frame, from_network_interface)
return
# If the frame comes from the DMZ port, it's considered as DMZ outbound traffic
elif from_network_interface == self.dmz_port:
self._process_dmz_outbound_frame(frame, from_network_interface)
return
def _process_external_inbound_frame(self, frame: Frame, from_network_interface: RouterInterface) -> None:
"""
Process frames arriving from the external network.
Determines the path for frames based on their destination IP addresses and ACL rules for the external inbound
interface. Frames destined for the DMZ or internal network are forwarded accordingly, if allowed by the ACL.
If a frame is permitted by the ACL, it is either passed to the session manager (if applicable) or forwarded to
the appropriate network zone (DMZ/internal). Denied frames are logged and dropped.
:param frame: The frame to be processed, containing network layer and transport layer information.
:param from_network_interface: The interface on the firewall through which the frame was received.
"""
# check if External Inbound ACL Rules permit frame
permitted, rule = self.external_inbound_acl.is_permitted(frame)
if not permitted:
self.sys_log.info(f"Frame blocked at interface {from_network_interface} by rule {rule}")
return
self.software_manager.arp.add_arp_cache_entry(
ip_address=frame.ip.src_ip_address,
mac_address=frame.ethernet.src_mac_addr,
network_interface=from_network_interface,
)
if self.check_send_frame_to_session_manager(frame):
# Port is open on this Router so pass Frame up to session manager first
self.session_manager.receive_frame(frame, from_network_interface)
else:
# If the destination IP is within the DMZ network, process the frame as DMZ inbound
if frame.ip.dst_ip_address in self.dmz_port.ip_network:
self._process_dmz_inbound_frame(frame, from_network_interface)
else:
# Otherwise, process the frame as internal inbound
self._process_internal_inbound_frame(frame, from_network_interface)
def _process_external_outbound_frame(self, frame: Frame, from_network_interface: RouterInterface) -> None:
"""
Process frames that are outbound towards the external network.
:param frame: The frame to be processed.
:param from_network_interface: The network interface controller from which the frame is coming.
:param re_attempt: Indicates if the processing is a re-attempt, defaults to False.
"""
# check if External Outbound ACL Rules permit frame
permitted, rule = self.external_outbound_acl.is_permitted(frame=frame)
if not permitted:
self.sys_log.info(f"Frame blocked at interface {from_network_interface} by rule {rule}")
return
self.process_frame(frame=frame, from_network_interface=from_network_interface)
def _process_internal_inbound_frame(self, frame: Frame, from_network_interface: RouterInterface) -> None:
"""
Process frames that are inbound towards the internal LAN.
This method is responsible for handling frames coming from either the external network or the DMZ towards
the internal LAN. It checks the frames against the internal inbound ACL to decide whether to allow or deny
the traffic, and take appropriate actions.
:param frame: The frame to be processed.
:param from_network_interface: The network interface controller from which the frame is coming.
:param re_attempt: Indicates if the processing is a re-attempt, defaults to False.
"""
# check if Internal Inbound ACL Rules permit frame
permitted, rule = self.internal_inbound_acl.is_permitted(frame=frame)
if not permitted:
self.sys_log.info(f"Frame blocked at interface {from_network_interface} by rule {rule}")
return
self.process_frame(frame=frame, from_network_interface=from_network_interface)
def _process_internal_outbound_frame(self, frame: Frame, from_network_interface: RouterInterface) -> None:
"""
Process frames that are outbound from the internal network.
This method handles frames that are leaving the internal network. Depending on the destination IP address,
the frame may be forwarded to the DMZ or to the external network.
:param frame: The frame to be processed.
:param from_network_interface: The network interface controller from which the frame is coming.
:param re_attempt: Indicates if the processing is a re-attempt, defaults to False.
"""
permitted, rule = self.internal_outbound_acl.is_permitted(frame)
if not permitted:
self.sys_log.info(f"Frame blocked at interface {from_network_interface} by rule {rule}")
return
self.software_manager.arp.add_arp_cache_entry(
ip_address=frame.ip.src_ip_address,
mac_address=frame.ethernet.src_mac_addr,
network_interface=from_network_interface,
)
if self.check_send_frame_to_session_manager(frame):
# Port is open on this Router so pass Frame up to session manager first
self.session_manager.receive_frame(frame, from_network_interface)
else:
# If the destination IP is within the DMZ network, process the frame as DMZ inbound
if frame.ip.dst_ip_address in self.dmz_port.ip_network:
self._process_dmz_inbound_frame(frame, from_network_interface)
else:
# If the destination IP is not within the DMZ network, process the frame as external outbound
self._process_external_outbound_frame(frame, from_network_interface)
def _process_dmz_inbound_frame(self, frame: Frame, from_network_interface: RouterInterface) -> None:
"""
Process frames that are inbound from the DMZ.
This method is responsible for handling frames coming from either the external network or the internal LAN
towards the DMZ. It checks the frames against the DMZ inbound ACL to decide whether to allow or deny the
traffic, and take appropriate actions.
:param frame: The frame to be processed.
:param from_network_interface: The network interface controller from which the frame is coming.
:param re_attempt: Indicates if the processing is a re-attempt, defaults to False.
"""
# check if DMZ Inbound ACL Rules permit frame
permitted, rule = self.dmz_inbound_acl.is_permitted(frame=frame)
if not permitted:
self.sys_log.info(f"Frame blocked at interface {from_network_interface} by rule {rule}")
return
self.process_frame(frame=frame, from_network_interface=from_network_interface)
def _process_dmz_outbound_frame(self, frame: Frame, from_network_interface: RouterInterface) -> None:
"""
Process frames that are outbound from the DMZ.
This method handles frames originating from the DMZ and determines their appropriate path based on the
destination IP address. It involves checking the DMZ outbound ACL, consulting the ARP cache and the routing
table to find the correct outbound NIC, and then forwarding the frame to either the internal network or the
external network.
:param frame: The frame to be processed.
:param from_network_interface: The network interface controller from which the frame is coming.
:param re_attempt: Indicates if the processing is a re-attempt, defaults to False.
"""
permitted, rule = self.dmz_outbound_acl.is_permitted(frame)
if not permitted:
self.sys_log.info(f"Frame blocked at interface {from_network_interface} by rule {rule}")
return
self.software_manager.arp.add_arp_cache_entry(
ip_address=frame.ip.src_ip_address,
mac_address=frame.ethernet.src_mac_addr,
network_interface=from_network_interface,
)
if self.check_send_frame_to_session_manager(frame):
# Port is open on this Router so pass Frame up to session manager first
self.session_manager.receive_frame(frame, from_network_interface)
else:
# Attempt to get the outbound NIC from the ARP cache using the destination IP address
outbound_nic = self.software_manager.arp.get_arp_cache_network_interface(frame.ip.dst_ip_address)
# If outbound NIC is not found in the ARP cache, consult the routing table to find the best route
if not outbound_nic:
route = self.route_table.find_best_route(frame.ip.dst_ip_address)
if route:
# If a route is found, get the corresponding outbound NIC from the ARP cache using the next-hop IP
# address
outbound_nic = self.software_manager.arp.get_arp_cache_network_interface(route.next_hop_ip_address)
# If an outbound NIC is determined
if outbound_nic:
if outbound_nic == self.external_port:
# If the outbound NIC is the external port, check the frame against the DMZ outbound ACL and
# process it as an external outbound frame
self._process_external_outbound_frame(frame, from_network_interface)
return
elif outbound_nic == self.internal_port:
# If the outbound NIC is the internal port, check the frame against the DMZ outbound ACL and
# process it as an internal inbound frame
self._process_internal_inbound_frame(frame, from_network_interface)
return
# TODO: What to do here? Destination unreachable? Send ICMP back?
return
@property
def external_port(self) -> RouterInterface:
"""
The external port of the firewall.
:return: The external port connecting the firewall to the external network.
"""
return self.network_interface[EXTERNAL_PORT_ID]
@validate_call()
def configure_external_port(self, ip_address: Union[IPV4Address, str], subnet_mask: Union[IPV4Address, str]):
"""
Configure the external port with an IP address and a subnet mask.
Enables the port once configured.
:param ip_address: The IP address to assign to the external port.
:param subnet_mask: The subnet mask to assign to the external port.
"""
# Configure the external port with the specified IP address and subnet mask
self.configure_port(EXTERNAL_PORT_ID, ip_address, subnet_mask)
self.external_port.enable()
@property
def internal_port(self) -> RouterInterface:
"""
The internal port of the firewall.
:return: The external port connecting the firewall to the internal LAN.
"""
return self.network_interface[INTERNAL_PORT_ID]
@validate_call()
def configure_internal_port(self, ip_address: Union[IPV4Address, str], subnet_mask: Union[IPV4Address, str]):
"""
Configure the internal port with an IP address and a subnet mask.
Enables the port once configured.
:param ip_address: The IP address to assign to the internal port.
:param subnet_mask: The subnet mask to assign to the internal port.
"""
self.configure_port(INTERNAL_PORT_ID, ip_address, subnet_mask)
self.internal_port.enable()
@property
def dmz_port(self) -> RouterInterface:
"""
The DMZ port of the firewall.
:return: The external port connecting the firewall to the DMZ.
"""
return self.network_interface[DMZ_PORT_ID]
@validate_call()
def configure_dmz_port(self, ip_address: Union[IPV4Address, str], subnet_mask: Union[IPV4Address, str]):
"""
Configure the DMZ port with an IP address and a subnet mask.
Enables the port once configured.
:param ip_address: The IP address to assign to the DMZ port.
:param subnet_mask: The subnet mask to assign to the DMZ port.
"""
self.configure_port(DMZ_PORT_ID, ip_address, subnet_mask)
self.dmz_port.enable()

30
src/primaite/simulator/network/hardware/nodes/network/network_node.py Normal file
View File

@@ -0,0 +1,30 @@
from abc import abstractmethod
from primaite.simulator.network.hardware.base import NetworkInterface, Node
from primaite.simulator.network.transmission.data_link_layer import Frame
class NetworkNode(Node):
"""
Represents an abstract base class for a network node that can receive and process network frames.
This class provides a common interface for network nodes such as routers and switches, defining the essential
behavior that allows these devices to handle incoming network traffic. Implementations of this class must
provide functionality for receiving and processing frames received on their network interfaces.
"""
@abstractmethod
def receive_frame(self, frame: Frame, from_network_interface: NetworkInterface):
"""
Abstract method that must be implemented by subclasses to define how to receive and process frames.
This method is called when a frame is received by a network interface belonging to this node. Subclasses
should implement the logic to process the frame, including examining its contents, making forwarding decisions,
or performing any necessary actions based on the frame's protocol and destination.
:param frame: The network frame that has been received.
:type frame: Frame
:param from_network_interface: The network interface on which the frame was received.
:type from_network_interface: NetworkInterface
"""
pass

1423
src/primaite/simulator/network/hardware/nodes/network/router.py Normal file
View File

File diff suppressed because it is too large Load Diff

200
src/primaite/simulator/network/hardware/nodes/network/switch.py Normal file
View File

@@ -0,0 +1,200 @@
from __future__ import annotations
from typing import Dict, Optional
from prettytable import MARKDOWN, PrettyTable
from primaite import getLogger
from primaite.exceptions import NetworkError
from primaite.simulator.network.hardware.base import Link, WiredNetworkInterface
from primaite.simulator.network.hardware.nodes.network.network_node import NetworkNode
from primaite.simulator.network.transmission.data_link_layer import Frame
_LOGGER = getLogger(__name__)
class SwitchPort(WiredNetworkInterface):
"""
Represents a Switch Port.
Switch ports connect devices within the same network. They operate at the data link layer (Layer 2) of the OSI model
and are responsible for receiving and forwarding frames based on MAC addresses. Despite operating at Layer 2,
they are an essential part of network infrastructure, enabling LAN segmentation, bandwidth management, and
the creation of VLANs.
Inherits from:
- WiredNetworkInterface: Provides properties and methods specific to wired connections.
Switch ports typically do not have IP addresses assigned to them as they function at Layer 2, but managed switches
can have management IP addresses for remote management and configuration purposes.
"""
_connected_node: Optional[Switch] = None
"The Switch to which the SwitchPort is connected."
def set_original_state(self):
"""Sets the original state."""
vals_to_include = {"port_num", "mac_address", "speed", "mtu", "enabled"}
self._original_state = self.model_dump(include=vals_to_include)
super().set_original_state()
def describe_state(self) -> Dict:
"""
Produce a dictionary describing the current state of this object.
:return: Current state of this object and child objects.
:rtype: Dict
"""
state = super().describe_state()
state.update(
{
"mac_address": self.mac_address,
"speed": self.speed,
"mtu": self.mtu,
"enabled": self.enabled,
}
)
return state
def send_frame(self, frame: Frame) -> bool:
"""
Attempts to send a network frame through the interface.
:param frame: The network frame to be sent.
:return: A boolean indicating whether the frame was successfully sent.
"""
if self.enabled:
self.pcap.capture_outbound(frame)
self._connected_link.transmit_frame(sender_nic=self, frame=frame)
return True
# Cannot send Frame as the SwitchPort is not enabled
return False
def receive_frame(self, frame: Frame) -> bool:
"""
Receives a network frame on the interface.
:param frame: The network frame being received.
:return: A boolean indicating whether the frame was successfully received.
"""
if self.enabled:
frame.decrement_ttl()
if frame.ip and frame.ip.ttl < 1:
self._connected_node.sys_log.info("Frame discarded as TTL limit reached")
return False
self.pcap.capture_inbound(frame)
self._connected_node.receive_frame(frame=frame, from_network_interface=self)
return True
return False
class Switch(NetworkNode):
"""
A class representing a Layer 2 network switch.
:ivar num_ports: The number of ports on the switch. Default is 24.
"""
num_ports: int = 24
"The number of ports on the switch."
network_interfaces: Dict[str, SwitchPort] = {}
"The SwitchPorts on the Switch."
network_interface: Dict[int, SwitchPort] = {}
"The SwitchPorts on the Switch by port id."
mac_address_table: Dict[str, SwitchPort] = {}
"A MAC address table mapping destination MAC addresses to corresponding SwitchPorts."
def __init__(self, **kwargs):
super().__init__(**kwargs)
if not self.network_interface:
self.network_interface = {i: SwitchPort() for i in range(1, self.num_ports + 1)}
for port_num, port in self.network_interface.items():
port._connected_node = self
port.port_num = port_num
port.parent = self
port.port_num = port_num
def show(self, markdown: bool = False):
"""
Prints a table of the SwitchPorts on the Switch.
:param markdown: If True, outputs the table in markdown format. Default is False.
"""
table = PrettyTable(["Port", "MAC Address", "Speed", "Status"])
if markdown:
table.set_style(MARKDOWN)
table.align = "l"
table.title = f"{self.hostname} Switch Ports"
for port_num, port in self.network_interface.items():
table.add_row([port_num, port.mac_address, port.speed, "Enabled" if port.enabled else "Disabled"])
print(table)
def describe_state(self) -> Dict:
"""
Produce a dictionary describing the current state of this object.
:return: Current state of this object and child objects.
"""
state = super().describe_state()
state["ports"] = {port_num: port.describe_state() for port_num, port in self.network_interface.items()}
state["num_ports"] = self.num_ports # redundant?
state["mac_address_table"] = {mac: port.port_num for mac, port in self.mac_address_table.items()}
return state
def _add_mac_table_entry(self, mac_address: str, switch_port: SwitchPort):
"""
Private method to add an entry to the MAC address table.
:param mac_address: MAC address to be added.
:param switch_port: Corresponding SwitchPort object.
"""
mac_table_port = self.mac_address_table.get(mac_address)
if not mac_table_port:
self.mac_address_table[mac_address] = switch_port
self.sys_log.info(f"Added MAC table entry: Port {switch_port.port_num} -> {mac_address}")
else:
if mac_table_port != switch_port:
self.mac_address_table.pop(mac_address)
self.sys_log.info(f"Removed MAC table entry: Port {mac_table_port.port_num} -> {mac_address}")
self._add_mac_table_entry(mac_address, switch_port)
def receive_frame(self, frame: Frame, from_network_interface: SwitchPort):
"""
Forward a frame to the appropriate port based on the destination MAC address.
:param frame: The Frame being received.
:param from_network_interface: The SwitchPort that received the frame.
"""
src_mac = frame.ethernet.src_mac_addr
dst_mac = frame.ethernet.dst_mac_addr
self._add_mac_table_entry(src_mac, from_network_interface)
outgoing_port = self.mac_address_table.get(dst_mac)
if outgoing_port and dst_mac.lower() != "ff:ff:ff:ff:ff:ff":
outgoing_port.send_frame(frame)
else:
# If the destination MAC is not in the table, flood to all ports except incoming
for port in self.network_interface.values():
if port.enabled and port != from_network_interface:
port.send_frame(frame)
def disconnect_link_from_port(self, link: Link, port_number: int):
"""
Disconnect a given link from the specified port number on the switch.
:param link: The Link object to be disconnected.
:param port_number: The port number on the switch from where the link should be disconnected.
:raise NetworkError: When an invalid port number is provided or the link does not match the connection.
"""
port = self.network_interface.get(port_number)
if port is None:
msg = f"Invalid port number {port_number} on the switch"
_LOGGER.error(msg)
raise NetworkError(msg)
if port._connected_link != link:
msg = f"The link does not match the connection at port number {port_number}"
_LOGGER.error(msg)
raise NetworkError(msg)
port.disconnect_link()

211
src/primaite/simulator/network/hardware/nodes/network/wireless_router.py Normal file
View File

@@ -0,0 +1,211 @@
from typing import Any, Dict, Union
from pydantic import validate_call
from primaite.simulator.network.airspace import AirSpaceFrequency, IPWirelessNetworkInterface
from primaite.simulator.network.hardware.nodes.network.router import Router, RouterInterface
from primaite.simulator.network.transmission.data_link_layer import Frame
from primaite.utils.validators import IPV4Address
class WirelessAccessPoint(IPWirelessNetworkInterface):
"""
Represents a Wireless Access Point (AP) in a network.
This class models a Wireless Access Point, a device that allows wireless devices to connect to a wired network
using Wi-Fi or other wireless standards. The Wireless Access Point bridges the wireless and wired segments of
the network, allowing wireless devices to communicate with other devices on the network.
As an integral component of wireless networking, a Wireless Access Point provides functionalities for network
management, signal broadcasting, security enforcement, and connection handling. It also possesses Layer 3
capabilities such as IP addressing and subnetting, allowing for network segmentation and routing.
Inherits from:
- WirelessNetworkInterface: Provides basic properties and methods specific to wireless interfaces.
- Layer3Interface: Provides Layer 3 properties like ip_address and subnet_mask, enabling the device to manage
network traffic and routing.
This class can be further specialised or extended to support specific features or standards related to wireless
networking, such as different Wi-Fi versions, frequency bands, or advanced security protocols.
"""
def model_post_init(self, __context: Any) -> None:
"""
Performs post-initialisation checks to ensure the model's IP configuration is valid.
This method is invoked after the initialisation of a network model object to validate its network settings,
particularly to ensure that the assigned IP address is not a network address. This validation is crucial for
maintaining the integrity of network simulations and avoiding configuration errors that could lead to
unrealistic or incorrect behavior.
:param __context: Contextual information or parameters passed to the method, used for further initializing or
validating the model post-creation.
:raises ValueError: If the IP address is the same as the network address, indicating an incorrect configuration.
"""
if self.ip_network.network_address == self.ip_address:
raise ValueError(f"{self.ip_address}/{self.subnet_mask} must not be a network address")
def describe_state(self) -> Dict:
"""
Produce a dictionary describing the current state of this object.
:return: Current state of this object and child objects.
:rtype: Dict
"""
return super().describe_state()
def receive_frame(self, frame: Frame) -> bool:
"""
Receives a network frame on the interface.
:param frame: The network frame being received.
:return: A boolean indicating whether the frame was successfully received.
"""
if self.enabled:
frame.decrement_ttl()
if frame.ip and frame.ip.ttl < 1:
self._connected_node.sys_log.info("Frame discarded as TTL limit reached")
return False
frame.set_received_timestamp()
self.pcap.capture_inbound(frame)
# If this destination or is broadcast
if frame.ethernet.dst_mac_addr == self.mac_address or frame.ethernet.dst_mac_addr == "ff:ff:ff:ff:ff:ff":
self._connected_node.receive_frame(frame=frame, from_network_interface=self)
return True
return False
def __str__(self) -> str:
"""
String representation of the NIC.
:return: A string combining the port number, MAC address and IP address of the NIC.
"""
return f"Port {self.port_num}: {self.mac_address}/{self.ip_address} ({self.frequency})"
class WirelessRouter(Router):
"""
A WirelessRouter class that extends the functionality of a standard Router to include wireless capabilities.
This class represents a network device that performs routing functions similar to a traditional router but also
includes the functionality of a wireless access point. This allows the WirelessRouter to not only direct traffic
between wired networks but also to manage and facilitate wireless network connections.
A WirelessRouter is instantiated and configured with both wired and wireless interfaces. The wired interfaces are
managed similarly to those in a standard Router, while the wireless interfaces require additional configuration
specific to wireless settings, such as setting the frequency band (e.g., 2.4 GHz or 5 GHz for Wi-Fi).
The WirelessRouter facilitates creating a network environment where devices can be interconnected via both
Ethernet (wired) and Wi-Fi (wireless), making it an essential component for simulating more complex and realistic
network topologies within PrimAITE.
Example:
>>> wireless_router = WirelessRouter(hostname="wireless_router_1")
>>> wireless_router.configure_router_interface(
... ip_address="192.168.1.1",
... subnet_mask="255.255.255.0"
... )
>>> wireless_router.configure_wireless_access_point(
... ip_address="10.10.10.1",
... subnet_mask="255.255.255.0"
... frequency=AirSpaceFrequency.WIFI_2_4
... )
"""
network_interfaces: Dict[str, Union[RouterInterface, WirelessAccessPoint]] = {}
network_interface: Dict[int, Union[RouterInterface, WirelessAccessPoint]] = {}
def __init__(self, hostname: str, **kwargs):
super().__init__(hostname=hostname, num_ports=0, **kwargs)
self.connect_nic(WirelessAccessPoint(ip_address="127.0.0.1", subnet_mask="255.0.0.0", gateway="0.0.0.0"))
self.connect_nic(RouterInterface(ip_address="127.0.0.1", subnet_mask="255.0.0.0", gateway="0.0.0.0"))
self.set_original_state()
@property
def wireless_access_point(self) -> WirelessAccessPoint:
"""
Retrieves the wireless access point interface associated with this wireless router.
This property provides direct access to the WirelessAccessPoint interface of the router, facilitating wireless
communications. Specifically, it returns the interface configured on port 1, dedicated to establishing and
managing wireless network connections. This interface is essential for enabling wireless connectivity,
allowing devices within connect to the network wirelessly.
:return: The WirelessAccessPoint instance representing the wireless connection interface on port 1 of the
wireless router.
"""
return self.network_interface[1]
@validate_call()
def configure_wireless_access_point(
self,
ip_address: IPV4Address,
subnet_mask: IPV4Address,
frequency: AirSpaceFrequency = AirSpaceFrequency.WIFI_2_4,
):
"""
Configures a wireless access point (WAP).
Sets its IP address, subnet mask, and operating frequency. This method ensures the wireless access point is
properly set up to manage wireless communication over the specified frequency band.
The method first disables the WAP to safely apply configuration changes. After configuring the IP settings,
it sets the WAP to operate on the specified frequency band and then re-enables the WAP for operation.
:param ip_address: The IP address to be assigned to the wireless access point.
:param subnet_mask: The subnet mask associated with the IP address
:param frequency: The operating frequency of the wireless access point, defined by the AirSpaceFrequency
enum. This determines the frequency band (e.g., 2.4 GHz or 5 GHz) the access point will use for wireless
communication. Default is AirSpaceFrequency.WIFI_2_4.
"""
self.wireless_access_point.disable() # Temporarily disable the WAP for reconfiguration
network_interface = self.network_interface[1]
network_interface.ip_address = ip_address
network_interface.subnet_mask = subnet_mask
self.sys_log.info(f"Configured WAP {network_interface}")
self.set_original_state()
self.wireless_access_point.frequency = frequency # Set operating frequency
self.wireless_access_point.enable() # Re-enable the WAP with new settings
@property
def router_interface(self) -> RouterInterface:
"""
Retrieves the router interface associated with this wireless router.
This property provides access to the router interface configured for wired connections. It specifically
returns the interface configured on port 2, which is reserved for wired LAN/WAN connections.
:return: The RouterInterface instance representing the wired LAN/WAN connection on port 2 of the wireless
router.
"""
return self.network_interface[2]
@validate_call()
def configure_router_interface(
self,
ip_address: IPV4Address,
subnet_mask: IPV4Address,
):
"""
Configures a router interface.
Sets its IP address and subnet mask.
The method first disables the router interface to safely apply configuration changes. After configuring the IP
settings, it re-enables the router interface for operation.
:param ip_address: The IP address to be assigned to the router interface.
:param subnet_mask: The subnet mask associated with the IP address
"""
self.router_interface.disable() # Temporarily disable the router interface for reconfiguration
super().configure_port(port=2, ip_address=ip_address, subnet_mask=subnet_mask) # Set IP configuration
self.router_interface.enable() # Re-enable the router interface with new settings
def configure_port(self, port: int, ip_address: Union[IPV4Address, str], subnet_mask: Union[IPV4Address, str]):
"""Not Implemented."""
raise NotImplementedError(
"Please use the 'configure_wireless_access_point' and 'configure_router_interface' functions."
)

1029
src/primaite/simulator/network/hardware/nodes/router.py
View File

File diff suppressed because it is too large Load Diff

120
src/primaite/simulator/network/hardware/nodes/switch.py
View File

@@ -1,120 +0,0 @@
from typing import Dict
from prettytable import MARKDOWN, PrettyTable
from primaite import getLogger
from primaite.exceptions import NetworkError
from primaite.simulator.network.hardware.base import Link, Node, SwitchPort
from primaite.simulator.network.transmission.data_link_layer import Frame
_LOGGER = getLogger(__name__)
class Switch(Node):
"""
A class representing a Layer 2 network switch.
:ivar num_ports: The number of ports on the switch. Default is 24.
"""
num_ports: int = 24
"The number of ports on the switch."
switch_ports: Dict[int, SwitchPort] = {}
"The SwitchPorts on the switch."
mac_address_table: Dict[str, SwitchPort] = {}
"A MAC address table mapping destination MAC addresses to corresponding SwitchPorts."
def __init__(self, **kwargs):
super().__init__(**kwargs)
if not self.switch_ports:
self.switch_ports = {i: SwitchPort() for i in range(1, self.num_ports + 1)}
for port_num, port in self.switch_ports.items():
port._connected_node = self
port._port_num_on_node = port_num
port.parent = self
port.port_num = port_num
def show(self, markdown: bool = False):
"""
Prints a table of the SwitchPorts on the Switch.
:param markdown: If True, outputs the table in markdown format. Default is False.
"""
table = PrettyTable(["Port", "MAC Address", "Speed", "Status"])
if markdown:
table.set_style(MARKDOWN)
table.align = "l"
table.title = f"{self.hostname} Switch Ports"
for port_num, port in self.switch_ports.items():
table.add_row([port_num, port.mac_address, port.speed, "Enabled" if port.enabled else "Disabled"])
print(table)
def describe_state(self) -> Dict:
"""
Produce a dictionary describing the current state of this object.
:return: Current state of this object and child objects.
"""
state = super().describe_state()
state["ports"] = {port_num: port.describe_state() for port_num, port in self.switch_ports.items()}
state["num_ports"] = self.num_ports # redundant?
state["mac_address_table"] = {mac: port.port_num for mac, port in self.mac_address_table.items()}
return state
def _add_mac_table_entry(self, mac_address: str, switch_port: SwitchPort):
"""
Private method to add an entry to the MAC address table.
:param mac_address: MAC address to be added.
:param switch_port: Corresponding SwitchPort object.
"""
mac_table_port = self.mac_address_table.get(mac_address)
if not mac_table_port:
self.mac_address_table[mac_address] = switch_port
self.sys_log.info(f"Added MAC table entry: Port {switch_port.port_num} -> {mac_address}")
else:
if mac_table_port != switch_port:
self.mac_address_table.pop(mac_address)
self.sys_log.info(f"Removed MAC table entry: Port {mac_table_port.port_num} -> {mac_address}")
self._add_mac_table_entry(mac_address, switch_port)
def forward_frame(self, frame: Frame, incoming_port: SwitchPort):
"""
Forward a frame to the appropriate port based on the destination MAC address.
:param frame: The Frame to be forwarded.
:param incoming_port: The port number from which the frame was received.
"""
src_mac = frame.ethernet.src_mac_addr
dst_mac = frame.ethernet.dst_mac_addr
self._add_mac_table_entry(src_mac, incoming_port)
outgoing_port = self.mac_address_table.get(dst_mac)
if outgoing_port and dst_mac.lower() != "ff:ff:ff:ff:ff:ff":
outgoing_port.send_frame(frame)
else:
# If the destination MAC is not in the table, flood to all ports except incoming
for port in self.switch_ports.values():
if port.enabled and port != incoming_port:
port.send_frame(frame)
def disconnect_link_from_port(self, link: Link, port_number: int):
"""
Disconnect a given link from the specified port number on the switch.
:param link: The Link object to be disconnected.
:param port_number: The port number on the switch from where the link should be disconnected.
:raise NetworkError: When an invalid port number is provided or the link does not match the connection.
"""
port = self.switch_ports.get(port_number)
if port is None:
msg = f"Invalid port number {port_number} on the switch"
_LOGGER.error(msg)
raise NetworkError(msg)
if port._connected_link != link:
msg = f"The link does not match the connection at port number {port_number}"
_LOGGER.error(msg)
raise NetworkError(msg)
port.disconnect_link()