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

#2925 Fixed TAP003 error and failing tests.

Browse Source
This commit is contained in:
Archer Bowen
2025-02-28 14:27:07 +00:00
committed by Marek Wolan
parent 908dbe4722
commit f532b5e0da
11 changed files with 67 additions and 56 deletions
Download Patch File Download Diff File Expand all files Collapse all files

21
src/primaite/game/agent/scripted_agents/TAP003.py
View File

@@ -257,7 +257,7 @@ class TAP003(AbstractTAP, discriminator="tap-003"):
self.logger.debug(f"Updating network knowledge. Changed {username}'s password to {password} on {hostname}.")
self._change_password_target_host = ""
# local password change
elif last_hist_item.action == "node-accounts-change-password" and last_hist_item.response.status == "success":
elif last_hist_item.action == "node-account-change-password" and last_hist_item.response.status == "success":
self.network_knowledge["current_session"] = {}
username = last_hist_item.request[6]
password = last_hist_item.request[8]
@@ -338,15 +338,17 @@ class TAP003(AbstractTAP, discriminator="tap-003"):
"""
if self.current_kill_chain_stage == self.selected_kill_chain.MANIPULATION:
if self._agent_trial_handler(self.config.agent_settings.kill_chain.MANIPULATION.probability):
self.logger.info(f"TAP003 reached the {self.current_kill_chain_stage.name}")
if self.current_stage_progress == KillChainStageProgress.PENDING:
self.logger.info(f"TAP003 reached the {self.current_kill_chain_stage.name}.")
self.current_stage_progress = KillChainStageProgress.IN_PROGRESS
self.current_host = self.starting_node
account_changes = self.config.agent_settings.kill_chain.MANIPULATION.account_changes
if len(account_changes) > 0:
if len(account_changes) > 0 or self._next_account_change:
if not self._next_account_change:
self._next_account_change = account_changes.pop(0)
if self._next_account_change["host"] == self.current_host:
# do a local password change
self.chosen_action = "node-accounts-change-password", {
self.chosen_action = "node-account-change-password", {
"node_name": self.current_host,
"username": self._next_account_change["username"],
"current_password": self.network_knowledge["credentials"][self.current_host]["password"],
@@ -382,14 +384,15 @@ class TAP003(AbstractTAP, discriminator="tap-003"):
],
}
self.logger.info(f"Changing password on remote node {hostname}")
try:
self._next_account_change = account_changes.pop(0)
self._change_password_target_host = hostname
if len(account_changes) == 0:
except IndexError:
self.logger.info("No further account changes required.")
self._next_account_change = None
self.logger.info("Finished changing passwords.")
self._change_password_target_host = hostname
if not self._next_account_change:
self.logger.info("Manipulation complete. Progressing to exploit...")
self._progress_kill_chain()
self.current_stage_progress = KillChainStageProgress.PENDING
else:
if self.config.agent_settings.repeat_kill_chain_stages == False:
self.current_kill_chain_stage = self.selected_kill_chain.FAILED

1
src/primaite/notebooks/UC7-TAP001-Kill-Chain-E2E.ipynb
View File

@@ -38,6 +38,7 @@
"# Importing the necessary PrimAITE libraries\n",
"from primaite.session.environment import PrimaiteGymEnv\n",
"import yaml\n",
"from primaite.game.agent.scripted_agents.TAP001 import TAP001, MobileMalwareKillChain\n",
"from primaite.config.load import load, _EXAMPLE_CFG\n",
"from deepdiff.diff import DeepDiff"
]

3
src/primaite/notebooks/UC7-TAP003-Kill-Chain-E2E.ipynb
View File

@@ -50,6 +50,7 @@
"from primaite.simulator.network.hardware.nodes.host.computer import Computer\n",
"from primaite.simulator.system.applications.web_browser import WebBrowser\n",
"from primaite.simulator.network.hardware.nodes.network.router import Router\n",
"from primaite.game.agent.scripted_agents.TAP003 import TAP003, InsiderKillChain, InsiderKillChainOptions\n",
"from primaite.config.load import load, _EXAMPLE_CFG\n",
"from pprint import pprint\n",
"from deepdiff.diff import DeepDiff\n",
@@ -1503,7 +1504,7 @@
"metadata": {},
"outputs": [],
"source": [
"account_changes = [{\"host\":\"ST_PROJ-A-PRV-PC-2\", \"ip_address\": \"192.168.230.2\", \"user_name\": user_username, \"old_password\": user_password, \"new_password\": tap003_new_password}]"
"account_changes = [{\"host\":\"ST_PROJ-A-PRV-PC-2\", \"ip_address\": \"192.168.230.2\", \"username\": user_username, \"old_password\": user_password, \"new_password\": tap003_new_password}]"
]
},
{

47
tests/e2e_integration_tests/threat_actor_profiles/test_abstract_tap.py
View File

@@ -10,22 +10,22 @@ from primaite.game.agent.scripted_agents.abstract_tap import (
KillChainStageOptions,
KillChainStageProgress,
)
from primaite.game.agent.scripted_agents.TAP001 import MobileMalwareKillChain
from primaite.game.agent.scripted_agents.TAP003 import InsiderKillChain
from primaite.game.agent.scripted_agents.TAP001 import MobileMalwareKillChain, TAP001
from primaite.game.agent.scripted_agents.TAP003 import InsiderKillChain, TAP003
from primaite.session.environment import PrimaiteGymEnv
START_STEP = 1 # The starting step of the agent.
FREQUENCY = 5 # The frequency of kill chain stage progression (E.g it's next attempt at "attacking").
VARIANCE = 0 # The timestep variance between kill chain progression (E.g Next timestep = Frequency +/- variance)
ATTACK_AGENT_INDEX = 32
def uc7_tap001_env() -> PrimaiteGymEnv:
with open(_EXAMPLE_CFG / "uc7_config.yaml", mode="r") as uc7_config:
cfg = yaml.safe_load(uc7_config)
for agents in cfg["agents"]:
if agents["ref"] == "attacker":
tap_cfg = agents
tap_cfg["agent_settings"]["start_step"] = 1
tap_cfg["agent_settings"]["frequency"] = 5
tap_cfg["agent_settings"]["variance"] = 0
cfg["agents"][ATTACK_AGENT_INDEX]["agent_settings"]["start_step"] = START_STEP
cfg["agents"][ATTACK_AGENT_INDEX]["agent_settings"]["frequency"] = FREQUENCY
cfg["agents"][ATTACK_AGENT_INDEX]["agent_settings"]["variance"] = VARIANCE
env = PrimaiteGymEnv(env_config=cfg)
@@ -42,27 +42,26 @@ def uc7_tap003_env(**kwargs) -> PrimaiteGymEnv:
"""
with open(_EXAMPLE_CFG / "uc7_config_tap003.yaml", "r") as uc7_config:
cfg = yaml.safe_load(uc7_config)
for agents in cfg["agents"]:
if agents["ref"] == "attacker":
tap_cfg = agents
tap_cfg["agent_settings"]["start_step"] = 1
tap_cfg["agent_settings"]["frequency"] = 5
tap_cfg["agent_settings"]["variance"] = 0
cfg["agents"][ATTACK_AGENT_INDEX]["agent_settings"]["start_step"] = START_STEP
cfg["agents"][ATTACK_AGENT_INDEX]["agent_settings"]["frequency"] = FREQUENCY
cfg["agents"][ATTACK_AGENT_INDEX]["agent_settings"]["variance"] = VARIANCE
if "repeat_kill_chain" in kwargs:
tap_cfg["agent_settings"]["repeat_kill_chain"] = kwargs["repeat_kill_chain"]
cfg["agents"][ATTACK_AGENT_INDEX]["agent_settings"]["repeat_kill_chain"] = kwargs["repeat_kill_chain"]
if "repeat_kill_chain_stages" in kwargs:
tap_cfg["agent_settings"]["repeat_kill_chain_stages"] = kwargs["repeat_kill_chain_stages"]
cfg["agents"][ATTACK_AGENT_INDEX]["agent_settings"]["repeat_kill_chain_stages"] = kwargs[
"repeat_kill_chain_stages"
]
if "planning_probability" in kwargs:
tap_cfg["agent_settings"]["kill_chain"]["PLANNING"]["probability"] = kwargs["planning_probability"]
cfg["agents"][ATTACK_AGENT_INDEX]["agent_settings"]["kill_chain"]["PLANNING"]["probability"] = kwargs[
"planning_probability"
]
if "custom_kill_chain" in kwargs:
tap_cfg["agent_settings"]["kill_chain"] = kwargs["custom_kill_chain"]
cfg["agents"][ATTACK_AGENT_INDEX]["agent_settings"]["kill_chain"] = kwargs["custom_kill_chain"]
if "starting_nodes" in kwargs:
tap_cfg["agent_settings"]["starting_nodes"] = kwargs["starting_nodes"]
cfg["agents"][ATTACK_AGENT_INDEX]["agent_settings"]["starting_nodes"] = kwargs["starting_nodes"]
if "target_nodes" in kwargs:
tap_cfg["agent_settings"]["target_nodes"] = kwargs["target_nodes"]
cfg["agents"][ATTACK_AGENT_INDEX]["agent_settings"]["target_nodes"] = kwargs["target_nodes"]
env = PrimaiteGymEnv(env_config=cfg)
return env

