Merge remote-tracking branch 'origin/dev' into 4.0.0-dev

CHANGELOG.md

@@ -42,6 +42,23 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [3.3.0] - 2024-09-04
 
+## [3.4.0]
+
+### Added
+-   Log observation space data by episode and step.
+-   Added `show_history` method to Agents, allowing you to view actions taken by an agent per step. By default, `DONOTHING` actions are omitted.
+-  New ``NODE_SEND_LOCAL_COMMAND`` action implemented which grants agents the ability to execute commands locally. (Previously limited to remote only)
+-  Added ability to set the observation threshold for NMNE, file access and application executions
+
+### Changed
+-   ACL's are no longer applied to layer-2 traffic.
+-   Random number seed values are recorded in simulation/seed.log if the seed is set in the config file
+    or `generate_seed_value` is set to `true`.
+-   ARP .show() method will now include the port number associated with each entry.
+-   Added `services_requires_scan` and `applications_requires_scan` to agent observation space config to allow the agents to be able to see actual health states of services and applications without requiring scans (Default `True`, set to `False` to allow agents to see actual health state without scanning).
+-   Updated the `Terminal` class to provide response information when sending remote command execution.
+
+## [3.3.0] - 2024-09-04
 ### Added
 -   Random Number Generator Seeding by specifying a random number seed in the config file.
 -   Implemented Terminal service class, providing a generic terminal simulation.

docs/source/request_system.rst

@@ -4,6 +4,8 @@
 
 .. _request_system:
 
+.. _request_system:
+
 Request System
 **************
 

src/primaite/game/agent/interface.py

@@ -133,6 +133,37 @@ class AbstractAgent(BaseModel, ABC):
                 table = self.add_agent_action(item=item, table=table)
         print(table)
 
+    def add_agent_action(self, item: AgentHistoryItem, table: PrettyTable) -> PrettyTable:
+        """Update the given table with information from given AgentHistoryItem."""
+        node, application = "unknown", "unknown"
+        if (node_id := item.parameters.get("node_id")) is not None:
+            node = self.action_manager.node_names[node_id]
+        if (application_id := item.parameters.get("application_id")) is not None:
+            application = self.action_manager.application_names[node_id][application_id]
+        if (application_name := item.parameters.get("application_name")) is not None:
+            application = application_name
+        table.add_row([item.timestep, item.action, node, application, item.response.status])
+        return table
+
+    def show_history(self, ignored_actions: Optional[list] = None):
+        """
+        Print an agent action provided it's not the DONOTHING action.
+
+        :param ignored_actions: OPTIONAL: List of actions to be ignored when displaying the history.
+                                If not provided, defaults to ignore DONOTHING actions.
+        """
+        if not ignored_actions:
+            ignored_actions = ["DONOTHING"]
+        table = PrettyTable()
+        table.field_names = ["Step", "Action", "Node", "Application", "Response"]
+        print(f"Actions for '{self.agent_name}':")
+        for item in self.history:
+            if item.action in ignored_actions:
+                pass
+            else:
+                table = self.add_agent_action(item=item, table=table)
+        print(table)
+
     def update_observation(self, state: Dict) -> ObsType:
         """
         Convert a state from the simulator into an observation for the agent using the observation space.

src/primaite/game/agent/observations/observation_manager.py

@@ -230,7 +230,7 @@ class ObservationManager(BaseModel):
         return self.obs.space
 
     @classmethod
-    def from_config(cls, config: Optional[Dict]) -> "ObservationManager":
+    def from_config(cls, config: Optional[Dict], thresholds: Optional[Dict] = {}) -> "ObservationManager":
         """
         Create observation space from a config.
 
@@ -241,6 +241,8 @@ class ObservationManager(BaseModel):
             AbstractObservation
             options: this must adhere to the chosen observation type's ConfigSchema nested class.
         :type config: Dict
