Merged PR 541: Config item that allows services and applications to return actual health states without the need to scan
## Summary Added: - services_requires_scan - applications_requires_scan To allow the agents to get the actual health state of services and applications without the need to scan ## Test process https://dev.azure.com/ma-dev-uk/PrimAITE/_git/PrimAITE/pullrequest/541?_a=files&path=/tests/unit_tests/_primaite/_game/_agent/test_observations.py ## Checklist - [X] PR is linked to a **work item** - [X] **acceptance criteria** of linked ticket are met - [X] performed **self-review** of the code - [ ] written **tests** for any new functionality added with this PR - [ ] updated the **documentation** if this PR changes or adds functionality - [ ] written/updated **design docs** if this PR implements new functionality - [X] updated the **change log** - [X] ran **pre-commit** checks for code style - [ ] attended to any **TO-DOs** left in the code Related work items: #2864
This commit is contained in:
Czar Echavez
@@ -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,
|
||||
)
|
||||
|
||||
|
||||
@@ -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())
|
||||
|
||||
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user