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

#2205 - Introduced a Firewall class for enhanced network security and control, extending Router functionalities. Updated ACLRule to support IP ranges via wildcard masking for refined traffic filtering. Includes documentation updates.

Browse Source
This commit is contained in:
Chris McCarthy
2024-02-09 23:29:06 +00:00
parent a960741f66
commit 58af58810d
9 changed files with 1759 additions and 181 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
src/primaite/simulator/__init__.py
View File

@@ -12,8 +12,8 @@ class _SimOutput:
self._path: Path = (
_PRIMAITE_ROOT.parent.parent / "simulation_output" / datetime.now().strftime("%Y-%m-%d_%H-%M-%S")
)
self.save_pcap_logs: bool = False
self.save_sys_logs: bool = False
self.save_pcap_logs: bool = True
self.save_sys_logs: bool = True
@property
def path(self) -> Path:

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

@@ -0,0 +1,492 @@
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",
}
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(),
}
)
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
else:
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
else:
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
else:
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()

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

@@ -6,6 +6,7 @@ from ipaddress import IPv4Address, IPv4Network
from typing import Any, Dict, List, Optional, Tuple, Union
from prettytable import MARKDOWN, PrettyTable
from pydantic import validate_call
from primaite.simulator.core import RequestManager, RequestType, SimComponent
from primaite.simulator.network.hardware.base import IPWiredNetworkInterface
@@ -19,6 +20,43 @@ from primaite.simulator.network.transmission.transport_layer import Port
from primaite.simulator.system.core.sys_log import SysLog
from primaite.simulator.system.services.arp.arp import ARP
from primaite.simulator.system.services.icmp.icmp import ICMP
from primaite.utils.validators import IPV4Address
@validate_call()
def ip_matches_masked_range(ip_to_check: IPV4Address, base_ip: IPV4Address, wildcard_mask: IPV4Address) -> bool:
"""
Determine if a given IP address matches a range defined by a base IP address and a wildcard mask.
The wildcard mask specifies which bits in the IP address should be ignored (1) and which bits must match (0).
The function applies the wildcard mask to both the base IP and the IP address to check by first negating the
wildcard mask and then performing a bitwise AND operation. This process effectively masks out the bits indicated
by the wildcard mask. If the resulting masked IP addresses are equal, it means the IP address to check falls within
the range defined by the base IP and wildcard mask.
:param IPv4Address ip_to_check: The IP address to be checked.
:param IPv4Address base_ip: The base IP address defining the start of the range.
:param IPv4Address wildcard_mask: The wildcard mask specifying which bits to ignore.
:return: A boolean value indicating whether the IP address matches the masked range.
:rtype: bool
Example usage:
>>> ip_matches_masked_range(ip_to_check="192.168.10.10", base_ip="192.168.1.1", wildcard_mask="0.0.255.255")
False
"""
# Convert the IP addresses from IPv4Address objects to integer representations for bitwise operations
base_ip_int = int(base_ip)
ip_to_check_int = int(ip_to_check)
wildcard_int = int(wildcard_mask)
# Negate the wildcard mask and apply it to both the base IP and the IP to check using bitwise AND
# This step masks out the bits to be ignored according to the wildcard mask
masked_base_ip = base_ip_int & ~wildcard_int
masked_ip_to_check = ip_to_check_int & ~wildcard_int
# Compare the masked IP addresses to determine if they match within the masked range
return masked_base_ip == masked_ip_to_check
class ACLAction(Enum):
@@ -30,22 +68,62 @@ class ACLAction(Enum):
class ACLRule(SimComponent):
"""
Represents an Access Control List (ACL) rule.
Represents an Access Control List (ACL) rule within a network device.
:ivar ACLAction action: Action to be performed (Permit/Deny). Default is DENY.
:ivar Optional[IPProtocol] protocol: Network protocol. Default is None.
:ivar Optional[IPv4Address] src_ip_address: Source IP address. Default is None.
:ivar Optional[Port] src_port: Source port number. Default is None.
:ivar Optional[IPv4Address] dst_ip_address: Destination IP address. Default is None.
:ivar Optional[Port] dst_port: Destination port number. Default is None.
Enables fine-grained control over network traffic based on specified criteria such as IP addresses, protocols,
and ports. ACL rules can be configured to permit or deny traffic, providing a powerful mechanism for enforcing
security policies and traffic flow.
ACL rules support specifying exact match conditions, ranges of IP addresses using wildcard masks, and
protocol types. This flexibility allows for complex traffic filtering scenarios, from blocking or allowing
specific types of traffic to entire subnets.
**Usage:**
- **Dedicated IP Addresses**: To match traffic from or to a specific IP address, set the `src_ip_address`
and/or `dst_ip_address` without a wildcard mask. This is useful for rules that apply to individual hosts.
- **IP Ranges with Wildcard Masks**: For rules that apply to a range of IP addresses, use the `src_wildcard_mask`
and/or `dst_wildcard_mask` in conjunction with the base IP address. Wildcard masks are a way to specify which
bits of the IP address should be matched exactly and which bits can vary. For example, a wildcard mask of
`0.0.0.255` applied to a base address of `192.168.1.0` allows for any address from `192.168.1.0` to
`192.168.1.255`.
- **Allowing All IP Traffic**: To mimic the Cisco ACL rule that permits all IP traffic from a specific range,
you may use wildcard masks with the rule action set to `PERMIT`. If your implementation includes an `ALL`
option in the `IPProtocol` enum, use it to allow all protocols; otherwise, consider the rule without a
specified protocol to apply to all IP traffic.
The combination of these attributes allows for the creation of granular rules to control traffic flow
effectively, enhancing network security and management.
:ivar ACLAction action: Specifies whether to `PERMIT` or `DENY` the traffic that matches the rule conditions.
The default action is `DENY`.
:ivar Optional[IPProtocol] protocol: The network protocol (e.g., TCP, UDP, ICMP) to match. If `None`, the rule
applies to all protocols.
:ivar Optional[IPv4Address] src_ip_address: The source IP address to match. If combined with `src_wildcard_mask`,
it specifies the start of an IP range.
:ivar Optional[IPv4Address] src_wildcard_mask: The wildcard mask for the source IP address, defining the range
of addresses to match.
:ivar Optional[IPv4Address] dst_ip_address: The destination IP address to match. If combined with
`dst_wildcard_mask`, it specifies the start of an IP range.
:ivar Optional[IPv4Address] dst_wildcard_mask: The wildcard mask for the destination IP address, defining the
range of addresses to match.
:ivar Optional[Port] src_port: The source port number to match. Relevant for TCP/UDP protocols.
:ivar Optional[Port] dst_port: The destination port number to match. Relevant for TCP/UDP protocols.
"""
action: ACLAction = ACLAction.DENY
protocol: Optional[IPProtocol] = None
src_ip_address: Optional[IPv4Address] = None
src_ip_address: Optional[IPV4Address] = None
src_wildcard_mask: Optional[IPV4Address] = None
dst_ip_address: Optional[IPV4Address] = None
dst_wildcard_mask: Optional[IPV4Address] = None
src_port: Optional[Port] = None
dst_ip_address: Optional[IPv4Address] = None
dst_port: Optional[Port] = None
hit_count: int = 0
def __str__(self) -> str:
rule_strings = []
@@ -76,24 +154,132 @@ class ACLRule(SimComponent):
state["src_port"] = self.src_port.name if self.src_port else None
state["dst_ip_address"] = str(self.dst_ip_address) if self.dst_ip_address else None
state["dst_port"] = self.dst_port.name if self.dst_port else None
state["hit_count"] = self.hit_count
return state
def permit_frame_check(self, frame: Frame) -> bool:
"""
Evaluates whether a given network frame should be permitted or denied based on this ACL rule.
This method checks the frame against the ACL rule's criteria, including protocol, source and destination IP
addresses (with support for wildcard masking), and source and destination ports. The method assumes that an
unspecified (None) criterion implies a match for any value in that category. For IP addresses, wildcard masking
can be used to specify ranges of addresses that match the rule.
The method follows these steps to determine if a frame is permitted:
1. Check if the frame's protocol matches the ACL rule's protocol.
2. For source and destination IP addresses:
1. If a wildcard mask is defined, check if the frame's IP address is within the range specified by the base
IP address and the wildcard mask.
2. If no wildcard mask is defined, directly compare the frame's IP address to the one specified in the rule.
3. Check if the frame's source and destination ports match those specified in the rule.
4. The frame is permitted if it matches all specified criteria and the rule's action is PERMIT. Conversely, it
is not permitted if any criterion does not match or if the rule's action is DENY.
:param frame (Frame): The network frame to be evaluated.
:return: True if the frame is permitted by this ACL rule, False otherwise.
"""
protocol_matches = self.protocol == frame.ip.protocol if self.protocol else True
src_ip_matches = self.src_ip_address is None # Assume match if no specific src IP is defined
if self.src_ip_address:
if self.src_wildcard_mask:
# If a src wildcard mask is provided, use it to check the range
src_ip_matches = ip_matches_masked_range(
ip_to_check=frame.ip.src_ip_address,
base_ip=self.src_ip_address,
wildcard_mask=self.src_wildcard_mask,
)
else:
# Direct comparison if no wildcard mask is defined
src_ip_matches = frame.ip.src_ip_address == self.src_ip_address
dst_ip_matches = self.dst_ip_address is None # Assume match if no specific dst IP is defined
if self.dst_ip_address:
if self.dst_wildcard_mask:
# If a dst wildcard mask is provided, use it to check the range
dst_ip_matches = ip_matches_masked_range(
ip_to_check=frame.ip.dst_ip_address,
base_ip=self.dst_ip_address,
wildcard_mask=self.dst_wildcard_mask,
)
else:
# Direct comparison if no wildcard mask is defined
dst_ip_matches = frame.ip.dst_ip_address == self.dst_ip_address
src_port = None
dst_port = None
if frame.tcp:
src_port = frame.tcp.src_port
dst_port = frame.tcp.dst_port
elif frame.udp:
src_port = frame.udp.src_port
dst_port = frame.udp.dst_port
src_port_matches = self.src_port == src_port if self.src_port else True
dst_port_matches = self.dst_port == dst_port if self.dst_port else True
# The frame is permitted if all conditions are met
if protocol_matches and src_ip_matches and dst_ip_matches and src_port_matches and dst_port_matches:
return self.action == ACLAction.PERMIT
else:
# If any condition is not met, the decision depends on the rule action
return False
class AccessControlList(SimComponent):
"""
Manages a list of ACLRules to filter network traffic.
:ivar SysLog sys_log: System logging instance.
:ivar ACLAction implicit_action: Default action for rules.
:ivar ACLRule implicit_rule: Implicit ACL rule, created based on implicit_action.
:ivar int max_acl_rules: Maximum number of ACL rules that can be added. Default is 25.
:ivar List[Optional[ACLRule]] _acl: A list containing the ACL rules.
Manages a list of ACLRule instances to filter network traffic based on predefined criteria. This class
provides functionalities to add, remove, and evaluate ACL rules, thereby controlling the flow of traffic
through a network device.
ACL rules can specify conditions based on source and destination IP addresses, IP protocols (TCP, UDP, ICMP),
and port numbers. Rules can be configured to permit or deny traffic that matches these conditions, offering
granular control over network security policies.
Usage:
- **Dedicated IP Addresses**: Directly specify the source and/or destination IP addresses in an ACL rule to
match traffic to or from specific hosts.
- **IP Ranges with Wildcard Masks**: Use wildcard masks along with base IP addresses to define ranges of IP
addresses that an ACL rule applies to. This is useful for specifying subnets or ranges of IP addresses.
- **Allowing All IP Traffic**: To mimic a Cisco-style ACL rule that allows all IP traffic from a specified
range, use the wildcard mask in conjunction with a permit action. If your system supports an `ALL` option
for the IP protocol, this can be used to allow all types of IP traffic; otherwise, the absence of a
specified protocol can be interpreted to mean all protocols.
Methods include functionalities to add and remove rules, reset to default configurations, and evaluate
whether specific frames are permitted or denied based on the current set of rules. The class also provides
utility functions to describe the current state and display the rules in a human-readable format.
Example:
>>> # To add a rule that permits all TCP traffic from the subnet 192.168.1.0/24 to 192.168.2.0/24:
>>> acl = AccessControlList()
>>> acl.add_rule(
... action=ACLAction.PERMIT,
... protocol=IPProtocol.TCP,
... src_ip_address="192.168.1.0",
... src_wildcard_mask="0.0.0.255",
... dst_ip_address="192.168.2.0",
... dst_wildcard_mask="0.0.0.255"
...)
This example demonstrates adding a rule with specific source and destination IP ranges, using wildcard masks
to allow a broad range of traffic while maintaining control over the flow of data for security and
management purposes.
:ivar ACLAction implicit_action: The default action (permit or deny) applied when no other rule matches.
Typically set to deny to follow the principle of least privilege.
:ivar int max_acl_rules: The maximum number of ACL rules that can be added to the list. Defaults to 25.
"""
sys_log: SysLog
implicit_action: ACLAction
implicit_rule: ACLRule
max_acl_rules: int = 25
name: str
_acl: List[Optional[ACLRule]] = [None] * 24
_default_config: Dict[int, dict] = {}
"""Config dict describing how the ACL list should look at episode start"""
@@ -210,13 +396,16 @@ class AccessControlList(SimComponent):
"""
return len([rule for rule in self._acl if rule is not None])
@validate_call()
def add_rule(
self,
action: ACLAction,
action: ACLAction = ACLAction.DENY,
protocol: Optional[IPProtocol] = None,
src_ip_address: Optional[Union[str, IPv4Address]] = None,
src_ip_address: Optional[IPV4Address] = None,
src_wildcard_mask: Optional[IPV4Address] = None,
dst_ip_address: Optional[IPV4Address] = None,
dst_wildcard_mask: Optional[IPV4Address] = None,
src_port: Optional[Port] = None,
dst_ip_address: Optional[Union[str, IPv4Address]] = None,
dst_port: Optional[Port] = None,
position: int = 0,
) -> None:
@@ -224,25 +413,25 @@ class AccessControlList(SimComponent):
Add a new ACL rule.
:param ACLAction action: Action to be performed (Permit/Deny).
:param Optional[IPProtocol] protocol: Network protocol.
:param Optional[Union[str, IPv4Address]] src_ip_address: Source IP address.
:param Optional[Port] src_port: Source port number.
:param Optional[Union[str, IPv4Address]] dst_ip_address: Destination IP address.
:param Optional[Port] dst_port: Destination port number.
:param int position: Position in the ACL list to insert the rule.
:param protocol: Network protocol. Optional, default is None.
:param src_ip_address: Source IP address. Optional, default is None.
:param src_wildcard_mask: Source IP wildcard mask. Optional, default is None.
:param src_port: Source port number. Optional, default is None.
:param dst_ip_address: Destination IP address. Optional, default is None.
:param dst_wildcard_mask: Destination IP wildcard mask. Optional, default is None.
:param dst_port: Destination port number. Optional, default is None.
:param int position: Position in the ACL list to insert the rule. Optional, default is 1.
:raises ValueError: When the position is out of bounds.
"""
if isinstance(src_ip_address, str):
src_ip_address = IPv4Address(src_ip_address)
if isinstance(dst_ip_address, str):
dst_ip_address = IPv4Address(dst_ip_address)
if 0 <= position < self.max_acl_rules:
if self._acl[position]:
self.sys_log.info(f"Overwriting ACL rule at position {position}")
self._acl[position] = ACLRule(
action=action,
src_ip_address=src_ip_address,
src_wildcard_mask=src_wildcard_mask,
dst_ip_address=dst_ip_address,
dst_wildcard_mask=dst_wildcard_mask,
protocol=protocol,
src_port=src_port,
dst_port=dst_port,
@@ -264,43 +453,25 @@ class AccessControlList(SimComponent):
else:
raise ValueError(f"Cannot remove ACL rule, position {position} is out of bounds.")
def is_permitted(
self,
protocol: IPProtocol,
src_ip_address: Union[str, IPv4Address],
src_port: Optional[Port],
dst_ip_address: Union[str, IPv4Address],
dst_port: Optional[Port],
) -> Tuple[bool, Optional[Union[str, ACLRule]]]:
"""
Check if a packet with the given properties is permitted through the ACL.
:param protocol: The protocol of the packet.
:param src_ip_address: Source IP address of the packet. Accepts string and IPv4Address.
:param src_port: Source port of the packet. Optional.
:param dst_ip_address: Destination IP address of the packet. Accepts string and IPv4Address.
:param dst_port: Destination port of the packet. Optional.
:return: A tuple with a boolean indicating if the packet is permitted and an optional rule or implicit action
string.
"""
if not isinstance(src_ip_address, IPv4Address):
src_ip_address = IPv4Address(src_ip_address)
if not isinstance(dst_ip_address, IPv4Address):
dst_ip_address = IPv4Address(dst_ip_address)
for rule in self._acl:
if not rule:
def is_permitted(self, frame: Frame) -> Tuple[bool, ACLRule]:
"""Check if a packet with the given properties is permitted through the ACL."""
permitted = False
rule: ACLRule = None
for _rule in self._acl:
if not _rule:
continue
if (
(rule.src_ip_address == src_ip_address or rule.src_ip_address is None)
and (rule.dst_ip_address == dst_ip_address or rule.dst_ip_address is None)
and (rule.protocol == protocol or rule.protocol is None)
and (rule.src_port == src_port or rule.src_port is None)
and (rule.dst_port == dst_port or rule.dst_port is None)
):
return rule.action == ACLAction.PERMIT, rule
if _rule.permit_frame_check(frame):
permitted = True
rule = _rule
break
if not rule:
permitted = self.implicit_action == ACLAction.PERMIT
rule = self.implicit_rule
return self.implicit_action == ACLAction.PERMIT, f"Implicit {self.implicit_action.name}"
rule.hit_count += 1
return permitted, rule
def get_relevant_rules(
self,
@@ -346,11 +517,25 @@ class AccessControlList(SimComponent):
:param markdown: Whether to display the table in Markdown format. Defaults to False.
"""
table = PrettyTable(["Index", "Action", "Protocol", "Src IP", "Src Port", "Dst IP", "Dst Port"])
table = PrettyTable(
[
"Index",
"Action",
"Protocol",
"Src IP",
"Src Wildcard",
"Src Port",
"Dst IP",
"Dst Wildcard",
"Dst Port",
"Hit Count",
]
)
if markdown:
table.set_style(MARKDOWN)
table.align = "l"
table.title = f"{self.sys_log.hostname} Access Control List"
table.title = f"{self.name} Access Control List"
for index, rule in enumerate(self.acl + [self.implicit_rule]):
if rule:
table.add_row(
@@ -359,22 +544,16 @@ class AccessControlList(SimComponent):
rule.action.name if rule.action else "ANY",
rule.protocol.name if rule.protocol else "ANY",
rule.src_ip_address if rule.src_ip_address else "ANY",
rule.src_wildcard_mask if rule.src_wildcard_mask else "ANY",
f"{rule.src_port.value} ({rule.src_port.name})" if rule.src_port else "ANY",
rule.dst_ip_address if rule.dst_ip_address else "ANY",
rule.dst_wildcard_mask if rule.dst_wildcard_mask else "ANY",
f"{rule.dst_port.value} ({rule.dst_port.name})" if rule.dst_port else "ANY",
rule.hit_count,
]
)
print(table)
@property
def num_rules(self) -> int:
"""
Get the number of rules in the ACL.
:return: The number of rules in the ACL.
"""
return len([rule for rule in self._acl if rule is not None])
class RouteEntry(SimComponent):
"""
@@ -880,7 +1059,7 @@ class Router(NetworkNode):
if not kwargs.get("sys_log"):
kwargs["sys_log"] = SysLog(hostname)
if not kwargs.get("acl"):
kwargs["acl"] = AccessControlList(sys_log=kwargs["sys_log"], implicit_action=ACLAction.DENY)
kwargs["acl"] = AccessControlList(sys_log=kwargs["sys_log"], implicit_action=ACLAction.DENY, name=hostname)
if not kwargs.get("route_table"):
kwargs["route_table"] = RouteTable(sys_log=kwargs["sys_log"])
super().__init__(hostname=hostname, num_ports=num_ports, **kwargs)
@@ -1008,6 +1187,36 @@ class Router(NetworkNode):
state["acl"] = self.acl.describe_state()
return state
def check_send_frame_to_session_manager(self, frame: Frame) -> bool:
"""
Determines whether a given network frame should be forwarded to the session manager.
his function evaluates whether the destination IP address of the frame corresponds to one of the router's
interface IP addresses. If so, it then checks if the frame is an ICMP packet or if the destination port matches
any of the ports that the router's software manager identifies as open. If either condition is met, the frame
is considered for further processing by the session manager, implying potential application-level handling or
response generation.
:param frame: The network frame to be evaluated.
:return: A boolean value indicating whether the frame should be sent to the session manager. ``True`` if the
frame's destination IP matches the router's interface and is directed to an open port or is an ICMP packet,
otherwise, ``False``.
"""
dst_ip_address = frame.ip.dst_ip_address
dst_port = None
if frame.ip.protocol == IPProtocol.TCP:
dst_port = frame.tcp.dst_port
elif frame.ip.protocol == IPProtocol.UDP:
dst_port = frame.udp.dst_port
if self.ip_is_router_interface(dst_ip_address) and (
frame.icmp or dst_port in self.software_manager.get_open_ports()
):
return True
return False
def receive_frame(self, frame: Frame, from_network_interface: RouterInterface):
"""
Processes an incoming frame received on one of the router's interfaces.
@@ -1021,26 +1230,8 @@ class Router(NetworkNode):
if self.operating_state != NodeOperatingState.ON:
return
protocol = frame.ip.protocol
src_ip_address = frame.ip.src_ip_address
dst_ip_address = frame.ip.dst_ip_address
src_port = None
dst_port = None
if frame.ip.protocol == IPProtocol.TCP:
src_port = frame.tcp.src_port
dst_port = frame.tcp.dst_port
elif frame.ip.protocol == IPProtocol.UDP:
src_port = frame.udp.src_port
dst_port = frame.udp.dst_port
# Check if it's permitted
permitted, rule = self.acl.is_permitted(
protocol=protocol,
src_ip_address=src_ip_address,
src_port=src_port,
dst_ip_address=dst_ip_address,
dst_port=dst_port,
)
permitted, rule = self.acl.is_permitted(frame)
if not permitted:
at_port = self._get_port_of_nic(from_network_interface)
@@ -1054,13 +1245,7 @@ class Router(NetworkNode):
network_interface=from_network_interface,
)
send_to_session_manager = False
if (frame.icmp and self.ip_is_router_interface(dst_ip_address)) or (
dst_port in self.software_manager.get_open_ports()
):
send_to_session_manager = True
if send_to_session_manager:
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:
@@ -1196,7 +1381,7 @@ class Router(NetworkNode):
def show(self, markdown: bool = False):
"""
Prints the state of the Ethernet interfaces on the Router.
Prints the state of the network interfaces on the Router.
:param markdown: Flag to indicate if the output should be in markdown format.
"""
@@ -1205,7 +1390,7 @@ class Router(NetworkNode):
if markdown:
table.set_style(MARKDOWN)
table.align = "l"
table.title = f"{self.hostname} Ethernet Interfaces"
table.title = f"{self.hostname} Network Interfaces"
for port, network_interface in self.network_interface.items():
table.add_row(
[

6
src/primaite/simulator/network/transmission/network_layer.py
View File

@@ -1,9 +1,9 @@
from enum import Enum
from ipaddress import IPv4Address
from pydantic import BaseModel
from primaite import getLogger
from primaite.utils.validators import IPV4Address
_LOGGER = getLogger(__name__)
@@ -73,9 +73,9 @@ class IPPacket(BaseModel):
... )
"""
src_ip_address: IPv4Address
src_ip_address: IPV4Address
"Source IP address."
dst_ip_address: IPv4Address
dst_ip_address: IPV4Address
"Destination IP address."
protocol: IPProtocol = IPProtocol.TCP
"IPProtocol."