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

Merge remote-tracking branch 'origin/dev' into feature/2445-make-observation-thresholds-configurable

Browse Source
This commit is contained in:
Czar Echavez
2024-09-17 12:42:57 +01:00
parent 4391d7cdd5 c589c3d3c2
commit 4f172634d0
53 changed files with 13101 additions and 200 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
tests/integration_tests/game_layer/observations/test_node_observations.py
View File

@@ -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,
)

11
tests/integration_tests/game_layer/observations/test_software_observations.py
View File

@@ -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())
@@ -76,6 +80,7 @@ def test_application_executions_categories(simulation):
app_obs = ApplicationObservation(
where=["network", "nodes", pc.hostname, "applications", "WebBrowser"],
applications_requires_scan=False,
thresholds={"app_executions": {"low": 3, "medium": 6, "high": 9}},
)
@@ -87,6 +92,7 @@ def test_application_executions_categories(simulation):
# should throw an error
ApplicationObservation(
where=["network", "nodes", pc.hostname, "applications", "WebBrowser"],
applications_requires_scan=False,
thresholds={"app_executions": {"low": 9, "medium": 6, "high": 9}},
)
@@ -94,5 +100,6 @@ def test_application_executions_categories(simulation):
# should throw an error
ApplicationObservation(
where=["network", "nodes", pc.hostname, "applications", "WebBrowser"],
applications_requires_scan=False,
thresholds={"app_executions": {"low": 3, "medium": 9, "high": 9}},
)

21
tests/integration_tests/game_layer/test_action_shapes.py Normal file
View File

@@ -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)

227
tests/unit_tests/_primaite/_game/_agent/test_observations.py
View File

