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

Merge 'origin/feature/1812' into feature/1924

Browse Source
This commit is contained in:
Marek Wolan
2023-10-09 18:25:46 +01:00
parent e85c5977d0 5a5710c6ae
commit 560e416eda
43 changed files with 1362 additions and 879 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
tests/e2e_integration_tests/test_uc2_data_manipulation_scenario.py
View File

@@ -1,11 +1,12 @@
from primaite.simulator.network.hardware.nodes.computer import Computer
from primaite.simulator.network.hardware.nodes.server import Server
from primaite.simulator.system.applications.database_client import DatabaseClient
from primaite.simulator.system.services.database_service import DatabaseService
from primaite.simulator.system.services.database.database_service import DatabaseService
from primaite.simulator.system.services.red_services.data_manipulation_bot import DataManipulationBot
def test_data_manipulation(uc2_network):
"""Tests the UC2 data manipulation scenario end-to-end. Is a work in progress."""
client_1: Computer = uc2_network.get_node_by_hostname("client_1")
db_manipulation_bot: DataManipulationBot = client_1.software_manager.software["DataManipulationBot"]
@@ -15,6 +16,8 @@ def test_data_manipulation(uc2_network):
web_server: Server = uc2_network.get_node_by_hostname("web_server")
db_client: DatabaseClient = web_server.software_manager.software["DatabaseClient"]
db_service.backup_database()
# First check that the DB client on the web_server can successfully query the users table on the database
assert db_client.query("SELECT * FROM user;")
@@ -23,3 +26,9 @@ def test_data_manipulation(uc2_network):
# Now check that the DB client on the web_server cannot query the users table on the database
assert not db_client.query("SELECT * FROM user;")
# Now restore the database
db_service.restore_backup()
# Now check that the DB client on the web_server can successfully query the users table on the database
assert db_client.query("SELECT * FROM user;")

10
tests/integration_tests/component_creation/test_action_integration.py
View File

