#2887 - Merge in changes on dev to resolve conflicts. All tests should now pass

This commit is contained in:
Charlie Crane
2025-02-03 11:18:34 +00:00
parent d1e932a57c
commit 3d01f52eea
10 changed files with 36 additions and 29 deletions

View File

@@ -1,6 +1,5 @@
# © Crown-owned copyright 2025, Defence Science and Technology Laboratory UK
"""PrimAITE game - Encapsulates the simulation and agents."""
from ipaddress import IPv4Address
from typing import Dict, List, Optional, Union
import numpy as np

View File

@@ -1493,6 +1493,7 @@ class Node(SimComponent, ABC):
:param hostname: The node hostname on the network.
:param operating_state: The node operating state, either ON or OFF.
"""
operating_state: NodeOperatingState = NodeOperatingState.OFF
"The hardware state of the node."
network_interfaces: Dict[str, NetworkInterface] = {}
@@ -1564,7 +1565,7 @@ class Node(SimComponent, ABC):
"Time steps until reveal to red scan is complete."
dns_server: Optional[IPv4Address] = None
"List of IP addresses of DNS servers used for name resolution."
"List of IP addresses of DNS servers used for name resolution."
default_gateway: Optional[IPV4Address] = None
"The default gateway IP address for forwarding network traffic to other networks."
@@ -1573,6 +1574,7 @@ class Node(SimComponent, ABC):
@property
def dns_server(self) -> Optional[IPv4Address]:
"""Convenience method to access the dns_server IP."""
return self.config.dns_server
@classmethod
@@ -1625,7 +1627,9 @@ class Node(SimComponent, ABC):
dns_server=kwargs["config"].dns_server,
)
super().__init__(**kwargs)
self.operating_state = NodeOperatingState.ON if not (p := kwargs["config"].operating_state) else NodeOperatingState[p.upper()]
self.operating_state = (
NodeOperatingState.ON if not (p := kwargs["config"].operating_state) else NodeOperatingState[p.upper()]
)
self._install_system_software()
self.session_manager.node = self
self.session_manager.software_manager = self.software_manager

View File

@@ -37,7 +37,7 @@ class Computer(HostNode, identifier="computer"):
SYSTEM_SOFTWARE: ClassVar[Dict] = {**HostNode.SYSTEM_SOFTWARE, "FTPClient": FTPClient}
config: "Computer.ConfigSchema" = Field(default_factory=lambda: Computer.ConfigSchema())
config: "Computer.ConfigSchema" = Field(default_factory=lambda: Computer.ConfigSchema())
class ConfigSchema(HostNode.ConfigSchema):
"""Configuration Schema for Computer class."""

View File

@@ -55,7 +55,10 @@ class HostARP(ARP):
:return: The NIC associated with the default gateway if it exists in the ARP cache; otherwise, None.
"""
if self.software_manager.node.config.default_gateway and self.software_manager.node.has_enabled_network_interface:
if (
self.software_manager.node.config.default_gateway
and self.software_manager.node.has_enabled_network_interface
):
return self.get_arp_cache_network_interface(self.software_manager.node.config.default_gateway)
def _get_arp_cache_mac_address(

View File

@@ -58,24 +58,28 @@ def client_server_routed() -> Network:
router_1.enable_port(2)
# Client 1
client_1 = Computer(config=dict(
hostname="client_1",
ip_address="192.168.2.2",
subnet_mask="255.255.255.0",
default_gateway="192.168.2.1",
start_up_duration=0,
))
client_1 = Computer(
config=dict(
hostname="client_1",
ip_address="192.168.2.2",
subnet_mask="255.255.255.0",
default_gateway="192.168.2.1",
start_up_duration=0,
)
)
client_1.power_on()
network.connect(endpoint_b=client_1.network_interface[1], endpoint_a=switch_2.network_interface[1])
# Server 1
server_1 = Server(config=dict(
hostname="server_1",
ip_address="192.168.1.2",
subnet_mask="255.255.255.0",
default_gateway="192.168.1.1",
start_up_duration=0,
))
server_1 = Server(
config=dict(
hostname="server_1",
ip_address="192.168.1.2",
subnet_mask="255.255.255.0",
default_gateway="192.168.1.1",
start_up_duration=0,
)
)
server_1.power_on()
network.connect(endpoint_b=server_1.network_interface[1], endpoint_a=switch_1.network_interface[1])

View File

@@ -43,7 +43,6 @@ class FTPServer(FTPServiceABC, identifier="FTPServer"):
"""Convenience method for accessing FTP server password."""
return self.config.server_password
def _process_ftp_command(self, payload: FTPPacket, session_id: Optional[str] = None, **kwargs) -> FTPPacket:
"""
Process the command in the FTP Packet.

View File

@@ -29,7 +29,6 @@ class NTPClient(Service, identifier="NTPClient"):
config: "NTPClient.ConfigSchema" = Field(default_factory=lambda: NTPClient.ConfigSchema())
time: Optional[datetime] = None
def __init__(self, **kwargs):

View File

@@ -232,7 +232,7 @@ def example_network() -> Network:
# Router 1
router_1_cfg = {"hostname": "router_1", "type": "router", "start_up_duration":0}
router_1_cfg = {"hostname": "router_1", "type": "router", "start_up_duration": 0}
# router_1 = Router(hostname="router_1", start_up_duration=0)
router_1 = Router.from_config(config=router_1_cfg)
@@ -253,7 +253,7 @@ def example_network() -> Network:
router_1.enable_port(1)
# Switch 2
switch_2_config = {"hostname": "switch_2", "type": "switch", "num_ports": 8, "start_up_duration":0}
switch_2_config = {"hostname": "switch_2", "type": "switch", "num_ports": 8, "start_up_duration": 0}
# switch_2 = Switch(hostname="switch_2", num_ports=8, start_up_duration=0)
switch_2 = Switch.from_config(config=switch_2_config)
switch_2.power_on()
@@ -387,7 +387,7 @@ def install_stuff_to_sim(sim: Simulation):
"ip_address": "10.0.1.2",
"subnet_mask": "255.255.255.0",
"default_gateway": "10.0.1.1",
"start_up_duration":0,
"start_up_duration": 0,
}
client_1: Computer = Computer.from_config(config=client_1_cfg)
client_1.power_on()

View File

@@ -3,8 +3,8 @@ from primaite.config.load import data_manipulation_config_path
from primaite.simulator.network.container import Network
from primaite.simulator.network.hardware.node_operating_state import NodeOperatingState
from primaite.simulator.network.hardware.nodes.host.computer import Computer
from primaite.simulator.network.hardware.nodes.network.wireless_router import WirelessRouter
from primaite.simulator.network.hardware.nodes.network.firewall import Firewall
from primaite.simulator.network.hardware.nodes.network.wireless_router import WirelessRouter
from tests.integration_tests.configuration_file_parsing import BASIC_CONFIG, DMZ_NETWORK, load_config

View File

@@ -8,10 +8,9 @@ from primaite.simulator.sim_container import Simulation
def test_file_observation():
sim = Simulation()
pc: Computer = Computer.from_config(config={"type":"computer",
"hostname":"beep",
"ip_address":"123.123.123.123",
"subnet_mask":"255.255.255.0"})
pc: Computer = Computer.from_config(
config={"type": "computer", "hostname": "beep", "ip_address": "123.123.123.123", "subnet_mask": "255.255.255.0"}
)
sim.network.add_node(pc)
f = pc.file_system.create_file(file_name="dog.png")