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

#2257: add firewall via config + fix router hop ip address + shuffling around tests

Browse Source
This commit is contained in:
Czar Echavez
2024-02-15 15:45:18 +00:00
parent ab7c7b9c06
commit b739823318
14 changed files with 322 additions and 173 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
src/primaite/game/game.py
View File

@@ -15,6 +15,7 @@ from primaite.simulator.network.hardware.base import NodeOperatingState
from primaite.simulator.network.hardware.nodes.host.computer import Computer
from primaite.simulator.network.hardware.nodes.host.host_node import NIC
from primaite.simulator.network.hardware.nodes.host.server import Server
from primaite.simulator.network.hardware.nodes.network.firewall import Firewall
from primaite.simulator.network.hardware.nodes.network.router import Router
from primaite.simulator.network.hardware.nodes.network.switch import Switch
from primaite.simulator.network.transmission.transport_layer import Port
@@ -252,6 +253,8 @@ class PrimaiteGame:
)
elif n_type == "router":
new_node = Router.from_config(node_cfg)
elif n_type == "firewall":
new_node = Firewall.from_config(node_cfg)
else:
_LOGGER.warning(f"invalid node type {n_type} in config")
if "services" in node_cfg:
@@ -264,12 +267,12 @@ class PrimaiteGame:
new_node.software_manager.install(SERVICE_TYPES_MAPPING[service_type])
new_service = new_node.software_manager.software[service_type]
game.ref_map_services[service_ref] = new_service.uuid
# start the service
new_service.start()
else:
_LOGGER.warning(f"service type not found {service_type}")
# start the service
new_service.start()
# service-dependent options
if service_type == "DNSClient":
if "options" in service_cfg:

4
src/primaite/simulator/__init__.py
View File

@@ -12,8 +12,8 @@ class _SimOutput:
self._path: Path = (
_PRIMAITE_ROOT.parent.parent / "simulation_output" / datetime.now().strftime("%Y-%m-%d_%H-%M-%S")
)
self.save_pcap_logs: bool = True
self.save_sys_logs: bool = True
self.save_pcap_logs: bool = False
self.save_sys_logs: bool = False
@property
def path(self) -> Path:

67
src/primaite/simulator/network/hardware/nodes/network/firewall.py
View File