4
tests/e2e_integration_tests/threat_actor_profiles/test_kill_chain_methods.py
View File

@@ -10,8 +10,8 @@ from primaite.game.agent.scripted_agents.abstract_tap import (
KillChainStageOptions,
KillChainStageProgress,
)
from primaite.game.agent.scripted_agents.TAP001 import MobileMalwareKillChain
from primaite.game.agent.scripted_agents.TAP003 import InsiderKillChain
from primaite.game.agent.scripted_agents.TAP001 import MobileMalwareKillChain, TAP001
from primaite.game.agent.scripted_agents.TAP003 import InsiderKillChain, TAP003
from primaite.session.environment import PrimaiteGymEnv
# Defining constants.

4
tests/e2e_integration_tests/threat_actor_profiles/test_tap001_kill_chain_repeat.py
View File

@@ -10,8 +10,8 @@ from primaite.game.agent.scripted_agents.abstract_tap import (
KillChainStageOptions,
KillChainStageProgress,
)
from primaite.game.agent.scripted_agents.TAP001 import MobileMalwareKillChain
from primaite.game.agent.scripted_agents.TAP003 import InsiderKillChain
from primaite.game.agent.scripted_agents.TAP001 import MobileMalwareKillChain, TAP001
from primaite.game.agent.scripted_agents.TAP003 import InsiderKillChain, TAP003
from primaite.session.environment import PrimaiteGymEnv
# Defining constants.

