Merge remote-tracking branch 'origin/dev' into feature/Updated-How-To-Guides

tests/integration_tests/configuration_file_parsing/threat_actor_profile_settings/__init__.py

@@ -0,0 +1 @@
+# © Crown-owned copyright 2025, Defence Science and Technology Laboratory UK

tests/integration_tests/configuration_file_parsing/threat_actor_profile_settings/test_kill_chain_settings.py

@@ -0,0 +1,25 @@
+# © Crown-owned copyright 2025, Defence Science and Technology Laboratory UK
+from pathlib import Path
+from typing import Union
+
+import yaml
+
+from primaite.config.load import _EXAMPLE_CFG
+from primaite.game.agent.scripted_agents.TAP003 import TAP003
+from primaite.game.game import PrimaiteGame
+
+ATTACK_AGENT_INDEX = 32
+
+
+def test_tap003_kill_chain_settings_load_config():
+    with open(_EXAMPLE_CFG / "uc7_config_tap003.yaml", mode="r") as uc7_config:
+        cfg = yaml.safe_load(uc7_config)
+        cfg["agents"][ATTACK_AGENT_INDEX]["agent_settings"]["kill_chain"]["MANIPULATION"]["probability"] = 0.5
+        cfg["agents"][ATTACK_AGENT_INDEX]["agent_settings"]["kill_chain"]["ACCESS"]["probability"] = 0.5
+        cfg["agents"][ATTACK_AGENT_INDEX]["agent_settings"]["kill_chain"]["PLANNING"]["probability"] = 0.5
+    game = PrimaiteGame.from_config(cfg)
+    tap: TAP003 = game.agents["attacker"]
+    kill_chain = tap.config.agent_settings.kill_chain
+    assert kill_chain.MANIPULATION.probability == 0.5
+    assert kill_chain.ACCESS.probability == 0.5
+    assert kill_chain.PLANNING.probability == 0.5

tests/integration_tests/configuration_file_parsing/threat_actor_profile_settings/test_threat_actor_profile_settings.py

@@ -0,0 +1,35 @@
+# © Crown-owned copyright 2025, Defence Science and Technology Laboratory UK
+from pathlib import Path
+from typing import Union
+
+import yaml
+
+from primaite.config.load import _EXAMPLE_CFG
+from primaite.game.agent.scripted_agents.TAP003 import TAP003
+from primaite.game.game import PrimaiteGame
+
+
+def test_threat_actor_profile_load_config():
+    """Test to check that threat actor profiles are able to be loaded."""
+    with open(_EXAMPLE_CFG / "uc7_config_tap003.yaml", mode="r") as uc7_config:
+        cfg = yaml.safe_load(uc7_config)
+
+    game = PrimaiteGame.from_config(cfg)
+    # tap003 is found and loaded TODO: Once tuple digestion is implemented, change to hardcoded 'tap003' test.
+    assert "attacker" in game.agents
+    assert isinstance(game.agents["attacker"], TAP003)
+    agent: TAP003 = game.agents["attacker"]
+    assert agent.config.agent_settings.start_step == 1
+    assert agent.config.agent_settings.frequency == 3
+    assert agent.config.agent_settings.variance == 0
+    assert not agent.config.agent_settings.repeat_kill_chain
+    assert agent.config.agent_settings.repeat_kill_chain_stages
+    assert agent.config.agent_settings.default_starting_node == "ST_PROJ-A-PRV-PC-1"
+    assert not agent.config.agent_settings.starting_nodes
+    assert agent.config.agent_settings.kill_chain.PLANNING.probability == 1
+    assert len(agent.config.agent_settings.kill_chain.PLANNING.starting_network_knowledge["credentials"]) == 6
+    assert agent.config.agent_settings.kill_chain.ACCESS.probability == 1
+    assert agent.config.agent_settings.kill_chain.MANIPULATION.probability == 1
+    assert len(agent.config.agent_settings.kill_chain.MANIPULATION.account_changes) == 3
+    assert agent.config.agent_settings.kill_chain.EXPLOIT.probability == 1
+    assert len(agent.config.agent_settings.kill_chain.EXPLOIT.malicious_acls) == 3

tests/integration_tests/game_layer/observations/test_acl_observations.py

@@ -47,7 +47,7 @@ def test_acl_observations(simulation):
     observation_space = acl_obs.observe(simulation.describe_state())
     assert observation_space.get(1) is not None
     rule_obs = observation_space.get(1)  # this is the ACL Rule added to allow NTP