@@ -1,8 +1,10 @@
from ipaddress import IPv4Address
from typing import Dict, Final, Optional, Union
from prettytable import MARKDOWN, PrettyTable
from pydantic import validate_call
from primaite.simulator.network.hardware.node_operating_state import NodeOperatingState
from primaite.simulator.network.hardware.nodes.network.router import (
AccessControlList,
ACLAction,
@@ -491,3 +493,68 @@ class Firewall(Router):
"""
self.configure_port(DMZ_PORT_ID, ip_address, subnet_mask)
self.dmz_port.enable()
@classmethod
def from_config(cls, cfg: dict) -> "Firewall":
"""Create a firewall based on a config dict."""
new = Firewall(hostname=cfg["hostname"], operating_state=NodeOperatingState.ON)
if "ports" in cfg:
internal_port = cfg["ports"]["internal_port"]
external_port = cfg["ports"]["external_port"]
dmz_port = cfg["ports"]["dmz_port"]
# configure internal port
new.configure_internal_port(
ip_address=IPV4Address(internal_port.get("ip_address")),
subnet_mask=IPV4Address(internal_port.get("subnet_mask")),
)
# configure external port
new.configure_external_port(
ip_address=IPV4Address(external_port.get("ip_address")),
subnet_mask=IPV4Address(external_port.get("subnet_mask")),
)
# configure dmz port
new.configure_dmz_port(
ip_address=IPV4Address(dmz_port.get("ip_address")), subnet_mask=IPV4Address(dmz_port.get("subnet_mask"))
)
if "acl" in cfg:
# acl rules for internal_inbound_acl
if cfg["acl"]["internal_inbound_acl"]:
new.internal_inbound_acl._default_config = cfg["acl"]["internal_inbound_acl"]
new.internal_inbound_acl._reset_rules_to_default()
# acl rules for internal_outbound_acl
if cfg["acl"]["internal_outbound_acl"]:
new.internal_outbound_acl._default_config = cfg["acl"]["internal_outbound_acl"]
new.internal_outbound_acl._reset_rules_to_default()
# acl rules for dmz_inbound_acl
if cfg["acl"]["dmz_inbound_acl"]:
new.dmz_inbound_acl._default_config = cfg["acl"]["dmz_inbound_acl"]
new.dmz_inbound_acl._reset_rules_to_default()
# acl rules for dmz_outbound_acl
if cfg["acl"]["dmz_outbound_acl"]:
new.dmz_outbound_acl._default_config = cfg["acl"]["dmz_outbound_acl"]
new.dmz_outbound_acl._reset_rules_to_default()
# acl rules for external_inbound_acl
if cfg["acl"]["external_inbound_acl"]:
new.external_inbound_acl._default_config = cfg["acl"]["external_inbound_acl"]
new.external_inbound_acl._reset_rules_to_default()
# acl rules for external_outbound_acl
if cfg["acl"]["external_outbound_acl"]:
new.external_outbound_acl._default_config = cfg["acl"]["external_outbound_acl"]
new.external_outbound_acl._reset_rules_to_default()
if "routes" in cfg:
for route in cfg.get("routes"):
new.route_table.add_route(
address=IPv4Address(route.get("address")),
subnet_mask=IPv4Address(route.get("subnet_mask")),
next_hop_ip_address=IPv4Address(route.get("next_hop_ip_address")),
metric=float(route.get("metric")),
)
return new

2
src/primaite/simulator/network/hardware/nodes/network/router.py
View File

@@ -1500,7 +1500,7 @@ class Router(NetworkNode):
new.route_table.add_route(
address=IPv4Address(route.get("address")),
subnet_mask=IPv4Address(route.get("subnet_mask")),
next_hop_ip_address=IPv4Address(route.get("subnet_mask")),
next_hop_ip_address=IPv4Address(route.get("next_hop_ip_address")),
metric=float(route.get("metric")),
)
return new

122
tests/assets/configs/dmz_network.yaml
View File

@@ -6,19 +6,19 @@
# . .
# . Internal Network .
# . .
# . -------------- -------------- -------------- .
# . | client_1 |------| switch_1 |------| router_1 | .
# . -------------- -------------- -------------- .
# . -------------- -------------- -------------- .
# . | client_1 |------| switch_1 |--------| router_1 | .
# . -------------- -------------- -------------- .
# . (Computer) | .
# ........................................................|.....................
# ........................................................|...................
# |
# |
# ........................................................|.....................
# ........................................................|...................
# . | .
# . DMZ Network | .
# . | .
# . ---------------- -------------- -------------- .
# . | dmz_server |------| switch_2 |------| router_2 | .
# . | dmz_server |------| switch_2 |------| firewall | .
# . ---------------- -------------- -------------- .
# . (Computer) | .
# ........................................................|...................
@@ -135,17 +135,17 @@ simulation:
action: PERMIT
protocol: ICMP
routes:
- address: 192.168.10.10
- address: 192.168.10.10 # route to dmz_server
subnet_mask: 255.255.255.0
next_hop_ip_address: 192.168.11.1
next_hop_ip_address: 192.168.1.2
metric: 0
- address: 192.168.20.10
- address: 192.168.20.10 # route to external_computer
subnet_mask: 255.255.255.0
next_hop_ip_address: 192.168.11.1
next_hop_ip_address: 192.168.1.2
metric: 0
- address: 192.168.20.11
- address: 192.168.20.11 # route to external_server
subnet_mask: 255.255.255.0
next_hop_ip_address: 192.168.11.1
next_hop_ip_address: 192.168.1.2
metric: 0
- ref: dmz_server
@@ -165,32 +165,72 @@ simulation:
start_up_duration: 0
shut_down_duration: 0
- ref: router_2
type: router
hostname: router_2
num_ports: 5
- ref: firewall
type: firewall
hostname: firewall
start_up_duration: 0
shut_down_duration: 0
ports:
1:
ip_address: 192.168.10.1
subnet_mask: 255.255.255.0
2:
ip_address: 192.168.11.1
subnet_mask: 255.255.255.0
3:
external_port: # port 1
ip_address: 192.168.20.1
subnet_mask: 255.255.255.0
internal_port: # port 2
ip_address: 192.168.1.2
subnet_mask: 255.255.255.0
dmz_port: # port 3
ip_address: 192.168.10.1
subnet_mask: 255.255.255.0
acl:
22:
action: PERMIT
src_port: ARP
dst_port: ARP
23:
action: PERMIT
protocol: ICMP
internal_inbound_acl:
22:
action: PERMIT
src_port: ARP
dst_port: ARP
23:
action: PERMIT
protocol: ICMP
internal_outbound_acl:
22:
action: PERMIT
src_port: ARP
dst_port: ARP
23:
action: PERMIT
protocol: ICMP
dmz_inbound_acl:
22:
action: PERMIT
src_port: ARP
dst_port: ARP
23:
action: PERMIT
protocol: ICMP
dmz_outbound_acl:
22:
action: PERMIT
src_port: ARP
dst_port: ARP
23:
action: PERMIT
protocol: ICMP
external_inbound_acl:
22:
action: PERMIT
src_port: ARP
dst_port: ARP
23:
action: PERMIT
protocol: ICMP
external_outbound_acl:
22:
action: PERMIT
src_port: ARP
dst_port: ARP
23:
action: PERMIT
protocol: ICMP
routes:
- address: 192.168.0.10
- address: 192.168.0.10 # route to client_1
subnet_mask: 255.255.255.0
next_hop_ip_address: 192.168.1.1
metric: 0
@@ -234,14 +274,14 @@ simulation:
endpoint_a_port: 1
endpoint_b_ref: switch_1
endpoint_b_port: 8
- ref: router_1___router_2
endpoint_a_ref: router_1
endpoint_a_port: 2
endpoint_b_ref: router_2
- ref: router_1___firewall
endpoint_a_ref: firewall
endpoint_a_port: 2 # internal firewall port
endpoint_b_ref: router_1
endpoint_b_port: 2
- ref: router_2___switch_2
endpoint_a_ref: router_2
endpoint_a_port: 1
- ref: firewall___switch_2
endpoint_a_ref: firewall
endpoint_a_port: 3 # dmz firewall port
endpoint_b_ref: switch_2
endpoint_b_port: 8
- ref: dmz_server___switch_2
@@ -249,9 +289,9 @@ simulation:
endpoint_a_port: 1
endpoint_b_ref: switch_2
endpoint_b_port: 1
- ref: router_2___switch_3
endpoint_a_ref: router_2
endpoint_a_port: 3
- ref: firewall___switch_3
endpoint_a_ref: firewall
endpoint_a_port: 1 # external firewall port
endpoint_b_ref: switch_3
endpoint_b_port: 8
- ref: external_computer___switch_3

19
tests/conftest.py
View File

@@ -1,9 +1,11 @@
# © Crown-owned copyright 2023, Defence Science and Technology Laboratory UK
from datetime import datetime
from pathlib import Path
from typing import Any, Dict, Tuple, Union
import pytest
import yaml
from _pytest.monkeypatch import MonkeyPatch
from primaite import getLogger, PRIMAITE_PATHS
from primaite.game.agent.actions import ActionManager
@@ -12,6 +14,7 @@ from primaite.game.agent.observations import ICSObservation, ObservationManager
from primaite.game.agent.rewards import RewardFunction
from primaite.game.game import PrimaiteGame
from primaite.session.session import PrimaiteSession
from primaite.simulator import SIM_OUTPUT
from primaite.simulator.file_system.file_system import FileSystem
from primaite.simulator.network.container import Network
from primaite.simulator.network.hardware.nodes.host.computer import Computer
@@ -29,6 +32,7 @@ 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 Service
from primaite.simulator.system.services.web_server.web_server import WebServer
from tests import TEST_ASSETS_ROOT
from tests.mock_and_patch.get_session_path_mock import temp_user_sessions_path
ACTION_SPACE_NODE_VALUES = 1
@@ -37,6 +41,21 @@ ACTION_SPACE_NODE_ACTION_VALUES = 1
_LOGGER = getLogger(__name__)
@pytest.fixture(scope="function", autouse=True)
def set_syslog_output_to_true():
"""Will be run before each test."""
monkeypatch = MonkeyPatch()
monkeypatch.setattr(
SIM_OUTPUT,
"path",
Path(TEST_ASSETS_ROOT.parent.parent / "simulation_output" / datetime.now().strftime("%Y-%m-%d_%H-%M-%S")),
)
monkeypatch.setattr(SIM_OUTPUT, "save_pcap_logs", True)
monkeypatch.setattr(SIM_OUTPUT, "save_sys_logs", True)
yield
class TestService(Service):
"""Test Service class"""

19
tests/integration_tests/configuration_file_parsing/__init__.py
View File

@@ -0,0 +1,19 @@
from pathlib import Path
from typing import Union
import yaml
from primaite.game.game import PrimaiteGame
from tests import TEST_ASSETS_ROOT
BASIC_CONFIG = TEST_ASSETS_ROOT / "configs/basic_switched_network.yaml"
DMZ_NETWORK = TEST_ASSETS_ROOT / "configs/dmz_network.yaml"
def load_config(config_path: Union[str, Path]) -> PrimaiteGame:
"""Returns a PrimaiteGame object which loads the contents of a given yaml path."""
with open(config_path, "r") as f:
cfg = yaml.safe_load(f)
return PrimaiteGame.from_config(cfg)

0
tests/integration_tests/configuration_file_parsing/nodes/__init__.py Normal file
View File

0
tests/integration_tests/configuration_file_parsing/nodes/network/__init__.py Normal file
View File

45
tests/integration_tests/configuration_file_parsing/nodes/network/test_firewall_config.py Normal file
View File

@@ -0,0 +1,45 @@
import pytest
from primaite.simulator.network.container import Network
from primaite.simulator.network.hardware.nodes.host.computer import Computer
from primaite.simulator.network.hardware.nodes.host.server import Server
from primaite.simulator.network.hardware.nodes.network.firewall import Firewall
from tests.integration_tests.configuration_file_parsing import DMZ_NETWORK, load_config
@pytest.fixture(scope="function")
def dmz_config() -> Network:
game = load_config(DMZ_NETWORK)
return game.simulation.network
def test_firewall_is_in_configuration(dmz_config):
"""Test that the firewall exists in the configuration file."""
network: Network = dmz_config
assert network.get_node_by_hostname("firewall")
def test_firewall_routes_are_correctly_added(dmz_config):
"""Test that the firewall routes have been correctly added to and configured in the network."""
network: Network = dmz_config
firewall: Firewall = network.get_node_by_hostname("firewall")
client_1: Computer = network.get_node_by_hostname("client_1")
dmz_server: Server = network.get_node_by_hostname("dmz_server")
external_computer: Computer = network.get_node_by_hostname("external_computer")
external_server: Server = network.get_node_by_hostname("external_server")
# there should be a route to client_1
assert firewall.route_table.find_best_route(client_1.network_interface[1].ip_address)
assert dmz_server.ping(client_1.network_interface[1].ip_address)
assert external_computer.ping(client_1.network_interface[1].ip_address)
assert external_server.ping(client_1.network_interface[1].ip_address)
def test_firewall_acl_rules_correctly_added():
"""
Test that makes sure that the firewall ACLs have been configured onto the firewall
node via configuration file.
"""
pass

54
tests/integration_tests/configuration_file_parsing/nodes/network/test_router_config.py Normal file
View File

@@ -0,0 +1,54 @@
import pytest
from primaite.simulator.network.container import Network
from primaite.simulator.network.hardware.nodes.host.computer import Computer
from primaite.simulator.network.hardware.nodes.host.server import Server
from primaite.simulator.network.hardware.nodes.network.router import Router
from tests.integration_tests.configuration_file_parsing import DMZ_NETWORK, load_config
@pytest.fixture(scope="function")
def dmz_config() -> Network:
game = load_config(DMZ_NETWORK)
return game.simulation.network
def test_router_is_in_configuration(dmz_config):
"""Test that the router exists in the configuration file."""
network: Network = dmz_config
assert network.get_node_by_hostname("router_1")
def test_router_routes_are_correctly_added(dmz_config):
"""Test that makes sure that router routes have been added from the configuration file."""
network: Network = dmz_config
router_1: Router = network.get_node_by_hostname("router_1")
client_1: Computer = network.get_node_by_hostname("client_1")
dmz_server: Server = network.get_node_by_hostname("dmz_server")
external_computer: Computer = network.get_node_by_hostname("external_computer")
external_server: Server = network.get_node_by_hostname("external_server")
# there should be a route to dmz_server
assert router_1.route_table.find_best_route(dmz_server.network_interface[1].ip_address)
assert client_1.ping(dmz_server.network_interface[1].ip_address)
assert external_computer.ping(dmz_server.network_interface[1].ip_address)
assert external_server.ping(dmz_server.network_interface[1].ip_address)
# there should be a route to external_computer
assert router_1.route_table.find_best_route(external_computer.network_interface[1].ip_address)
assert client_1.ping(external_computer.network_interface[1].ip_address)
assert dmz_server.ping(external_computer.network_interface[1].ip_address)
assert external_server.ping(external_computer.network_interface[1].ip_address)
# there should be a route to external_server
assert router_1.route_table.find_best_route(external_server.network_interface[1].ip_address)
assert client_1.ping(external_server.network_interface[1].ip_address)
assert dmz_server.ping(external_server.network_interface[1].ip_address)
assert external_computer.ping(external_server.network_interface[1].ip_address)
def test_router_acl_rules_correctly_added():
"""Test that makes sure that the router ACLs have been configured onto the router node via configuration file."""
pass

26
tests/integration_tests/configuration_file_parsing/nodes/test_node_config.py Normal file
View File

@@ -0,0 +1,26 @@
from primaite.config.load import example_config_path
from primaite.simulator.network.container import Network
from tests.integration_tests.configuration_file_parsing import DMZ_NETWORK, load_config
def test_example_config():
"""Test that the example config can be parsed properly."""
game = load_config(example_config_path())
network: Network = game.simulation.network
assert len(network.nodes) == 10 # 10 nodes in example network
assert len(network.routers) == 1 # 1 router in network
assert len(network.switches) == 2 # 2 switches in network
assert len(network.servers) == 5 # 5 servers in network
def test_dmz_config():
"""Test that the DMZ network config can be parsed properly."""
game = load_config(DMZ_NETWORK)
network: Network = game.simulation.network
assert len(network.nodes) == 9 # 9 nodes in network
assert len(network.routers) == 2 # 2 routers in network
assert len(network.switches) == 3 # 3 switches in network
assert len(network.servers) == 2 # 2 servers in network

76
tests/integration_tests/configuration_file_parsing/router_game_configuration.py
View File

@@ -1,76 +0,0 @@
from pathlib import Path
from typing import Union
import yaml
from primaite.game.game import PrimaiteGame
from primaite.simulator.network.container import Network
from primaite.simulator.network.hardware.nodes.host.computer import Computer
from primaite.simulator.network.hardware.nodes.host.server import Server
from primaite.simulator.network.hardware.nodes.network.router import Router
from tests import TEST_ASSETS_ROOT
DMZ_NETWORK = TEST_ASSETS_ROOT / "configs/dmz_network.yaml"
def load_config(config_path: Union[str, Path]) -> PrimaiteGame:
"""Returns a PrimaiteGame object which loads the contents of a given yaml path."""
with open(config_path, "r") as f:
cfg = yaml.safe_load(f)
return PrimaiteGame.from_config(cfg)
def test_dmz_config():
"""Test that the DMZ network config can be parsed properly."""
game = load_config(DMZ_NETWORK)
network: Network = game.simulation.network
assert len(network.nodes) == 9 # 9 nodes in network
assert len(network.routers) == 2 # 2 routers in network
assert len(network.switches) == 3 # 3 switches in network
assert len(network.servers) == 2 # 2 servers in network
def test_router_routes_are_correctly_added():
"""Test that makes sure that router routes have been added from the configuration file."""
game = load_config(DMZ_NETWORK)
network: Network = game.simulation.network
router_1: Router = network.get_node_by_hostname("router_1")
client_1: Computer = network.get_node_by_hostname("client_1")
dmz_server: Server = network.get_node_by_hostname("dmz_server")
external_computer: Computer = network.get_node_by_hostname("external_computer")
external_server: Server = network.get_node_by_hostname("external_server")
# test that client_1 has a route to the DMZ and external nodes - they are on a second router
# there should be a route to the dmz server
assert router_1.route_table.find_best_route(dmz_server.network_interface[1].ip_address)
# ping DMZ server
# assert client_1.ping(dmz_server.network_interface[1].ip_address)
def test_firewall_node_added_to_network():
"""Test that the firewall has been correctly added to and configured in the network."""
pass
def test_router_acl_rules_correctly_added():
"""Test that makes sure that the router ACLs have been configured onto the router node via configuration file."""
pass
def test_firewall_routes_are_correctly_added():
"""Test that the firewall routes have been correctly added to and configured in the network."""
pass
def test_firewall_acl_rules_correctly_added():
"""
Test that makes sure that the firewall ACLs have been configured onto the firewall
node via configuration file.
"""
pass

52
tests/integration_tests/configuration_file_parsing/software_installation_and_configuration.py
View File

@@ -1,14 +1,6 @@
from ipaddress import IPv4Address
from pathlib import Path
from typing import Union
import yaml
from primaite.config.load import example_config_path
from primaite.game.agent.data_manipulation_bot import DataManipulationAgent
from primaite.game.agent.interface import ProxyAgent, RandomAgent
from primaite.game.game import APPLICATION_TYPES_MAPPING, PrimaiteGame, SERVICE_TYPES_MAPPING
from primaite.simulator.network.container import Network
from primaite.game.game import APPLICATION_TYPES_MAPPING, SERVICE_TYPES_MAPPING
from primaite.simulator.network.hardware.nodes.host.computer import Computer
from primaite.simulator.system.applications.database_client import DatabaseClient
from primaite.simulator.system.applications.red_applications.data_manipulation_bot import DataManipulationBot
@@ -22,47 +14,7 @@ from primaite.simulator.system.services.ftp.ftp_server import FTPServer
from primaite.simulator.system.services.ntp.ntp_client import NTPClient
from primaite.simulator.system.services.ntp.ntp_server import NTPServer
from primaite.simulator.system.services.web_server.web_server import WebServer
from tests import TEST_ASSETS_ROOT
BASIC_CONFIG = TEST_ASSETS_ROOT / "configs/basic_switched_network.yaml"
def load_config(config_path: Union[str, Path]) -> PrimaiteGame:
"""Returns a PrimaiteGame object which loads the contents of a given yaml path."""
with open(config_path, "r") as f:
cfg = yaml.safe_load(f)
return PrimaiteGame.from_config(cfg)
def test_example_config():
"""Test that the example config can be parsed properly."""
game = load_config(example_config_path())
assert len(game.agents) == 4 # red, blue and 2 green agents
# green agent 1
assert game.agents[0].agent_name == "client_2_green_user"
assert isinstance(game.agents[0], RandomAgent)
# green agent 2
assert game.agents[1].agent_name == "client_1_green_user"
assert isinstance(game.agents[1], RandomAgent)
# red agent
assert game.agents[2].agent_name == "client_1_data_manipulation_red_bot"
assert isinstance(game.agents[2], DataManipulationAgent)
# blue agent
assert game.agents[3].agent_name == "defender"
assert isinstance(game.agents[3], ProxyAgent)
network: Network = game.simulation.network
assert len(network.nodes) == 10 # 10 nodes in example network
assert len(network.routers) == 1 # 1 router in network
assert len(network.switches) == 2 # 2 switches in network
assert len(network.servers) == 5 # 5 servers in network
from tests.integration_tests.configuration_file_parsing import BASIC_CONFIG, load_config
def test_node_software_install():