4
tests/e2e_integration_tests/threat_actor_profiles/test_tap001_kill_chain_stages.py
View File

@@ -10,8 +10,8 @@ from primaite.game.agent.scripted_agents.abstract_tap import (
KillChainStageOptions,
KillChainStageProgress,
)
from primaite.game.agent.scripted_agents.TAP001 import MobileMalwareKillChain
from primaite.game.agent.scripted_agents.TAP003 import InsiderKillChain
from primaite.game.agent.scripted_agents.TAP001 import MobileMalwareKillChain, TAP001
from primaite.game.agent.scripted_agents.TAP003 import InsiderKillChain, TAP003
from primaite.session.environment import PrimaiteGymEnv
from primaite.simulator.file_system.file_system_item_abc import FileSystemItemHealthStatus
from primaite.simulator.system.applications.red_applications.c2.c2_beacon import C2Beacon

4
tests/e2e_integration_tests/threat_actor_profiles/test_tap001_propagate_stage.py
View File

@@ -10,8 +10,8 @@ from primaite.game.agent.scripted_agents.abstract_tap import (
KillChainStageOptions,
KillChainStageProgress,
)
from primaite.game.agent.scripted_agents.TAP001 import MobileMalwareKillChain
from primaite.game.agent.scripted_agents.TAP003 import InsiderKillChain
from primaite.game.agent.scripted_agents.TAP001 import MobileMalwareKillChain, TAP001
from primaite.game.agent.scripted_agents.TAP003 import InsiderKillChain, TAP003
from primaite.session.environment import PrimaiteGymEnv
from primaite.simulator.file_system.file_system_item_abc import FileSystemItemHealthStatus
from primaite.simulator.system.applications.red_applications.c2.c2_beacon import C2Beacon

4
tests/e2e_integration_tests/threat_actor_profiles/test_tap003_kill_chain_repeat.py
View File

@@ -10,8 +10,8 @@ from primaite.game.agent.scripted_agents.abstract_tap import (
KillChainStageOptions,
KillChainStageProgress,
)
from primaite.game.agent.scripted_agents.TAP001 import MobileMalwareKillChain
from primaite.game.agent.scripted_agents.TAP003 import InsiderKillChain
from primaite.game.agent.scripted_agents.TAP001 import MobileMalwareKillChain, TAP001
from primaite.game.agent.scripted_agents.TAP003 import InsiderKillChain, TAP003
from primaite.session.environment import PrimaiteGymEnv
# Defining constants.

25
tests/e2e_integration_tests/threat_actor_profiles/test_tap003_kill_chain_stages.py
View File