+        :param thresholds: Dictionary containing the observation thresholds.
+        :type thresholds: Optional[Dict]
         """
         if config is None:
             return cls(NullObservation())

tests/integration_tests/game_layer/observations/test_nic_observations.py

@@ -85,6 +85,14 @@ def test_nic(simulation):
     # in the NICObservation class so we set it now.
     nic_obs.capture_nmne = True
 
+    # The Simulation object created by the fixture also creates the
+    # NICObservation class with the NICObservation.capture_nmnme class variable
+    # set to False. Under normal (non-test) circumstances this class variable
+    # is set from a config file such as data_manipulation.yaml. So although
+    # capture_nmne is set to True in the NetworkInterface class it's still False
+    # in the NICObservation class so we set it now.
+    nic_obs.capture_nmne = True
+
     # Set the NMNE configuration to capture DELETE/ENCRYPT queries as MNEs
     nmne_config = {
         "capture_nmne": True,  # Enable the capture of MNEs

tests/integration_tests/game_layer/test_action_shapes.py

@@ -0,0 +1,21 @@
+# © Crown-owned copyright 2024, Defence Science and Technology Laboratory UK
+from typing import Tuple
+
+from primaite.game.agent.interface import ProxyAgent
+from primaite.game.game import PrimaiteGame
+from tests import TEST_ASSETS_ROOT
+
+FIREWALL_ACTIONS_NETWORK = TEST_ASSETS_ROOT / "configs/firewall_actions_network.yaml"
+
+
+def test_router_acl_add_rule_action_shape(game_and_agent: Tuple[PrimaiteGame, ProxyAgent]):
+    """Test to check ROUTER_ADD_ACL_RULE has the expected action shape."""
+    game, agent = game_and_agent
+
+    # assert that the shape of the actions is correct
+    router_acl_add_rule_action = agent.action_manager.actions.get("ROUTER_ACL_ADDRULE")
+    assert router_acl_add_rule_action.shape.get("source_ip_id") == len(agent.action_manager.ip_address_list)
+    assert router_acl_add_rule_action.shape.get("dest_ip_id") == len(agent.action_manager.ip_address_list)
+    assert router_acl_add_rule_action.shape.get("source_port_id") == len(agent.action_manager.ports)
+    assert router_acl_add_rule_action.shape.get("dest_port_id") == len(agent.action_manager.ports)
+    assert router_acl_add_rule_action.shape.get("protocol_id") == len(agent.action_manager.protocols)

tests/integration_tests/game_layer/test_actions.py

@@ -108,7 +108,7 @@ def test_router_acl_addrule_integration(game_and_agent: Tuple[PrimaiteGame, Prox
     """
     Test that the RouterACLAddRuleAction can form a request and that it is accepted by the simulation.
 
-    The ACL starts off with 4 rules, and we add a rule, and check that the ACL now has 5 rules.
+    The ACL starts off with 3 rules, and we add a rule, and check that the ACL now has 4 rules.
     """
     game, agent = game_and_agent
 
@@ -117,7 +117,7 @@ def test_router_acl_addrule_integration(game_and_agent: Tuple[PrimaiteGame, Prox
     server_1 = game.simulation.network.get_node_by_hostname("server_1")
     server_2 = game.simulation.network.get_node_by_hostname("server_2")
     router = game.simulation.network.get_node_by_hostname("router")
-    assert router.acl.num_rules == 4
+    assert router.acl.num_rules == 3
     assert client_1.ping("10.0.2.3")  # client_1 can ping server_2
     assert server_2.ping("10.0.1.2")  # server_2 can ping client_1
 
@@ -167,8 +167,8 @@ def test_router_acl_addrule_integration(game_and_agent: Tuple[PrimaiteGame, Prox
     agent.store_action(action)
     game.step()
 
-    # 5: Check that the ACL now has 6 rules, but that server_1 can still ping server_2
-    assert router.acl.num_rules == 6
+    # 5: Check that the ACL now has 5 rules, but that server_1 can still ping server_2
+    assert router.acl.num_rules == 5
     assert server_1.ping("10.0.2.3")  # Can ping server_2
 
 
@@ -198,8 +198,8 @@ def test_router_acl_removerule_integration(game_and_agent: Tuple[PrimaiteGame, P
     agent.store_action(action)
     game.step()
 
-    # 3: Check that the ACL now has 3 rules, and that client 1 cannot access example.com
-    assert router.acl.num_rules == 3
+    # 3: Check that the ACL now has 2 rules, and that client 1 cannot access example.com
+    assert router.acl.num_rules == 2
     assert not browser.get_webpage()
     client_1.software_manager.software.get("dns-client").dns_cache.clear()
     assert client_1.ping("10.0.2.2")  # pinging still works because ICMP is allowed