Merge branch 'dev' into feature/2041_2042-Add-NTP-Services
This commit is contained in:
Nick Todd
@@ -0,0 +1,180 @@
|
||||
from ipaddress import IPv4Address
|
||||
from typing import Tuple
|
||||
|
||||
import pytest
|
||||
|
||||
from primaite.simulator.network.container import Network
|
||||
from primaite.simulator.network.hardware.nodes.computer import Computer
|
||||
from primaite.simulator.network.hardware.nodes.router import ACLAction, Router
|
||||
from primaite.simulator.network.hardware.nodes.server import Server
|
||||
from primaite.simulator.network.transmission.transport_layer import Port
|
||||
from primaite.simulator.system.applications.application import ApplicationOperatingState
|
||||
from primaite.simulator.system.applications.database_client import DatabaseClient
|
||||
from primaite.simulator.system.applications.red_applications.dos_bot import DoSAttackStage, DoSBot
|
||||
from primaite.simulator.system.services.database.database_service import DatabaseService
|
||||
from primaite.simulator.system.software import SoftwareHealthState
|
||||
|
||||
|
||||
@pytest.fixture(scope="function")
|
||||
def dos_bot_and_db_server(client_server) -> Tuple[DoSBot, Computer, DatabaseService, Server]:
|
||||
computer, server = client_server
|
||||
|
||||
# Install DoSBot on computer
|
||||
computer.software_manager.install(DoSBot)
|
||||
|
||||
dos_bot: DoSBot = computer.software_manager.software.get("DoSBot")
|
||||
dos_bot.configure(
|
||||
target_ip_address=IPv4Address(server.nics.get(next(iter(server.nics))).ip_address),
|
||||
target_port=Port.POSTGRES_SERVER,
|
||||
)
|
||||
|
||||
# Install DB Server service on server
|
||||
server.software_manager.install(DatabaseService)
|
||||
db_server_service: DatabaseService = server.software_manager.software.get("DatabaseService")
|
||||
db_server_service.start()
|
||||
|
||||
return dos_bot, computer, db_server_service, server
|
||||
|
||||
|
||||
@pytest.fixture(scope="function")
|
||||
def dos_bot_db_server_green_client(example_network) -> Network:
|
||||
network: Network = example_network
|
||||
|
||||
router_1: Router = example_network.get_node_by_hostname("router_1")
|
||||
router_1.acl.add_rule(
|
||||
action=ACLAction.PERMIT, src_port=Port.POSTGRES_SERVER, dst_port=Port.POSTGRES_SERVER, position=0
|
||||
)
|
||||
|
||||
client_1: Computer = network.get_node_by_hostname("client_1")
|
||||
client_2: Computer = network.get_node_by_hostname("client_2")
|
||||
server: Server = network.get_node_by_hostname("server_1")
|
||||
|
||||
# install DoS bot on client 1
|
||||
client_1.software_manager.install(DoSBot)
|
||||
|
||||
dos_bot: DoSBot = client_1.software_manager.software.get("DoSBot")
|
||||
dos_bot.configure(
|
||||
target_ip_address=IPv4Address(server.nics.get(next(iter(server.nics))).ip_address),
|
||||
target_port=Port.POSTGRES_SERVER,
|
||||
)
|
||||
|
||||
# install db server service on server
|
||||
server.software_manager.install(DatabaseService)
|
||||
db_server_service: DatabaseService = server.software_manager.software.get("DatabaseService")
|
||||
db_server_service.start()
|
||||
|
||||
# Install DB client (green) on client 2
|
||||
client_2.software_manager.install(DatabaseClient)
|
||||
|
||||
database_client: DatabaseClient = client_2.software_manager.software.get("DatabaseClient")
|
||||
database_client.configure(server_ip_address=IPv4Address("192.168.0.1"))
|
||||
database_client.run()
|
||||
|
||||
return network
|
||||
|
||||
|
||||
def test_repeating_dos_attack(dos_bot_and_db_server):
|
||||
dos_bot, computer, db_server_service, server = dos_bot_and_db_server
|
||||
|
||||
assert db_server_service.health_state_actual is SoftwareHealthState.GOOD
|
||||
|
||||
dos_bot.port_scan_p_of_success = 1
|
||||
dos_bot.repeat = True
|
||||
dos_bot.run()
|
||||
|
||||
assert len(dos_bot.connections) == db_server_service.max_sessions
|
||||
assert len(db_server_service.connections) == db_server_service.max_sessions
|
||||
assert len(dos_bot.connections) == db_server_service.max_sessions
|
||||
|
||||
assert dos_bot.attack_stage is DoSAttackStage.NOT_STARTED
|
||||
assert db_server_service.health_state_actual is SoftwareHealthState.OVERWHELMED
|
||||
|
||||
db_server_service.clear_connections()
|
||||
db_server_service.health_state_actual = SoftwareHealthState.GOOD
|
||||
assert len(db_server_service.connections) == 0
|
||||
|
||||
computer.apply_timestep(timestep=1)
|
||||
server.apply_timestep(timestep=1)
|
||||
|
||||
assert len(dos_bot.connections) == db_server_service.max_sessions
|
||||
assert len(db_server_service.connections) == db_server_service.max_sessions
|
||||
assert len(dos_bot.connections) == db_server_service.max_sessions
|
||||
|
||||
assert dos_bot.attack_stage is DoSAttackStage.NOT_STARTED
|
||||
assert db_server_service.health_state_actual is SoftwareHealthState.OVERWHELMED
|
||||
|
||||
|
||||
def test_non_repeating_dos_attack(dos_bot_and_db_server):
|
||||
dos_bot, computer, db_server_service, server = dos_bot_and_db_server
|
||||
|
||||
assert db_server_service.health_state_actual is SoftwareHealthState.GOOD
|
||||
|
||||
dos_bot.port_scan_p_of_success = 1
|
||||
dos_bot.repeat = False
|
||||
dos_bot.run()
|
||||
|
||||
assert len(dos_bot.connections) == db_server_service.max_sessions
|
||||
assert len(db_server_service.connections) == db_server_service.max_sessions
|
||||
assert len(dos_bot.connections) == db_server_service.max_sessions
|
||||
|
||||
assert dos_bot.attack_stage is DoSAttackStage.COMPLETED
|
||||
assert db_server_service.health_state_actual is SoftwareHealthState.OVERWHELMED
|
||||
|
||||
db_server_service.clear_connections()
|
||||
db_server_service.health_state_actual = SoftwareHealthState.GOOD
|
||||
assert len(db_server_service.connections) == 0
|
||||
|
||||
computer.apply_timestep(timestep=1)
|
||||
server.apply_timestep(timestep=1)
|
||||
|
||||
assert len(dos_bot.connections) == 0
|
||||
assert len(db_server_service.connections) == 0
|
||||
assert len(dos_bot.connections) == 0
|
||||
|
||||
assert dos_bot.attack_stage is DoSAttackStage.COMPLETED
|
||||
assert db_server_service.health_state_actual is SoftwareHealthState.GOOD
|
||||
|
||||
|
||||
def test_dos_bot_database_service_connection(dos_bot_and_db_server):
|
||||
dos_bot, computer, db_server_service, server = dos_bot_and_db_server
|
||||
|
||||
dos_bot.operating_state = ApplicationOperatingState.RUNNING
|
||||
dos_bot.attack_stage = DoSAttackStage.PORT_SCAN
|
||||
dos_bot._perform_dos()
|
||||
|
||||
assert len(dos_bot.connections) == db_server_service.max_sessions
|
||||
assert len(db_server_service.connections) == db_server_service.max_sessions
|
||||
assert len(dos_bot.connections) == db_server_service.max_sessions
|
||||
|
||||
|
||||
def test_dos_blocks_green_agent_connection(dos_bot_db_server_green_client):
|
||||
network: Network = dos_bot_db_server_green_client
|
||||
|
||||
client_1: Computer = network.get_node_by_hostname("client_1")
|
||||
dos_bot: DoSBot = client_1.software_manager.software.get("DoSBot")
|
||||
|
||||
client_2: Computer = network.get_node_by_hostname("client_2")
|
||||
green_db_client: DatabaseClient = client_2.software_manager.software.get("DatabaseClient")
|
||||
|
||||
server: Server = network.get_node_by_hostname("server_1")
|
||||
db_server_service: DatabaseService = server.software_manager.software.get("DatabaseService")
|
||||
|
||||
assert db_server_service.health_state_actual is SoftwareHealthState.GOOD
|
||||
|
||||
dos_bot.port_scan_p_of_success = 1
|
||||
dos_bot.repeat = False
|
||||
dos_bot.run()
|
||||
|
||||
# DoS bot fills up connection of db server service
|
||||
assert len(dos_bot.connections) == db_server_service.max_sessions
|
||||
assert len(db_server_service.connections) == db_server_service.max_sessions
|
||||
assert len(dos_bot.connections) == db_server_service.max_sessions
|
||||
assert len(green_db_client.connections) == 0
|
||||
|
||||
assert dos_bot.attack_stage is DoSAttackStage.COMPLETED
|
||||
# db server service is overwhelmed
|
||||
assert db_server_service.health_state_actual is SoftwareHealthState.OVERWHELMED
|
||||
|
||||
# green agent tries to connect but fails because service is overwhelmed
|
||||
assert green_db_client.connect() is False
|
||||
assert len(green_db_client.connections) == 0
|
||||
@@ -1,6 +1,9 @@
|
||||
from ipaddress import IPv4Address
|
||||
from typing import Tuple
|
||||
|
||||
from primaite.simulator.network.hardware.node_operating_state import NodeOperatingState
|
||||
import pytest
|
||||
|
||||
from primaite.simulator.network.hardware.base import Link, NIC, Node, NodeOperatingState
|
||||
from primaite.simulator.network.hardware.nodes.server import Server
|
||||
from primaite.simulator.system.applications.database_client import DatabaseClient
|
||||
from primaite.simulator.system.services.database.database_service import DatabaseService
|
||||
@@ -8,57 +11,109 @@ from primaite.simulator.system.services.ftp.ftp_server import FTPServer
|
||||
from primaite.simulator.system.services.service import ServiceOperatingState
|
||||
|
||||
|
||||
def test_database_client_server_connection(uc2_network):
|
||||
web_server: Server = uc2_network.get_node_by_hostname("web_server")
|
||||
db_client: DatabaseClient = web_server.software_manager.software.get("DatabaseClient")
|
||||
@pytest.fixture(scope="function")
|
||||
def peer_to_peer() -> Tuple[Node, Node]:
|
||||
node_a = Node(hostname="node_a", operating_state=NodeOperatingState.ON)
|
||||
nic_a = NIC(ip_address="192.168.0.10", subnet_mask="255.255.255.0", operating_state=NodeOperatingState.ON)
|
||||
node_a.connect_nic(nic_a)
|
||||
node_a.software_manager.get_open_ports()
|
||||
|
||||
db_server: Server = uc2_network.get_node_by_hostname("database_server")
|
||||
db_service: DatabaseService = db_server.software_manager.software.get("DatabaseService")
|
||||
node_b = Node(hostname="node_b", operating_state=NodeOperatingState.ON)
|
||||
nic_b = NIC(ip_address="192.168.0.11", subnet_mask="255.255.255.0")
|
||||
node_b.connect_nic(nic_b)
|
||||
|
||||
Link(endpoint_a=nic_a, endpoint_b=nic_b)
|
||||
|
||||
assert node_a.ping("192.168.0.11")
|
||||
|
||||
node_a.software_manager.install(DatabaseClient)
|
||||
node_a.software_manager.software["DatabaseClient"].configure(server_ip_address=IPv4Address("192.168.0.11"))
|
||||
node_a.software_manager.software["DatabaseClient"].run()
|
||||
|
||||
node_b.software_manager.install(DatabaseService)
|
||||
database_service: DatabaseService = node_b.software_manager.software["DatabaseService"] # noqa
|
||||
database_service.start()
|
||||
return node_a, node_b
|
||||
|
||||
|
||||
@pytest.fixture(scope="function")
|
||||
def peer_to_peer_secure_db() -> Tuple[Node, Node]:
|
||||
node_a = Node(hostname="node_a", operating_state=NodeOperatingState.ON)
|
||||
nic_a = NIC(ip_address="192.168.0.10", subnet_mask="255.255.255.0", operating_state=NodeOperatingState.ON)
|
||||
node_a.connect_nic(nic_a)
|
||||
node_a.software_manager.get_open_ports()
|
||||
|
||||
node_b = Node(hostname="node_b", operating_state=NodeOperatingState.ON)
|
||||
nic_b = NIC(ip_address="192.168.0.11", subnet_mask="255.255.255.0")
|
||||
node_b.connect_nic(nic_b)
|
||||
|
||||
Link(endpoint_a=nic_a, endpoint_b=nic_b)
|
||||
|
||||
assert node_a.ping("192.168.0.11")
|
||||
|
||||
node_a.software_manager.install(DatabaseClient)
|
||||
node_a.software_manager.software["DatabaseClient"].configure(server_ip_address=IPv4Address("192.168.0.11"))
|
||||
node_a.software_manager.software["DatabaseClient"].run()
|
||||
|
||||
node_b.software_manager.install(DatabaseService)
|
||||
database_service: DatabaseService = node_b.software_manager.software["DatabaseService"] # noqa
|
||||
database_service.password = "12345"
|
||||
database_service.start()
|
||||
return node_a, node_b
|
||||
|
||||
|
||||
def test_database_client_server_connection(peer_to_peer):
|
||||
node_a, node_b = peer_to_peer
|
||||
|
||||
db_client: DatabaseClient = node_a.software_manager.software["DatabaseClient"]
|
||||
|
||||
db_service: DatabaseService = node_b.software_manager.software["DatabaseService"]
|
||||
|
||||
db_client.connect()
|
||||
assert len(db_client.connections) == 1
|
||||
assert len(db_service.connections) == 1
|
||||
|
||||
db_client.disconnect()
|
||||
assert len(db_client.connections) == 0
|
||||
assert len(db_service.connections) == 0
|
||||
|
||||
|
||||
def test_database_client_server_correct_password(uc2_network):
|
||||
web_server: Server = uc2_network.get_node_by_hostname("web_server")
|
||||
db_client: DatabaseClient = web_server.software_manager.software.get("DatabaseClient")
|
||||
def test_database_client_server_correct_password(peer_to_peer_secure_db):
|
||||
node_a, node_b = peer_to_peer_secure_db
|
||||
|
||||
db_server: Server = uc2_network.get_node_by_hostname("database_server")
|
||||
db_service: DatabaseService = db_server.software_manager.software.get("DatabaseService")
|
||||
db_client: DatabaseClient = node_a.software_manager.software["DatabaseClient"]
|
||||
|
||||
db_client.disconnect()
|
||||
|
||||
db_client.configure(server_ip_address=IPv4Address("192.168.1.14"), server_password="12345")
|
||||
db_service.password = "12345"
|
||||
|
||||
assert db_client.connect()
|
||||
db_service: DatabaseService = node_b.software_manager.software["DatabaseService"]
|
||||
|
||||
db_client.configure(server_ip_address=IPv4Address("192.168.0.11"), server_password="12345")
|
||||
db_client.connect()
|
||||
assert len(db_client.connections) == 1
|
||||
assert len(db_service.connections) == 1
|
||||
|
||||
|
||||
def test_database_client_server_incorrect_password(uc2_network):
|
||||
web_server: Server = uc2_network.get_node_by_hostname("web_server")
|
||||
db_client: DatabaseClient = web_server.software_manager.software.get("DatabaseClient")
|
||||
def test_database_client_server_incorrect_password(peer_to_peer_secure_db):
|
||||
node_a, node_b = peer_to_peer_secure_db
|
||||
|
||||
db_server: Server = uc2_network.get_node_by_hostname("database_server")
|
||||
db_service: DatabaseService = db_server.software_manager.software.get("DatabaseService")
|
||||
db_client: DatabaseClient = node_a.software_manager.software["DatabaseClient"]
|
||||
|
||||
db_client.disconnect()
|
||||
db_client.configure(server_ip_address=IPv4Address("192.168.1.14"), server_password="54321")
|
||||
db_service.password = "12345"
|
||||
db_service: DatabaseService = node_b.software_manager.software["DatabaseService"]
|
||||
|
||||
assert not db_client.connect()
|
||||
# should fail
|
||||
db_client.connect()
|
||||
assert len(db_client.connections) == 0
|
||||
assert len(db_service.connections) == 0
|
||||
|
||||
db_client.configure(server_ip_address=IPv4Address("192.168.0.11"), server_password="wrongpass")
|
||||
db_client.connect()
|
||||
assert len(db_client.connections) == 0
|
||||
assert len(db_service.connections) == 0
|
||||
|
||||
|
||||
def test_database_client_query(uc2_network):
|
||||
"""Tests DB query across the network returns HTTP status 200 and date."""
|
||||
web_server: Server = uc2_network.get_node_by_hostname("web_server")
|
||||
db_client: DatabaseClient = web_server.software_manager.software.get("DatabaseClient")
|
||||
|
||||
assert db_client.connected
|
||||
db_client: DatabaseClient = web_server.software_manager.software["DatabaseClient"]
|
||||
db_client.connect()
|
||||
|
||||
assert db_client.query("SELECT")
|
||||
|
||||
@@ -66,13 +121,13 @@ def test_database_client_query(uc2_network):
|
||||
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.get("DatabaseService")
|
||||
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.get("FTPServer")
|
||||
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
|
||||
@@ -81,7 +136,7 @@ def test_create_database_backup(uc2_network):
|
||||
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.get("DatabaseService")
|
||||
db_service: DatabaseService = db_server.software_manager.software["DatabaseService"]
|
||||
|
||||
# create a back up
|
||||
assert db_service.backup_database() is True
|
||||
@@ -107,7 +162,7 @@ def test_database_client_cannot_query_offline_database_server(uc2_network):
|
||||
|
||||
web_server: Server = uc2_network.get_node_by_hostname("web_server")
|
||||
db_client: DatabaseClient = web_server.software_manager.software.get("DatabaseClient")
|
||||
assert db_client.connected
|
||||
assert len(db_client.connections)
|
||||
|
||||
assert db_client.query("SELECT") is True
|
||||
|
||||
|
||||
Reference in New Issue
Block a user