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

Merge remote-tracking branch 'origin/dev' into feature/2258-dosbot-cannot-be-added-via-configuration-file

Browse Source
This commit is contained in:
Czar Echavez
2024-02-13 16:04:02 +00:00
parent 7beacfd95f 9835b31dc5
commit 98a4fbecb5
17 changed files with 2610 additions and 274 deletions
Download Patch File Download Diff File Expand all files Collapse all files

280
tests/integration_tests/network/test_firewall.py Normal file
View File

@@ -0,0 +1,280 @@
from ipaddress import IPv4Address
import pytest
from primaite.simulator.network.container import Network
from primaite.simulator.network.hardware.nodes.host.computer import Computer
from primaite.simulator.network.hardware.nodes.network.firewall import Firewall
from primaite.simulator.network.hardware.nodes.network.router import ACLAction
from primaite.simulator.network.transmission.network_layer import IPProtocol
from primaite.simulator.network.transmission.transport_layer import Port
from primaite.simulator.system.services.ntp.ntp_client import NTPClient
from primaite.simulator.system.services.ntp.ntp_server import NTPServer
@pytest.fixture(scope="function")
def dmz_external_internal_network() -> Network:
"""
Fixture for setting up a simulated network with a firewall, external node, internal node, and DMZ node. This
configuration is designed to test firewall rules and their impact on traffic between these network segments.
-------------- -------------- --------------
| external |---------| firewall |---------| internal |
-------------- -------------- --------------
|
|
---------
| DMZ |
---------
The network is set up as follows:
- An external node simulates an entity outside the organization's network.
- An internal node represents a device within the organization's LAN.
- A DMZ (Demilitarized Zone) node acts as a server or service exposed to external traffic.
- A firewall node controls traffic between these nodes based on ACL (Access Control List) rules.
The firewall is configured to allow ICMP and ARP traffic across all interfaces to ensure basic connectivity
for the tests. Specific tests will modify ACL rules to test various traffic filtering scenarios.
:return: A `Network` instance with the described nodes and configurations.
"""
network = Network()
firewall_node: Firewall = Firewall(hostname="firewall_1", start_up_duration=0)
firewall_node.power_on()
# configure firewall ports
firewall_node.configure_external_port(
ip_address=IPv4Address("192.168.10.1"), subnet_mask=IPv4Address("255.255.255.0")
)
firewall_node.configure_dmz_port(ip_address=IPv4Address("192.168.1.1"), subnet_mask=IPv4Address("255.255.255.0"))
firewall_node.configure_internal_port(
ip_address=IPv4Address("192.168.0.1"), subnet_mask=IPv4Address("255.255.255.0")
)
# Allow ICMP
firewall_node.internal_inbound_acl.add_rule(action=ACLAction.PERMIT, protocol=IPProtocol.ICMP, position=23)
firewall_node.internal_outbound_acl.add_rule(action=ACLAction.PERMIT, protocol=IPProtocol.ICMP, position=23)
firewall_node.external_inbound_acl.add_rule(action=ACLAction.PERMIT, protocol=IPProtocol.ICMP, position=23)
firewall_node.external_outbound_acl.add_rule(action=ACLAction.PERMIT, protocol=IPProtocol.ICMP, position=23)
firewall_node.dmz_inbound_acl.add_rule(action=ACLAction.PERMIT, protocol=IPProtocol.ICMP, position=23)
firewall_node.dmz_outbound_acl.add_rule(action=ACLAction.PERMIT, protocol=IPProtocol.ICMP, position=23)
# Allow ARP
firewall_node.internal_inbound_acl.add_rule(
action=ACLAction.PERMIT, src_port=Port.ARP, dst_port=Port.ARP, position=22
)
firewall_node.internal_outbound_acl.add_rule(
action=ACLAction.PERMIT, src_port=Port.ARP, dst_port=Port.ARP, position=22
)
firewall_node.external_inbound_acl.add_rule(
action=ACLAction.PERMIT, src_port=Port.ARP, dst_port=Port.ARP, position=22
)
firewall_node.external_outbound_acl.add_rule(
action=ACLAction.PERMIT, src_port=Port.ARP, dst_port=Port.ARP, position=22
)
firewall_node.dmz_inbound_acl.add_rule(action=ACLAction.PERMIT, src_port=Port.ARP, dst_port=Port.ARP, position=22)
firewall_node.dmz_outbound_acl.add_rule(action=ACLAction.PERMIT, src_port=Port.ARP, dst_port=Port.ARP, position=22)
# external node
external_node = Computer(
hostname="external_node",
ip_address="192.168.10.2",
subnet_mask="255.255.255.0",
default_gateway="192.168.10.1",
start_up_duration=0,
)
external_node.power_on()
external_node.software_manager.install(NTPServer)
ntp_service: NTPServer = external_node.software_manager.software["NTPServer"]
ntp_service.start()
# connect external node to firewall node
network.connect(endpoint_b=external_node.network_interface[1], endpoint_a=firewall_node.external_port)
# internal node
internal_node = Computer(
hostname="internal_node",
ip_address="192.168.0.2",
subnet_mask="255.255.255.0",
default_gateway="192.168.0.1",
start_up_duration=0,
)
internal_node.power_on()
internal_node.software_manager.install(NTPClient)
internal_ntp_client: NTPClient = internal_node.software_manager.software["NTPClient"]
internal_ntp_client.configure(external_node.network_interface[1].ip_address)
internal_ntp_client.start()
# connect external node to firewall node
network.connect(endpoint_b=internal_node.network_interface[1], endpoint_a=firewall_node.internal_port)
# dmz node
dmz_node = Computer(
hostname="dmz_node",
ip_address="192.168.1.2",
subnet_mask="255.255.255.0",
default_gateway="192.168.1.1",
start_up_duration=0,
)
dmz_node.power_on()
dmz_ntp_client: NTPClient = dmz_node.software_manager.software["NTPClient"]
dmz_ntp_client.configure(external_node.network_interface[1].ip_address)
dmz_ntp_client.start()
# connect external node to firewall node
network.connect(endpoint_b=dmz_node.network_interface[1], endpoint_a=firewall_node.dmz_port)
return network
def test_firewall_can_ping_nodes(dmz_external_internal_network):
"""
Tests the firewall's ability to ping the external, internal, and DMZ nodes in the network.
Verifies that the firewall has connectivity to all nodes within the network by performing a ping operation.
Successful pings indicate proper network setup and basic ICMP traffic passage through the firewall.
"""
firewall = dmz_external_internal_network.get_node_by_hostname("firewall_1")
# ping from the firewall
assert firewall.ping("192.168.0.2") # firewall to internal
assert firewall.ping("192.168.1.2") # firewall to dmz
assert firewall.ping("192.168.10.2") # firewall to external
def test_nodes_can_ping_default_gateway(dmz_external_internal_network):
"""
Checks if the external, internal, and DMZ nodes can ping their respective default gateways.
This test confirms that each node is correctly configured with a route to its default gateway and that the
firewall permits ICMP traffic for these basic connectivity checks.
"""
external_node = dmz_external_internal_network.get_node_by_hostname("external_node")
internal_node = dmz_external_internal_network.get_node_by_hostname("internal_node")
dmz_node = dmz_external_internal_network.get_node_by_hostname("dmz_node")
assert internal_node.ping(internal_node.default_gateway) # default gateway internal
assert dmz_node.ping(dmz_node.default_gateway) # default gateway dmz
assert external_node.ping(external_node.default_gateway) # default gateway external
def test_nodes_can_ping_default_gateway_on_another_subnet(dmz_external_internal_network):
"""
Verifies that nodes can ping default gateways located in a different subnet, facilitated by the firewall.
This test assesses the routing and firewall ACL configurations that allow ICMP traffic between different
network segments, ensuring that nodes can reach default gateways outside their local subnet.
"""
external_node = dmz_external_internal_network.get_node_by_hostname("external_node")
internal_node = dmz_external_internal_network.get_node_by_hostname("internal_node")
dmz_node = dmz_external_internal_network.get_node_by_hostname("dmz_node")
assert internal_node.ping(external_node.default_gateway) # internal node to external default gateway
assert internal_node.ping(dmz_node.default_gateway) # internal node to dmz default gateway
assert dmz_node.ping(internal_node.default_gateway) # dmz node to internal default gateway
assert dmz_node.ping(external_node.default_gateway) # dmz node to external default gateway
assert external_node.ping(external_node.default_gateway) # external node to internal default gateway
assert external_node.ping(dmz_node.default_gateway) # external node to dmz default gateway
def test_nodes_can_ping_each_other(dmz_external_internal_network):
"""
Evaluates the ability of each node (external, internal, DMZ) to ping the other nodes within the network.
This comprehensive connectivity test checks if the firewall's current ACL configuration allows for inter-node
communication via ICMP pings, highlighting the effectiveness of the firewall rules in place.
"""
external_node = dmz_external_internal_network.get_node_by_hostname("external_node")
internal_node = dmz_external_internal_network.get_node_by_hostname("internal_node")
dmz_node = dmz_external_internal_network.get_node_by_hostname("dmz_node")
# test that nodes can ping each other
assert internal_node.ping(external_node.network_interface[1].ip_address)
assert internal_node.ping(dmz_node.network_interface[1].ip_address)
assert external_node.ping(internal_node.network_interface[1].ip_address)
assert external_node.ping(dmz_node.network_interface[1].ip_address)
assert dmz_node.ping(internal_node.network_interface[1].ip_address)
assert dmz_node.ping(external_node.network_interface[1].ip_address)
def test_service_blocked(dmz_external_internal_network):
"""
Tests the firewall's default blocking stance on NTP service requests from internal and DMZ nodes.
Initially, without specific ACL rules to allow NTP traffic, this test confirms that NTP clients on both the
internal and DMZ nodes are unable to update their time, demonstrating the firewall's effective blocking of
unspecified services.
"""
firewall = dmz_external_internal_network.get_node_by_hostname("firewall_1")
internal_node = dmz_external_internal_network.get_node_by_hostname("internal_node")
dmz_node = dmz_external_internal_network.get_node_by_hostname("dmz_node")
internal_ntp_client: NTPClient = internal_node.software_manager.software["NTPClient"]
dmz_ntp_client: NTPClient = dmz_node.software_manager.software["NTPClient"]
assert not internal_ntp_client.time
internal_ntp_client.request_time()
assert not internal_ntp_client.time
assert not dmz_ntp_client.time
dmz_ntp_client.request_time()
assert not dmz_ntp_client.time
firewall.show_rules()
def test_service_allowed_with_rule(dmz_external_internal_network):
"""
Tests that NTP service requests are allowed through the firewall based on ACL rules.
This test verifies the functionality of the firewall in a network scenario where both an internal node and
a node in the DMZ attempt to access NTP services. Initially, no NTP traffic is allowed. The test then
configures ACL rules on the firewall to permit NTP traffic and checks if the NTP clients on the internal
node and DMZ node can successfully request and receive time updates.
Procedure:
1. Assert that the internal node's NTP client initially has no time information due to ACL restrictions.
2. Add ACL rules to the firewall to permit outbound and inbound NTP traffic from the internal network.
3. Trigger an NTP time request from the internal node and assert that it successfully receives time
information.
4. Assert that the DMZ node's NTP client initially has no time information.
5. Add ACL rules to the firewall to permit outbound and inbound NTP traffic from the DMZ.
6. Trigger an NTP time request from the DMZ node and assert that it successfully receives time information.
Asserts:
- The internal node's NTP client has no time information before ACL rules are applied.
- The internal node's NTP client successfully receives time information after the appropriate ACL rules
are applied.
- The DMZ node's NTP client has no time information before ACL rules are applied for the DMZ.
- The DMZ node's NTP client successfully receives time information after the appropriate ACL rules for
the DMZ are applied.
"""
firewall = dmz_external_internal_network.get_node_by_hostname("firewall_1")
internal_node = dmz_external_internal_network.get_node_by_hostname("internal_node")
dmz_node = dmz_external_internal_network.get_node_by_hostname("dmz_node")
internal_ntp_client: NTPClient = internal_node.software_manager.software["NTPClient"]
dmz_ntp_client: NTPClient = dmz_node.software_manager.software["NTPClient"]
assert not internal_ntp_client.time
firewall.internal_outbound_acl.add_rule(action=ACLAction.PERMIT, src_port=Port.NTP, dst_port=Port.NTP, position=1)
firewall.internal_inbound_acl.add_rule(action=ACLAction.PERMIT, src_port=Port.NTP, dst_port=Port.NTP, position=1)
internal_ntp_client.request_time()
assert internal_ntp_client.time
assert not dmz_ntp_client.time
firewall.dmz_outbound_acl.add_rule(action=ACLAction.PERMIT, src_port=Port.NTP, dst_port=Port.NTP, position=1)
firewall.dmz_inbound_acl.add_rule(action=ACLAction.PERMIT, src_port=Port.NTP, dst_port=Port.NTP, position=1)
dmz_ntp_client.request_time()
assert dmz_ntp_client.time
firewall.show_rules()

