Merge remote-tracking branch 'origin/dev' into feature/2840_LOCAL_TERMINAL_COMMAND

tests/integration_tests/game_layer/actions/test_user_account_actions.py

@@ -0,0 +1,176 @@
+# © Crown-owned copyright 2024, Defence Science and Technology Laboratory UK
+import pytest
+
+from primaite.simulator.network.hardware.nodes.host.computer import Computer
+from primaite.simulator.network.hardware.nodes.network.router import ACLAction
+from primaite.simulator.network.transmission.transport_layer import Port
+
+
+@pytest.fixture
+def game_and_agent_fixture(game_and_agent):
+    """Create a game with a simple agent that can be controlled by the tests."""
+    game, agent = game_and_agent
+
+    client_1: Computer = game.simulation.network.get_node_by_hostname("client_1")
+    client_1.start_up_duration = 3
+
+    return (game, agent)
+
+
+def test_user_account_add_user_action(game_and_agent_fixture):
+    """Tests the add user account action."""
+    game, agent = game_and_agent_fixture
+    client_1 = game.simulation.network.get_node_by_hostname("client_1")
+
+    assert len(client_1.user_manager.users) == 1  # admin is created by default
+    assert len(client_1.user_manager.admins) == 1
+
+    # add admin account
+    action = (
+        "NODE_ACCOUNTS_ADD_USER",
+        {"node_id": 0, "username": "admin_2", "password": "e-tronic-boogaloo", "is_admin": True},
+    )
+    agent.store_action(action)
+    game.step()
+
+    assert len(client_1.user_manager.users) == 2  # new user added
+    assert len(client_1.user_manager.admins) == 2
+
+    # add non admin account
+    action = (
+        "NODE_ACCOUNTS_ADD_USER",
+        {"node_id": 0, "username": "leeroy.jenkins", "password": "no_plan_needed", "is_admin": False},
+    )
+    agent.store_action(action)
+    game.step()
+
+    assert len(client_1.user_manager.users) == 3  # new user added
+    assert len(client_1.user_manager.admins) == 2
+
+
+def test_user_account_disable_user_action(game_and_agent_fixture):
+    """Tests the disable user account action."""
+    game, agent = game_and_agent_fixture
+    client_1 = game.simulation.network.get_node_by_hostname("client_1")
+
+    client_1.user_manager.add_user(username="test", password="password", is_admin=True)
+    assert len(client_1.user_manager.users) == 2  # new user added
+    assert len(client_1.user_manager.admins) == 2
+
+    test_user = client_1.user_manager.users.get("test")
+    assert test_user
+    assert test_user.disabled is not True
+
+    # disable test account
+    action = (
+        "NODE_ACCOUNTS_DISABLE_USER",
+        {
+            "node_id": 0,
+            "username": "test",
+        },
+    )
+    agent.store_action(action)
+    game.step()
+    assert test_user.disabled
+
+
+def test_user_account_change_password_action(game_and_agent_fixture):
+    """Tests the change password user account action."""
+    game, agent = game_and_agent_fixture
+    client_1 = game.simulation.network.get_node_by_hostname("client_1")
+
+    client_1.user_manager.add_user(username="test", password="password", is_admin=True)
+
+    test_user = client_1.user_manager.users.get("test")
+    assert test_user.password == "password"
+
+    # change account password
+    action = (
+        "NODE_ACCOUNTS_CHANGE_PASSWORD",
+        {"node_id": 0, "username": "test", "current_password": "password", "new_password": "2Hard_2_Hack"},
+    )
+    agent.store_action(action)
+    game.step()
+
+    assert test_user.password == "2Hard_2_Hack"
+
+
+def test_user_account_create_terminal_action(game_and_agent_fixture):
+    """Tests that agents can use the terminal to create new users."""
+    game, agent = game_and_agent_fixture
+
+    router = game.simulation.network.get_node_by_hostname("router")
+    router.acl.add_rule(action=ACLAction.PERMIT, src_port=Port.SSH, dst_port=Port.SSH, position=4)
+
+    server_1 = game.simulation.network.get_node_by_hostname("server_1")
+    server_1_usm = server_1.software_manager.software["UserManager"]
+    server_1_usm.add_user("user123", "password", is_admin=True)
+
+    action = (
+        "SSH_TO_REMOTE",
+        {
+            "node_id": 0,
+            "username": "user123",
+            "password": "password",
+            "remote_ip": str(server_1.network_interface[1].ip_address),
+        },
+    )
+    agent.store_action(action)
+    game.step()
+    assert agent.history[-1].response.status == "success"
+
+    # Create a new user account via terminal.
+    action = (
+        "NODE_SEND_REMOTE_COMMAND",
+        {
+            "node_id": 0,
+            "remote_ip": str(server_1.network_interface[1].ip_address),
+            "command": ["service", "UserManager", "add_user", "new_user", "new_pass", True],
+        },
+    )
+    agent.store_action(action)
+    game.step()
+    new_user = server_1.user_manager.users.get("new_user")
+    assert new_user
+    assert new_user.password == "new_pass"
+    assert new_user.disabled is not True
+
+
+def test_user_account_disable_terminal_action(game_and_agent_fixture):
+    """Tests that agents can use the terminal to disable users."""
+    game, agent = game_and_agent_fixture
+    router = game.simulation.network.get_node_by_hostname("router")
+    router.acl.add_rule(action=ACLAction.PERMIT, src_port=Port.SSH, dst_port=Port.SSH, position=4)
+
+    server_1 = game.simulation.network.get_node_by_hostname("server_1")
+    server_1_usm = server_1.software_manager.software["UserManager"]
+    server_1_usm.add_user("user123", "password", is_admin=True)
+
+    action = (
+        "SSH_TO_REMOTE",
+        {
+            "node_id": 0,
+            "username": "user123",
+            "password": "password",
+            "remote_ip": str(server_1.network_interface[1].ip_address),
+        },
+    )
+    agent.store_action(action)
+    game.step()
+    assert agent.history[-1].response.status == "success"
+
+    # Disable a user via terminal
+    action = (
+        "NODE_SEND_REMOTE_COMMAND",
+        {
+            "node_id": 0,
+            "remote_ip": str(server_1.network_interface[1].ip_address),
+            "command": ["service", "UserManager", "disable_user", "user123"],
+        },
+    )
+    agent.store_action(action)
+    game.step()
+
+    new_user = server_1.user_manager.users.get("user123")
+    assert new_user
+    assert new_user.disabled is True

