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

#2402 add firewall acl actions

Browse Source
This commit is contained in:
Cristian-VM2
2024-03-29 16:30:39 +00:00
parent 2eb900746b
commit 7299a12c64
5 changed files with 793 additions and 83 deletions
Download Patch File Download Diff File Expand all files Collapse all files

154
src/primaite/game/agent/actions.py
View File

@@ -458,7 +458,7 @@ class RouterACLAddRuleAction(AbstractAction):
permission_str = "UNUSED"
return ["do_nothing"] # NOT SUPPORTED, JUST DO NOTHING IF WE COME ACROSS THIS
elif permission == 1:
permission_str = "ALLOW"
permission_str = "PERMIT"
elif permission == 2:
permission_str = "DENY"
else:
@@ -540,6 +540,156 @@ class RouterACLRemoveRuleAction(AbstractAction):
return ["network", "node", target_router_nodename, "acl", "remove_rule", position]
class FirewallACLAddRuleAction(AbstractAction):
"""Action which adds a rule to a firewall port's ACL."""
def __init__(
self,
manager: "ActionManager",
max_acl_rules: int,
num_ips: int,
num_ports: int,
num_protocols: int,
**kwargs,
) -> None:
"""Init method for FirewallACLAddRuleAction.
:param manager: Reference to the ActionManager which created this action.
:type manager: ActionManager
:param max_acl_rules: Maximum number of ACL rules that can be added to the router.
:type max_acl_rules: int
:param num_ips: Number of IP addresses in the simulation.
:type num_ips: int
:param num_ports: Number of ports in the simulation.
:type num_ports: int
:param num_protocols: Number of protocols in the simulation.
:type num_protocols: int
"""
super().__init__(manager=manager)
num_permissions = 3
self.shape: Dict[str, int] = {
"position": max_acl_rules,
"permission": num_permissions,
"source_ip_id": num_ips,
"dest_ip_id": num_ips,
"source_port_id": num_ports,
"dest_port_id": num_ports,
"protocol_id": num_protocols,
}
def form_request(
self,
target_firewall_nodename: str,
firewall_port_name: str,
firewall_port_direction: str,
position: int,
permission: int,
source_ip_id: int,
dest_ip_id: int,
source_port_id: int,
dest_port_id: int,
protocol_id: int,
) -> List[str]:
"""Return the action formatted as a request which can be ingested by the PrimAITE simulation."""
if permission == 0:
permission_str = "UNUSED"
return ["do_nothing"] # NOT SUPPORTED, JUST DO NOTHING IF WE COME ACROSS THIS
elif permission == 1:
permission_str = "PERMIT"
elif permission == 2:
permission_str = "DENY"
else:
_LOGGER.warning(f"{self.__class__} received permission {permission}, expected 0 or 1.")
if protocol_id == 0:
return ["do_nothing"] # NOT SUPPORTED, JUST DO NOTHING IF WE COME ACROSS THIS
if protocol_id == 1:
protocol = "ALL"
else:
protocol = self.manager.get_internet_protocol_by_idx(protocol_id - 2)
# subtract 2 to account for UNUSED=0 and ALL=1.
if source_ip_id == 0:
return ["do_nothing"] # invalid formulation
elif source_ip_id == 1:
src_ip = "ALL"
else:
src_ip = self.manager.get_ip_address_by_idx(source_ip_id - 2)
# subtract 2 to account for UNUSED=0, and ALL=1
if source_port_id == 0:
return ["do_nothing"] # invalid formulation
elif source_port_id == 1:
src_port = "ALL"
else:
src_port = self.manager.get_port_by_idx(source_port_id - 2)
# subtract 2 to account for UNUSED=0, and ALL=1
if source_ip_id == 0:
return ["do_nothing"] # invalid formulation
elif dest_ip_id == 1:
dst_ip = "ALL"
else:
dst_ip = self.manager.get_ip_address_by_idx(dest_ip_id - 2)
# subtract 2 to account for UNUSED=0, and ALL=1
if dest_port_id == 0:
return ["do_nothing"] # invalid formulation
elif dest_port_id == 1:
dst_port = "ALL"
else:
dst_port = self.manager.get_port_by_idx(dest_port_id - 2)
# subtract 2 to account for UNUSED=0, and ALL=1
return [
"network",
"node",
target_firewall_nodename,
firewall_port_name,
firewall_port_direction,
"acl",
"add_rule",
permission_str,
protocol,
str(src_ip),
src_port,
str(dst_ip),
dst_port,
position,
]
class FirewallACLRemoveRuleAction(AbstractAction):
"""Action which removes a rule from a firewall port's ACL."""
def __init__(self, manager: "ActionManager", max_acl_rules: int, **kwargs) -> None:
"""Init method for RouterACLRemoveRuleAction.
:param manager: Reference to the ActionManager which created this action.
:type manager: ActionManager
:param max_acl_rules: Maximum number of ACL rules that can be added to the router.
:type max_acl_rules: int
"""
super().__init__(manager=manager)
self.shape: Dict[str, int] = {"position": max_acl_rules}
def form_request(
self, target_firewall_nodename: str, firewall_port_name: str, firewall_port_direction: str, position: int
) -> List[str]:
"""Return the action formatted as a request which can be ingested by the PrimAITE simulation."""
return [
"network",
"node",
target_firewall_nodename,
firewall_port_name,
firewall_port_direction,
"acl",
"remove_rule",
position,
]
class NetworkNICAbstractAction(AbstractAction):
"""
Abstract base class for NIC actions.
@@ -668,6 +818,8 @@ class ActionManager:
"NODE_RESET": NodeResetAction,
"ROUTER_ACL_ADDRULE": RouterACLAddRuleAction,
"ROUTER_ACL_REMOVERULE": RouterACLRemoveRuleAction,
"FIREWALL_ACL_ADDRULE": FirewallACLAddRuleAction,
"FIREWALL_ACL_REMOVERULE": FirewallACLRemoveRuleAction,
"NETWORK_NIC_ENABLE": NetworkNICEnableAction,
"NETWORK_NIC_DISABLE": NetworkNICDisableAction,
"NETWORK_PORT_ENABLE": NetworkPortEnableAction,

170
src/primaite/simulator/network/hardware/nodes/network/firewall.py
View File

@@ -1,10 +1,10 @@
from ipaddress import IPv4Address
from typing import Dict, Final, Optional, Union
from typing import Dict, Final, Union
from prettytable import MARKDOWN, PrettyTable
from pydantic import validate_call
from pydantic import Field, validate_call
# from primaite.simulator.core import RequestManager, RequestType
from primaite.simulator.core import RequestManager, RequestType
from primaite.simulator.network.hardware.node_operating_state import NodeOperatingState
from primaite.simulator.network.hardware.nodes.network.router import (
AccessControlList,
@@ -68,22 +68,34 @@ class Firewall(Router):
:ivar str hostname: The Firewall hostname.
"""
internal_inbound_acl: Optional[AccessControlList] = None
internal_inbound_acl: AccessControlList = Field(
default_factory=lambda: AccessControlList(name="Internal Inbound", implicit_action=ACLAction.DENY)
)
"""Access Control List for managing entering the internal network."""
internal_outbound_acl: Optional[AccessControlList] = None
internal_outbound_acl: AccessControlList = Field(
default_factory=lambda: AccessControlList(name="Internal Outbound", implicit_action=ACLAction.DENY)
)
"""Access Control List for managing traffic leaving the internal network."""
dmz_inbound_acl: Optional[AccessControlList] = None
dmz_inbound_acl: AccessControlList = Field(
default_factory=lambda: AccessControlList(name="DMZ Inbound", implicit_action=ACLAction.DENY)
)
"""Access Control List for managing traffic entering the DMZ."""
dmz_outbound_acl: Optional[AccessControlList] = None
dmz_outbound_acl: AccessControlList = Field(
default_factory=lambda: AccessControlList(name="DMZ Outbound", implicit_action=ACLAction.DENY)
)
"""Access Control List for managing traffic leaving the DMZ."""
external_inbound_acl: Optional[AccessControlList] = None
external_inbound_acl: AccessControlList = Field(
default_factory=lambda: AccessControlList(name="External Inbound", implicit_action=ACLAction.PERMIT)
)
"""Access Control List for managing traffic entering from an external network."""
external_outbound_acl: Optional[AccessControlList] = None
external_outbound_acl: AccessControlList = Field(
default_factory=lambda: AccessControlList(name="External Outbound", implicit_action=ACLAction.PERMIT)
)
"""Access Control List for managing traffic leaving towards an external network."""
def __init__(self, hostname: str, **kwargs):
@@ -101,88 +113,84 @@ class Firewall(Router):
self.connect_nic(
RouterInterface(ip_address="127.0.0.1", subnet_mask="255.0.0.0", gateway="0.0.0.0", port_name="dmz")
)
# 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_inbound_acl.sys_log = kwargs["sys_log"]
self.internal_inbound_acl.name = f"{hostname} - Internal Inbound"
self.internal_outbound_acl.sys_log = kwargs["sys_log"]
self.internal_outbound_acl.name = f"{hostname} - Internal Outbound"
self.dmz_inbound_acl.sys_log = kwargs["sys_log"]
self.dmz_inbound_acl.name = f"{hostname} - DMZ Inbound"
self.dmz_outbound_acl.sys_log = kwargs["sys_log"]
self.dmz_outbound_acl.name = f"{hostname} - DMZ Outbound"
self.external_inbound_acl.sys_log = kwargs["sys_log"]
self.external_inbound_acl.name = f"{hostname} - External Inbound"
self.external_outbound_acl.sys_log = kwargs["sys_log"]
self.external_outbound_acl.name = f"{hostname} - External Outbound"
def _init_request_manager(self) -> RequestManager:
"""
Initialise the request manager.
More information in user guide and docstring for SimComponent._init_request_manager.
"""
rm = super()._init_request_manager()
self._internal_acl_request_manager = RequestManager()
rm.add_request("internal", RequestType(func=self._internal_acl_request_manager))
self._dmz_acl_request_manager = RequestManager()
rm.add_request("dmz", RequestType(func=self._dmz_acl_request_manager))
self._external_acl_request_manager = RequestManager()
rm.add_request("external", RequestType(func=self._external_acl_request_manager))
self._internal_inbound_acl_request_manager = RequestManager()
self._internal_outbound_acl_request_manager = RequestManager()
self._internal_acl_request_manager.add_request(
"inbound", RequestType(func=self._internal_inbound_acl_request_manager)
)
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"
self._internal_acl_request_manager.add_request(
"outbound", RequestType(func=self._internal_outbound_acl_request_manager)
)
# 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.dmz_inbound_acl_request_manager = RequestManager()
self.dmz_outbound_acl_request_manager = RequestManager()
self._dmz_acl_request_manager.add_request("inbound", RequestType(func=self.dmz_inbound_acl_request_manager))
self._dmz_acl_request_manager.add_request("outbound", RequestType(func=self.dmz_outbound_acl_request_manager))
self.external_inbound_acl_request_manager = RequestManager()
self.external_outbound_acl_request_manager = RequestManager()
self._external_acl_request_manager.add_request(
"inbound", RequestType(func=self.external_inbound_acl_request_manager)
)
self.external_outbound_acl = AccessControlList(
sys_log=kwargs["sys_log"], implicit_action=ACLAction.PERMIT, name=f"{hostname} - External Outbound"
self._external_acl_request_manager.add_request(
"outbound", RequestType(func=self.external_outbound_acl_request_manager)
)
# def _init_request_manager(self) -> RequestManager:
# """
# Initialise the request manager.
self._internal_inbound_acl_request_manager.add_request(
"acl", RequestType(func=self.internal_inbound_acl._request_manager)
)
self._internal_outbound_acl_request_manager.add_request(
"acl", RequestType(func=self.internal_outbound_acl._request_manager)
)
# More information in user guide and docstring for SimComponent._init_request_manager.
# """
# rm = super()._init_request_manager()
# self._internal_acl_request_manager = RequestManager()
# rm.add_request("internal", RequestType(func=self._internal_acl_request_manager))
self.dmz_inbound_acl_request_manager.add_request("acl", RequestType(func=self.dmz_inbound_acl._request_manager))
self.dmz_outbound_acl_request_manager.add_request(
"acl", RequestType(func=self.dmz_outbound_acl._request_manager)
)
# self._dmz_acl_request_manager = RequestManager()
# rm.add_request("dmz", RequestType(func=self._dmz_acl_request_manager))
self.external_inbound_acl_request_manager.add_request(
"acl", RequestType(func=self.external_inbound_acl._request_manager)
)
self.external_outbound_acl_request_manager.add_request(
"acl", RequestType(func=self.external_outbound_acl._request_manager)
)
# self._external_acl_request_manager = RequestManager()
# rm.add_request("external", RequestType(func=self._external_acl_request_manager))
# self._internal_inbound_acl_request_manager = RequestManager()
# self._internal_outbound_acl_request_manager = RequestManager()
# self._internal_acl_request_manager.add_request(
# "inbound", RequestType(func=self._internal_inbound_acl_request_manager)
# )
# self._internal_acl_request_manager.add_request(
# "outbound", RequestType(func=self._internal_outbound_acl_request_manager)
# )
# self.dmz_inbound_acl_request_manager = RequestManager()
# self.dmz_outbound_acl_request_manager = RequestManager()
# self._dmz_acl_request_manager.add_request("inbound", RequestType(func=self.dmz_inbound_acl_request_manager))
# self._dmz_acl_request_manager.add_request("outbound", RequestType(func=self.dmz_outbound_acl_request_manager))
# self.external_inbound_acl_request_manager = RequestManager()
# self.external_outbound_acl_request_manager = RequestManager()
# self._external_acl_request_manager.add_request(
# "inbound", RequestType(func=self.external_inbound_acl_request_manager)
# )
# self._external_acl_request_manager.add_request(
# "outbound", RequestType(func=self.external_outbound_acl_request_manager)
# )
# self._internal_inbound_acl_request_manager.add_request(
# "acl", RequestType(func=self.internal_inbound_acl._request_manager)
# )
# self._internal_outbound_acl_request_manager.add_request(
# "acl", RequestType(func=self.internal_outbound_acl._request_manager)
# )
# self.dmz_inbound_acl_request_manager.add_request("acl", RequestType(func=self.dmz_inbound_acl._request_manager))
# self.dmz_outbound_acl_request_manager.add_request(
# "acl", RequestType(func=self.dmz_outbound_acl._request_manager)
# )
# self.external_inbound_acl_request_manager.add_request(
# "acl", RequestType(func=self.external_inbound_acl._request_manager)
# )
# self.external_outbound_acl_request_manager.add_request(
# "acl", RequestType(func=self.external_outbound_acl._request_manager)
# )
# return rm
return rm
def describe_state(self) -> Dict:
"""

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

@@ -275,7 +275,7 @@ class AccessControlList(SimComponent):
:ivar int max_acl_rules: The maximum number of ACL rules that can be added to the list. Defaults to 25.
"""
sys_log: SysLog
sys_log: Optional[SysLog] = None
implicit_action: ACLAction
implicit_rule: ACLRule
max_acl_rules: int = 25