@@ -10,10 +10,10 @@ from primaite.game.agent.scripted_agents.abstract_tap import (
KillChainStageOptions,
KillChainStageProgress,
)
from primaite.game.agent.scripted_agents.TAP001 import MobileMalwareKillChain
from primaite.game.agent.scripted_agents.TAP003 import InsiderKillChain
from primaite.game.agent.scripted_agents.TAP003 import InsiderKillChain, TAP003
from primaite.session.environment import PrimaiteGymEnv
from primaite.simulator.network.hardware.nodes.network.router import ACLAction
from primaite.simulator.network.hardware.nodes.network.firewall import Firewall
from primaite.simulator.network.hardware.nodes.network.router import ACLAction, Router
# Defining constants.
@@ -31,6 +31,7 @@ def uc7_tap003_env() -> PrimaiteGymEnv:
with open(_EXAMPLE_CFG / "uc7_config_tap003.yaml", mode="r") as uc7_config:
cfg = yaml.safe_load(uc7_config)
cfg["io_settings"]["save_sys_logs"] = False
cfg["io_settings"]["save_agent_logs"] = True
cfg["agents"][ATTACK_AGENT_INDEX]["agent_settings"]["start_step"] = START_STEP
cfg["agents"][ATTACK_AGENT_INDEX]["agent_settings"]["frequency"] = FREQUENCY
cfg["agents"][ATTACK_AGENT_INDEX]["agent_settings"]["variance"] = VARIANCE
@@ -94,8 +95,13 @@ def test_tap003_kill_chain_stage_planning():
env = environment_step(i=2, env=env)
# Testing that the stage successfully impacted the simulation - User is logged in
# TODO: Add an assert for this.
# Testing that the stage successful - TAP003 has loaded it's starting network knowledge into it's network knowledge.
# At this point TAP003 will parse it's starting network knowledge config into it's a private attribute (`network_knowledge`)
assert (
tap003.network_knowledge["credentials"]
== tap003.config.agent_settings.kill_chain.PLANNING.starting_network_knowledge["credentials"]
)
def test_tap003_kill_chain_stage_access():
@@ -123,9 +129,7 @@ def test_tap003_kill_chain_stage_access():
def test_tap003_kill_chain_stage_manipulation():
"""Tests the successful/failed handlers in the manipulation stage in the InsiderKillChain"""
env = uc7_tap003_env()
env.reset()
tap003: TAP003 = env.game.agents["attacker"]
assert tap003.current_kill_chain_stage == BaseKillChain.NOT_STARTED
@@ -152,10 +156,12 @@ def test_tap003_kill_chain_stage_manipulation():
# Testing that the stage successfully impacted the simulation - Accounts Altered
env = environment_step(i=5, env=env)
st_intra_prv_rt_dr_1: Router = env.game.simulation.network.get_node_by_hostname("ST_INTRA-PRV-RT-DR-1")
assert tap003.current_kill_chain_stage.name == InsiderKillChain.MANIPULATION.name
assert st_intra_prv_rt_dr_1.user_manager.admins["admin"].password == "red_pass"
env = environment_step(i=5, env=env)
st_intra_prv_rt_cr: Router = env.game.simulation.network.get_node_by_hostname("ST_INTRA-PRV-RT-CR")
assert tap003.current_kill_chain_stage.name == InsiderKillChain.MANIPULATION.name
assert st_intra_prv_rt_cr.user_manager.admins["admin"].password == "red_pass"
env = environment_step(i=5, env=env)
@@ -172,6 +178,7 @@ def test_tap003_kill_chain_stage_exploit():
st_intra_prv_rt_dr_1: Router = env.game.simulation.network.get_node_by_hostname("ST_INTRA-PRV-RT-DR-1")
st_intra_prv_rt_cr: Router = env.game.simulation.network.get_node_by_hostname("ST_INTRA-PRV-RT-CR")
rem_pub_rt_dr: Router = env.game.simulation.network.get_node_by_hostname("REM-PUB-RT-DR")
assert tap003.current_kill_chain_stage == BaseKillChain.NOT_STARTED
env = environment_step(i=2, env=env)
@@ -189,12 +196,12 @@ def test_tap003_kill_chain_stage_exploit():
assert tap003.current_kill_chain_stage.name == InsiderKillChain.ACCESS.name
assert tap003.next_kill_chain_stage.name == InsiderKillChain.MANIPULATION.name
env = environment_step(i=9, env=env)
env = environment_step(i=16, env=env)
assert tap003.current_kill_chain_stage.name == InsiderKillChain.EXPLOIT.name
# Testing that the stage successfully impacted the simulation - Malicious ACL Added:
for _ in range(ATTACK_AGENT_INDEX):
for _ in range(14):
env.step(0)
# Tests that the ACL has been added and that the action is deny.

4
tests/e2e_integration_tests/threat_actor_profiles/test_tap003_multiple_rules.py
View File

@@ -12,8 +12,8 @@ from primaite.game.agent.scripted_agents.abstract_tap import (
KillChainStageOptions,
KillChainStageProgress,
)
from primaite.game.agent.scripted_agents.TAP001 import MobileMalwareKillChain
from primaite.game.agent.scripted_agents.TAP003 import InsiderKillChain
from primaite.game.agent.scripted_agents.TAP001 import MobileMalwareKillChain, TAP001
from primaite.game.agent.scripted_agents.TAP003 import InsiderKillChain, TAP003
from primaite.session.environment import PrimaiteGymEnv
from primaite.simulator.network.hardware.nodes.network.router import ACLAction, Router
from primaite.utils.validation.ip_protocol import PROTOCOL_LOOKUP