tests/integration_tests/game_layer/observations/test_nic_observations.py

@@ -77,6 +77,14 @@ def test_nic(simulation):
 
     nic_obs = NICObservation(where=["network", "nodes", pc.hostname, "NICs", 1], include_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/observations/test_node_observations.py

@@ -39,6 +39,8 @@ def test_host_observation(simulation):
         folders=[],
         network_interfaces=[],
         file_system_requires_scan=True,
+        services_requires_scan=True,
+        applications_requires_scan=True,
         include_users=False,
     )
 

tests/integration_tests/game_layer/observations/test_software_observations.py

@@ -29,7 +29,9 @@ def test_service_observation(simulation):
     ntp_server = pc.software_manager.software.get("NTPServer")
     assert ntp_server
 
-    service_obs = ServiceObservation(where=["network", "nodes", pc.hostname, "services", "NTPServer"])
+    service_obs = ServiceObservation(
+        where=["network", "nodes", pc.hostname, "services", "NTPServer"], services_requires_scan=True
+    )
 
     assert service_obs.space["operating_status"] == spaces.Discrete(7)
     assert service_obs.space["health_status"] == spaces.Discrete(5)
@@ -54,7 +56,9 @@ def test_application_observation(simulation):
     web_browser: WebBrowser = pc.software_manager.software.get("WebBrowser")
     assert web_browser
 
-    app_obs = ApplicationObservation(where=["network", "nodes", pc.hostname, "applications", "WebBrowser"])
+    app_obs = ApplicationObservation(
+        where=["network", "nodes", pc.hostname, "applications", "WebBrowser"], applications_requires_scan=True
+    )
 
     web_browser.close()
     observation_state = app_obs.observe(simulation.describe_state())

tests/integration_tests/game_layer/test_RNG_seed.py

@@ -7,6 +7,7 @@ import yaml
 from primaite.config.load import data_manipulation_config_path
 from primaite.game.agent.interface import AgentHistoryItem
 from primaite.session.environment import PrimaiteGymEnv
+from primaite.simulator import SIM_OUTPUT
 
 
 @pytest.fixture()
@@ -33,6 +34,11 @@ def test_rng_seed_set(create_env):
 
     assert a == b
 
+    # Check that seed log file was created.
+    path = SIM_OUTPUT.path / "seed.log"
+    with open(path, "r") as file:
+        assert file
+
 
 def test_rng_seed_unset(create_env):
     """Test with no RNG seed."""
@@ -48,3 +54,19 @@ def test_rng_seed_unset(create_env):
     b = [item.timestep for item in env.game.agents["client_2_green_user"].history if item.action != "DONOTHING"]
 
     assert a != b
+
+
+def test_for_generated_seed():
+    """
+    Show that setting generate_seed_value to true producess a valid seed.
+    """
+    with open(data_manipulation_config_path(), "r") as f:
+        cfg = yaml.safe_load(f)
+
+    cfg["game"]["generate_seed_value"] = True
+    PrimaiteGymEnv(env_config=cfg)
+    path = SIM_OUTPUT.path / "seed.log"
+    with open(path, "r") as file:
+        data = file.read()
+
+    assert data.split(" ")[3] != None

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

@@ -106,7 +106,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
 
@@ -115,7 +115,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
 
@@ -138,8 +138,8 @@ def test_router_acl_addrule_integration(game_and_agent: Tuple[PrimaiteGame, Prox
     agent.store_action(action)
     game.step()
 
-    # 3: Check that the ACL now has 5 rules, and that client 1 cannot ping server 2
-    assert router.acl.num_rules == 5
+    # 3: Check that the ACL now has 4 rules, and that client 1 cannot ping server 2
+    assert router.acl.num_rules == 4
     assert not client_1.ping("10.0.2.3")  # Cannot ping server_2
     assert client_1.ping("10.0.2.2")  # Can ping server_1
     assert not server_2.ping(
@@ -165,8 +165,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
 
 
@@ -195,8 +195,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("DNSClient").dns_cache.clear()
     assert client_1.ping("10.0.2.2")  # pinging still works because ICMP is allowed