@@ -1,10 +1,11 @@
# © Crown-owned copyright 2024, Defence Science and Technology Laboratory UK
import json
from typing import List
import pytest
import yaml
from primaite.game.agent.observations import ObservationManager
from primaite.game.agent.observations import ApplicationObservation, ObservationManager, ServiceObservation
from primaite.game.agent.observations.file_system_observations import FileObservation, FolderObservation
from primaite.game.agent.observations.host_observations import HostObservation
@@ -130,3 +131,227 @@ class TestFileSystemRequiresScan:
[], files=[], num_files=0, include_num_access=False, file_system_requires_scan=False
)
assert obs_not_requiring_scan.observe(folder_state)["health_status"] == 3
class TestServicesRequiresScan:
@pytest.mark.parametrize(
("yaml_option_string", "expected_val"),
(
("services_requires_scan: true", True),
("services_requires_scan: false", False),
(" ", True),
),
)
def test_obs_config(self, yaml_option_string, expected_val):
"""Check that the default behaviour is to set service_requires_scan to True."""
obs_cfg_yaml = f"""
type: CUSTOM
options:
components:
- type: NODES
label: NODES
options:
hosts:
- hostname: domain_controller
- hostname: web_server
services:
- service_name: WebServer
- service_name: DNSClient
- hostname: database_server
folders:
- folder_name: database
files:
- file_name: database.db
- hostname: backup_server
services:
- service_name: FTPServer
- hostname: security_suite
- hostname: client_1
- hostname: client_2
num_services: 3
num_applications: 0
num_folders: 1
num_files: 1
num_nics: 2
include_num_access: false
{yaml_option_string}
include_nmne: true
monitored_traffic:
icmp:
- NONE
tcp:
- DNS
routers:
- hostname: router_1
num_ports: 0
ip_list:
- 192.168.1.10
- 192.168.1.12
- 192.168.1.14
- 192.168.1.16
- 192.168.1.110
- 192.168.10.21
- 192.168.10.22
- 192.168.10.110
wildcard_list:
- 0.0.0.1
port_list:
- 80
- 5432
protocol_list:
- ICMP
- TCP
- UDP
num_rules: 10
- type: LINKS
label: LINKS
options:
link_references:
- router_1:eth-1<->switch_1:eth-8
- router_1:eth-2<->switch_2:eth-8
- switch_1:eth-1<->domain_controller:eth-1
- switch_1:eth-2<->web_server:eth-1
- switch_1:eth-3<->database_server:eth-1
- switch_1:eth-4<->backup_server:eth-1
- switch_1:eth-7<->security_suite:eth-1
- switch_2:eth-1<->client_1:eth-1
- switch_2:eth-2<->client_2:eth-1
- switch_2:eth-7<->security_suite:eth-2
- type: "NONE"
label: ICS
options: {{}}
"""
cfg = yaml.safe_load(obs_cfg_yaml)
manager = ObservationManager.from_config(cfg)
hosts: List[HostObservation] = manager.obs.components["NODES"].hosts
for i, host in enumerate(hosts):
services: List[ServiceObservation] = host.services
for j, service in enumerate(services):
val = service.services_requires_scan
print(f"host {i} service {j} {val}")
assert val == expected_val # Make sure services require scan by default
def test_services_requires_scan(self):
state = {"health_state_actual": 3, "health_state_visible": 1, "operating_state": 1}
obs_requiring_scan = ServiceObservation([], services_requires_scan=True)
assert obs_requiring_scan.observe(state)["health_status"] == 1 # should be visible value
obs_not_requiring_scan = ServiceObservation([], services_requires_scan=False)
assert obs_not_requiring_scan.observe(state)["health_status"] == 3 # should be actual value
class TestApplicationsRequiresScan:
@pytest.mark.parametrize(
("yaml_option_string", "expected_val"),
(
("applications_requires_scan: true", True),
("applications_requires_scan: false", False),
(" ", True),
),
)
def test_obs_config(self, yaml_option_string, expected_val):
"""Check that the default behaviour is to set applications_requires_scan to True."""
obs_cfg_yaml = f"""
type: CUSTOM
options:
components:
- type: NODES
label: NODES
options:
hosts:
- hostname: domain_controller
- hostname: web_server
- hostname: database_server
folders:
- folder_name: database
files:
- file_name: database.db
- hostname: backup_server
- hostname: security_suite
- hostname: client_1
applications:
- application_name: WebBrowser
- hostname: client_2
applications:
- application_name: WebBrowser
- application_name: DatabaseClient
num_services: 0
num_applications: 3
num_folders: 1
num_files: 1
num_nics: 2
include_num_access: false
{yaml_option_string}
include_nmne: true
monitored_traffic:
icmp:
- NONE
tcp:
- DNS
routers:
- hostname: router_1
num_ports: 0
ip_list:
- 192.168.1.10
- 192.168.1.12
- 192.168.1.14
- 192.168.1.16
- 192.168.1.110
- 192.168.10.21
- 192.168.10.22
- 192.168.10.110
wildcard_list:
- 0.0.0.1
port_list:
- 80
- 5432
protocol_list:
- ICMP
- TCP
- UDP
num_rules: 10
- type: LINKS
label: LINKS
options:
link_references:
- router_1:eth-1<->switch_1:eth-8
- router_1:eth-2<->switch_2:eth-8
- switch_1:eth-1<->domain_controller:eth-1
- switch_1:eth-2<->web_server:eth-1
- switch_1:eth-3<->database_server:eth-1
- switch_1:eth-4<->backup_server:eth-1
- switch_1:eth-7<->security_suite:eth-1
- switch_2:eth-1<->client_1:eth-1
- switch_2:eth-2<->client_2:eth-1
- switch_2:eth-7<->security_suite:eth-2
- type: "NONE"
label: ICS
options: {{}}
"""
cfg = yaml.safe_load(obs_cfg_yaml)
manager = ObservationManager.from_config(cfg)
hosts: List[HostObservation] = manager.obs.components["NODES"].hosts
for i, host in enumerate(hosts):
services: List[ServiceObservation] = host.services
for j, service in enumerate(services):
val = service.services_requires_scan
print(f"host {i} service {j} {val}")
assert val == expected_val # Make sure applications require scan by default
def test_applications_requires_scan(self):
state = {"health_state_actual": 3, "health_state_visible": 1, "operating_state": 1, "num_executions": 1}
obs_requiring_scan = ApplicationObservation([], applications_requires_scan=True)
assert obs_requiring_scan.observe(state)["health_status"] == 1 # should be visible value
obs_not_requiring_scan = ApplicationObservation([], applications_requires_scan=False)
assert obs_not_requiring_scan.observe(state)["health_status"] == 3 # should be actual value