87
tests/integration_tests/network/test_wireless_router.py Normal file
View File

@@ -0,0 +1,87 @@
import pytest
from primaite.simulator.network.airspace import AIR_SPACE, AirSpaceFrequency
from primaite.simulator.network.container import Network
from primaite.simulator.network.hardware.nodes.host.computer import Computer
from primaite.simulator.network.hardware.nodes.network.router import ACLAction
from primaite.simulator.network.hardware.nodes.network.wireless_router import WirelessRouter
from primaite.simulator.network.transmission.network_layer import IPProtocol
from primaite.simulator.network.transmission.transport_layer import Port
@pytest.fixture(scope="function")
def setup_network():
network = Network()
# Configure PC A
pc_a = Computer(
hostname="pc_a",
ip_address="192.168.0.2",
subnet_mask="255.255.255.0",
default_gateway="192.168.0.1",
start_up_duration=0,
)
pc_a.power_on()
network.add_node(pc_a)
# Configure Router 1
router_1 = WirelessRouter(hostname="router_1", start_up_duration=0)
router_1.power_on()
network.add_node(router_1)
# Configure the connection between PC A and Router 1 port 2
router_1.configure_router_interface("192.168.0.1", "255.255.255.0")
network.connect(pc_a.network_interface[1], router_1.network_interface[2])
# Configure Router 1 ACLs
router_1.acl.add_rule(action=ACLAction.PERMIT, src_port=Port.ARP, dst_port=Port.ARP, position=22)
router_1.acl.add_rule(action=ACLAction.PERMIT, protocol=IPProtocol.ICMP, position=23)
# Configure PC B
pc_b = Computer(
hostname="pc_b",
ip_address="192.168.2.2",
subnet_mask="255.255.255.0",
default_gateway="192.168.2.1",
start_up_duration=0,
)
pc_b.power_on()
network.add_node(pc_b)
# Configure Router 2
router_2 = WirelessRouter(hostname="router_2", start_up_duration=0)
router_2.power_on()
network.add_node(router_2)
# Configure the connection between PC B and Router 2 port 2
router_2.configure_router_interface("192.168.2.1", "255.255.255.0")
network.connect(pc_b.network_interface[1], router_2.network_interface[2])
# Configure Router 2 ACLs
# Configure the wireless connection between Router 1 port 1 and Router 2 port 1
router_1.configure_wireless_access_point("192.168.1.1", "255.255.255.0")
router_2.configure_wireless_access_point("192.168.1.2", "255.255.255.0")
AIR_SPACE.show()
router_1.route_table.add_route(
address="192.168.2.0", subnet_mask="255.255.255.0", next_hop_ip_address="192.168.1.2"
)
# Configure Route from Router 2 to PC A subnet
router_2.route_table.add_route(
address="192.168.0.2", subnet_mask="255.255.255.0", next_hop_ip_address="192.168.1.1"
)
return pc_a, pc_b, router_1, router_2
def test_cross_router_connectivity(setup_network):
pc_a, pc_b, router_1, router_2 = setup_network
# Ensure that PCs can ping across routers before any frequency change
assert pc_a.ping(pc_a.default_gateway), "PC A should ping its default gateway successfully."
assert pc_b.ping(pc_b.default_gateway), "PC B should ping its default gateway successfully."
assert pc_a.ping(pc_b.network_interface[1].ip_address), "PC A should ping PC B across routers successfully."
assert pc_b.ping(pc_a.network_interface[1].ip_address), "PC B should ping PC A across routers successfully."

