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

#2257: added common node attributes page + ability to set node operating state via config + tests

Browse Source
This commit is contained in:
Czar Echavez
2024-02-29 15:20:54 +00:00
parent eefc2739c8
commit 49a4e1fb56
18 changed files with 227 additions and 70 deletions
Download Patch File Download Diff File Expand all files Collapse all files

16
src/primaite/game/game.py
View File

@@ -234,7 +234,9 @@ class PrimaiteGame:
subnet_mask=IPv4Address(node_cfg.get("subnet_mask", "255.255.255.0")),
default_gateway=node_cfg["default_gateway"],
dns_server=node_cfg.get("dns_server", None),
operating_state=NodeOperatingState.ON,
operating_state=NodeOperatingState.ON
if not (p := node_cfg.get("operating_state"))
else NodeOperatingState[p.upper()],
)
elif n_type == "server":
new_node = Server(
@@ -243,13 +245,17 @@ class PrimaiteGame:
subnet_mask=IPv4Address(node_cfg.get("subnet_mask", "255.255.255.0")),
default_gateway=node_cfg["default_gateway"],
dns_server=node_cfg.get("dns_server", None),
operating_state=NodeOperatingState.ON,
operating_state=NodeOperatingState.ON
if not (p := node_cfg.get("operating_state"))
else NodeOperatingState[p.upper()],
)
elif n_type == "switch":
new_node = Switch(
hostname=node_cfg["hostname"],
num_ports=int(node_cfg.get("num_ports", "8")),
operating_state=NodeOperatingState.ON,
operating_state=NodeOperatingState.ON
if not (p := node_cfg.get("operating_state"))
else NodeOperatingState[p.upper()],
)
elif n_type == "router":
new_node = Router.from_config(node_cfg)
@@ -359,7 +365,9 @@ class PrimaiteGame:
new_node.shut_down_duration = 0
net.add_node(new_node)
new_node.power_on()
# run through the power on step if the node is to be turned on at the start
if new_node.operating_state == NodeOperatingState.ON:
new_node.power_on()
game.ref_map_nodes[node_ref] = new_node.uuid
# set start up and shut down duration

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

25
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
@@ -253,17 +253,6 @@ class NIC(IPWiredNetworkInterface):
return f"Port {self.port_num}: {self.mac_address}/{self.ip_address}"
SYSTEM_SOFTWARE = {
"HostARP": HostARP,
"ICMP": ICMP,
"DNSClient": DNSClient,
"FTPClient": FTPClient,
"NTPClient": NTPClient,
"WebBrowser": WebBrowser,
}
"""List of system software that is automatically installed on nodes."""
class HostNode(Node):
"""
Represents a host node in the network.
@@ -308,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] = {}
@@ -324,7 +323,7 @@ class HostNode(Node):
This method equips the host with essential network services and applications, preparing it for various
network-related tasks and operations.
"""
for _, software_class in SYSTEM_SOFTWARE.items():
for _, software_class in self.SYSTEM_SOFTWARE.items():
self.software_manager.install(software_class)
super()._install_system_software()

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

@@ -12,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
@@ -479,7 +481,12 @@ class Firewall(Router):
@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)
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"]
@@ -505,34 +512,82 @@ class Firewall(Router):
if "acl" in cfg:
# acl rules for internal_inbound_acl
if cfg["acl"]["internal_inbound_acl"]:
firewall.internal_inbound_acl.max_acl_rules
firewall.internal_inbound_acl._default_config = cfg["acl"]["internal_inbound_acl"]
firewall.internal_inbound_acl._reset_rules_to_default()
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"]:
firewall.internal_outbound_acl._default_config = cfg["acl"]["internal_outbound_acl"]
firewall.internal_outbound_acl._reset_rules_to_default()
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"]:
firewall.dmz_inbound_acl._default_config = cfg["acl"]["dmz_inbound_acl"]
firewall.dmz_inbound_acl._reset_rules_to_default()
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"]:
firewall.dmz_outbound_acl._default_config = cfg["acl"]["dmz_outbound_acl"]
firewall.dmz_outbound_acl._reset_rules_to_default()
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"]:
firewall.external_inbound_acl._default_config = cfg["acl"]["external_inbound_acl"]
firewall.external_inbound_acl._reset_rules_to_default()
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"]:
firewall.external_outbound_acl._default_config = cfg["acl"]["external_outbound_acl"]
firewall.external_outbound_acl._reset_rules_to_default()
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(

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

@@ -1401,7 +1401,9 @@ class Router(NetworkNode):
router = Router(
hostname=cfg["hostname"],
num_ports=int(cfg.get("num_ports", "5")),
operating_state=NodeOperatingState.ON,
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():