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

Merge remote-tracking branch 'origin/dev' into feature/2319-database-admin

Browse Source
This commit is contained in:
Marek Wolan
2024-03-01 09:23:00 +00:00
parent bd0b2e0033 b849bfd6e2
commit 9ff8adab1a
88 changed files with 3595 additions and 668 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 = True
self.save_sys_logs: bool = True
self.save_pcap_logs: bool = False
self.save_sys_logs: bool = False
@property
def path(self) -> Path:

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

@@ -1,7 +1,7 @@
# flake8: noqa
"""Core of the PrimAITE Simulator."""
from abc import ABC, abstractmethod
from typing import Callable, ClassVar, Dict, List, Optional, Union
from abc import abstractmethod
from typing import Callable, Dict, List, Optional, Union
from uuid import uuid4
from pydantic import BaseModel, ConfigDict, Field

4
src/primaite/simulator/network/airspace.py
View File

@@ -168,7 +168,9 @@ class WirelessNetworkInterface(NetworkInterface, ABC):
self.enabled = True
self._connected_node.sys_log.info(f"Network Interface {self} enabled")
self.pcap = PacketCapture(hostname=self._connected_node.hostname, interface_num=self.port_num)
self.pcap = PacketCapture(
hostname=self._connected_node.hostname, port_num=self.port_num, port_name=self.port_name
)
AIR_SPACE.add_wireless_interface(self)
def disable(self):

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

@@ -94,6 +94,9 @@ class NetworkInterface(SimComponent, ABC):
port_num: Optional[int] = None
"The port number assigned to this interface on the connected node."
port_name: Optional[str] = None
"The port name assigned to this interface on the connected node."
pcap: Optional[PacketCapture] = None
"A PacketCapture instance for capturing and analysing packets passing through this interface."
@@ -248,7 +251,7 @@ class NetworkInterface(SimComponent, ABC):
:return: A string combining the port number and the mac address
"""
return f"Port {self.port_num}: {self.mac_address}"
return f"Port {self.port_name if self.port_name else self.port_num}: {self.mac_address}"
class WiredNetworkInterface(NetworkInterface, ABC):
@@ -293,7 +296,9 @@ class WiredNetworkInterface(NetworkInterface, ABC):
self.enabled = True
self._connected_node.sys_log.info(f"Network Interface {self} enabled")
self.pcap = PacketCapture(hostname=self._connected_node.hostname, interface_num=self.port_num)
self.pcap = PacketCapture(
hostname=self._connected_node.hostname, port_num=self.port_num, port_name=self.port_name
)
if self._connected_link:
self._connected_link.endpoint_up()
@@ -1024,7 +1029,7 @@ class Node(SimComponent):
self.sys_log.info("Resetting")
self.power_off()
def connect_nic(self, network_interface: NetworkInterface):
def connect_nic(self, network_interface: NetworkInterface, port_name: Optional[str] = None):
"""
Connect a Network Interface to the node.
@@ -1036,7 +1041,9 @@ class Node(SimComponent):
new_nic_num = len(self.network_interfaces)
self.network_interface[new_nic_num] = network_interface
network_interface._connected_node = self
network_interface._port_num_on_node = new_nic_num
network_interface.port_num = new_nic_num
if port_name:
network_interface.port_name = port_name
network_interface.parent = self
self.sys_log.info(f"Connected Network Interface {network_interface}")
if self.operating_state == NodeOperatingState.ON:

2
src/primaite/simulator/network/hardware/network_interface/wireless/wireless_access_point.py
View File

@@ -83,4 +83,4 @@ class WirelessAccessPoint(IPWirelessNetworkInterface):
: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}"
return f"Port {self.port_name if self.port_name else self.port_num}: {self.mac_address}/{self.ip_address}"

2
src/primaite/simulator/network/hardware/network_interface/wireless/wireless_nic.py
View File

@@ -80,4 +80,4 @@ class WirelessNIC(IPWirelessNetworkInterface):
: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}"
return f"Port {self.port_name if self.port_name else self.port_num}: {self.mac_address}/{self.ip_address}"

33
src/primaite/simulator/network/hardware/nodes/host/host_node.py
View File