-    assert rule_obs.get("position") == 0  # rule was put at position 1 (0 because counting from 1 instead of 1)
+    assert rule_obs.get("position") == 1  # rule was put at position 1
     assert rule_obs.get("permission") == 1  # permit = 1 deny = 2
     assert rule_obs.get("source_ip_id") == 1  # applies to all source nodes
     assert rule_obs.get("dest_ip_id") == 1  # applies to all destination nodes
@@ -60,7 +60,7 @@ def test_acl_observations(simulation):
     observation_space = acl_obs.observe(simulation.describe_state())
     assert observation_space.get(1) is not None
     rule_obs = observation_space.get(1)  # this is the ACL Rule added to allow NTP
-    assert rule_obs.get("position") == 0
+    assert rule_obs.get("position") == 1
     assert rule_obs.get("permission") == 0
     assert rule_obs.get("source_ip_id") == 0
     assert rule_obs.get("dest_ip_id") == 0

tests/integration_tests/game_layer/observations/test_firewall_observation.py

@@ -11,15 +11,15 @@ from primaite.utils.validation.port import PORT_LOOKUP
 
 def check_default_rules(acl_obs):
     assert len(acl_obs) == 7
-    assert all(acl_obs[i]["position"] == i - 1 for i in range(1, 8))
-    assert all(acl_obs[i]["permission"] == 0 for i in range(1, 8))
-    assert all(acl_obs[i]["source_ip_id"] == 0 for i in range(1, 8))
-    assert all(acl_obs[i]["source_wildcard_id"] == 0 for i in range(1, 8))
-    assert all(acl_obs[i]["source_port_id"] == 0 for i in range(1, 8))
-    assert all(acl_obs[i]["dest_ip_id"] == 0 for i in range(1, 8))
-    assert all(acl_obs[i]["dest_wildcard_id"] == 0 for i in range(1, 8))
-    assert all(acl_obs[i]["dest_port_id"] == 0 for i in range(1, 8))
-    assert all(acl_obs[i]["protocol_id"] == 0 for i in range(1, 8))
+    assert all(acl_obs[i]["position"] == i for i in range(7))
+    assert all(acl_obs[i]["permission"] == 0 for i in range(7))
+    assert all(acl_obs[i]["source_ip_id"] == 0 for i in range(7))
+    assert all(acl_obs[i]["source_wildcard_id"] == 0 for i in range(7))
+    assert all(acl_obs[i]["source_port_id"] == 0 for i in range(7))
+    assert all(acl_obs[i]["dest_ip_id"] == 0 for i in range(7))
+    assert all(acl_obs[i]["dest_wildcard_id"] == 0 for i in range(7))
+    assert all(acl_obs[i]["dest_port_id"] == 0 for i in range(7))
+    assert all(acl_obs[i]["protocol_id"] == 0 for i in range(7))
 
 
 def test_firewall_observation():
@@ -75,7 +75,7 @@ def test_firewall_observation():
 
     observation = firewall_observation.observe(firewall.describe_state())
     observed_rule = observation["ACL"]["INTERNAL"]["INBOUND"][5]
-    assert observed_rule["position"] == 4
+    assert observed_rule["position"] == 5
     assert observed_rule["permission"] == 2
     assert observed_rule["source_ip_id"] == 2
     assert observed_rule["source_wildcard_id"] == 3

tests/integration_tests/game_layer/observations/test_router_observation.py

@@ -53,7 +53,7 @@ def test_router_observation():
     # Observe the state using the RouterObservation instance
     observed_output = router_observation.observe(router.describe_state())
     observed_rule = observed_output["ACL"][5]
-    assert observed_rule["position"] == 4
+    assert observed_rule["position"] == 5
     assert observed_rule["permission"] == 2
     assert observed_rule["source_ip_id"] == 2
     assert observed_rule["source_wildcard_id"] == 3
@@ -77,7 +77,7 @@ def test_router_observation():
     )
     observed_output = router_observation.observe(router.describe_state())
     observed_rule = observed_output["ACL"][2]
-    assert observed_rule["position"] == 1
+    assert observed_rule["position"] == 2
     assert observed_rule["permission"] == 1
     assert observed_rule["source_ip_id"] == 1
     assert observed_rule["source_wildcard_id"] == 1