330
tests/unit_tests/_primaite/_simulator/_network/_hardware/nodes/test_acl.py
View File

@@ -1,111 +1,293 @@
from ipaddress import IPv4Address
import pytest
from primaite.simulator.network.hardware.base import generate_mac_address
from primaite.simulator.network.hardware.nodes.network.router import ACLAction, Router
from primaite.simulator.network.transmission.network_layer import IPProtocol
from primaite.simulator.network.transmission.transport_layer import Port
from primaite.simulator.network.protocols.icmp import ICMPPacket
from primaite.simulator.network.transmission.data_link_layer import EthernetHeader, Frame
from primaite.simulator.network.transmission.network_layer import IPPacket, IPProtocol
from primaite.simulator.network.transmission.transport_layer import Port, TCPHeader, UDPHeader
def test_add_rule():
@pytest.fixture(scope="function")
def router_with_acl_rules():
"""
Provides a router instance with predefined ACL rules for testing.
:Setup:
1. Creates a Router object named "Router".
2. Adds a PERMIT rule for TCP traffic from 192.168.1.1:HTTPS to 192.168.1.2:HTTP.
3. Adds a DENY rule for TCP traffic from 192.168.1.3:8080 to 192.168.1.4:80.
:return: A configured Router object with ACL rules.
"""
router = Router("Router")
acl = router.acl
# Add rules here as needed
acl.add_rule(
action=ACLAction.PERMIT,
protocol=IPProtocol.TCP,
src_ip_address=IPv4Address("192.168.1.1"),
src_ip_address="192.168.1.1",
src_port=Port.HTTPS,
dst_ip_address="192.168.1.2",
dst_port=Port.HTTP,
position=1,
)
acl.add_rule(
action=ACLAction.DENY,
protocol=IPProtocol.TCP,
src_ip_address="192.168.1.3",
src_port=Port(8080),
dst_ip_address=IPv4Address("192.168.1.2"),
dst_ip_address="192.168.1.4",
dst_port=Port(80),
position=2,
)
return router
@pytest.fixture(scope="function")
def router_with_wildcard_acl():
"""
Provides a router instance with ACL rules that include wildcard masking for testing.
:Setup:
1. Creates a Router object named "Router".
2. Adds a PERMIT rule for TCP traffic from 192.168.1.1:8080 to 10.1.1.2:80.
3. Adds a DENY rule with a wildcard mask for TCP traffic from the 192.168.1.0/24 network to 10.1.1.3:443.
4. Adds a PERMIT rule for any traffic to the 10.2.0.0/16 network.
:return: A Router object with configured ACL rules, including rules with wildcard masking.
"""
router = Router("Router")
acl = router.acl
# Rule to permit traffic from a specific source IP and port to a specific destination IP and port
acl.add_rule(
action=ACLAction.PERMIT,
protocol=IPProtocol.TCP,
src_ip_address="192.168.1.1",
src_port=Port(8080),
dst_ip_address="10.1.1.2",
dst_port=Port(80),
position=1,
)
# Rule to deny traffic from an IP range to a specific destination IP and port
acl.add_rule(
action=ACLAction.DENY,
protocol=IPProtocol.TCP,
src_ip_address="192.168.1.0",
src_wildcard_mask="0.0.0.255",
dst_ip_address="10.1.1.3",
dst_port=Port(443),
position=2,
)
# Rule to permit any traffic to a range of destination IPs
acl.add_rule(
action=ACLAction.PERMIT,
protocol=None,
src_ip_address=None,
dst_ip_address="10.2.0.0",
dst_wildcard_mask="0.0.255.255",
position=3,
)
return router
def test_add_rule(router_with_acl_rules):
"""
Tests that an ACL rule is added correctly to the router's ACL.
Asserts:
- The action of the added rule is PERMIT.
- The protocol of the added rule is TCP.
- The source IP address matches "192.168.1.1".
- The source port is HTTPS.
- The destination IP address matches "192.168.1.2".
- The destination port is HTTP.
"""
acl = router_with_acl_rules.acl
assert acl.acl[1].action == ACLAction.PERMIT
assert acl.acl[1].protocol == IPProtocol.TCP
assert acl.acl[1].src_ip_address == IPv4Address("192.168.1.1")
assert acl.acl[1].src_port == Port(8080)
assert acl.acl[1].src_port == Port.HTTPS
assert acl.acl[1].dst_ip_address == IPv4Address("192.168.1.2")
assert acl.acl[1].dst_port == Port(80)
assert acl.acl[1].dst_port == Port.HTTP
def test_remove_rule():
router = Router("Router")
acl = router.acl
acl.add_rule(
action=ACLAction.PERMIT,
protocol=IPProtocol.TCP,
src_ip_address=IPv4Address("192.168.1.1"),
src_port=Port(8080),
dst_ip_address=IPv4Address("192.168.1.2"),
dst_port=Port(80),
position=1,
)
def test_remove_rule(router_with_acl_rules):
"""
Tests the removal of an ACL rule from the router's ACL.
Asserts that accessing the removed rule index in the ACL returns None.
"""
acl = router_with_acl_rules.acl
acl.remove_rule(1)
assert not acl.acl[1]
assert acl.acl[1] is None
def test_rules():
router = Router("Router")
acl = router.acl
acl.add_rule(
action=ACLAction.PERMIT,
protocol=IPProtocol.TCP,
src_ip_address=IPv4Address("192.168.1.1"),
src_port=Port(8080),
dst_ip_address=IPv4Address("192.168.1.2"),
dst_port=Port(80),
position=1,
)
acl.add_rule(
action=ACLAction.DENY,
protocol=IPProtocol.TCP,
src_ip_address=IPv4Address("192.168.1.3"),
src_port=Port(8080),
dst_ip_address=IPv4Address("192.168.1.4"),
dst_port=Port(80),
position=2,
)
is_permitted, rule = acl.is_permitted(
protocol=IPProtocol.TCP,
src_ip_address=IPv4Address("192.168.1.1"),
src_port=Port(8080),
dst_ip_address=IPv4Address("192.168.1.2"),
dst_port=Port(80),
def test_traffic_permitted_by_specific_rule(router_with_acl_rules):
"""
Verifies that traffic matching a specific ACL rule is correctly permitted.
Asserts traffic that matches a permit rule is allowed through the ACL.
"""
acl = router_with_acl_rules.acl
permitted_frame = Frame(
ethernet=EthernetHeader(src_mac_addr=generate_mac_address(), dst_mac_addr=generate_mac_address()),
ip=IPPacket(src_ip_address="192.168.1.1", dst_ip_address="192.168.1.2", protocol=IPProtocol.TCP),
tcp=TCPHeader(src_port=Port.HTTPS, dst_port=Port.HTTP),
)
is_permitted, _ = acl.is_permitted(permitted_frame)
assert is_permitted
is_permitted, rule = acl.is_permitted(
protocol=IPProtocol.TCP,
src_ip_address=IPv4Address("192.168.1.3"),
src_port=Port(8080),
dst_ip_address=IPv4Address("192.168.1.4"),
dst_port=Port(80),
def test_traffic_denied_by_specific_rule(router_with_acl_rules):
"""
Verifies that traffic matching a specific ACL rule is correctly denied.
Asserts traffic that matches a deny rule is blocked by the ACL.
"""
acl = router_with_acl_rules.acl
not_permitted_frame = Frame(
ethernet=EthernetHeader(src_mac_addr=generate_mac_address(), dst_mac_addr=generate_mac_address()),
ip=IPPacket(src_ip_address="192.168.1.3", dst_ip_address="192.168.1.4", protocol=IPProtocol.TCP),
tcp=TCPHeader(src_port=Port(8080), dst_port=Port(80)),
)
is_permitted, _ = acl.is_permitted(not_permitted_frame)
assert not is_permitted
def test_default_rule():
def test_default_rule(router_with_acl_rules):
"""
Tests the default deny rule of the ACL.
This test verifies that traffic which does not match any explicit permit rule in the ACL
is correctly denied, as per the common "default deny" security stance that ACLs implement.
Asserts the frame does not match any of the predefined ACL rules and is therefore not permitted by the ACL,
illustrating the default deny behavior when no explicit permit rule is matched.
"""
acl = router_with_acl_rules.acl
not_permitted_frame = Frame(
ethernet=EthernetHeader(src_mac_addr=generate_mac_address(), dst_mac_addr=generate_mac_address()),
ip=IPPacket(src_ip_address="192.168.1.5", dst_ip_address="192.168.1.12", protocol=IPProtocol.UDP),
udp=UDPHeader(src_port=Port.HTTPS, dst_port=Port.HTTP),
)
is_permitted, rule = acl.is_permitted(not_permitted_frame)
assert not is_permitted
def test_direct_ip_match_with_acl(router_with_wildcard_acl):
"""
Tests ACL functionality for a direct IP address match.
Asserts direct IP address match traffic is permitted by the ACL rule.
"""
acl = router_with_wildcard_acl.acl
frame = Frame(
ethernet=EthernetHeader(src_mac_addr=generate_mac_address(), dst_mac_addr=generate_mac_address()),
ip=IPPacket(src_ip_address="192.168.1.1", dst_ip_address="10.1.1.2", protocol=IPProtocol.TCP),
tcp=TCPHeader(src_port=Port(8080), dst_port=Port(80)),
)
assert acl.is_permitted(frame)[0], "Direct IP match should be permitted."
def test_ip_range_match_denied_with_acl(router_with_wildcard_acl):
"""
Tests ACL functionality for denying traffic from an IP range using wildcard masking.
Asserts traffic from the specified IP range is correctly denied by the ACL rule.
"""
acl = router_with_wildcard_acl.acl
frame = Frame(
ethernet=EthernetHeader(src_mac_addr=generate_mac_address(), dst_mac_addr=generate_mac_address()),
ip=IPPacket(src_ip_address="192.168.1.100", dst_ip_address="10.1.1.3", protocol=IPProtocol.TCP),
tcp=TCPHeader(src_port=Port(8080), dst_port=Port(443)),
)
assert not acl.is_permitted(frame)[0], "IP range match with wildcard mask should be denied."
def test_traffic_permitted_to_destination_range_with_acl(router_with_wildcard_acl):
"""
Tests ACL functionality for permitting traffic to a destination IP range using wildcard masking.
Asserts traffic to the specified destination IP range is correctly permitted by the ACL rule.
"""
acl = router_with_wildcard_acl.acl
frame = Frame(
ethernet=EthernetHeader(src_mac_addr=generate_mac_address(), dst_mac_addr=generate_mac_address()),
ip=IPPacket(src_ip_address="192.168.1.50", dst_ip_address="10.2.200.200", protocol=IPProtocol.UDP),
udp=UDPHeader(src_port=Port(1433), dst_port=Port(1433)),
)
assert acl.is_permitted(frame)[0], "Traffic to destination IP range should be permitted."
def test_ip_traffic_from_specific_subnet():
"""
Tests that the ACL permits or denies IP traffic from specific subnets, mimicking a Cisco ACL rule for IP traffic.
This test verifies the ACL's ability to permit all IP traffic from a specific subnet (192.168.1.0/24) while denying
traffic from other subnets. The test mimics a Cisco ACL rule that allows IP traffic from a specified range using
wildcard masking.
The test frames are constructed with varying protocols (TCP, UDP, ICMP) and source IP addresses, to demonstrate the
rule's general applicability to all IP protocols and its enforcement based on source IP address range.
Asserts
- Traffic from within the 192.168.1.0/24 subnet is permitted.
- Traffic from outside the 192.168.1.0/24 subnet is denied.
"""
router = Router("Router")
acl = router.acl
# Add rules here as needed
acl.add_rule(
action=ACLAction.PERMIT,
protocol=IPProtocol.TCP,
src_ip_address=IPv4Address("192.168.1.1"),
src_port=Port(8080),
dst_ip_address=IPv4Address("192.168.1.2"),
dst_port=Port(80),
src_ip_address="192.168.1.0",
src_wildcard_mask="0.0.0.255",
position=1,
)
acl.add_rule(
action=ACLAction.DENY,
protocol=IPProtocol.TCP,
src_ip_address=IPv4Address("192.168.1.3"),
src_port=Port(8080),
dst_ip_address=IPv4Address("192.168.1.4"),
dst_port=Port(80),
position=2,
permitted_frame_1 = Frame(
ethernet=EthernetHeader(src_mac_addr=generate_mac_address(), dst_mac_addr=generate_mac_address()),
ip=IPPacket(src_ip_address="192.168.1.50", dst_ip_address="10.2.200.200", protocol=IPProtocol.TCP),
tcp=TCPHeader(src_port=Port.POSTGRES_SERVER, dst_port=Port.POSTGRES_SERVER),
)
is_permitted, rule = acl.is_permitted(
protocol=IPProtocol.UDP,
src_ip_address=IPv4Address("192.168.1.5"),
src_port=Port(8080),
dst_ip_address=IPv4Address("192.168.1.12"),
dst_port=Port(80),
assert acl.is_permitted(permitted_frame_1)[0]
permitted_frame_2 = Frame(
ethernet=EthernetHeader(src_mac_addr=generate_mac_address(), dst_mac_addr=generate_mac_address()),
ip=IPPacket(src_ip_address="192.168.1.10", dst_ip_address="85.199.214.101", protocol=IPProtocol.UDP),
udp=UDPHeader(src_port=Port.NTP, dst_port=Port.NTP),
)
assert not is_permitted
assert acl.is_permitted(permitted_frame_2)[0]
permitted_frame_3 = Frame(
ethernet=EthernetHeader(src_mac_addr=generate_mac_address(), dst_mac_addr=generate_mac_address()),
ip=IPPacket(src_ip_address="192.168.1.200", dst_ip_address="192.168.1.1", protocol=IPProtocol.ICMP),
icmp=ICMPPacket(identifier=1),
)
assert acl.is_permitted(permitted_frame_3)[0]
not_permitted_frame_1 = Frame(
ethernet=EthernetHeader(src_mac_addr=generate_mac_address(), dst_mac_addr=generate_mac_address()),
ip=IPPacket(src_ip_address="192.168.0.50", dst_ip_address="10.2.200.200", protocol=IPProtocol.TCP),
tcp=TCPHeader(src_port=Port.POSTGRES_SERVER, dst_port=Port.POSTGRES_SERVER),
)
assert not acl.is_permitted(not_permitted_frame_1)[0]
not_permitted_frame_2 = Frame(
ethernet=EthernetHeader(src_mac_addr=generate_mac_address(), dst_mac_addr=generate_mac_address()),
ip=IPPacket(src_ip_address="192.168.2.10", dst_ip_address="85.199.214.101", protocol=IPProtocol.UDP),
udp=UDPHeader(src_port=Port.NTP, dst_port=Port.NTP),
)
assert not acl.is_permitted(not_permitted_frame_2)[0]
acl.show()