@@ -1,7 +1,7 @@
from __future__ import annotations
from ipaddress import IPv4Address
from typing import Any, Dict, Optional
from typing import Any, ClassVar, Dict, Optional
from primaite import getLogger
from primaite.simulator.network.hardware.base import IPWiredNetworkInterface, Link, Node
@@ -250,7 +250,7 @@ class NIC(IPWiredNetworkInterface):
: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}"
return f"Port {self.port_name if self.port_name else self.port_num}: {self.mac_address}/{self.ip_address}"
class HostNode(Node):
@@ -297,6 +297,16 @@ class HostNode(Node):
* Web Browser: Provides web browsing capabilities.
"""
SYSTEM_SOFTWARE: ClassVar[Dict] = {
"HostARP": HostARP,
"ICMP": ICMP,
"DNSClient": DNSClient,
"FTPClient": FTPClient,
"NTPClient": NTPClient,
"WebBrowser": WebBrowser,
}
"""List of system software that is automatically installed on nodes."""
network_interfaces: Dict[str, NIC] = {}
"The Network Interfaces on the node."
network_interface: Dict[int, NIC] = {}
@@ -313,23 +323,8 @@ class HostNode(Node):
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)
for _, software_class in self.SYSTEM_SOFTWARE.items():
self.software_manager.install(software_class)
super()._install_system_software()

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

@@ -1,8 +1,10 @@
from ipaddress import IPv4Address
from typing import Dict, Final, Optional, Union
from prettytable import MARKDOWN, PrettyTable
from pydantic import validate_call
from primaite.simulator.network.hardware.node_operating_state import NodeOperatingState
from primaite.simulator.network.hardware.nodes.network.router import (
AccessControlList,
ACLAction,
@@ -10,6 +12,8 @@ from primaite.simulator.network.hardware.nodes.network.router import (
RouterInterface,
)
from primaite.simulator.network.transmission.data_link_layer import Frame
from primaite.simulator.network.transmission.network_layer import IPProtocol
from primaite.simulator.network.transmission.transport_layer import Port
from primaite.simulator.system.core.sys_log import SysLog
from primaite.utils.validators import IPV4Address
@@ -85,7 +89,17 @@ class Firewall(Router):
if not kwargs.get("sys_log"):
kwargs["sys_log"] = SysLog(hostname)
super().__init__(hostname=hostname, num_ports=3, **kwargs)
super().__init__(hostname=hostname, num_ports=0, **kwargs)
self.connect_nic(
RouterInterface(ip_address="127.0.0.1", subnet_mask="255.0.0.0", gateway="0.0.0.0", port_name="external")
)
self.connect_nic(
RouterInterface(ip_address="127.0.0.1", subnet_mask="255.0.0.0", gateway="0.0.0.0", port_name="internal")
)
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(
@@ -473,3 +487,123 @@ class Firewall(Router):
"""
self.configure_port(DMZ_PORT_ID, ip_address, subnet_mask)
self.dmz_port.enable()
@classmethod
def from_config(cls, cfg: dict) -> "Firewall":
"""Create a firewall based on a config dict."""
firewall = Firewall(
hostname=cfg["hostname"],
operating_state=NodeOperatingState.ON
if not (p := cfg.get("operating_state"))
else NodeOperatingState[p.upper()],
)
if "ports" in cfg:
internal_port = cfg["ports"]["internal_port"]
external_port = cfg["ports"]["external_port"]
dmz_port = cfg["ports"]["dmz_port"]
# configure internal port
firewall.configure_internal_port(
ip_address=IPV4Address(internal_port.get("ip_address")),
subnet_mask=IPV4Address(internal_port.get("subnet_mask", "255.255.255.0")),
)
# configure external port
firewall.configure_external_port(
ip_address=IPV4Address(external_port.get("ip_address")),
subnet_mask=IPV4Address(external_port.get("subnet_mask", "255.255.255.0")),
)
# configure dmz port
firewall.configure_dmz_port(
ip_address=IPV4Address(dmz_port.get("ip_address")),
subnet_mask=IPV4Address(dmz_port.get("subnet_mask", "255.255.255.0")),
)
if "acl" in cfg:
# acl rules for internal_inbound_acl
if cfg["acl"]["internal_inbound_acl"]:
for r_num, r_cfg in cfg["acl"]["internal_inbound_acl"].items():
firewall.internal_inbound_acl.add_rule(
action=ACLAction[r_cfg["action"]],
src_port=None if not (p := r_cfg.get("src_port")) else Port[p],
dst_port=None if not (p := r_cfg.get("dst_port")) else Port[p],
protocol=None if not (p := r_cfg.get("protocol")) else IPProtocol[p],
src_ip_address=r_cfg.get("src_ip"),
dst_ip_address=r_cfg.get("dst_ip"),
position=r_num,
)
# acl rules for internal_outbound_acl
if cfg["acl"]["internal_outbound_acl"]:
for r_num, r_cfg in cfg["acl"]["internal_outbound_acl"].items():
firewall.internal_outbound_acl.add_rule(
action=ACLAction[r_cfg["action"]],
src_port=None if not (p := r_cfg.get("src_port")) else Port[p],
dst_port=None if not (p := r_cfg.get("dst_port")) else Port[p],
protocol=None if not (p := r_cfg.get("protocol")) else IPProtocol[p],
src_ip_address=r_cfg.get("src_ip"),
dst_ip_address=r_cfg.get("dst_ip"),
position=r_num,
)
# acl rules for dmz_inbound_acl
if cfg["acl"]["dmz_inbound_acl"]:
for r_num, r_cfg in cfg["acl"]["dmz_inbound_acl"].items():
firewall.dmz_inbound_acl.add_rule(
action=ACLAction[r_cfg["action"]],
src_port=None if not (p := r_cfg.get("src_port")) else Port[p],
dst_port=None if not (p := r_cfg.get("dst_port")) else Port[p],
protocol=None if not (p := r_cfg.get("protocol")) else IPProtocol[p],
src_ip_address=r_cfg.get("src_ip"),
dst_ip_address=r_cfg.get("dst_ip"),
position=r_num,
)
# acl rules for dmz_outbound_acl
if cfg["acl"]["dmz_outbound_acl"]:
for r_num, r_cfg in cfg["acl"]["dmz_outbound_acl"].items():
firewall.dmz_outbound_acl.add_rule(
action=ACLAction[r_cfg["action"]],
src_port=None if not (p := r_cfg.get("src_port")) else Port[p],
dst_port=None if not (p := r_cfg.get("dst_port")) else Port[p],
protocol=None if not (p := r_cfg.get("protocol")) else IPProtocol[p],
src_ip_address=r_cfg.get("src_ip"),
dst_ip_address=r_cfg.get("dst_ip"),
position=r_num,
)
# acl rules for external_inbound_acl
if cfg["acl"]["external_inbound_acl"]:
for r_num, r_cfg in cfg["acl"]["external_inbound_acl"].items():
firewall.external_inbound_acl.add_rule(
action=ACLAction[r_cfg["action"]],
src_port=None if not (p := r_cfg.get("src_port")) else Port[p],
dst_port=None if not (p := r_cfg.get("dst_port")) else Port[p],
protocol=None if not (p := r_cfg.get("protocol")) else IPProtocol[p],
src_ip_address=r_cfg.get("src_ip"),
dst_ip_address=r_cfg.get("dst_ip"),
position=r_num,
)
# acl rules for external_outbound_acl
if cfg["acl"]["external_outbound_acl"]:
for r_num, r_cfg in cfg["acl"]["external_outbound_acl"].items():
firewall.external_outbound_acl.add_rule(
action=ACLAction[r_cfg["action"]],
src_port=None if not (p := r_cfg.get("src_port")) else Port[p],
dst_port=None if not (p := r_cfg.get("dst_port")) else Port[p],
protocol=None if not (p := r_cfg.get("protocol")) else IPProtocol[p],
src_ip_address=r_cfg.get("src_ip"),
dst_ip_address=r_cfg.get("dst_ip"),
position=r_num,
)
if "routes" in cfg:
for route in cfg.get("routes"):
firewall.route_table.add_route(
address=IPv4Address(route.get("address")),
subnet_mask=IPv4Address(route.get("subnet_mask", "255.255.255.0")),
next_hop_ip_address=IPv4Address(route.get("next_hop_ip_address")),
metric=float(route.get("metric", 0)),
)
return firewall

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