@@ -1,11 +1,11 @@
import pytest
from primaite.simulator.core import Action
from primaite.simulator.core import RequestType
from primaite.simulator.network.hardware.nodes.computer import Computer
from primaite.simulator.network.hardware.nodes.server import Server
from primaite.simulator.network.hardware.nodes.switch import Switch
from primaite.simulator.sim_container import Simulation
from primaite.simulator.system.services.database_service import DatabaseService
from primaite.simulator.system.services.database.database_service import DatabaseService
def test_passing_actions_down(monkeypatch) -> None:
@@ -32,7 +32,7 @@ def test_passing_actions_down(monkeypatch) -> None:
sim.network.connect(s1.switch_ports[3], srv.ethernet_port[1])
# call this method to make sure no errors occur.
sim._action_manager.get_action_tree()
sim._request_manager.get_request_types_recursively()
# patch the action to do something which we can check the result of.
action_invoked = False
@@ -42,13 +42,13 @@ def test_passing_actions_down(monkeypatch) -> None:
action_invoked = True
monkeypatch.setitem(
downloads_folder._action_manager.actions, "repair", Action(func=lambda request, context: succeed())
downloads_folder._request_manager.request_types, "repair", RequestType(func=lambda request, context: succeed())
)
assert not action_invoked
# call the patched method
sim.apply_action(
sim.apply_request(
["network", "node", pc1.uuid, "file_system", "folder", pc1.file_system.get_folder("downloads").uuid, "repair"]
)

44
tests/integration_tests/component_creation/test_permission_system.py
View File

@@ -3,7 +3,7 @@ from typing import Dict, List, Literal
import pytest
from primaite.simulator.core import Action, ActionManager, AllowAllValidator, SimComponent
from primaite.simulator.core import AllowAllValidator, RequestManager, RequestType, SimComponent
from primaite.simulator.domain.controller import AccountGroup, GroupMembershipValidator
@@ -29,11 +29,11 @@ def test_group_action_validation() -> None:
def __init__(self, **kwargs):
super().__init__(**kwargs)
self._action_manager = ActionManager()
self._request_manager = RequestManager()
self._action_manager.add_action(
self._request_manager.add_request(
"create_folder",
Action(
RequestType(
func=lambda request, context: self.create_folder(request[0]),
validator=GroupMembershipValidator([AccountGroup.LOCAL_ADMIN, AccountGroup.DOMAIN_ADMIN]),
),
@@ -52,13 +52,13 @@ def test_group_action_validation() -> None:
# check that the folder is created when a local admin tried to do it
permitted_context = {"request_source": {"agent": "BLUE", "account": "User1", "groups": ["LOCAL_ADMIN"]}}
my_node = Node(uuid="0000-0000-1234", name="pc")
my_node.apply_action(["create_folder", "memes"], context=permitted_context)
my_node.apply_request(["create_folder", "memes"], context=permitted_context)
assert len(my_node.folders) == 1
assert my_node.folders[0].name == "memes"
# check that the number of folders is still 1 even after attempting to create a second one without permissions
invalid_context = {"request_source": {"agent": "BLUE", "account": "User1", "groups": ["LOCAL_USER", "DOMAIN_USER"]}}
my_node.apply_action(["create_folder", "memes2"], context=invalid_context)
my_node.apply_request(["create_folder", "memes2"], context=invalid_context)
assert len(my_node.folders) == 1
assert my_node.folders[0].name == "memes"
@@ -79,32 +79,32 @@ def test_hierarchical_action_with_validation() -> None:
def __init__(self, **kwargs):
super().__init__(**kwargs)
self.action_manager = ActionManager()
self.request_manager = RequestManager()
self.action_manager.add_action(
self.request_manager.add_request(
"turn_on",
Action(
RequestType(
func=lambda request, context: self.turn_on(),
validator=AllowAllValidator(),
),
)
self.action_manager.add_action(
self.request_manager.add_request(
"turn_off",
Action(
RequestType(
func=lambda request, context: self.turn_off(),
validator=AllowAllValidator(),
),
)
self.action_manager.add_action(
self.request_manager.add_request(
"disable",
Action(
RequestType(
func=lambda request, context: self.disable(),
validator=GroupMembershipValidator([AccountGroup.LOCAL_ADMIN, AccountGroup.DOMAIN_ADMIN]),
),
)
self.action_manager.add_action(
self.request_manager.add_request(
"enable",
Action(
RequestType(
func=lambda request, context: self.enable(),
validator=GroupMembershipValidator([AccountGroup.LOCAL_ADMIN, AccountGroup.DOMAIN_ADMIN]),
),
@@ -135,11 +135,11 @@ def test_hierarchical_action_with_validation() -> None:
def __init__(self, **kwargs):
super().__init__(**kwargs)
self.action_manager = ActionManager()
self.request_manager = RequestManager()
self.action_manager.add_action(
self.request_manager.add_request(
"apps",
Action(
RequestType(
func=lambda request, context: self.send_action_to_app(request.pop(0), request, context),
validator=AllowAllValidator(),
),
@@ -155,7 +155,7 @@ def test_hierarchical_action_with_validation() -> None:
def send_action_to_app(self, app_name: str, options: List[str], context: Dict):
for app in self.apps:
if app_name == app.name:
app.apply_action(options, context)
app.apply_request(options, context)
break
else:
msg = f"Node has no app with name {app_name}"
@@ -178,15 +178,15 @@ def test_hierarchical_action_with_validation() -> None:
}
# check that a non-admin can't disable this app
my_node.apply_action(["apps", "Chrome", "disable"], non_admin_context)
my_node.apply_request(["apps", "Chrome", "disable"], non_admin_context)
assert my_node.apps[0].name == "Chrome" # if failure occurs on this line, the test itself is broken
assert my_node.apps[0].state == "off"
# check that a non-admin can turn this app on
my_node.apply_action(["apps", "Firefox", "turn_on"], non_admin_context)
my_node.apply_request(["apps", "Firefox", "turn_on"], non_admin_context)
assert my_node.apps[1].name == "Firefox" # if failure occurs on this line, the test itself is broken
assert my_node.apps[1].state == "on"
# check that an admin can disable this app
my_node.apply_action(["apps", "Chrome", "disable"], admin_context)
my_node.apply_request(["apps", "Chrome", "disable"], admin_context)
assert my_node.apps[0].state == "disabled"

37
tests/integration_tests/system/test_database_on_node.py
View File

@@ -2,7 +2,8 @@ from ipaddress import IPv4Address
from primaite.simulator.network.hardware.nodes.server import Server
from primaite.simulator.system.applications.database_client import DatabaseClient
from primaite.simulator.system.services.database_service import DatabaseService
from primaite.simulator.system.services.database.database_service import DatabaseService
from primaite.simulator.system.services.ftp.ftp_server import FTPServer
def test_database_client_server_connection(uc2_network):
@@ -57,3 +58,37 @@ def test_database_client_query(uc2_network):
db_client.connect()
assert db_client.query("SELECT * FROM user;")
def test_create_database_backup(uc2_network):
"""Run the backup_database method and check if the FTP server has the relevant file."""
db_server: Server = uc2_network.get_node_by_hostname("database_server")
db_service: DatabaseService = db_server.software_manager.software["DatabaseService"]
# back up should be created
assert db_service.backup_database() is True
backup_server: Server = uc2_network.get_node_by_hostname("backup_server")
ftp_server: FTPServer = backup_server.software_manager.software["FTPServer"]
# backup file should exist in the backup server
assert ftp_server.file_system.get_file(folder_name=db_service.uuid, file_name="database.db") is not None
def test_restore_backup(uc2_network):
"""Run the restore_backup method and check if the backup is properly restored."""
db_server: Server = uc2_network.get_node_by_hostname("database_server")
db_service: DatabaseService = db_server.software_manager.software["DatabaseService"]
# create a back up
assert db_service.backup_database() is True
# delete database locally
db_service.file_system.delete_file(folder_name="database", file_name="database.db")
assert db_service.file_system.get_file(folder_name="database", file_name="database.db") is None
# back up should be restored
assert db_service.restore_backup() is True
assert db_service.file_system.get_file(folder_name="database", file_name="database.db") is not None

6
tests/integration_tests/system/test_dns_client_server.py
View File

@@ -1,9 +1,7 @@
from ipaddress import IPv4Address
from primaite.simulator.network.hardware.nodes.computer import Computer
from primaite.simulator.network.hardware.nodes.server import Server
from primaite.simulator.system.services.dns_client import DNSClient
from primaite.simulator.system.services.dns_server import DNSServer
from primaite.simulator.system.services.dns.dns_client import DNSClient
from primaite.simulator.system.services.dns.dns_server import DNSServer
from primaite.simulator.system.services.service import ServiceOperatingState

62
tests/integration_tests/system/test_ftp_client_server.py Normal file
View File

@@ -0,0 +1,62 @@
from ipaddress import IPv4Address
from primaite.simulator.network.hardware.nodes.computer import Computer
from primaite.simulator.network.hardware.nodes.server import Server
from primaite.simulator.system.services.ftp.ftp_client import FTPClient
from primaite.simulator.system.services.ftp.ftp_server import FTPServer
from primaite.simulator.system.services.service import ServiceOperatingState
def test_ftp_client_store_file_in_server(uc2_network):
"""
Test checks to see if the client is able to store files in the backup server.
"""
client_1: Computer = uc2_network.get_node_by_hostname("client_1")
backup_server: Server = uc2_network.get_node_by_hostname("backup_server")
ftp_client: FTPClient = client_1.software_manager.software["FTPClient"]
ftp_server: FTPServer = backup_server.software_manager.software["FTPServer"]
assert ftp_client.operating_state == ServiceOperatingState.RUNNING
assert ftp_server.operating_state == ServiceOperatingState.RUNNING
# create file on ftp client
ftp_client.file_system.create_file(file_name="test_file.txt")
assert ftp_client.send_file(
src_folder_name="root",
src_file_name="test_file.txt",
dest_folder_name="client_1_backup",
dest_file_name="test_file.txt",
dest_ip_address=backup_server.nics.get(next(iter(backup_server.nics))).ip_address,
)
assert ftp_server.file_system.get_file(folder_name="client_1_backup", file_name="test_file.txt")
def test_ftp_client_retrieve_file_from_server(uc2_network):
"""
Test checks to see if the client is able to retrieve files from the backup server.
"""
client_1: Computer = uc2_network.get_node_by_hostname("client_1")
backup_server: Server = uc2_network.get_node_by_hostname("backup_server")
ftp_client: FTPClient = client_1.software_manager.software["FTPClient"]
ftp_server: FTPServer = backup_server.software_manager.software["FTPServer"]
assert ftp_client.operating_state == ServiceOperatingState.RUNNING
assert ftp_server.operating_state == ServiceOperatingState.RUNNING
# create file on ftp server
ftp_server.file_system.create_file(file_name="test_file.txt", folder_name="file_share")
assert ftp_client.request_file(
src_folder_name="file_share",
src_file_name="test_file.txt",
dest_folder_name="downloads",
dest_file_name="test_file.txt",
dest_ip_address=backup_server.nics.get(next(iter(backup_server.nics))).ip_address,
)
# client should have retrieved the file
assert ftp_client.file_system.get_file(folder_name="downloads", file_name="test_file.txt")

2
tests/unit_tests/_primaite/_simulator/_domain/test_account.py
View File

@@ -13,6 +13,6 @@ def test_account_deserialise():
"""Test that an account can be deserialised. The test fails if pydantic throws an error."""
acct_json = (
'{"uuid":"dfb2bcaa-d3a1-48fd-af3f-c943354622b4","num_logons":0,"num_logoffs":0,"num_group_changes":0,'
'"username":"Jake","password":"JakePass1!","account_type":2,"status":2,"action_manager":null}'
'"username":"Jake","password":"JakePass1!","account_type":2,"status":2,"request_manager":null}'
)
acct = Account.model_validate_json(acct_json)

4
tests/unit_tests/_primaite/_simulator/_system/_services/test_database.py
View File

@@ -1,9 +1,7 @@
import json
import pytest
from primaite.simulator.network.hardware.base import Node
from primaite.simulator.system.services.database_service import DatabaseService
from primaite.simulator.system.services.database.database_service import DatabaseService
@pytest.fixture(scope="function")

16
tests/unit_tests/_primaite/_simulator/_system/_services/test_dns.py
View File

@@ -3,16 +3,20 @@ from ipaddress import IPv4Address
import pytest
from primaite.simulator.network.hardware.base import Node
from primaite.simulator.network.hardware.nodes.computer import Computer
from primaite.simulator.network.hardware.nodes.server import Server
from primaite.simulator.network.protocols.dns import DNSPacket, DNSReply, DNSRequest
from primaite.simulator.network.transmission.network_layer import IPProtocol
from primaite.simulator.network.transmission.transport_layer import Port
from primaite.simulator.system.services.dns_client import DNSClient
from primaite.simulator.system.services.dns_server import DNSServer
from primaite.simulator.system.services.dns.dns_client import DNSClient
from primaite.simulator.system.services.dns.dns_server import DNSServer
@pytest.fixture(scope="function")
def dns_server() -> Node:
node = Node(hostname="dns_server")
node = Server(
hostname="dns_server", ip_address="192.168.1.10", subnet_mask="255.255.255.0", default_gateway="192.168.1.1"
)
node.software_manager.install(software_class=DNSServer)
node.software_manager.software["DNSServer"].start()
return node
@@ -20,9 +24,9 @@ def dns_server() -> Node:
@pytest.fixture(scope="function")
def dns_client() -> Node:
node = Node(hostname="dns_client")
node.software_manager.install(software_class=DNSClient)
node.software_manager.software["DNSClient"].start()
node = Computer(
hostname="dns_client", ip_address="192.168.1.11", subnet_mask="255.255.255.0", default_gateway="192.168.1.1"
)
return node

86
tests/unit_tests/_primaite/_simulator/_system/_services/test_ftp.py Normal file
View File

@@ -0,0 +1,86 @@
from ipaddress import IPv4Address
import pytest
from primaite.simulator.network.hardware.base import Node
from primaite.simulator.network.hardware.nodes.computer import Computer
from primaite.simulator.network.hardware.nodes.server import Server
from primaite.simulator.network.protocols.ftp import FTPCommand, FTPPacket
from primaite.simulator.network.transmission.network_layer import IPProtocol
from primaite.simulator.network.transmission.transport_layer import Port
from primaite.simulator.system.services.ftp.ftp_client import FTPClient
from primaite.simulator.system.services.ftp.ftp_server import FTPServer
@pytest.fixture(scope="function")
def ftp_server() -> Node:
node = Server(
hostname="ftp_server", ip_address="192.168.1.10", subnet_mask="255.255.255.0", default_gateway="192.168.1.1"
)
node.software_manager.install(software_class=FTPServer)
node.software_manager.software["FTPServer"].start()
return node
@pytest.fixture(scope="function")
def ftp_client() -> Node:
node = Computer(
hostname="ftp_client", ip_address="192.168.1.11", subnet_mask="255.255.255.0", default_gateway="192.168.1.1"
)
return node
def test_create_ftp_server(ftp_server):
assert ftp_server is not None
ftp_server_service: FTPServer = ftp_server.software_manager.software["FTPServer"]
assert ftp_server_service.name is "FTPServer"
assert ftp_server_service.port is Port.FTP
assert ftp_server_service.protocol is IPProtocol.TCP
def test_create_ftp_client(ftp_client):
assert ftp_client is not None
ftp_client_service: FTPClient = ftp_client.software_manager.software["FTPClient"]
assert ftp_client_service.name is "FTPClient"
assert ftp_client_service.port is Port.FTP
assert ftp_client_service.protocol is IPProtocol.TCP
def test_ftp_server_store_file(ftp_server):
"""Test to make sure the FTP Server knows how to deal with request responses."""
assert ftp_server.file_system.get_file(folder_name="downloads", file_name="file.txt") is None
response: FTPPacket = FTPPacket(
ftp_command=FTPCommand.STOR,
ftp_command_args={
"dest_folder_name": "downloads",
"dest_file_name": "file.txt",
"file_size": 24,
},
packet_payload_size=24,
)
ftp_server_service: FTPServer = ftp_server.software_manager.software["FTPServer"]
ftp_server_service.receive(response)
assert ftp_server.file_system.get_file(folder_name="downloads", file_name="file.txt")
def test_ftp_client_store_file(ftp_client):
"""Test to make sure the FTP Client knows how to deal with request responses."""
assert ftp_client.file_system.get_file(folder_name="downloads", file_name="file.txt") is None
response: FTPPacket = FTPPacket(
ftp_command=FTPCommand.STOR,
ftp_command_args={
"dest_folder_name": "downloads",
"dest_file_name": "file.txt",
"file_size": 24,
},
packet_payload_size=24,
)
ftp_client_service: FTPClient = ftp_client.software_manager.software["FTPClient"]
ftp_client_service.receive(response)
assert ftp_client.file_system.get_file(folder_name="downloads", file_name="file.txt")