@@ -998,7 +998,7 @@ class RouterInterface(IPWiredNetworkInterface):
: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}"
return f"Port {self.port_name if self.port_name else self.port_num}: {self.mac_address}/{self.ip_address}"
class Router(NetworkNode):
@@ -1398,21 +1398,23 @@ class Router(NetworkNode):
:return: Configured router.
:rtype: Router
"""
new = Router(
router = Router(
hostname=cfg["hostname"],
num_ports=cfg.get("num_ports"),
operating_state=NodeOperatingState.ON,
num_ports=int(cfg.get("num_ports", "5")),
operating_state=NodeOperatingState.ON
if not (p := cfg.get("operating_state"))
else NodeOperatingState[p.upper()],
)
if "ports" in cfg:
for port_num, port_cfg in cfg["ports"].items():
new.configure_port(
router.configure_port(
port=port_num,
ip_address=port_cfg["ip_address"],
subnet_mask=port_cfg["subnet_mask"],
subnet_mask=IPv4Address(port_cfg.get("subnet_mask", "255.255.255.0")),
)
if "acl" in cfg:
for r_num, r_cfg in cfg["acl"].items():
new.acl.add_rule(
router.acl.add_rule(
action=ACLAction[r_cfg["action"]],
src_port=None if not (p := r_cfg.get("src_port")) else Port[p],
dst_port=None if not (p := r_cfg.get("dst_port")) else Port[p],
@@ -1421,4 +1423,12 @@ class Router(NetworkNode):
dst_ip_address=r_cfg.get("dst_ip"),
position=r_num,
)
return new
if "routes" in cfg:
for route in cfg.get("routes"):
router.route_table.add_route(
address=IPv4Address(route.get("address")),
subnet_mask=IPv4Address(route.get("subnet_mask", "255.255.255.0")),
next_hop_ip_address=IPv4Address(route.get("next_hop_ip_address")),
metric=float(route.get("metric", 0)),
)
return router

5
src/primaite/simulator/network/hardware/nodes/network/wireless_router.py
View File

@@ -80,7 +80,10 @@ class WirelessAccessPoint(IPWirelessNetworkInterface):
: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})"
return (
f"Port {self.port_name if self.port_name else self.port_num}: "
f"{self.mac_address}/{self.ip_address} ({self.frequency})"
)
class WirelessRouter(Router):

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

@@ -9,11 +9,18 @@ _LOGGER = getLogger(__name__)
class IPProtocol(Enum):
"""Enum representing transport layer protocols in IP header."""
"""
Enum representing transport layer protocols in IP header.
.. _List of IPProtocols:
"""
TCP = "tcp"
"""Transmission Control Protocol."""
UDP = "udp"
"""User Datagram Protocol."""
ICMP = "icmp"
"""Internet Control Message Protocol."""
class Precedence(Enum):

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

@@ -5,7 +5,11 @@ from pydantic import BaseModel
class Port(Enum):
"""Enumeration of common known TCP/UDP ports used by protocols for operation of network applications."""
"""
Enumeration of common known TCP/UDP ports used by protocols for operation of network applications.
.. _List of Ports:
"""
NONE = 0
"Place holder for a non-port."

19
src/primaite/simulator/system/core/packet_capture.py
View File

@@ -21,7 +21,13 @@ class PacketCapture:
The PCAPs are logged to: <simulation output directory>/<hostname>/<hostname>_<ip address>_pcap.log
"""
def __init__(self, hostname: str, ip_address: Optional[str] = None, interface_num: Optional[int] = None):
def __init__(
self,
hostname: str,
ip_address: Optional[str] = None,
port_num: Optional[int] = None,
port_name: Optional[str] = None,
):
"""
Initialize the PacketCapture process.
@@ -32,9 +38,12 @@ class PacketCapture:
"The hostname for which PCAP logs are being recorded."
self.ip_address: str = ip_address
"The IP address associated with the PCAP logs."
self.interface_num = interface_num
self.port_num = port_num
"The interface num on the Node."
self.port_name = port_name
"The interface name on the Node."
self.inbound_logger = None
self.outbound_logger = None
@@ -79,10 +88,12 @@ class PacketCapture:
def _get_logger_name(self, outbound: bool = False) -> str:
"""Get PCAP the logger name."""
if self.port_name:
return f"{self.hostname}_{self.port_name}_{'outbound' if outbound else 'inbound'}_pcap"
if self.ip_address:
return f"{self.hostname}_{self.ip_address}_{'outbound' if outbound else 'inbound'}_pcap"
if self.interface_num:
return f"{self.hostname}_port-{self.interface_num}_{'outbound' if outbound else 'inbound'}_pcap"
if self.port_num:
return f"{self.hostname}_port-{self.port_num}_{'outbound' if outbound else 'inbound'}_pcap"
return f"{self.hostname}_{'outbound' if outbound else 'inbound'}_pcap"
def _get_log_path(self, outbound: bool = False) -> Path:

1
src/primaite/simulator/system/services/database/database_service.py
View File

@@ -23,6 +23,7 @@ class DatabaseService(Service):
"""
password: Optional[str] = None
"""Password that needs to be provided by clients if they want to connect to the DatabaseService."""
backup_server_ip: IPv4Address = None
"""IP address of the backup server."""