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

Merged PR 504: Command and Control Full PR

Browse Source 
## Summary
Implements the Command and Control applications to the quality and capability needed for the TAP001 expansion and lays the foundations for all the features required for TAP002 (Next Release).

The C2C suite contains three new applications:

**1. Abstract C2**

Base class for the C2 Server and the C2 Beacon. Controls the main internal logic of both applications but with a couple of abstract methods which each class defines differently.

**2. C2 Server**

The C2 Server takes red agent actions and converts the action options into C2 Commands which are then passed to the C2 Beacon.
The output of these commands is sent back to the C2 Server and then returned back to the red agent.

**3. C2 Beacon**

The C2 beacon uses the Terminal and the Ransomware Applications to perform different commands which it receives these commands and executes them and returns the output.

The C2 beacon can also be configured by the Red Agent to configure the current networking behaviour.

For a much more detailed description please refer to the .rst documentation and the notebook which demonstrate and describe the functionality very explicitly.

Lastly the wiki page also provides more information around the design work for this feature.

[Command and Control](/Welcome-to-PrimAITE!/Design/[~In-Progress~]/Command-and-Control)

Worth noting that some changes were needed that were unseen during the design page but the overall goals of the feature have been accomplished.

## Test process
Tested via notebooks and a series of e2e tests.

## Checklist
- [x] PR is linked to a **work item**
- [x] **acceptance criteria** of linked ticket are met
- [x] performed **self-review** of the code
- [x] written **tests** for any new functionality added with this PR
- [x] updated the **documentation** if this PR changes or adds functionality
- [x] written/updated **design docs** if this PR implements new functionality
- [x] updated the **change log**
- [x] ran **pre-commit** checks for code style
- [x] attended to any **TO-DOs** left in the code (One remaining but unsure if it should be handled in this PR)

Related work items: #2689, #2720, #2721, #2779
This commit is contained in:
Archer Bowen
2024-08-20 13:16:22 +00:00
parent 23194e1ba8 c9d62d512c
commit bceedbf72d
22 changed files with 5199 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
CHANGELOG.md
View File

@@ -14,6 +14,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
- Added actions to change users' passwords.
- Added a `listen_on_ports` set in the `IOSoftware` class to enable software listening on ports in addition to the
main port they're assigned.
- Added two new red applications: ``C2Beacon`` and ``C2Server`` which aim to simulate malicious network infrastructure.
Refer to the ``Command and Control Application Suite E2E Demonstration`` notebook for more information.
- Added reward calculation details to AgentHistoryItem.
### Changed

BIN
docs/_static/c2_sequence.png vendored Normal file
View File

Binary file not shown.
Side by Side

After

Width:  |  Height:  |  Size: 54 KiB

319
docs/source/simulation_components/system/applications/c2_suite.rst Normal file
View File

@@ -0,0 +1,319 @@
.. only:: comment
© Crown-owned copyright 2024, Defence Science and Technology Laboratory UK
.. _C2_Suite:
Command and Control Application Suite
#####################################
Comprising of two applications, the Command and Control (C2) suites intends to introduce
malicious network architecture and begin to further the realism of red agents within primAITE.
Overview:
=========
These two new classes give red agents a cyber realistic way of leveraging the capabilities of the ``Terminal`` application whilst introducing more opportunities for the blue agent(s) to notice and subvert a red agent during an episode.
For a more in-depth look at the command and control applications then please refer to the ``C2-E2E-Notebook``.
``C2 Server``
"""""""""""""
The C2 Server application is intended to represent the malicious infrastructure already under the control of an adversary.
The C2 Server is configured to listen and await ``keep alive`` traffic from a C2 beacon. Once received the C2 Server is able to send and receive C2 commands.
Currently, the C2 Server offers three commands:
+---------------------+---------------------------------------------------------------------------+
|C2 Command | Meaning |
+=====================+===========================================================================+
|RANSOMWARE_CONFIGURE | Configures an installed ransomware script based on the passed parameters. |
+---------------------+---------------------------------------------------------------------------+
|RANSOMWARE_LAUNCH | Launches the installed ransomware script. |
+---------------------+---------------------------------------------------------------------------+
|DATA_EXFILTRATION | Copies a target file from a remote node to the C2 Beacon & Server via FTP |
+---------------------+---------------------------------------------------------------------------+
|TERMINAL | Executes a command via the terminal installed on the C2 Beacons Host. |
+---------------------+---------------------------------------------------------------------------+
It's important to note that in order to keep PrimAITE realistic from a cyber perspective,
The C2 Server application should never be visible or actionable upon directly by the blue agent.
This is because in the real world, C2 servers are hosted on ephemeral public domains that would not be accessible by private network blue agent.
Therefore granting blue agent(s) the ability to perform counter measures directly against the application would be unrealistic.
It is more accurate to see the host that the C2 Server is installed on as being able to route to the C2 Server (Internet Access).
``C2 Beacon``
"""""""""""""
The C2 Beacon application is intended to represent malware that is used to establish and maintain contact to a C2 Server within a compromised network.
A C2 Beacon will need to be first configured with the C2 Server IP Address which can be done via the ``configure`` method.
Once installed and configured; the c2 beacon can establish connection with the C2 Server via executing the application.
This will send an initial ``keep alive`` to the given C2 Server (The C2 Server IPv4Address must be given upon C2 Beacon configuration).
Which is then resolved and responded by another ``Keep Alive`` by the c2 server back to the C2 beacon to confirm connection.
The C2 Beacon will send out periodic keep alive based on it's configuration parameters to configure it's active connection with the c2 server.
It's recommended that a C2 Beacon is installed and configured mid episode by a Red Agent for a more cyber realistic simulation.
Usage
=====
As mentioned, the C2 Suite is intended to grant Red Agents further flexibility whilst also expanding a blue agent's observation_space.
Adding to this, the following behaviour of the C2 beacon can be configured by users for increased domain randomisation:
+---------------------+---------------------------------------------------------------------------+
|Configuration Option | Option Meaning |
+=====================+===========================================================================+
|c2_server_ip_address | The IP Address of the C2 Server. (The C2 Server must be running) |
+---------------------+---------------------------------------------------------------------------+
|keep_alive_frequency | How often should the C2 Beacon confirm it's connection in timesteps. |
+---------------------+---------------------------------------------------------------------------+
|masquerade_protocol | What protocol should the C2 traffic masquerade as? (HTTP, FTP or DNS) |
+---------------------+---------------------------------------------------------------------------+
|masquerade_port | What port should the C2 traffic use? (TCP or UDP) |
+---------------------+---------------------------------------------------------------------------+
Implementation
==============
Both applications inherit from an abstract C2 which handles the keep alive functionality and main logic.
However, each host implements it's own receive methods.
- The ``C2 Beacon`` is responsible for the following logic:
- Establishes and confirms connection to the C2 Server via sending ``C2Payload.KEEP_ALIVE``.
- Receives and executes C2 Commands given by the C2 Server via ``C2Payload.INPUT``.
- Returns the RequestResponse of the C2 Commands executed back the C2 Server via ``C2Payload.OUTPUT``.
- The ``C2 Server`` is responsible for the following logic:
- Listens and resolves connection to a C2 Beacon via responding to ``C2Payload.KEEP_ALIVE``.
- Sends C2 Commands to the C2 Beacon via ``C2Payload.INPUT``.
- Receives the RequestResponse of the C2 Commands executed by C2 Beacon via ``C2Payload.OUTPUT``.
The sequence diagram below clarifies the functionality of both applications:
.. image:: ../../../../_static/c2_sequence.png
:width: 1000
:align: center
For further details and more in-depth examples please refer to the ``Command-&-Control notebook``
Examples
========
Python
""""""
.. code-block:: python
from primaite.simulator.network.container import Network
from primaite.simulator.network.hardware.nodes.host.computer import Computer
from primaite.simulator.network.hardware.nodes.network.switch import Switch
from primaite.simulator.system.applications.database_client import DatabaseClient
from primaite.simulator.system.applications.red_applications.ransomware_script import RansomwareScript
from primaite.simulator.system.services.database.database_service import DatabaseService
from primaite.simulator.system.applications.red_applications.c2.c2_server import C2Command, C2Server
from primaite.simulator.system.applications.red_applications.c2.c2_beacon import C2Beacon
# Network Setup
network = Network()
switch = Switch(hostname="switch", start_up_duration=0, num_ports=4)
switch.power_on()
node_a = Computer(hostname="node_a", ip_address="192.168.0.10", subnet_mask="255.255.255.0", start_up_duration=0)
node_a.power_on()
network.connect(node_a.network_interface[1], switch.network_interface[1])
node_b = Computer(hostname="node_b", ip_address="192.168.0.11", subnet_mask="255.255.255.0", start_up_duration=0)
node_b.power_on()
network.connect(node_b.network_interface[1], switch.network_interface[2])
node_c = Computer(hostname="node_c", ip_address="192.168.0.12", subnet_mask="255.255.255.0", start_up_duration=0)
node_c.power_on()
network.connect(node_c.network_interface[1], switch.network_interface[3])
node_c.software_manager.install(software_class=DatabaseService)
node_b.software_manager.install(software_class=DatabaseClient)
node_b.software_manager.install(software_class=RansomwareScript)
node_a.software_manager.install(software_class=C2Server)
# C2 Application objects
c2_server_host: Computer = network.get_node_by_hostname("node_a")
c2_beacon_host: Computer = network.get_node_by_hostname("node_b")
c2_server: C2Server = c2_server_host.software_manager.software["C2Server"]
c2_beacon: C2Beacon = c2_beacon_host.software_manager.software["C2Beacon"]
# Configuring the C2 Beacon
c2_beacon.configure(c2_server_ip_address="192.168.0.10", keep_alive_frequency=5)
# Launching the C2 Server (Needs to be running in order to listen for connections)
c2_server.run()
# Establishing connection
c2_beacon.establish()
# Example command: Creating a file
file_create_command = {
"commands": [
["file_system", "create", "folder", "test_folder"],
["file_system", "create", "file", "test_folder", "example_file", "True"],
],
"username": "admin",
"password": "admin",
"ip_address": None,
}
c2_server.send_command(C2Command.TERMINAL, command_options=file_create_command)
# Example command: Installing and configuring Ransomware:
ransomware_installation_command = { "commands": [
["software_manager","application","install","RansomwareScript"],
],
"username": "admin",
"password": "admin",
"ip_address": None,
}
c2_server.send_command(given_command=C2Command.TERMINAL, command_options=ransomware_installation_command)
ransomware_config = {"server_ip_address": "192.168.0.12"}
c2_server.send_command(given_command=C2Command.RANSOMWARE_CONFIGURE, command_options=ransomware_config)
c2_beacon_host.software_manager.show()
# Example command: File Exfiltration
data_exfil_options = {
"username": "admin",
"password": "admin",
"ip_address": None,
"target_ip_address": "192.168.0.12",
"target_file_name": "database.db",
"target_folder_name": "database",
}
c2_server.send_command(given_command=C2Command.DATA_EXFILTRATION, command_options=data_exfil_options)
# Example command: Launching Ransomware
c2_server.send_command(given_command=C2Command.RANSOMWARE_LAUNCH, command_options={})
Via Configuration
"""""""""""""""""
.. code-block:: yaml
simulation:
network:
nodes:
- ref: example_computer_1
hostname: computer_a
type: computer
...
applications:
type: C2Server
...
hostname: computer_b
type: computer
...
# A C2 Beacon will not automatically connection to a C2 Server.
# Either an agent must use application_execute.
# Or a if using the simulation layer - .establish().
applications:
type: C2Beacon
options:
c2_server_ip_address: ...
keep_alive_frequency: 5
masquerade_protocol: tcp
masquerade_port: http
listen_on_ports:
- 80
- 53
- 21
C2 Beacon Configuration
=======================
.. include:: ../common/common_configuration.rst
.. |SOFTWARE_NAME| replace:: C2Beacon
.. |SOFTWARE_NAME_BACKTICK| replace:: ``C2Beacon``
``c2_server_ip_address``
""""""""""""""""""""""""
IP address of the ``C2Server`` that the C2 Beacon will use to establish connection.
This must be a valid octet i.e. in the range of ``0.0.0.0`` and ``255.255.255.255``.
``Keep Alive Frequency``
""""""""""""""""""""""""
How often should the C2 Beacon confirm it's connection in timesteps.
For example, if the keep alive Frequency is set to one then every single timestep
the C2 connection will be confirmed.
It's worth noting that this may be useful option when investigating
network blue agent observation space.
This must be a valid integer i.e ``10``. Defaults to ``5``.
``Masquerade Protocol``
"""""""""""""""""""""""
The protocol that the C2 Beacon will use to communicate to the C2 Server with.
Currently only ``TCP`` and ``UDP`` are valid masquerade protocol options.
It's worth noting that this may be useful option to bypass ACL rules.
This must be a string i.e *UDP*. Defaults to ``TCP``.
*Please refer to the ``IPProtocol`` class for further reference.*
``Masquerade Port``
"""""""""""""""""""
What port that the C2 Beacon will use to communicate to the C2 Server with.
Currently only ``FTP``, ``HTTP`` and ``DNS`` are valid masquerade port options.
It's worth noting that this may be useful option to bypass ACL rules.
This must be a string i.e ``DNS``. Defaults to ``HTTP``.
*Please refer to the ``IPProtocol`` class for further reference.*
C2 Server Configuration
=======================
*The C2 Server does not currently offer any unique configuration options and will configure itself to match the C2 beacon's network behaviour.*
.. include:: ../common/common_configuration.rst
.. |SOFTWARE_NAME| replace:: C2Server
.. |SOFTWARE_NAME_BACKTICK| replace:: ``C2Server``

17
docs/source/simulation_components/system/services/terminal.rst
View File

@@ -5,26 +5,26 @@
.. _Terminal:
Terminal
========
########
The ``Terminal.py`` class provides a generic terminal simulation, by extending the base Service class within PrimAITE. The aim of this is to act as the primary entrypoint for Nodes within the environment.
Overview
--------
========
The Terminal service uses Secure Socket (SSH) as the communication method between terminals. They operate on port 22, and are part of the services automatically
installed on Nodes when they are instantiated.
Key capabilities
================
""""""""""""""""
- Ensures packets are matched to an existing session
- Simulates common Terminal processes/commands.
- Leverages the Service base class for install/uninstall, status tracking etc.
Usage
=====
"""""
- Pre-Installs on any `Node` (component with the exception of `Switches`).
- Terminal Clients connect, execute commands and disconnect from remote nodes.
@@ -32,7 +32,7 @@ Usage
- Service runs on SSH port 22 by default.
Implementation
==============
""""""""""""""
- Manages remote connections in a dictionary by session ID.
- Processes commands, forwarding to the ``RequestManager`` or ``SessionManager`` where appropriate.
@@ -67,7 +67,7 @@ Python
terminal: Terminal = client.software_manager.software.get("Terminal")
Creating Remote Terminal Connection
"""""""""""""""""""""""""""
"""""""""""""""""""""""""""""""""""
.. code-block:: python
@@ -93,7 +93,7 @@ Creating Remote Terminal Connection
Executing a basic application install command
"""""""""""""""""""""""""""""""""
"""""""""""""""""""""""""""""""""""""""""""""
.. code-block:: python
@@ -121,7 +121,7 @@ Executing a basic application install command
Creating a folder on a remote node
""""""""""""""""""""""""""""""""
""""""""""""""""""""""""""""""""""
.. code-block:: python
@@ -148,6 +148,7 @@ Creating a folder on a remote node
Disconnect from Remote Node
"""""""""""""""""""""""""""
.. code-block:: python

169
src/primaite/game/agent/actions.py
View File

@@ -1071,6 +1071,51 @@ class NodeNetworkServiceReconAction(AbstractAction):
]
class ConfigureC2BeaconAction(AbstractAction):
"""Action which configures a C2 Beacon based on the parameters given."""
class _Opts(BaseModel):
"""Schema for options that can be passed to this action."""
c2_server_ip_address: str
keep_alive_frequency: int = Field(default=5, ge=1)
masquerade_protocol: str = Field(default="TCP")
masquerade_port: str = Field(default="HTTP")
@field_validator(
"c2_server_ip_address",
"keep_alive_frequency",
"masquerade_protocol",
"masquerade_port",
mode="before",
)
@classmethod
def not_none(cls, v: str, info: ValidationInfo) -> int:
"""If None is passed, use the default value instead."""
if v is None:
return cls.model_fields[info.field_name].default
return v
def __init__(self, manager: "ActionManager", **kwargs) -> None:
super().__init__(manager=manager)
def form_request(self, node_id: int, config: Dict) -> RequestFormat:
"""Return the action formatted as a request that can be ingested by the simulation."""
node_name = self.manager.get_node_name_by_idx(node_id)
if node_name is None:
return ["do_nothing"]
config = ConfigureC2BeaconAction._Opts(
c2_server_ip_address=config["c2_server_ip_address"],
keep_alive_frequency=config["keep_alive_frequency"],
masquerade_port=config["masquerade_port"],
masquerade_protocol=config["masquerade_protocol"],
)
ConfigureC2BeaconAction._Opts.model_validate(config) # check that options adhere to schema
return ["network", "node", node_name, "application", "C2Beacon", "configure", config.__dict__]
class NodeAccountsChangePasswordAction(AbstractAction):
"""Action which changes the password for a user."""
@@ -1127,6 +1172,79 @@ class NodeSessionsRemoteLogoutAction(AbstractAction):
return ["network", "node", node_name, "service", "Terminal", "remote_logoff", remote_ip]
class RansomwareConfigureC2ServerAction(AbstractAction):
"""Action which sends a command from the C2 Server to the C2 Beacon which configures a local RansomwareScript."""
def __init__(self, manager: "ActionManager", **kwargs) -> None:
super().__init__(manager=manager)
def form_request(self, node_id: int, config: Dict) -> RequestFormat:
"""Return the action formatted as a request that can be ingested by the simulation."""
node_name = self.manager.get_node_name_by_idx(node_id)
if node_name is None:
return ["do_nothing"]
# Using the ransomware scripts model to validate.
ConfigureRansomwareScriptAction._Opts.model_validate(config) # check that options adhere to schema
return ["network", "node", node_name, "application", "C2Server", "ransomware_configure", config]
class RansomwareLaunchC2ServerAction(AbstractAction):
"""Action which causes the C2 Server to send a command to the C2 Beacon to launch the RansomwareScript."""
def __init__(self, manager: "ActionManager", **kwargs) -> None:
super().__init__(manager=manager)
def form_request(self, node_id: int) -> RequestFormat:
"""Return the action formatted as a request that can be ingested by the simulation."""
node_name = self.manager.get_node_name_by_idx(node_id)
if node_name is None:
return ["do_nothing"]
# This action currently doesn't require any further configuration options.
return ["network", "node", node_name, "application", "C2Server", "ransomware_launch"]
class ExfiltrationC2ServerAction(AbstractAction):
"""Action which exfiltrates a target file from a certain node onto the C2 beacon and then the C2 Server."""
class _Opts(BaseModel):
"""Schema for options that can be passed to this action."""
username: Optional[str]
password: Optional[str]
target_ip_address: str
target_file_name: str
target_folder_name: str
exfiltration_folder_name: Optional[str]
def __init__(self, manager: "ActionManager", **kwargs) -> None:
super().__init__(manager=manager)
def form_request(
self,
node_id: int,
account: dict,
target_ip_address: str,
target_file_name: str,
target_folder_name: str,
exfiltration_folder_name: Optional[str],
) -> RequestFormat:
"""Return the action formatted as a request that can be ingested by the simulation."""
node_name = self.manager.get_node_name_by_idx(node_id)
if node_name is None:
return ["do_nothing"]
command_model = {
"target_file_name": target_file_name,
"target_folder_name": target_folder_name,
"exfiltration_folder_name": exfiltration_folder_name,
"target_ip_address": target_ip_address,
"username": account["username"],
"password": account["password"],
}
ExfiltrationC2ServerAction._Opts.model_validate(command_model)
return ["network", "node", node_name, "application", "C2Server", "exfiltrate", command_model]
class NodeSendRemoteCommandAction(AbstractAction):
"""Action which sends a terminal command to a remote node via SSH."""
@@ -1148,6 +1266,52 @@ class NodeSendRemoteCommandAction(AbstractAction):
]
class TerminalC2ServerAction(AbstractAction):
"""Action which causes the C2 Server to send a command to the C2 Beacon to execute the terminal command passed."""
class _Opts(BaseModel):
"""Schema for options that can be passed to this action."""
commands: Union[List[RequestFormat], RequestFormat]
ip_address: Optional[str]
username: Optional[str]
password: Optional[str]
def __init__(self, manager: "ActionManager", **kwargs) -> None:
super().__init__(manager=manager)
def form_request(self, node_id: int, commands: List, ip_address: Optional[str], account: dict) -> RequestFormat:
"""Return the action formatted as a request that can be ingested by the simulation."""
node_name = self.manager.get_node_name_by_idx(node_id)
if node_name is None:
return ["do_nothing"]
command_model = {
"commands": commands,
"ip_address": ip_address,
"username": account["username"],
"password": account["password"],
}
TerminalC2ServerAction._Opts.model_validate(command_model)
return ["network", "node", node_name, "application", "C2Server", "terminal_command", command_model]
class RansomwareLaunchC2ServerAction(AbstractAction):
"""Action which causes the C2 Server to send a command to the C2 Beacon to launch the RansomwareScript."""
def __init__(self, manager: "ActionManager", **kwargs) -> None:
super().__init__(manager=manager)
def form_request(self, node_id: int) -> RequestFormat:
"""Return the action formatted as a request that can be ingested by the simulation."""
node_name = self.manager.get_node_name_by_idx(node_id)
if node_name is None:
return ["do_nothing"]
# This action currently doesn't require any further configuration options.
return ["network", "node", node_name, "application", "C2Server", "ransomware_launch"]
class ActionManager:
"""Class which manages the action space for an agent."""
@@ -1199,6 +1363,11 @@ class ActionManager:
"CONFIGURE_DATABASE_CLIENT": ConfigureDatabaseClientAction,
"CONFIGURE_RANSOMWARE_SCRIPT": ConfigureRansomwareScriptAction,
"CONFIGURE_DOSBOT": ConfigureDoSBotAction,
"CONFIGURE_C2_BEACON": ConfigureC2BeaconAction,
"C2_SERVER_RANSOMWARE_LAUNCH": RansomwareLaunchC2ServerAction,
"C2_SERVER_RANSOMWARE_CONFIGURE": RansomwareConfigureC2ServerAction,
"C2_SERVER_TERMINAL_COMMAND": TerminalC2ServerAction,
"C2_SERVER_DATA_EXFILTRATE": ExfiltrationC2ServerAction,
"NODE_ACCOUNTS_CHANGE_PASSWORD": NodeAccountsChangePasswordAction,
"SSH_TO_REMOTE": NodeSessionsRemoteLoginAction,
"SESSIONS_REMOTE_LOGOFF": NodeSessionsRemoteLogoutAction,

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

@@ -27,10 +27,13 @@ from primaite.simulator.network.hardware.nodes.network.router import Router
from primaite.simulator.network.hardware.nodes.network.switch import Switch
from primaite.simulator.network.hardware.nodes.network.wireless_router import WirelessRouter
from primaite.simulator.network.nmne import NMNEConfig
from primaite.simulator.network.transmission.network_layer import IPProtocol
from primaite.simulator.network.transmission.transport_layer import Port
from primaite.simulator.sim_container import Simulation
from primaite.simulator.system.applications.application import Application
from primaite.simulator.system.applications.database_client import DatabaseClient # noqa: F401
from primaite.simulator.system.applications.red_applications.c2.c2_beacon import C2Beacon # noqa: F401
from primaite.simulator.system.applications.red_applications.c2.c2_server import C2Server # noqa: F401
from primaite.simulator.system.applications.red_applications.data_manipulation_bot import ( # noqa: F401
DataManipulationBot,
)
@@ -453,6 +456,15 @@ class PrimaiteGame:
dos_intensity=float(opt.get("dos_intensity", "1.0")),
max_sessions=int(opt.get("max_sessions", "1000")),
)
elif application_type == "C2Beacon":
if "options" in application_cfg:
opt = application_cfg["options"]
new_application.configure(
c2_server_ip_address=IPv4Address(opt.get("c2_server_ip_address")),
keep_alive_frequency=(opt.get("keep_alive_frequency", 5)),
masquerade_protocol=IPProtocol[(opt.get("masquerade_protocol", IPProtocol.TCP))],
masquerade_port=Port[(opt.get("masquerade_port", Port.HTTP))],
)
if "network_interfaces" in node_cfg:
for nic_num, nic_cfg in node_cfg["network_interfaces"].items():
new_node.connect_nic(NIC(ip_address=nic_cfg["ip_address"], subnet_mask=nic_cfg["subnet_mask"]))

1826
src/primaite/notebooks/Command-&-Control-E2E-Demonstration.ipynb Normal file
View File

File diff suppressed because it is too large Load Diff

9
src/primaite/simulator/network/hardware/base.py
View File

@@ -1281,13 +1281,18 @@ class UserSessionManager(Service):
def pre_timestep(self, timestep: int) -> None:
"""Apply any pre-timestep logic that helps make sure we have the correct observations."""
self.current_timestep = timestep
inactive_sessions: list = []
if self.local_session:
if self.local_session.last_active_step + self.local_session_timeout_steps <= timestep:
self._timeout_session(self.local_session)
inactive_sessions.append(self.local_session)
for session in self.remote_sessions:
remote_session = self.remote_sessions[session]
if remote_session.last_active_step + self.remote_session_timeout_steps <= timestep:
self._timeout_session(remote_session)
inactive_sessions.append(remote_session)
for sessions in inactive_sessions:
self._timeout_session(sessions)
def _timeout_session(self, session: UserSession) -> None:
"""

23
src/primaite/simulator/network/protocols/masquerade.py Normal file
View File

@@ -0,0 +1,23 @@
# © Crown-owned copyright 2024, Defence Science and Technology Laboratory UK
from enum import Enum
from typing import Optional
from primaite.simulator.network.protocols.packet import DataPacket
class MasqueradePacket(DataPacket):
"""Represents an generic malicious packet that is masquerading as another protocol."""
masquerade_protocol: Enum # The 'Masquerade' protocol that is currently in use
masquerade_port: Enum # The 'Masquerade' port that is currently in use
class C2Packet(MasqueradePacket):
"""Represents C2 suite communications packets."""
payload_type: Enum # The type of C2 traffic (e.g keep alive, command or command out)
command: Optional[Enum] = None # Used to pass the actual C2 Command in C2 INPUT
keep_alive_frequency: int

18
src/primaite/simulator/system/applications/application.py
View File

@@ -214,3 +214,21 @@ class Application(IOSoftware):
f"Cannot perform request on application '{self.application.name}' because it is not in the "
f"{self.state.name} state."
)
def _can_perform_network_action(self) -> bool:
"""
Checks if the application can perform outbound network actions.
First confirms application suitability via the can_perform_action method.
Then confirms that the host has an enabled NIC that can be used for outbound traffic.
:return: True if outbound network actions can be performed, otherwise False.
:rtype bool:
"""
if not super()._can_perform_action():
return False
for nic in self.software_manager.node.network_interface.values():
if nic.enabled:
return True
return False

64
src/primaite/simulator/system/applications/red_applications/c2/__init__.py Normal file
View File

@@ -0,0 +1,64 @@
# © Crown-owned copyright 2024, Defence Science and Technology Laboratory UK
from typing import Optional, Union
from pydantic import BaseModel, Field, field_validator, ValidationInfo
from primaite.interface.request import RequestFormat
class CommandOpts(BaseModel):
"""A C2 Pydantic Schema acting as a base class for all C2 Commands."""
@field_validator("payload", "exfiltration_folder_name", "ip_address", mode="before", check_fields=False)
@classmethod
def not_none(cls, v: str, info: ValidationInfo) -> int:
"""If None is passed, use the default value instead."""
if v is None:
return cls.model_fields[info.field_name].default
return v
class RansomwareOpts(CommandOpts):
"""A Pydantic Schema for the Ransomware Configuration command options."""
server_ip_address: str
"""The IP Address of the target database that the RansomwareScript will attack."""
payload: str = Field(default="ENCRYPT")
"""The malicious payload to be used to attack the target database."""
class RemoteOpts(CommandOpts):
"""A base C2 Pydantic Schema for all C2 Commands that require a terminal connection."""
ip_address: Optional[str] = Field(default=None)
"""The IP address of a remote host. If this field defaults to None then a local session is used."""
username: str
"""A Username of a valid user account. Used to login into both remote and local hosts."""
password: str
"""A Password of a valid user account. Used to login into both remote and local hosts."""
class ExfilOpts(RemoteOpts):
"""A Pydantic Schema for the C2 Data Exfiltration command options."""
target_ip_address: str
"""The IP address of the target host that will be the target of the exfiltration."""
target_file_name: str
"""The name of the file that is attempting to be exfiltrated."""
target_folder_name: str
"""The name of the remote folder which contains the target file."""
exfiltration_folder_name: str = Field(default="exfiltration_folder")
"""The name of C2 Suite folder used to store the target file. Defaults to ``exfiltration_folder``"""
class TerminalOpts(RemoteOpts):
"""A Pydantic Schema for the C2 Terminal command options."""
commands: Union[list[RequestFormat], RequestFormat]
"""A list or individual Terminal Command. Please refer to the RequestResponse system for further info."""

487
src/primaite/simulator/system/applications/red_applications/c2/abstract_c2.py Normal file
View File

@@ -0,0 +1,487 @@
# © Crown-owned copyright 2024, Defence Science and Technology Laboratory UK
from abc import abstractmethod
from enum import Enum
from ipaddress import IPv4Address
from typing import Dict, Optional, Union
from pydantic import BaseModel, Field, validate_call
from primaite.interface.request import RequestResponse
from primaite.simulator.file_system.file_system import FileSystem, Folder
from primaite.simulator.network.protocols.masquerade import C2Packet
from primaite.simulator.network.transmission.network_layer import IPProtocol
from primaite.simulator.network.transmission.transport_layer import Port
from primaite.simulator.system.applications.application import Application, ApplicationOperatingState
from primaite.simulator.system.core.session_manager import Session
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
from primaite.simulator.system.software import SoftwareHealthState
class C2Command(Enum):
"""Enumerations representing the different commands the C2 suite currently supports."""
RANSOMWARE_CONFIGURE = "Ransomware Configure"
"Instructs the c2 beacon to configure the ransomware with the provided options."
RANSOMWARE_LAUNCH = "Ransomware Launch"
"Instructs the c2 beacon to execute the installed ransomware."
DATA_EXFILTRATION = "Data Exfiltration"
"Instructs the c2 beacon to attempt to return a file to the C2 Server."
TERMINAL = "Terminal"
"Instructs the c2 beacon to execute the provided terminal command."
class C2Payload(Enum):
"""Represents the different types of command and control payloads."""
KEEP_ALIVE = "keep_alive"
"""C2 Keep Alive payload. Used by the C2 beacon and C2 Server to confirm their connection."""
INPUT = "input_command"
"""C2 Input Command payload. Used by the C2 Server to send a command to the c2 beacon."""
OUTPUT = "output_command"
"""C2 Output Command. Used by the C2 Beacon to send the results of a Input command to the c2 server."""
class AbstractC2(Application, identifier="AbstractC2"):
"""
An abstract command and control (c2) application.
Extends the application class to provide base functionality for c2 suite applications
such as c2 beacons and c2 servers.
Provides the base methods for handling ``Keep Alive`` connections, configuring masquerade ports and protocols
as well as providing the abstract methods for sending, receiving and parsing commands.
Defaults to masquerading as HTTP (Port 80) via TCP.
Please refer to the Command-&-Control notebook for an in-depth example of the C2 Suite.
"""
c2_connection_active: bool = False
"""Indicates if the c2 server and c2 beacon are currently connected."""
c2_remote_connection: IPv4Address = None
"""The IPv4 Address of the remote c2 connection. (Either the IP of the beacon or the server)."""
c2_session: Session = None
"""The currently active session that the C2 Traffic is using. Set after establishing connection."""
keep_alive_inactivity: int = 0
"""Indicates how many timesteps since the last time the c2 application received a keep alive."""
class _C2Opts(BaseModel):
"""A Pydantic Schema for the different C2 configuration options."""
keep_alive_frequency: int = Field(default=5, ge=1)
"""The frequency at which ``Keep Alive`` packets are sent to the C2 Server from the C2 Beacon."""
masquerade_protocol: IPProtocol = Field(default=IPProtocol.TCP)
"""The currently chosen protocol that the C2 traffic is masquerading as. Defaults as TCP."""
masquerade_port: Port = Field(default=Port.HTTP)
"""The currently chosen port that the C2 traffic is masquerading as. Defaults at HTTP."""
c2_config: _C2Opts = _C2Opts()
"""
Holds the current configuration settings of the C2 Suite.
The C2 beacon initialise this class through it's internal configure method.
The C2 Server when receiving a keep alive will initialise it's own configuration
to match that of the configuration settings passed in the keep alive through _resolve keep alive.
If the C2 Beacon is reconfigured then a new keep alive is set which causes the
C2 beacon to reconfigure it's configuration settings.
"""
def _craft_packet(
self, c2_payload: C2Payload, c2_command: Optional[C2Command] = None, command_options: Optional[Dict] = {}
) -> C2Packet:
"""
Creates and returns a Masquerade Packet using the parameters given.
The packet uses the current c2 configuration and parameters given
to construct the base networking information such as the masquerade
protocol/port. Additionally all C2 Traffic packets pass the currently
in use C2 configuration. This ensures that the all C2 applications
can keep their configuration in sync.
:param c2_payload: The type of C2 Traffic ot be sent
:type c2_payload: C2Payload
:param c2_command: The C2 command to be sent to the C2 Beacon.
:type c2_command: C2Command.
:param command_options: The relevant C2 Beacon parameters.F
:type command_options: Dict
:return: Returns the construct C2Packet
:rtype: C2Packet
"""
constructed_packet = C2Packet(
masquerade_protocol=self.c2_config.masquerade_protocol,
masquerade_port=self.c2_config.masquerade_port,
keep_alive_frequency=self.c2_config.keep_alive_frequency,
payload_type=c2_payload,
command=c2_command,
payload=command_options,
)
return constructed_packet
def describe_state(self) -> Dict:
"""
Describe the state of the C2 application.
:return: A dictionary representation of the C2 application's state.
:rtype: Dict
"""
return super().describe_state()
def __init__(self, **kwargs):
"""Initialise the C2 applications to by default listen for HTTP traffic."""
kwargs["listen_on_ports"] = {Port.HTTP, Port.FTP, Port.DNS}
kwargs["port"] = Port.NONE
kwargs["protocol"] = IPProtocol.TCP
super().__init__(**kwargs)
@property
def _host_ftp_client(self) -> Optional[FTPClient]:
"""Return the FTPClient that is installed C2 Application's host.
This method confirms that the FTP Client is functional via the ._can_perform_action
method. If the FTP Client service is not in a suitable state (e.g disabled/pause)
then this method will return None.
(The FTP Client service is installed by default)
:return: An FTPClient object is successful, else None
:rtype: union[FTPClient, None]
"""
ftp_client: Union[FTPClient, None] = self.software_manager.software.get("FTPClient")
if ftp_client is None:
self.sys_log.warning(f"{self.__class__.__name__}: No FTPClient. Attempting to install.")
self.software_manager.install(FTPClient)
ftp_client = self.software_manager.software.get("FTPClient")
# Force start if the service is stopped.
if ftp_client.operating_state == ServiceOperatingState.STOPPED:
if not ftp_client.start():
self.sys_log.warning(f"{self.__class__.__name__}: cannot start the FTP Client.")
if not ftp_client._can_perform_action():
self.sys_log.error(f"{self.__class__.__name__}: is unable to use the FTP service on its host.")
return
return ftp_client
@property
def _host_ftp_server(self) -> Optional[FTPServer]:
"""
Returns the FTP Server that is installed C2 Application's host.
If a FTPServer is not installed then this method will attempt to install one.
:return: An FTPServer object is successful, else None
:rtype: Optional[FTPServer]
"""
ftp_server: Optional[FTPServer] = self.software_manager.software.get("FTPServer")
if ftp_server is None:
self.sys_log.warning(f"{self.__class__.__name__}:No FTPServer installed. Attempting to install FTPServer.")
self.software_manager.install(FTPServer)
ftp_server = self.software_manager.software.get("FTPServer")
# Force start if the service is stopped.
if ftp_server.operating_state == ServiceOperatingState.STOPPED:
if not ftp_server.start():
self.sys_log.warning(f"{self.__class__.__name__}: cannot start the FTP Server.")
if not ftp_server._can_perform_action():
self.sys_log.error(f"{self.__class__.__name__}: is unable use FTP Server service on its host.")
return
return ftp_server
# Getter property for the get_exfiltration_folder method ()
@property
def _host_file_system(self) -> FileSystem:
"""Return the C2 Host's filesystem (Used for exfiltration related commands) ."""
host_file_system: FileSystem = self.software_manager.file_system
if host_file_system is None:
self.sys_log.error(f"{self.__class__.__name__}: does not seem to have a file system!")
return host_file_system
def get_exfiltration_folder(self, folder_name: Optional[str] = "exfiltration_folder") -> Optional[Folder]:
"""Return a folder used for storing exfiltrated data. Otherwise returns None."""
if self._host_file_system is None:
return
exfiltration_folder: Union[Folder, None] = self._host_file_system.get_folder(folder_name)
if exfiltration_folder is None:
self.sys_log.info(f"{self.__class__.__name__}: Creating a exfiltration folder.")
return self._host_file_system.create_folder(folder_name=folder_name)
return exfiltration_folder
# Validate call ensures we are only handling Masquerade Packets.
@validate_call
def _handle_c2_payload(self, payload: C2Packet, session_id: Optional[str] = None) -> bool:
"""Handles masquerade payloads for both c2 beacons and c2 servers.
Currently, the C2 application suite can handle the following payloads:
KEEP ALIVE:
Establishes or confirms connection from the C2 Beacon to the C2 server.
Sent by both C2 beacons and C2 Servers.
INPUT:
Contains a c2 command which must be executed by the C2 beacon.
Sent by C2 Servers and received by C2 Beacons.
OUTPUT:
Contains the output of a c2 command which must be returned to the C2 Server.
Sent by C2 Beacons and received by C2 Servers
The payload is passed to a different method dependant on the payload type.
:param payload: The C2 Payload to be parsed and handled.
:return: True if the c2 payload was handled successfully, False otherwise.
:rtype: Bool
"""
if payload.payload_type == C2Payload.KEEP_ALIVE:
self.sys_log.info(f"{self.name} received a KEEP ALIVE payload.")
return self._handle_keep_alive(payload, session_id)
elif payload.payload_type == C2Payload.INPUT:
self.sys_log.info(f"{self.name} received an INPUT COMMAND payload.")
return self._handle_command_input(payload, session_id)
elif payload.payload_type == C2Payload.OUTPUT:
self.sys_log.info(f"{self.name} received an OUTPUT COMMAND payload.")
return self._handle_command_output(payload)
else:
self.sys_log.warning(
f"{self.name} received an unexpected c2 payload:{payload.payload_type}. Dropping Packet."
)
return False
@abstractmethod
def _handle_command_output(payload):
"""Abstract Method: Used in C2 server to parse and receive the output of commands sent to the c2 beacon."""
pass
@abstractmethod
def _handle_command_input(payload):
"""Abstract Method: Used in C2 beacon to parse and handle commands received from the c2 server."""
pass
@abstractmethod
def _handle_keep_alive(self, payload: C2Packet, session_id: Optional[str]) -> bool:
"""Abstract Method: Each C2 suite handles ``C2Payload.KEEP_ALIVE`` differently."""
pass
# from_network_interface=from_network_interface
def receive(self, payload: any, session_id: Optional[str] = None, **kwargs) -> bool:
"""Receives masquerade packets. Used by both c2 server and c2 beacon.
Defining the `Receive` method so that the application can receive packets via the session manager.
These packets are then immediately handed to ._handle_c2_payload.
:param payload: The Masquerade Packet to be received.
:type payload: C2Packet
:param session_id: The transport session_id that the payload is originating from.
:type session_id: str
:return: Returns a bool if the traffic was received correctly (See _handle_c2_payload.)
:rtype: bool
"""
if not isinstance(payload, C2Packet):
self.sys_log.warning(f"{self.name}: Payload is not an C2Packet")
self.sys_log.debug(f"{self.name}: {payload}")
return False
return self._handle_c2_payload(payload, session_id)
def _send_keep_alive(self, session_id: Optional[str]) -> bool:
"""Sends a C2 keep alive payload to the self.remote_connection IPv4 Address.
Used by both the c2 client and the s2 server for establishing and confirming connection.
This method also contains some additional validation to ensure that the C2 applications
are correctly configured before sending any traffic.
:param session_id: The transport session_id that the payload is originating from.
:type session_id: str
:returns: Returns True if a send alive was successfully sent. False otherwise.
:rtype bool:
"""
# Checking that the c2 application is capable of connecting to remote.
# Purely a safety guard clause.
if not (connection_status := self._check_connection()[0]):
self.sys_log.warning(
f"{self.name}: Unable to send keep alive due to c2 connection status: {connection_status}."
)
return False
# Passing our current C2 configuration in remain in sync.
keep_alive_packet = self._craft_packet(c2_payload=C2Payload.KEEP_ALIVE)
# Sending the keep alive via the .send() method (as with all other applications.)
if self.send(
payload=keep_alive_packet,
dest_ip_address=self.c2_remote_connection,
dest_port=self.c2_config.masquerade_port,
ip_protocol=self.c2_config.masquerade_protocol,
session_id=session_id,
):
# Setting the keep_alive_sent guard condition to True. This is used to prevent packet storms.
# This prevents the _resolve_keep_alive method from calling this method again (until the next timestep.)
self.keep_alive_sent = True
self.sys_log.info(f"{self.name}: Keep Alive sent to {self.c2_remote_connection}")
self.sys_log.debug(
f"{self.name}: Keep Alive sent to {self.c2_remote_connection} "
f"Masquerade Port: {self.c2_config.masquerade_port} "
f"Masquerade Protocol: {self.c2_config.masquerade_protocol} "
)
return True
else:
self.sys_log.warning(
f"{self.name}: Failed to send a Keep Alive. The node may be unable to access networking resources."
)
return False
def _resolve_keep_alive(self, payload: C2Packet, session_id: Optional[str]) -> bool:
"""
Parses the Masquerade Port/Protocol within the received Keep Alive packet.
Used to dynamically set the Masquerade Port and Protocol based on incoming traffic.
Returns True on successfully extracting and configuring the masquerade port/protocols.
Returns False otherwise.
:param payload: The Keep Alive payload received.
:type payload: C2Packet
:param session_id: The transport session_id that the payload is originating from.
:type session_id: str
:return: True on successful configuration, false otherwise.
:rtype: bool
"""
# Validating that they are valid Enums.
if not isinstance(payload.masquerade_port, Port) or not isinstance(payload.masquerade_protocol, IPProtocol):
self.sys_log.warning(
f"{self.name}: Received invalid Masquerade Values within Keep Alive."
f"Port: {payload.masquerade_port} Protocol: {payload.masquerade_protocol}."
)
return False
# Updating the C2 Configuration attribute.
self.c2_config.masquerade_port = payload.masquerade_port
self.c2_config.masquerade_protocol = payload.masquerade_protocol
self.c2_config.keep_alive_frequency = payload.keep_alive_frequency
self.sys_log.debug(
f"{self.name}: C2 Config Resolved Config from Keep Alive:"
f"Masquerade Port: {self.c2_config.masquerade_port}"
f"Masquerade Protocol: {self.c2_config.masquerade_protocol}"
f"Keep Alive Frequency: {self.c2_config.keep_alive_frequency}"
)
# This statement is intended to catch on the C2 Application that is listening for connection.
if self.c2_remote_connection is None:
self.sys_log.debug(f"{self.name}: Attempting to configure remote C2 connection based off received output.")
self.c2_remote_connection = IPv4Address(self.c2_session.with_ip_address)
self.c2_connection_active = True # Sets the connection to active
self.keep_alive_inactivity = 0 # Sets the keep alive inactivity to zero
return True
def _reset_c2_connection(self) -> None:
"""
Resets all currently established C2 communications to their default setting.
This method is called once a C2 application considers their remote connection
severed and reverts back to default settings. Worth noting that that this will
revert any non-default configuration that a user/agent may have set.
"""
self.c2_connection_active = False
self.c2_session = None
self.keep_alive_inactivity = 0
self.keep_alive_frequency = 5
self.c2_remote_connection = None
self.c2_config.masquerade_port = Port.HTTP
self.c2_config.masquerade_protocol = IPProtocol.TCP
@abstractmethod
def _confirm_remote_connection(self, timestep: int) -> bool:
"""
Abstract method - Confirms the suitability of the current C2 application remote connection.
Each application will have perform different behaviour to confirm the remote connection.
:return: Boolean. True if remote connection is confirmed, false otherwise.
"""
def apply_timestep(self, timestep: int) -> None:
"""Apply a timestep to the c2_server & c2 beacon.
Used to keep track of when the c2 server should consider a beacon dead
and set it's c2_remote_connection attribute to false.
1. Each timestep the keep_alive_inactivity is increased.
2. If the keep alive inactivity eclipses that of the keep alive frequency then another keep alive is sent.
3. If a keep alive response packet is received then the ``keep_alive_inactivity`` attribute is reset.
Therefore, if ``keep_alive_inactivity`` attribute is not 0 after a keep alive is sent
then the connection is considered severed and c2 beacon will shut down.
:param timestep: The current timestep of the simulation.
:type timestep: Int
:return bool: Returns false if connection was lost. Returns True if connection is active or re-established.
:rtype bool:
"""
if (
self.operating_state is ApplicationOperatingState.RUNNING
and self.health_state_actual is SoftwareHealthState.GOOD
):
self.keep_alive_inactivity += 1
self._confirm_remote_connection(timestep)
return super().apply_timestep(timestep=timestep)
def _check_connection(self) -> tuple[bool, RequestResponse]:
"""
Validation method: Checks that the C2 application is capable of sending C2 Command input/output.
Performs a series of connection validation to ensure that the C2 application is capable of
sending and responding to the remote c2 connection. This method is used to confirm connection
before carrying out Agent Commands hence why this method also returns a tuple
containing both a success boolean as well as RequestResponse.
:return: A tuple containing a boolean True/False and a corresponding Request Response
:rtype: tuple[bool, RequestResponse]
"""
if not self._can_perform_network_action():
self.sys_log.warning(f"{self.name}: Unable to make leverage networking resources. Rejecting Command.")
return (
False,
RequestResponse(
status="failure", data={"Reason": "Unable to access networking resources. Unable to send command."}
),
)
if self.c2_remote_connection is None:
self.sys_log.warning(f"{self.name}: C2 Application has yet to establish connection. Rejecting command.")
return (
False,
RequestResponse(
status="failure",
data={"Reason": "C2 Application has yet to establish connection. Unable to send command."},
),
)
return (
True,
RequestResponse(status="success", data={"Reason": "C2 Application is able to send connections."}),
)

636
src/primaite/simulator/system/applications/red_applications/c2/c2_beacon.py Normal file
View File

@@ -0,0 +1,636 @@
# © Crown-owned copyright 2024, Defence Science and Technology Laboratory UK
from enum import Enum
from ipaddress import IPv4Address
from typing import Dict, Optional
from prettytable import MARKDOWN, PrettyTable
from pydantic import validate_call
from primaite.interface.request import RequestFormat, RequestResponse
from primaite.simulator.core import RequestManager, RequestType
from primaite.simulator.network.protocols.masquerade import C2Packet
from primaite.simulator.network.transmission.network_layer import IPProtocol
from primaite.simulator.network.transmission.transport_layer import Port
from primaite.simulator.system.applications.red_applications.c2 import ExfilOpts, RansomwareOpts, TerminalOpts
from primaite.simulator.system.applications.red_applications.c2.abstract_c2 import AbstractC2, C2Command, C2Payload
from primaite.simulator.system.applications.red_applications.ransomware_script import RansomwareScript
from primaite.simulator.system.services.terminal.terminal import Terminal, TerminalClientConnection
class C2Beacon(AbstractC2, identifier="C2Beacon"):
"""
C2 Beacon Application.
Represents a vendor generic C2 beacon is used in conjunction with the C2 Server
to simulate malicious communications and infrastructure within primAITE.
Must be configured with the C2 Server's IP Address upon installation.
Please refer to the _configure method for further information.
Extends the Abstract C2 application to include the following:
1. Receiving commands from the C2 Server (Command input)
2. Leveraging the terminal application to execute requests (dependent on the command given)
3. Sending the RequestResponse back to the C2 Server (Command output)
Please refer to the Command-&-Control notebook for an in-depth example of the C2 Suite.
"""
keep_alive_attempted: bool = False
"""Indicates if a keep alive has been attempted to be sent this timestep. Used to prevent packet storms."""
terminal_session: TerminalClientConnection = None
"The currently in use terminal session."
@property
def _host_terminal(self) -> Optional[Terminal]:
"""Return the Terminal that is installed on the same machine as the C2 Beacon."""
host_terminal: Terminal = self.software_manager.software.get("Terminal")
if host_terminal is None:
self.sys_log.warning(f"{self.__class__.__name__} cannot find a terminal on its host.")
return host_terminal
@property
def _host_ransomware_script(self) -> RansomwareScript:
"""Return the RansomwareScript that is installed on the same machine as the C2 Beacon."""
ransomware_script: RansomwareScript = self.software_manager.software.get("RansomwareScript")
if ransomware_script is None:
self.sys_log.warning(f"{self.__class__.__name__} cannot find installed ransomware on its host.")
return ransomware_script
def _set_terminal_session(self, username: str, password: str, ip_address: Optional[IPv4Address] = None) -> bool:
"""
Attempts to create and a terminal session using the parameters given.
If an IP Address is passed then this method will attempt to create a remote terminal
session. Otherwise a local terminal session will be created.
:return: Returns true if a terminal session was successfully set. False otherwise.
:rtype: Bool
"""
self.terminal_session is None
host_terminal: Terminal = self._host_terminal
self.terminal_session = host_terminal.login(username=username, password=password, ip_address=ip_address)
return self.terminal_session is not None
def _init_request_manager(self) -> RequestManager:
"""
Initialise the request manager.
More information in user guide and docstring for SimComponent._init_request_manager.
"""
rm = super()._init_request_manager()
rm.add_request(
name="execute",
request_type=RequestType(func=lambda request, context: RequestResponse.from_bool(self.establish())),
)
def _configure(request: RequestFormat, context: Dict) -> RequestResponse:
"""
Request for configuring the C2 Beacon.
:param request: Request with one element containing a dict of parameters for the configure method.
:type request: RequestFormat
:param context: additional context for resolving this action, currently unused
:type context: dict
:return: RequestResponse object with a success code reflecting whether the configuration could be applied.
:rtype: RequestResponse
"""
c2_remote_ip = request[-1].get("c2_server_ip_address")
if c2_remote_ip is None:
self.sys_log.error(f"{self.name}: Did not receive C2 Server IP in configuration parameters.")
RequestResponse(
status="failure", data={"No C2 Server IP given to C2 beacon. Unable to configure C2 Beacon"}
)
c2_remote_ip = IPv4Address(c2_remote_ip)
frequency = request[-1].get("keep_alive_frequency")
protocol = request[-1].get("masquerade_protocol")
port = request[-1].get("masquerade_port")
return RequestResponse.from_bool(
self.configure(
c2_server_ip_address=c2_remote_ip,
keep_alive_frequency=frequency,
masquerade_protocol=IPProtocol[protocol],
masquerade_port=Port[port],
)
)
rm.add_request("configure", request_type=RequestType(func=_configure))
return rm
def __init__(self, **kwargs):
kwargs["name"] = "C2Beacon"
super().__init__(**kwargs)
# Configure is practically setter method for the ``c2.config`` attribute that also ties into the request manager.
@validate_call
def configure(
self,
c2_server_ip_address: IPv4Address = None,
keep_alive_frequency: int = 5,
masquerade_protocol: Enum = IPProtocol.TCP,
masquerade_port: Enum = Port.HTTP,
) -> bool:
"""
Configures the C2 beacon to communicate with the C2 server.
The C2 Beacon has four different configuration options which can be used to
modify the networking behaviour between the C2 Server and the C2 Beacon.
Configuration Option | Option Meaning
---------------------|------------------------
c2_server_ip_address | The IP Address of the C2 Server. (The C2 Server must be running)
keep_alive_frequency | How often should the C2 Beacon confirm it's connection in timesteps.
masquerade_protocol | What protocol should the C2 traffic masquerade as? (HTTP, FTP or DNS)
masquerade_port | What port should the C2 traffic use? (TCP or UDP)
These configuration options are used to reassign the fields in the inherited inner class
``c2_config``.
If a connection is already in progress then this method also sends a keep alive to the C2
Server in order for the C2 Server to sync with the new configuration settings.
:param c2_server_ip_address: The IP Address of the C2 Server. Used to establish connection.
:type c2_server_ip_address: IPv4Address
:param keep_alive_frequency: The frequency (timesteps) at which the C2 beacon will send keep alive(s).
:type keep_alive_frequency: Int
:param masquerade_protocol: The Protocol that C2 Traffic will masquerade as. Defaults to TCP.
:type masquerade_protocol: Enum (IPProtocol)
:param masquerade_port: The Port that the C2 Traffic will masquerade as. Defaults to FTP.
:type masquerade_port: Enum (Port)
:return: Returns True if the configuration was successful, False otherwise.
"""
self.c2_remote_connection = IPv4Address(c2_server_ip_address)
self.c2_config.keep_alive_frequency = keep_alive_frequency
self.c2_config.masquerade_port = masquerade_port
self.c2_config.masquerade_protocol = masquerade_protocol
self.sys_log.info(
f"{self.name}: Configured {self.name} with remote C2 server connection: {c2_server_ip_address=}."
)
self.sys_log.debug(
f"{self.name}: configured with the following settings:"
f"Remote C2 Server: {c2_server_ip_address}"
f"Keep Alive Frequency {keep_alive_frequency}"
f"Masquerade Protocol: {masquerade_protocol}"
f"Masquerade Port: {masquerade_port}"
)
# Send a keep alive to the C2 Server if we already have a keep alive.
if self.c2_connection_active is True:
self.sys_log.info(f"{self.name}: Updating C2 Server with updated C2 configuration.")
return self._send_keep_alive(self.c2_session.uuid if not None else None)
return True
def establish(self) -> bool:
"""Establishes connection to the C2 server via a send alive. The C2 Beacon must already be configured."""
if self.c2_remote_connection is None:
self.sys_log.info(f"{self.name}: Failed to establish connection. C2 Beacon has not been configured.")
return False
self.run()
self.num_executions += 1
# Creates a new session if using the establish method.
return self._send_keep_alive(session_id=None)
def _handle_command_input(self, payload: C2Packet, session_id: Optional[str]) -> bool:
"""
Handles the parsing of C2 Commands from C2 Traffic (Masquerade Packets).
Dependant the C2 Command parsed from the payload, the following methods are called and returned:
C2 Command | Internal Method
---------------------|------------------------
RANSOMWARE_CONFIGURE | self._command_ransomware_config()
RANSOMWARE_LAUNCH | self._command_ransomware_launch()
DATA_EXFILTRATION | self._command_data_exfiltration()
TERMINAL | self._command_terminal()
Please see each method individually for further information regarding
the implementation of these commands.
:param payload: The INPUT C2 Payload
:type payload: C2Packet
:return: The Request Response provided by the terminal execute method.
:rtype Request Response:
"""
command = payload.command
if not isinstance(command, C2Command):
self.sys_log.warning(f"{self.name}: Received unexpected C2 command. Unable to resolve command")
return self._return_command_output(
command_output=RequestResponse(
status="failure",
data={"Reason": "C2 Beacon received unexpected C2Command. Unable to resolve command."},
),
session_id=session_id,
)
if command == C2Command.RANSOMWARE_CONFIGURE:
self.sys_log.info(f"{self.name}: Received a ransomware configuration C2 command.")
return self._return_command_output(
command_output=self._command_ransomware_config(payload), session_id=session_id
)
elif command == C2Command.RANSOMWARE_LAUNCH:
self.sys_log.info(f"{self.name}: Received a ransomware launch C2 command.")
return self._return_command_output(
command_output=self._command_ransomware_launch(payload), session_id=session_id
)
elif command == C2Command.TERMINAL:
self.sys_log.info(f"{self.name}: Received a terminal C2 command.")
return self._return_command_output(command_output=self._command_terminal(payload), session_id=session_id)
elif command == C2Command.DATA_EXFILTRATION:
self.sys_log.info(f"{self.name}: Received a Data Exfiltration C2 command.")
return self._return_command_output(
command_output=self._command_data_exfiltration(payload), session_id=session_id
)
else:
self.sys_log.error(f"{self.name}: Received an C2 command: {command} but was unable to resolve command.")
return self._return_command_output(
RequestResponse(status="failure", data={"Reason": "Unexpected Behaviour. Unable to resolve command."})
)
def _return_command_output(self, command_output: RequestResponse, session_id: Optional[str] = None) -> bool:
"""Responsible for responding to the C2 Server with the output of the given command.
:param command_output: The RequestResponse returned by the terminal application's execute method.
:type command_output: Request Response
:param session_id: The current session established with the C2 Server.
:type session_id: Str
"""
output_packet = self._craft_packet(c2_payload=C2Payload.OUTPUT, command_options=command_output)
if self.send(
payload=output_packet,
dest_ip_address=self.c2_remote_connection,
dest_port=self.c2_config.masquerade_port,
ip_protocol=self.c2_config.masquerade_protocol,
session_id=session_id,
):
self.sys_log.info(f"{self.name}: Command output sent to {self.c2_remote_connection}")
self.sys_log.debug(
f"{self.name}: on {self.c2_config.masquerade_port} via {self.c2_config.masquerade_protocol}"
)
return True
else:
self.sys_log.warning(
f"{self.name}: failed to send a output packet. The node may be unable to access the network."
)
return False
def _command_ransomware_config(self, payload: C2Packet) -> RequestResponse:
"""
C2 Command: Ransomware Configuration.
Calls the locally installed RansomwareScript application's configure method
and passes the given parameters.
The class attribute self._host_ransomware_script will return None if the host
does not have an instance of the RansomwareScript.
:payload C2Packet: The incoming INPUT command.
:type Masquerade Packet: C2Packet.
:return: Returns the Request Response returned by the Terminal execute method.
:rtype: Request Response
"""
command_opts = RansomwareOpts.model_validate(payload.payload)
if self._host_ransomware_script is None:
return RequestResponse(
status="failure",
data={"Reason": "Cannot find any instances of a RansomwareScript. Have you installed one?"},
)
return RequestResponse.from_bool(
self._host_ransomware_script.configure(
server_ip_address=command_opts.server_ip_address, payload=command_opts.payload
)
)
def _command_ransomware_launch(self, payload: C2Packet) -> RequestResponse:
"""
C2 Command: Ransomware Launch.
Uses the RansomwareScript's public method .attack() to carry out the
ransomware attack and uses the .from_bool method to return a RequestResponse
:payload C2Packet: The incoming INPUT command.
:type Masquerade Packet: C2Packet.
:return: Returns the Request Response returned by the Terminal execute method.
:rtype: Request Response
"""
if self._host_ransomware_script is None:
return RequestResponse(
status="failure",
data={"Reason": "Cannot find any instances of a RansomwareScript. Have you installed one?"},
)
return RequestResponse.from_bool(self._host_ransomware_script.attack())
def _command_data_exfiltration(self, payload: C2Packet) -> RequestResponse:
"""
C2 Command: Data Exfiltration.
Uses the FTP Client & Server services to perform the data exfiltration.
This command instructs the C2 Beacon to ssh into the target ip
and execute a command which causes the FTPClient service to send a
target file will be moved from the target IP address onto the C2 Beacon's host
file system.
However, if no IP is given, then the command will move the target file from this
machine onto the C2 server. (This logic is performed on the C2)
:payload C2Packet: The incoming INPUT command.
:type Masquerade Packet: C2Packet.
:return: Returns a tuple containing Request Response returned by the Terminal execute method.
:rtype: Request Response
"""
if self._host_ftp_server is None:
self.sys_log.warning(f"{self.name}: C2 Beacon unable to the FTP Server. Unable to resolve command.")
return RequestResponse(
status="failure",
data={"Reason": "Cannot find any instances of both a FTP Server & Client. Are they installed?"},
)
command_opts = ExfilOpts.model_validate(payload.payload)
# Setting up the terminal session and the ftp server
if not self._set_terminal_session(
username=command_opts.username, password=command_opts.password, ip_address=command_opts.target_ip_address
):
return RequestResponse(
status="failure", data={"Reason": "Cannot create a terminal session. Are the credentials correct?"}
)
# Using the terminal to start the FTP Client on the remote machine.
self.terminal_session.execute(command=["service", "start", "FTPClient"])
# Need to supply to the FTP Client the C2 Beacon's host IP.
host_network_interfaces = self.software_manager.node.network_interfaces
local_ip = host_network_interfaces.get(next(iter(host_network_interfaces))).ip_address
# Creating the FTP creation options.
ftp_opts = {
"dest_ip_address": str(local_ip),
"src_folder_name": command_opts.target_folder_name,
"src_file_name": command_opts.target_file_name,
"dest_folder_name": command_opts.exfiltration_folder_name,
"dest_file_name": command_opts.target_file_name,
}
attempt_exfiltration: tuple[bool, RequestResponse] = self._perform_exfiltration(ftp_opts)
if attempt_exfiltration[0] is False:
self.sys_log.error(f"{self.name}: File Exfiltration Attempt Failed: {attempt_exfiltration[1].data}")
return attempt_exfiltration[1]
# Sending the transferred target data back to the C2 Server to successfully exfiltrate the data out the network.
return RequestResponse.from_bool(
self._host_ftp_client.send_file(
dest_ip_address=self.c2_remote_connection,
src_folder_name=command_opts.exfiltration_folder_name, # The Exfil folder is inherited attribute.
src_file_name=command_opts.target_file_name,
dest_folder_name=command_opts.exfiltration_folder_name,
dest_file_name=command_opts.target_file_name,
)
)
def _perform_exfiltration(self, ftp_opts: dict) -> tuple[bool, RequestResponse]:
"""
Attempts to exfiltrate a target file from a target using the parameters given.
Uses the current terminal_session to send a command to the
remote host's FTP Client passing the ExfilOpts as command options.
This will instruct the FTP client to send the target file to the
dest_ip_address's destination folder.
This method assumes that the following:
1. The self.terminal_session is the remote target.
2. The target has a functioning FTP Client Service.
:ExfilOpts: A Pydantic model containing the require configuration options
:type ExfilOpts: ExfilOpts
:return: Returns a tuple containing a success boolean and a Request Response..
:rtype: tuple[bool, RequestResponse
"""
# Creating the exfiltration folder .
exfiltration_folder = self.get_exfiltration_folder(ftp_opts.get("dest_folder_name"))
# Using the terminal to send the target data back to the C2 Beacon.
exfil_response: RequestResponse = RequestResponse.from_bool(
self.terminal_session.execute(command=["service", "FTPClient", "send", ftp_opts])
)
# Validating that we successfully received the target data.
if exfil_response.status == "failure":
self.sys_log.warning(f"{self.name}: Remote connection failure. failed to transfer the target data via FTP.")
return [False, exfil_response]
# Target file:
target_file: str = ftp_opts.get("src_file_name")
if exfiltration_folder.get_file(target_file) is None:
self.sys_log.warning(
f"{self.name}: Unable to locate exfiltrated file on local filesystem. "
f"Perhaps the file transfer failed?"
)
return [
False,
RequestResponse(status="failure", data={"reason": "Unable to locate exfiltrated data on file system."}),
]
if self._host_ftp_client is None:
self.sys_log.warning(f"{self.name}: C2 Beacon unable to the FTP Server. Unable to resolve command.")
return [
False,
RequestResponse(
status="failure",
data={"Reason": "Cannot find any instances of both a FTP Server & Client. Are they installed?"},
),
]
return [
True,
RequestResponse(
status="success",
data={"Reason": "Located the target file on local file system. Data exfiltration successful."},
),
]
def _command_terminal(self, payload: C2Packet) -> RequestResponse:
"""
C2 Command: Terminal.
Creates a request that executes a terminal command.
This request is then sent to the terminal service in order to be executed.
:payload C2Packet: The incoming INPUT command.
:type Masquerade Packet: C2Packet.
:return: Returns the Request Response returned by the Terminal execute method.
:rtype: Request Response
"""
command_opts = TerminalOpts.model_validate(payload.payload)
if self._host_terminal is None:
return RequestResponse(
status="failure",
data={"Reason": "Host does not seem to have terminal installed. Unable to resolve command."},
)
terminal_output: Dict[int, RequestResponse] = {}
# Creating a remote terminal session if given an IP Address, otherwise using a local terminal session.
if not self._set_terminal_session(
username=command_opts.username, password=command_opts.password, ip_address=command_opts.ip_address
):
return RequestResponse(
status="failure",
data={"Reason": "Cannot create a terminal session. Are the credentials correct?"},
)
# Converts a singular terminal command: [RequestFormat] into a list with one element [[RequestFormat]]
# Checks the first element - if this element is a str then there must be multiple commands.
command_opts.commands = (
[command_opts.commands] if isinstance(command_opts.commands[0], str) else command_opts.commands
)
for index, given_command in enumerate(command_opts.commands):
# A try catch exception ladder was used but was considered not the best approach
# as it can end up obscuring visibility of actual bugs (Not the expected ones) and was a temporary solution.
# TODO: Refactor + add further validation to ensure that a request is correct. (maybe a pydantic method?)
terminal_output[index] = self.terminal_session.execute(given_command)
# Reset our remote terminal session.
self.terminal_session is None
return RequestResponse(status="success", data=terminal_output)
def _handle_keep_alive(self, payload: C2Packet, session_id: Optional[str]) -> bool:
"""
Handles receiving and sending keep alive payloads. This method is only called if we receive a keep alive.
In the C2 Beacon implementation of this method the c2 connection active boolean
is set to true and the keep alive inactivity is reset only after sending a keep alive
as wel as receiving a response back from the C2 Server.
This is because the C2 Server is the listener and thus will only ever receive packets from
the C2 Beacon rather than the other way around. (The C2 Beacon is akin to a reverse shell)
Therefore, we need a response back from the listener (C2 Server)
before the C2 beacon is able to confirm it's connection.
Returns False if a keep alive was unable to be sent.
Returns True if a keep alive was successfully sent or already has been sent this timestep.
:return: True if successfully handled, false otherwise.
:rtype: Bool
"""
self.sys_log.info(f"{self.name}: Keep Alive Received from {self.c2_remote_connection}.")
# Using this guard clause to prevent packet storms and recognise that we've achieved a connection.
# This guard clause triggers on the c2 suite that establishes connection.
if self.keep_alive_attempted is True:
self.c2_connection_active = True # Sets the connection to active
self.keep_alive_inactivity = 0 # Sets the keep alive inactivity to zero
self.c2_session = self.software_manager.session_manager.sessions_by_uuid[session_id]
# We set keep alive_attempted here to show that we've achieved connection.
self.keep_alive_attempted = False
self.sys_log.warning(f"{self.name}: Connection successfully Established with C2 Server.")
return True
# If we've reached this part of the method then we've received a keep alive but haven't sent a reply.
# Therefore we also need to configure the masquerade attributes based off the keep alive sent.
if self._resolve_keep_alive(payload, session_id) is False:
self.sys_log.warning(f"{self.name}: Keep Alive Could not be resolved correctly. Refusing Keep Alive.")
return False
self.keep_alive_attempted = True
# If this method returns true then we have sent successfully sent a keep alive.
return self._send_keep_alive(session_id)
def _confirm_remote_connection(self, timestep: int) -> bool:
"""Checks the suitability of the current C2 Server connection.
If a connection cannot be confirmed then this method will return false otherwise true.
:param timestep: The current timestep of the simulation.
:type timestep: Int
:return: Returns False if connection was lost. Returns True if connection is active or re-established.
:rtype bool:
"""
self.keep_alive_attempted = False # Resetting keep alive sent.
if self.keep_alive_inactivity == self.c2_config.keep_alive_frequency:
self.sys_log.info(
f"{self.name}: Attempting to Send Keep Alive to {self.c2_remote_connection} at timestep {timestep}."
)
self._send_keep_alive(session_id=self.c2_session.uuid)
if self.keep_alive_inactivity != 0:
self.sys_log.warning(
f"{self.name}: Did not receive keep alive from c2 Server. Connection considered severed."
)
self._reset_c2_connection()
self.close()
return False
return True
# Defining this abstract method from Abstract C2
def _handle_command_output(self, payload: C2Packet):
"""C2 Beacons currently does not need to handle output commands coming from the C2 Servers."""
self.sys_log.warning(f"{self.name}: C2 Beacon received an unexpected OUTPUT payload: {payload}.")
pass
def show(self, markdown: bool = False):
"""
Prints a table of the current status of the C2 Beacon.
Displays the current values of the following C2 attributes:
``C2 Connection Active``:
If the C2 Beacon is currently connected to the C2 Server
``C2 Remote Connection``:
The IP of the C2 Server. (Configured by upon installation)
``Keep Alive Inactivity``:
How many timesteps have occurred since the last keep alive.
``Keep Alive Frequency``:
How often should the C2 Beacon attempt a keep alive?
``Current Masquerade Protocol``:
The current protocol that the C2 Traffic is using. (e.g TCP/UDP)
``Current Masquerade Port``:
The current port that the C2 Traffic is using. (e.g HTTP (Port 80))
:param markdown: If True, outputs the table in markdown format. Default is False.
"""
table = PrettyTable(
[
"C2 Connection Active",
"C2 Remote Connection",
"Keep Alive Inactivity",
"Keep Alive Frequency",
"Current Masquerade Protocol",
"Current Masquerade Port",
]
)
if markdown:
table.set_style(MARKDOWN)
table.align = "l"
table.title = f"{self.name} Running Status"
table.add_row(
[
self.c2_connection_active,
self.c2_remote_connection,
self.keep_alive_inactivity,
self.c2_config.keep_alive_frequency,
self.c2_config.masquerade_protocol,
self.c2_config.masquerade_port,
]
)
print(table)

396
src/primaite/simulator/system/applications/red_applications/c2/c2_server.py Normal file
View File

@@ -0,0 +1,396 @@
# © Crown-owned copyright 2024, Defence Science and Technology Laboratory UK
from typing import Dict, Optional
from prettytable import MARKDOWN, PrettyTable
from pydantic import validate_call
from primaite.interface.request import RequestFormat, RequestResponse
from primaite.simulator.core import RequestManager, RequestType
from primaite.simulator.network.protocols.masquerade import C2Packet
from primaite.simulator.system.applications.red_applications.c2 import (
CommandOpts,
ExfilOpts,
RansomwareOpts,
TerminalOpts,
)
from primaite.simulator.system.applications.red_applications.c2.abstract_c2 import AbstractC2, C2Command, C2Payload
class C2Server(AbstractC2, identifier="C2Server"):
"""
C2 Server Application.
Represents a vendor generic C2 Server used in conjunction with the C2 beacon
to simulate malicious communications and infrastructure within primAITE.
The C2 Server must be installed and be in a running state before it's able to receive
red agent actions and send commands to the C2 beacon.
Extends the Abstract C2 application to include the following:
1. Sending commands to the C2 Beacon. (Command input)
2. Parsing terminal RequestResponses back to the Agent.
Please refer to the Command-&-Control notebook for an in-depth example of the C2 Suite.
"""
current_command_output: RequestResponse = None
"""The Request Response by the last command send. This attribute is updated by the method _handle_command_output."""
def _init_request_manager(self) -> RequestManager:
"""
Initialise the request manager.
More information in user guide and docstring for SimComponent._init_request_manager.
"""
rm = super()._init_request_manager()
def _configure_ransomware_action(request: RequestFormat, context: Dict) -> RequestResponse:
"""Requests - Sends a RANSOMWARE_CONFIGURE C2Command to the C2 Beacon with the given parameters.
:param request: Request with one element containing a dict of parameters for the configure method.
:type request: RequestFormat
:param context: additional context for resolving this action, currently unused
:type context: dict
:return: RequestResponse object with a success code reflecting whether the configuration could be applied.
:rtype: RequestResponse
"""
command_payload = {
"server_ip_address": request[-1].get("server_ip_address"),
"payload": request[-1].get("payload"),
}
return self.send_command(given_command=C2Command.RANSOMWARE_CONFIGURE, command_options=command_payload)
def _launch_ransomware_action(request: RequestFormat, context: Dict) -> RequestResponse:
"""Agent Action - Sends a RANSOMWARE_LAUNCH C2Command to the C2 Beacon with the given parameters.
:param request: Request with one element containing a dict of parameters for the configure method.
:type request: RequestFormat
:param context: additional context for resolving this action, currently unused
:type context: dict
:return: RequestResponse object with a success code reflecting whether the ransomware was launched.
:rtype: RequestResponse
"""
return self.send_command(given_command=C2Command.RANSOMWARE_LAUNCH, command_options={})
def _data_exfiltration_action(request: RequestFormat, context: Dict) -> RequestResponse:
"""Agent Action - Sends a Data Exfiltration C2Command to the C2 Beacon with the given parameters.
:param request: Request with one element containing a dict of parameters for the configure method.
:type request: RequestFormat
:param context: additional context for resolving this action, currently unused
:type context: dict
:return: RequestResponse object with a success code reflecting whether the ransomware was launched.
:rtype: RequestResponse
"""
command_payload = request[-1]
return self.send_command(given_command=C2Command.DATA_EXFILTRATION, command_options=command_payload)
def _remote_terminal_action(request: RequestFormat, context: Dict) -> RequestResponse:
"""Agent Action - Sends a TERMINAL C2Command to the C2 Beacon with the given parameters.
:param request: Request with one element containing a dict of parameters for the configure method.
:type request: RequestFormat
:param context: additional context for resolving this action, currently unused
:type context: dict
:return: RequestResponse object with a success code reflecting whether the ransomware was launched.
:rtype: RequestResponse
"""
command_payload = request[-1]
return self.send_command(given_command=C2Command.TERMINAL, command_options=command_payload)
rm.add_request(
name="ransomware_configure",
request_type=RequestType(func=_configure_ransomware_action),
)
rm.add_request(
name="ransomware_launch",
request_type=RequestType(func=_launch_ransomware_action),
)
rm.add_request(
name="terminal_command",
request_type=RequestType(func=_remote_terminal_action),
)
rm.add_request(
name="exfiltrate",
request_type=RequestType(func=_data_exfiltration_action),
)
return rm
def __init__(self, **kwargs):
kwargs["name"] = "C2Server"
super().__init__(**kwargs)
self.run()
def _handle_command_output(self, payload: C2Packet) -> bool:
"""
Handles the parsing of C2 Command Output from C2 Traffic (Masquerade Packets).
Parses the Request Response from the given C2Packet's payload attribute (Inherited from Data packet).
This RequestResponse is then stored in the C2 Server class attribute self.current_command_output.
If the payload attribute does not contain a RequestResponse, then an error will be raised in syslog and
the self.current_command_output is updated to reflect the error.
:param payload: The OUTPUT C2 Payload
:type payload: C2Packet
:return: Returns True if the self.current_command_output was updated, false otherwise.
:rtype Bool:
"""
self.sys_log.info(f"{self.name}: Received command response from C2 Beacon: {payload}.")
command_output = payload.payload
if not isinstance(command_output, RequestResponse):
self.sys_log.warning(f"{self.name}: C2 Server received invalid command response: {command_output}.")
self.current_command_output = RequestResponse(
status="failure", data={"Reason": "Received unexpected C2 Response."}
)
return False
self.current_command_output = command_output
return True
def _handle_keep_alive(self, payload: C2Packet, session_id: Optional[str]) -> bool:
"""
Handles receiving and sending keep alive payloads. This method is only called if we receive a keep alive.
Abstract method inherited from abstract C2.
In the C2 Server implementation of this method the following logic is performed:
1. The ``self.c2_connection_active`` is set to True. (Indicates that we're received a connection)
2. The received keep alive (Payload parameter) is then resolved by _resolve_keep_alive.
3. After the keep alive is resolved, a keep alive is sent back to confirm connection.
This is because the C2 Server is the listener and thus will only ever receive packets from
the C2 Beacon rather than the other way around.
The C2 Beacon/Server communication is akin to that of a real-world reverse shells.
Returns False if a keep alive was unable to be sent.
Returns True if a keep alive was successfully sent or already has been sent this timestep.
:param payload: The Keep Alive payload received.
:type payload: C2Packet
:param session_id: The transport session_id that the payload originates from.
:type session_id: str
:return: True if the keep alive was successfully handled, false otherwise.
:rtype: Bool
"""
self.sys_log.info(f"{self.name}: Keep Alive Received. Attempting to resolve the remote connection details.")
self.c2_connection_active = True # Sets the connection to active
self.c2_session = self.software_manager.session_manager.sessions_by_uuid[session_id]
if self._resolve_keep_alive(payload, session_id) == False:
self.sys_log.warning(f"{self.name}: Keep Alive Could not be resolved correctly. Refusing Keep Alive.")
return False
self.sys_log.info(f"{self.name}: Remote connection successfully established: {self.c2_remote_connection}.")
self.sys_log.debug(f"{self.name}: Attempting to send Keep Alive response back to {self.c2_remote_connection}.")
# If this method returns true then we have sent successfully sent a keep alive response back.
return self._send_keep_alive(session_id)
@validate_call
def send_command(self, given_command: C2Command, command_options: Dict) -> RequestResponse:
"""
Sends a C2 command to the C2 Beacon using the given parameters.
C2 Command | Command Synopsis
---------------------|------------------------
RANSOMWARE_CONFIGURE | Configures an installed ransomware script based on the passed parameters.
RANSOMWARE_LAUNCH | Launches the installed ransomware script.
DATA_EXFILTRATION | Utilises the FTP Service to exfiltrate data back to the C2 Server.
TERMINAL | Executes a command via the terminal installed on the C2 Beacons Host.
Currently, these commands leverage the pre-existing capability of other applications.
However, the commands are sent via the network rather than the game layer which
grants more opportunity to the blue agent to prevent attacks.
Additionally, future editions of primAITE may expand the C2 repertoire to allow for
more complex red agent behaviour such as establishing further fall back channels
or introduce red applications that are only installable via C2 Servers. (T1105)
For more information on the impact of these commands please refer to the terminal
and the ransomware applications.
:param given_command: The C2 command to be sent to the C2 Beacon.
:type given_command: C2Command.
:param command_options: The relevant C2 Beacon parameters.
:type command_options: Dict
:return: Returns the Request Response of the C2 Beacon's host terminal service execute method.
:rtype: RequestResponse
"""
if not isinstance(given_command, C2Command):
self.sys_log.warning(f"{self.name}: Received unexpected C2 command. Unable to send command.")
return RequestResponse(
status="failure", data={"Reason": "Received unexpected C2Command. Unable to send command."}
)
connection_status: tuple[bool, RequestResponse] = self._check_connection()
if connection_status[0] is False:
return connection_status[1]
setup_success, command_options = self._command_setup(given_command, command_options)
if setup_success is False:
self.sys_log.warning(
f"{self.name}: Failed to perform necessary C2 Server setup for given command: {given_command}."
)
return RequestResponse(
status="failure", data={"Reason": "Failed to perform necessary C2 Server setup for given command."}
)
self.sys_log.info(f"{self.name}: Attempting to send command {given_command}.")
command_packet = self._craft_packet(
c2_payload=C2Payload.INPUT, c2_command=given_command, command_options=command_options.model_dump()
)
if self.send(
payload=command_packet,
dest_ip_address=self.c2_remote_connection,
session_id=self.c2_session.uuid,
dest_port=self.c2_config.masquerade_port,
ip_protocol=self.c2_config.masquerade_protocol,
):
self.sys_log.info(f"{self.name}: Successfully sent {given_command}.")
self.sys_log.info(f"{self.name}: Awaiting command response {given_command}.")
# If the command output was handled currently, the self.current_command_output will contain the RequestResponse.
if self.current_command_output is None:
return RequestResponse(
status="failure", data={"Reason": "Command sent to the C2 Beacon but no response was ever received."}
)
return self.current_command_output
def _command_setup(self, given_command: C2Command, command_options: dict) -> tuple[bool, CommandOpts]:
"""
Performs any necessary C2 Server setup needed to perform certain commands.
This includes any option validation and any other required setup.
The following table details any C2 Server prequisites for following commands.
C2 Command | Command Service/Application Requirements
---------------------|-----------------------------------------
RANSOMWARE_CONFIGURE | N/A
RANSOMWARE_LAUNCH | N/A
DATA_EXFILTRATION | FTP Server & File system folder
TERMINAL | N/A
Currently, only the data exfiltration command require the C2 Server
to perform any necessary setup. Specifically, the Data Exfiltration command requires
the C2 Server to have an running FTP Server service as well as a folder for
storing any exfiltrated data.
:param given_command: Any C2 Command.
:type given_command: C2Command.
:param command_options: The relevant command parameters.
:type command_options: Dict
:returns: Tuple containing a success bool if the setup was successful and the validated c2 opts.
:rtype: tuple[bool, CommandOpts]
"""
server_setup_success: bool = True
if given_command == C2Command.DATA_EXFILTRATION: # Data exfiltration setup
# Validating command options
command_options = ExfilOpts.model_validate(command_options)
if self._host_ftp_server is None:
self.sys_log.warning(f"{self.name}: Unable to setup the FTP Server for data exfiltration")
server_setup_success = False
if self.get_exfiltration_folder(command_options.exfiltration_folder_name) is None:
self.sys_log.warning(f"{self.name}: Unable to create a folder for storing exfiltration data.")
server_setup_success = False
if given_command == C2Command.TERMINAL:
# Validating command options
command_options = TerminalOpts.model_validate(command_options)
if given_command == C2Command.RANSOMWARE_CONFIGURE:
# Validating command options
command_options = RansomwareOpts.model_validate(command_options)
if given_command == C2Command.RANSOMWARE_LAUNCH:
# Validating command options
command_options = CommandOpts.model_validate(command_options)
return [server_setup_success, command_options]
def _confirm_remote_connection(self, timestep: int) -> bool:
"""Checks the suitability of the current C2 Beacon connection.
Inherited Abstract Method.
If a C2 Server has not received a keep alive within the current set
keep alive frequency (self._keep_alive_frequency) then the C2 beacons
connection is considered dead and any commands will be rejected.
This method is called on each timestep (Called by .apply_timestep)
:param timestep: The current timestep of the simulation.
:type timestep: Int
:return: Returns False if the C2 beacon is considered dead. Otherwise True.
:rtype bool:
"""
if self.keep_alive_inactivity > self.c2_config.keep_alive_frequency:
self.sys_log.info(f"{self.name}: C2 Beacon connection considered dead due to inactivity.")
self.sys_log.debug(
f"{self.name}: Did not receive expected keep alive connection from {self.c2_remote_connection}"
f"{self.name}: Expected at timestep: {timestep} due to frequency: {self.c2_config.keep_alive_frequency}"
f"{self.name}: Last Keep Alive received at {(timestep - self.keep_alive_inactivity)}"
)
self._reset_c2_connection()
return False
return True
# Abstract method inherited from abstract C2.
# C2 Servers do not currently receive any input commands from the C2 beacon.
def _handle_command_input(self, payload: C2Packet) -> None:
"""Defining this method (Abstract method inherited from abstract C2) in order to instantiate the class.
C2 Servers currently do not receive input commands coming from the C2 Beacons.
:param payload: The incoming C2Packet
:type payload: C2Packet.
"""
self.sys_log.warning(f"{self.name}: C2 Server received an unexpected INPUT payload: {payload}")
pass
def show(self, markdown: bool = False):
"""
Prints a table of the current C2 attributes on a C2 Server.
Displays the current values of the following C2 attributes:
``C2 Connection Active``:
If the C2 Server has established connection with a C2 Beacon.
``C2 Remote Connection``:
The IP of the C2 Beacon. (Configured by upon receiving a keep alive.)
``Current Masquerade Protocol``:
The current protocol that the C2 Traffic is using. (e.g TCP/UDP)
``Current Masquerade Port``:
The current port that the C2 Traffic is using. (e.g HTTP (Port 80))
:param markdown: If True, outputs the table in markdown format. Default is False.
"""
table = PrettyTable(
["C2 Connection Active", "C2 Remote Connection", "Current Masquerade Protocol", "Current Masquerade Port"]
)
if markdown:
table.set_style(MARKDOWN)
table.align = "l"
table.title = f"{self.name} Running Status"
table.add_row(
[
self.c2_connection_active,
self.c2_remote_connection,
self.c2_config.masquerade_protocol,
self.c2_config.masquerade_port,
]
)
print(table)

24
src/primaite/simulator/system/applications/red_applications/ransomware_script.py
View File

@@ -2,6 +2,8 @@
from ipaddress import IPv4Address
from typing import Dict, Optional
from prettytable import MARKDOWN, PrettyTable
from primaite.interface.request import RequestFormat, RequestResponse
from primaite.simulator.core import RequestManager, RequestType
from primaite.simulator.network.transmission.network_layer import IPProtocol
@@ -169,3 +171,25 @@ class RansomwareScript(Application, identifier="RansomwareScript"):
else:
self.sys_log.warning("Attack Attempted to launch too quickly")
return False
def show(self, markdown: bool = False):
"""
Prints a table of the current status of the Ransomware Script.
Displays the current values of the following Ransomware Attributes:
``server_ip_address`:
The IP of the target database.
``payload``:
The payload (type of attack) to be sent to the database.
:param markdown: If True, outputs the table in markdown format. Default is False.
"""
table = PrettyTable(["Target Server IP Address", "Payload"])
if markdown:
table.set_style(MARKDOWN)
table.align = "l"
table.title = f"{self.name} Running Status"
table.add_row([self.server_ip_address, self.payload])
print(table)

56
src/primaite/simulator/system/services/ftp/ftp_client.py
View File

@@ -1,8 +1,10 @@
# © Crown-owned copyright 2024, Defence Science and Technology Laboratory UK
from ipaddress import IPv4Address
from typing import Optional
from typing import Dict, Optional
from primaite import getLogger
from primaite.interface.request import RequestFormat, RequestResponse
from primaite.simulator.core import RequestManager, RequestType
from primaite.simulator.file_system.file_system import File
from primaite.simulator.network.protocols.ftp import FTPCommand, FTPPacket, FTPStatusCode
from primaite.simulator.network.transmission.network_layer import IPProtocol
@@ -28,6 +30,58 @@ class FTPClient(FTPServiceABC):
super().__init__(**kwargs)
self.start()
def _init_request_manager(self) -> RequestManager:
"""
Initialise the request manager.
More information in user guide and docstring for SimComponent._init_request_manager.
"""
rm = super()._init_request_manager()
def _send_data_request(request: RequestFormat, context: Dict) -> RequestResponse:
"""
Request for sending data via the ftp_client using the request options parameters.
:param request: Request with one element containing a dict of parameters for the send method.
:type request: RequestFormat
:param context: additional context for resolving this action, currently unused
:type context: dict
:return: RequestResponse object with a success code reflecting whether the configuration could be applied.
:rtype: RequestResponse
"""
dest_ip = request[-1].get("dest_ip_address")
dest_ip = None if dest_ip is None else IPv4Address(dest_ip)
# Missing FTP Options results is an automatic failure.
src_folder = request[-1].get("src_folder_name", None)
src_file_name = request[-1].get("src_file_name", None)
dest_folder = request[-1].get("dest_folder_name", None)
dest_file_name = request[-1].get("dest_file_name", None)
if not self.file_system.access_file(folder_name=src_folder, file_name=src_file_name):
self.sys_log.debug(
f"{self.name}: Received a FTP Request to transfer file: {src_file_name} to Remote IP: {dest_ip}."
)
return RequestResponse(
status="failure",
data={
"reason": "Unable to locate given file on local file system. Perhaps given options are invalid?"
},
)
return RequestResponse.from_bool(
self.send_file(
dest_ip_address=dest_ip,
src_folder_name=src_folder,
src_file_name=src_file_name,
dest_folder_name=dest_folder,
dest_file_name=dest_file_name,
)
)
rm.add_request("send", request_type=RequestType(func=_send_data_request)),
return rm
def _process_ftp_command(self, payload: FTPPacket, session_id: Optional[str] = None, **kwargs) -> FTPPacket:
"""
Process the command in the FTP Packet.

76
tests/assets/configs/basic_c2_setup.yaml Normal file
View File

@@ -0,0 +1,76 @@
# Basic Switched network
#
# -------------- -------------- --------------
# | node_a |------| switch_1 |------| node_b |
# -------------- -------------- --------------
#
io_settings:
save_step_metadata: false
save_pcap_logs: true
save_sys_logs: true
sys_log_level: WARNING
agent_log_level: INFO
save_agent_logs: true
write_agent_log_to_terminal: True
game:
max_episode_length: 256
ports:
- ARP
- DNS
- HTTP
- POSTGRES_SERVER
protocols:
- ICMP
- TCP
- UDP
simulation:
network:
nodes:
- type: switch
hostname: switch_1
num_ports: 8
- hostname: node_a
type: computer
ip_address: 192.168.10.21
subnet_mask: 255.255.255.0
default_gateway: 192.168.10.1
applications:
- type: C2Server
options:
listen_on_ports:
- 80
- 53
- 21
- hostname: node_b
type: computer
ip_address: 192.168.10.22
subnet_mask: 255.255.255.0
default_gateway: 192.168.10.1
applications:
- type: C2Beacon
options:
c2_server_ip_address: 192.168.10.21
keep_alive_frequency: 5
masquerade_protocol: TCP
masquerade_port: HTTP
listen_on_ports:
- 80
- 53
- 21
links:
- endpoint_a_hostname: switch_1
endpoint_a_port: 1
endpoint_b_hostname: node_a
endpoint_b_port: 1
bandwidth: 200
- endpoint_a_hostname: switch_1
endpoint_a_port: 2
endpoint_b_hostname: node_b
endpoint_b_port: 1
bandwidth: 200

9
tests/conftest.py
View File

@@ -458,6 +458,11 @@ def game_and_agent():
{"type": "HOST_NIC_DISABLE"},
{"type": "NETWORK_PORT_ENABLE"},
{"type": "NETWORK_PORT_DISABLE"},
{"type": "CONFIGURE_C2_BEACON"},
{"type": "C2_SERVER_RANSOMWARE_LAUNCH"},
{"type": "C2_SERVER_RANSOMWARE_CONFIGURE"},
{"type": "C2_SERVER_TERMINAL_COMMAND"},
{"type": "C2_SERVER_DATA_EXFILTRATE"},
{"type": "NODE_ACCOUNTS_CHANGE_PASSWORD"},
{"type": "SSH_TO_REMOTE"},
{"type": "SESSIONS_REMOTE_LOGOFF"},
@@ -472,12 +477,14 @@ def game_and_agent():
"applications": [
{"application_name": "WebBrowser"},
{"application_name": "DoSBot"},
{"application_name": "C2Server"},
],
"folders": [{"folder_name": "downloads", "files": [{"file_name": "cat.png"}]}],
},
{
"node_name": "server_1",
"services": [{"service_name": "DNSServer"}],
"applications": [{"application_name": "C2Beacon"}],
},
{"node_name": "server_2", "services": [{"service_name": "WebServer"}]},
{"node_name": "router"},
@@ -485,7 +492,7 @@ def game_and_agent():
max_folders_per_node=2,
max_files_per_folder=2,
max_services_per_node=2,
max_applications_per_node=2,
max_applications_per_node=3,
max_nics_per_node=2,
max_acl_rules=10,
protocols=["TCP", "UDP", "ICMP"],

203
tests/integration_tests/game_layer/actions/test_c2_suite_actions.py Normal file
View File

@@ -0,0 +1,203 @@
# © Crown-owned copyright 2024, Defence Science and Technology Laboratory UK
from ipaddress import IPv4Address
from typing import Tuple
import pytest
from primaite.game.agent.interface import ProxyAgent
from primaite.game.game import PrimaiteGame
from primaite.simulator.file_system.file_system_item_abc import FileSystemItemHealthStatus
from primaite.simulator.network.hardware.base import UserManager
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 ACLAction
from primaite.simulator.network.transmission.transport_layer import Port
from primaite.simulator.system.applications.red_applications.c2.c2_beacon import C2Beacon
from primaite.simulator.system.applications.red_applications.c2.c2_server import C2Command, C2Server
from primaite.simulator.system.services.database.database_service import DatabaseService
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
@pytest.fixture
def game_and_agent_fixture(game_and_agent):
"""Create a game with a simple agent that can be controlled by the tests."""
game, agent = game_and_agent
router = game.simulation.network.get_node_by_hostname("router")
router.acl.add_rule(action=ACLAction.PERMIT, src_port=Port.HTTP, dst_port=Port.HTTP, position=4)
router.acl.add_rule(action=ACLAction.PERMIT, src_port=Port.DNS, dst_port=Port.DNS, position=5)
router.acl.add_rule(action=ACLAction.PERMIT, src_port=Port.FTP, dst_port=Port.FTP, position=6)
c2_server_host = game.simulation.network.get_node_by_hostname("client_1")
c2_server_host.software_manager.install(software_class=C2Server)
c2_server: C2Server = c2_server_host.software_manager.software["C2Server"]
c2_server.run()
return (game, agent)
def test_c2_beacon_default(game_and_agent_fixture: Tuple[PrimaiteGame, ProxyAgent]):
"""Tests that a Red Agent can install, configure and establish a C2 Beacon (default params)."""
game, agent = game_and_agent_fixture
# Installing C2 Beacon on Server_1
server_1: Server = game.simulation.network.get_node_by_hostname("server_1")
action = (
"NODE_APPLICATION_INSTALL",
{"node_id": 1, "application_name": "C2Beacon"},
)
agent.store_action(action)
game.step()
assert agent.history[-1].response.status == "success"
action = (
"CONFIGURE_C2_BEACON",
{
"node_id": 1,
"config": {
"c2_server_ip_address": "10.0.1.2",
"keep_alive_frequency": 5,
"masquerade_protocol": "TCP",
"masquerade_port": "HTTP",
},
},
)
agent.store_action(action)
game.step()
assert agent.history[-1].response.status == "success"
action = (
"NODE_APPLICATION_EXECUTE",
{"node_id": 1, "application_id": 0},
)
agent.store_action(action)
game.step()
assert agent.history[-1].response.status == "success"
# Asserting that we've confirmed our connection
c2_beacon: C2Beacon = server_1.software_manager.software["C2Beacon"]
assert c2_beacon.c2_connection_active == True
def test_c2_server_ransomware(game_and_agent_fixture: Tuple[PrimaiteGame, ProxyAgent]):
"""Tests that a Red Agent can install a RansomwareScript, Configure and launch all via C2 Server actions."""
game, agent = game_and_agent_fixture
# Installing a C2 Beacon on server_1
server_1: Server = game.simulation.network.get_node_by_hostname("server_1")
server_1.software_manager.install(C2Beacon)
# Installing a database on Server_2 for the ransomware to attack
server_2: Server = game.simulation.network.get_node_by_hostname("server_2")
server_2.software_manager.install(DatabaseService)
server_2.software_manager.software["DatabaseService"].start()
# Configuring the C2 to connect to client 1 (C2 Server)
c2_beacon: C2Beacon = server_1.software_manager.software["C2Beacon"]
c2_beacon.configure(c2_server_ip_address=IPv4Address("10.0.1.2"))
c2_beacon.establish()
assert c2_beacon.c2_connection_active == True
# C2 Action 1: Installing the RansomwareScript & Database client via Terminal
action = (
"C2_SERVER_TERMINAL_COMMAND",
{
"node_id": 0,
"ip_address": None,
"account": {
"username": "admin",
"password": "admin",
},
"commands": [
["software_manager", "application", "install", "RansomwareScript"],
["software_manager", "application", "install", "DatabaseClient"],
],
},
)
agent.store_action(action)
game.step()
assert agent.history[-1].response.status == "success"
action = (
"C2_SERVER_RANSOMWARE_CONFIGURE",
{
"node_id": 0,
"config": {"server_ip_address": "10.0.2.3", "payload": "ENCRYPT"},
},
)
agent.store_action(action)
game.step()
assert agent.history[-1].response.status == "success"
# Stepping a few timesteps to allow for the RansowmareScript to finish installing.
action = ("DONOTHING", {})
agent.store_action(action)
game.step()
game.step()
game.step()
action = (
"C2_SERVER_RANSOMWARE_LAUNCH",
{
"node_id": 0,
},
)
agent.store_action(action)
game.step()
assert agent.history[-1].response.status == "success"
database_file = server_2.software_manager.file_system.get_file("database", "database.db")
assert database_file.health_status == FileSystemItemHealthStatus.CORRUPT
def test_c2_server_data_exfiltration(game_and_agent_fixture: Tuple[PrimaiteGame, ProxyAgent]):
"""Tests that a Red Agent can extract a database.db file via C2 Server actions."""
game, agent = game_and_agent_fixture
# Installing a C2 Beacon on server_1
server_1: Server = game.simulation.network.get_node_by_hostname("server_1")
server_1.software_manager.install(C2Beacon)
# Installing a database on Server_2 (creates a database.db file.)
server_2: Server = game.simulation.network.get_node_by_hostname("server_2")
server_2.software_manager.install(DatabaseService)
server_2.software_manager.software["DatabaseService"].start()
# Configuring the C2 to connect to client 1 (C2 Server)
c2_beacon: C2Beacon = server_1.software_manager.software["C2Beacon"]
c2_beacon.configure(c2_server_ip_address=IPv4Address("10.0.1.2"))
c2_beacon.establish()
assert c2_beacon.c2_connection_active == True
# Selecting a target file to steal: database.db
# Server 2 ip : 10.0.2.3
database_file = server_2.software_manager.file_system.get_file(folder_name="database", file_name="database.db")
assert database_file is not None
# C2 Action: Data exfiltrate.
action = (
"C2_SERVER_DATA_EXFILTRATE",
{
"node_id": 0,
"target_file_name": "database.db",
"target_folder_name": "database",
"exfiltration_folder_name": "spoils",
"target_ip_address": "10.0.2.3",
"account": {
"username": "admin",
"password": "admin",
},
},
)
agent.store_action(action)
game.step()
assert server_1.file_system.access_file(folder_name="spoils", file_name="database.db")
client_1 = game.simulation.network.get_node_by_hostname("client_1")
assert client_1.file_system.access_file(folder_name="spoils", file_name="database.db")

554
tests/integration_tests/system/red_applications/test_c2_suite_integration.py Normal file
View File

@@ -0,0 +1,554 @@
# © Crown-owned copyright 2024, Defence Science and Technology Laboratory UK
from ipaddress import IPv4Address
from typing import Tuple
import pytest
import yaml
from primaite.game.agent.interface import ProxyAgent
from primaite.game.game import PrimaiteGame
from primaite.simulator.file_system.file_system_item_abc import FileSystemItemHealthStatus
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 AccessControlList, ACLAction, Router
from primaite.simulator.network.hardware.nodes.network.switch import Switch
from primaite.simulator.network.transmission.network_layer import IPProtocol
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.c2.c2_beacon import C2Beacon
from primaite.simulator.system.applications.red_applications.c2.c2_server import C2Command, C2Server
from primaite.simulator.system.applications.red_applications.ransomware_script import RansomwareScript
from primaite.simulator.system.services.database.database_service import DatabaseService
from primaite.simulator.system.services.dns.dns_server import DNSServer
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.web_server.web_server import WebServer
from tests import TEST_ASSETS_ROOT
@pytest.fixture(scope="function")
def basic_network() -> Network:
network = Network()
# Creating two generic nodes for the C2 Server and the C2 Beacon.
node_a = Computer(
hostname="node_a",
ip_address="192.168.0.2",
subnet_mask="255.255.255.252",
default_gateway="192.168.0.1",
start_up_duration=0,
)
node_a.power_on()
node_a.software_manager.get_open_ports()
node_a.software_manager.install(software_class=C2Server)
node_b = Computer(
hostname="node_b",
ip_address="192.168.255.2",
subnet_mask="255.255.255.248",
default_gateway="192.168.255.1",
start_up_duration=0,
)
node_b.power_on()
node_b.software_manager.install(software_class=C2Beacon)
# Creating a generic computer for testing remote terminal connections.
node_c = Computer(
hostname="node_c",
ip_address="192.168.255.3",
subnet_mask="255.255.255.248",
default_gateway="192.168.255.1",
start_up_duration=0,
)
node_c.power_on()
# Creating a router to sit between node 1 and node 2.
router = Router(hostname="router", num_ports=3, start_up_duration=0)
# Default allow all.
router.acl.add_rule(action=ACLAction.PERMIT)
router.power_on()
# Creating switches for each client.
switch_1 = Switch(hostname="switch_1", num_ports=6, start_up_duration=0)
switch_1.power_on()
# Connecting the switches to the router.
router.configure_port(port=1, ip_address="192.168.0.1", subnet_mask="255.255.255.252")
network.connect(endpoint_a=router.network_interface[1], endpoint_b=switch_1.network_interface[6])
switch_2 = Switch(hostname="switch_2", num_ports=6, start_up_duration=0)
switch_2.power_on()
network.connect(endpoint_a=router.network_interface[2], endpoint_b=switch_2.network_interface[6])
router.configure_port(port=2, ip_address="192.168.255.1", subnet_mask="255.255.255.248")
router.enable_port(1)
router.enable_port(2)
# Connecting the node to each switch
network.connect(node_a.network_interface[1], switch_1.network_interface[1])
network.connect(node_b.network_interface[1], switch_2.network_interface[1])
network.connect(node_c.network_interface[1], switch_2.network_interface[2])
return network
def setup_c2(given_network: Network):
"""Installs the C2 Beacon & Server, configures and then returns."""
computer_a: Computer = given_network.get_node_by_hostname("node_a")
c2_server: C2Server = computer_a.software_manager.software.get("C2Server")
computer_a.software_manager.install(DatabaseService)
computer_a.software_manager.software["DatabaseService"].start()
computer_b: Computer = given_network.get_node_by_hostname("node_b")
c2_beacon: C2Beacon = computer_b.software_manager.software.get("C2Beacon")
computer_b.software_manager.install(DatabaseClient)
computer_b.software_manager.software["DatabaseClient"].configure(server_ip_address=IPv4Address("192.168.0.2"))
computer_b.software_manager.software["DatabaseClient"].run()
c2_beacon.configure(c2_server_ip_address="192.168.0.2", keep_alive_frequency=2)
c2_server.run()
c2_beacon.establish()
return given_network, computer_a, c2_server, computer_b, c2_beacon
def test_c2_suite_setup_receive(basic_network):
"""Test that C2 Beacon can successfully establish connection with the C2 Server."""
network: Network = basic_network
network, computer_a, c2_server, computer_b, c2_beacon = setup_c2(network)
# Asserting that the c2 beacon has established a c2 connection
assert c2_beacon.c2_connection_active is True
# Asserting that the c2 server has established a c2 connection.
assert c2_server.c2_connection_active is True
assert c2_server.c2_remote_connection == IPv4Address("192.168.255.2")
for i in range(50):
network.apply_timestep(i)
assert c2_beacon.c2_connection_active is True
assert c2_server.c2_connection_active is True
def test_c2_suite_keep_alive_inactivity(basic_network):
"""Tests that C2 Beacon disconnects from the C2 Server after inactivity."""
network: Network = basic_network
network, computer_a, c2_server, computer_b, c2_beacon = setup_c2(network)
c2_beacon.apply_timestep(0)
assert c2_beacon.keep_alive_inactivity == 1
# Keep Alive successfully sent and received upon the 2nd timestep.
c2_beacon.apply_timestep(1)
assert c2_beacon.keep_alive_inactivity == 0
assert c2_beacon.c2_connection_active == True
# Now we turn off the c2 server (Thus preventing a keep alive)
c2_server.close()
c2_beacon.apply_timestep(2)
assert c2_beacon.keep_alive_inactivity == 1
c2_beacon.apply_timestep(3)
# C2 Beacon resets it's connections back to default.
assert c2_beacon.keep_alive_inactivity == 0
assert c2_beacon.c2_connection_active == False
assert c2_beacon.operating_state == ApplicationOperatingState.CLOSED
def test_c2_suite_configure_request(basic_network):
"""Tests that the request system can be used to successfully setup a c2 suite."""
network: Network = basic_network
network, computer_a, c2_server, computer_b, c2_beacon = setup_c2(network)
# Testing Via Requests:
c2_server.run()
network.apply_timestep(0)
c2_beacon_config = {
"c2_server_ip_address": "192.168.0.2",
"keep_alive_frequency": 5,
"masquerade_protocol": "TCP",
"masquerade_port": "HTTP",
}
network.apply_request(["node", "node_b", "application", "C2Beacon", "configure", c2_beacon_config])
network.apply_timestep(0)
network.apply_request(["node", "node_b", "application", "C2Beacon", "execute"])
assert c2_beacon.c2_connection_active is True
assert c2_server.c2_connection_active is True
assert c2_server.c2_remote_connection == IPv4Address("192.168.255.2")
def test_c2_suite_ransomware_commands(basic_network):
"""Tests the Ransomware commands can be used to configure & launch ransomware via Requests."""
# Setting up the network:
network: Network = basic_network
network, computer_a, c2_server, computer_b, c2_beacon = setup_c2(network)
# Testing Via Requests:
computer_b.software_manager.install(software_class=RansomwareScript)
ransomware_config = {"server_ip_address": "192.168.0.2"}
network.apply_request(["node", "node_a", "application", "C2Server", "ransomware_configure", ransomware_config])
ransomware_script: RansomwareScript = computer_b.software_manager.software["RansomwareScript"]
assert ransomware_script.server_ip_address == "192.168.0.2"
network.apply_request(["node", "node_a", "application", "C2Server", "ransomware_launch"])
database_file = computer_a.software_manager.file_system.get_file("database", "database.db")
assert database_file.health_status == FileSystemItemHealthStatus.CORRUPT
def test_c2_suite_acl_block(basic_network):
"""Tests that C2 Beacon disconnects from the C2 Server after blocking ACL rules."""
network: Network = basic_network
network, computer_a, c2_server, computer_b, c2_beacon = setup_c2(network)
computer_b.software_manager.install(software_class=RansomwareScript)
ransomware_config = {"server_ip_address": "192.168.0.2"}
router: Router = network.get_node_by_hostname("router")
c2_beacon.apply_timestep(0)
assert c2_beacon.keep_alive_inactivity == 1
# Keep Alive successfully sent and received upon the 2nd timestep.
c2_beacon.apply_timestep(1)
assert c2_beacon.keep_alive_inactivity == 0
assert c2_beacon.c2_connection_active == True
# Now we add a HTTP blocking acl (Thus preventing a keep alive)
router.acl.add_rule(action=ACLAction.DENY, src_port=Port.HTTP, dst_port=Port.HTTP, position=0)
c2_beacon.apply_timestep(2)
c2_beacon.apply_timestep(3)
# C2 Beacon resets after unable to maintain contact.
assert c2_beacon.keep_alive_inactivity == 0
assert c2_beacon.c2_connection_active == False
assert c2_beacon.operating_state == ApplicationOperatingState.CLOSED
def test_c2_suite_terminal_command_file_creation(basic_network):
"""Tests the C2 Terminal command can be used on local and remote."""
network: Network = basic_network
network, computer_a, c2_server, computer_b, c2_beacon = setup_c2(network)
computer_c: Computer = network.get_node_by_hostname("node_c")
# Asserting to demonstrate that the test files don't exist:
assert (
computer_c.software_manager.file_system.access_file(folder_name="test_folder", file_name="test_file") == False
)
assert (
computer_b.software_manager.file_system.access_file(folder_name="test_folder", file_name="test_file") == False
)
# Testing that we can create the test file and folders via the terminal command (Local C2 Terminal).
# Local file/folder creation commands.
folder_create_command = {
"commands": ["file_system", "create", "folder", "test_folder"],
"username": "admin",
"password": "admin",
"ip_address": None,
}
c2_server.send_command(C2Command.TERMINAL, command_options=folder_create_command)
file_create_command = {
"commands": ["file_system", "create", "file", "test_folder", "test_file", "True"],
"username": "admin",
"password": "admin",
"ip_address": None,
}
c2_server.send_command(C2Command.TERMINAL, command_options=file_create_command)
assert computer_b.software_manager.file_system.access_file(folder_name="test_folder", file_name="test_file") == True
assert c2_beacon.terminal_session is not None
# Testing that we can create the same test file/folders via on node 3 via a remote terminal.
file_remote_create_command = {
"commands": [
["file_system", "create", "folder", "test_folder"],
["file_system", "create", "file", "test_folder", "test_file", "True"],
],
"username": "admin",
"password": "admin",
"ip_address": "192.168.255.3",
}
c2_server.send_command(C2Command.TERMINAL, command_options=file_remote_create_command)
assert computer_c.software_manager.file_system.access_file(folder_name="test_folder", file_name="test_file") == True
assert c2_beacon.terminal_session is not None
def test_c2_suite_acl_bypass(basic_network):
"""Tests that C2 Beacon can be reconfigured to connect C2 Server to bypass blocking ACL rules.
1. This Test first configures a router to block HTTP traffic and asserts the following:
1. C2 Beacon and C2 Server are unable to maintain connection
2. Traffic is confirmed to be blocked by the ACL rule.
2. Next the C2 Beacon is re-configured to use FTP which is permitted by the ACL and asserts the following;
1. The C2 Beacon and C2 Server re-establish connection
2. The ACL rule has not prevent any further traffic.
3. A test file create command is sent & it's output confirmed
3. The ACL is then re-configured to block FTP traffic and asserts the following:
1. C2 Beacon and C2 Server are unable to maintain connection
2. Traffic is confirmed to be blocked by the ACL rule.
4. Next the C2 Beacon is re-configured to use HTTP which is permitted by the ACL and asserts the following;
1. The C2 Beacon and C2 Server re-establish connection
2. The ACL rule has not prevent any further traffic.
3. A test file create command is sent & it's output confirmed
"""
network: Network = basic_network
network, computer_a, c2_server, computer_b, c2_beacon = setup_c2(network)
router: Router = network.get_node_by_hostname("router")
################ Confirm Default Setup #########################
# Permitting all HTTP & FTP traffic
router.acl.add_rule(action=ACLAction.PERMIT, src_port=Port.HTTP, dst_port=Port.HTTP, position=0)
router.acl.add_rule(action=ACLAction.PERMIT, src_port=Port.FTP, dst_port=Port.FTP, position=1)
c2_beacon.apply_timestep(0)
assert c2_beacon.keep_alive_inactivity == 1
# Keep Alive successfully sent and received upon the 2nd timestep.
c2_beacon.apply_timestep(1)
assert c2_beacon.keep_alive_inactivity == 0
assert c2_beacon.c2_connection_active == True
################ Denying HTTP Traffic #########################
# Now we add a HTTP blocking acl (Thus preventing a keep alive)
router.acl.add_rule(action=ACLAction.DENY, src_port=Port.HTTP, dst_port=Port.HTTP, position=0)
blocking_acl: AccessControlList = router.acl.acl[0]
# Asserts to show the C2 Suite is unable to maintain connection:
network.apply_timestep(2)
network.apply_timestep(3)
c2_packets_blocked = blocking_acl.match_count
assert c2_packets_blocked != 0
assert c2_beacon.c2_connection_active is False
# Stepping one more time to confirm that the C2 server drops its connection
network.apply_timestep(4)
assert c2_server.c2_connection_active is False
################ Configuring C2 to use FTP #####################
# Reconfiguring the c2 beacon to now use FTP
c2_beacon.configure(
c2_server_ip_address="192.168.0.2",
keep_alive_frequency=2,
masquerade_port=Port.FTP,
masquerade_protocol=IPProtocol.TCP,
)
c2_beacon.establish()
################ Confirming connection via FTP #####################
# Confirming we've re-established connection
assert c2_beacon.c2_connection_active is True
assert c2_server.c2_connection_active is True
# Confirming that we can send commands:
ftp_file_create_command = {
"commands": [
["file_system", "create", "folder", "test_folder"],
["file_system", "create", "file", "test_folder", "ftp_test_file", "True"],
],
"username": "admin",
"password": "admin",
"ip_address": None,
}
c2_server.send_command(C2Command.TERMINAL, command_options=ftp_file_create_command)
assert (
computer_b.software_manager.file_system.access_file(folder_name="test_folder", file_name="ftp_test_file")
== True
)
# Confirming we can maintain connection
# Stepping twenty timesteps in the network
i = 4 # We're already at the 4th timestep (starting at timestep 4)
for i in range(20):
network.apply_timestep(i)
# Confirming HTTP ACL ineffectiveness (C2 Bypass)
# Asserting that the ACL hasn't caught more traffic and the c2 connection is still active
assert c2_packets_blocked == blocking_acl.match_count
assert c2_server.c2_connection_active is True
assert c2_beacon.c2_connection_active is True
################ Denying FTP Traffic & Enable HTTP #########################
# Blocking FTP and re-permitting HTTP:
router.acl.add_rule(action=ACLAction.PERMIT, src_port=Port.HTTP, dst_port=Port.HTTP, position=0)
router.acl.add_rule(action=ACLAction.DENY, src_port=Port.FTP, dst_port=Port.FTP, position=1)
blocking_acl: AccessControlList = router.acl.acl[1]
# Asserts to show the C2 Suite is unable to maintain connection:
network.apply_timestep(25)
network.apply_timestep(26)
c2_packets_blocked = blocking_acl.match_count
assert c2_packets_blocked != 0
assert c2_beacon.c2_connection_active is False
# Stepping one more time to confirm that the C2 server drops its connection
network.apply_timestep(27)
assert c2_server.c2_connection_active is False
################ Configuring C2 to use HTTP #####################
# Reconfiguring the c2 beacon to now use HTTP Again
c2_beacon.configure(
c2_server_ip_address="192.168.0.2",
keep_alive_frequency=2,
masquerade_port=Port.HTTP,
masquerade_protocol=IPProtocol.TCP,
)
c2_beacon.establish()
################ Confirming connection via HTTP #####################
# Confirming we've re-established connection
assert c2_beacon.c2_connection_active is True
assert c2_server.c2_connection_active is True
# Confirming that we can send commands
http_folder_create_command = {
"commands": ["file_system", "create", "folder", "test_folder"],
"username": "admin",
"password": "admin",
"ip_address": None,
}
c2_server.send_command(C2Command.TERMINAL, command_options=http_folder_create_command)
http_file_create_command = {
"commands": ["file_system", "create", "file", "test_folder", "http_test_file", "true"],
"username": "admin",
"password": "admin",
"ip_address": None,
}
c2_server.send_command(C2Command.TERMINAL, command_options=http_file_create_command)
assert (
computer_b.software_manager.file_system.access_file(folder_name="test_folder", file_name="http_test_file")
== True
)
assert c2_beacon.c2_connection_active is True
assert c2_server.c2_connection_active is True
# Confirming we can maintain connection
# Stepping twenty timesteps in the network
i = 28 # We're already at the 28th timestep
for i in range(20):
network.apply_timestep(i)
# Confirming FTP ACL ineffectiveness (C2 Bypass)
# Asserting that the ACL hasn't caught more traffic and the c2 connection is still active
assert c2_packets_blocked == blocking_acl.match_count
assert c2_server.c2_connection_active is True
assert c2_beacon.c2_connection_active is True
def test_c2_suite_yaml():
"""Tests that the C2 Suite is can be configured correctly via the Yaml."""
with open(TEST_ASSETS_ROOT / "configs" / "basic_c2_setup.yaml") as f:
cfg = yaml.safe_load(f)
game = PrimaiteGame.from_config(cfg)
yaml_network = game.simulation.network
computer_a: Computer = yaml_network.get_node_by_hostname("node_a")
c2_server: C2Server = computer_a.software_manager.software.get("C2Server")
computer_b: Computer = yaml_network.get_node_by_hostname("node_b")
c2_beacon: C2Beacon = computer_b.software_manager.software.get("C2Beacon")
assert c2_server.operating_state == ApplicationOperatingState.RUNNING
assert c2_beacon.c2_remote_connection == IPv4Address("192.168.10.21")
c2_beacon.establish()
# Asserting that the c2 beacon has established a c2 connection
assert c2_beacon.c2_connection_active is True
# Asserting that the c2 server has established a c2 connection.
assert c2_server.c2_connection_active is True
assert c2_server.c2_remote_connection == IPv4Address("192.168.10.22")
for i in range(50):
yaml_network.apply_timestep(i)
assert c2_beacon.c2_connection_active is True
assert c2_server.c2_connection_active is True
def test_c2_suite_file_extraction(basic_network):
"""Test that C2 Beacon can successfully exfiltrate a target file."""
network: Network = basic_network
network, computer_a, c2_server, computer_b, c2_beacon = setup_c2(network)
# Asserting that the c2 beacon has established a c2 connection
assert c2_beacon.c2_connection_active is True
# Asserting that the c2 server has established a c2 connection.
assert c2_server.c2_connection_active is True
assert c2_server.c2_remote_connection == IPv4Address("192.168.255.2")
# Creating the target file on computer_c
computer_c: Computer = network.get_node_by_hostname("node_c")
computer_c.file_system.create_folder("important_files")
computer_c.file_system.create_file(file_name="secret.txt", folder_name="important_files")
assert computer_c.file_system.access_file(folder_name="important_files", file_name="secret.txt")
# Installing an FTP Server on the same node as C2 Beacon via the terminal:
# Attempting to exfiltrate secret.txt from computer c to the C2 Server
c2_server.send_command(
given_command=C2Command.DATA_EXFILTRATION,
command_options={
"username": "admin",
"password": "admin",
"target_ip_address": "192.168.255.3",
"target_folder_name": "important_files",
"exfiltration_folder_name": "yoinked_files",
"target_file_name": "secret.txt",
},
)
# Asserting that C2 Beacon has managed to get the file
assert c2_beacon._host_file_system.access_file(folder_name="yoinked_files", file_name="secret.txt")
# Asserting that the C2 Beacon can relay it back to the C2 Server
assert c2_server._host_file_system.access_file(folder_name="yoinked_files", file_name="secret.txt")

63
tests/integration_tests/system/test_ftp_client_server.py
View File

@@ -3,6 +3,7 @@ from typing import Tuple
import pytest
from primaite.interface.request import RequestResponse
from primaite.simulator.network.hardware.nodes.host.computer import Computer
from primaite.simulator.network.hardware.nodes.host.server import Server
from primaite.simulator.system.services.ftp.ftp_client import FTPClient
@@ -105,3 +106,65 @@ def test_ftp_client_tries_to_connect_to_offline_server(ftp_client_and_ftp_server
# client should have retrieved the file
assert ftp_client.file_system.get_file(folder_name="downloads", file_name="test_file.txt") is None
def test_ftp_client_store_file_in_server_via_request_success(ftp_client_and_ftp_server):
"""
Test checks to see if the client can successfully store files in the backup server via the request manager.
"""
ftp_client, computer, ftp_server, server = ftp_client_and_ftp_server
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")
ftp_opts = {
"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": server.network_interfaces.get(next(iter(server.network_interfaces))).ip_address,
}
ftp_client.apply_request(["send", ftp_opts])
assert ftp_server.file_system.get_file(folder_name="client_1_backup", file_name="test_file.txt")
def test_ftp_client_store_file_in_server_via_request_failure(ftp_client_and_ftp_server):
"""
Test checks to see if the client fails to store files in the backup server via the request manager.
"""
ftp_client, computer, ftp_server, server = ftp_client_and_ftp_server
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")
# Purposefully misconfigured FTP Options
ftp_opts = {
"src_folder_name": "root",
"dest_ip_address": server.network_interfaces.get(next(iter(server.network_interfaces))).ip_address,
}
request_response: RequestResponse = ftp_client.apply_request(["send", ftp_opts])
assert request_response.status == "failure"
# Purposefully misconfigured FTP Options
ftp_opts = {
"src_folder_name": "root",
"src_file_name": "not_a_real_file.txt",
"dest_folder_name": "client_1_backup",
"dest_file_name": "test_file.txt",
"dest_ip_address": server.network_interfaces.get(next(iter(server.network_interfaces))).ip_address,
}
request_response: RequestResponse = ftp_client.apply_request(["send", ftp_opts])
assert request_response.status == "failure"

248
tests/unit_tests/_primaite/_simulator/_system/_applications/_red_applications/test_c2_suite.py Normal file
View File

@@ -0,0 +1,248 @@
# © Crown-owned copyright 2024, Defence Science and Technology Laboratory UK
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.transmission.network_layer import IPProtocol
from primaite.simulator.network.transmission.transport_layer import Port
from primaite.simulator.system.applications.application import ApplicationOperatingState
from primaite.simulator.system.applications.red_applications.c2.c2_beacon import C2Beacon
from primaite.simulator.system.applications.red_applications.c2.c2_server import C2Command, C2Server
@pytest.fixture(scope="function")
def basic_c2_network() -> Network:
network = Network()
# Creating two generic nodes for the C2 Server and the C2 Beacon.
computer_a = Computer(
hostname="computer_a",
ip_address="192.168.0.1",
subnet_mask="255.255.255.252",
start_up_duration=0,
)
computer_a.power_on()
computer_a.software_manager.install(software_class=C2Server)
computer_b = Computer(
hostname="computer_b", ip_address="192.168.0.2", subnet_mask="255.255.255.252", start_up_duration=0
)
computer_b.power_on()
computer_b.software_manager.install(software_class=C2Beacon)
network.connect(endpoint_a=computer_a.network_interface[1], endpoint_b=computer_b.network_interface[1])
return network
def setup_c2(given_network: Network):
"""Installs the C2 Beacon & Server, configures and then returns."""
network: Network = given_network
computer_a: Computer = network.get_node_by_hostname("computer_a")
computer_b: Computer = network.get_node_by_hostname("computer_b")
c2_beacon: C2Beacon = computer_b.software_manager.software.get("C2Beacon")
c2_server: C2Server = computer_a.software_manager.software.get("C2Server")
c2_beacon.configure(c2_server_ip_address="192.168.0.1", keep_alive_frequency=2)
c2_server.run()
c2_beacon.establish()
return network, computer_a, c2_server, computer_b, c2_beacon
def test_c2_handle_server_disconnect(basic_c2_network):
"""Tests that the C2 suite will be able handle the c2 server application closing."""
network: Network = basic_c2_network
network, computer_a, c2_server, computer_b, c2_beacon = setup_c2(network)
assert c2_beacon.c2_connection_active is True
##### C2 Server disconnecting.
# Closing the C2 Server
c2_server.close()
# Applying 10 timesteps to trigger C2 beacon keep alive
for i in range(10):
network.apply_timestep(i)
assert c2_beacon.c2_connection_active is False
assert c2_beacon.operating_state is ApplicationOperatingState.CLOSED
# C2 Beacon disconnected.
network: Network = basic_c2_network
network, computer_a, c2_server, computer_b, c2_beacon = setup_c2(network)
def test_c2_handle_beacon_disconnect(basic_c2_network):
"""Tests that the C2 suite will be able handle the c2 beacon application closing."""
network: Network = basic_c2_network
network, computer_a, c2_server, computer_b, c2_beacon = setup_c2(network)
assert c2_server.c2_connection_active is True
# Closing the C2 beacon
c2_beacon.close()
assert c2_beacon.operating_state is ApplicationOperatingState.CLOSED
# Attempting a simple C2 Server command:
file_create_command = {
"commands": [["file_system", "create", "folder", "test_folder"]],
"username": "admin",
"password": "admin",
"ip_address": None,
}
command_request_response = c2_server.send_command(C2Command.TERMINAL, command_options=file_create_command)
assert command_request_response.status == "failure"
# Despite the command failing - The C2 Server will still consider the beacon alive
# Until it does not respond within the keep alive frequency set in the last keep_alive.
assert c2_server.c2_connection_active is True
# Stepping 6 timesteps in order for the C2 server to consider the beacon dead.
for i in range(6):
network.apply_timestep(i)
assert c2_server.c2_connection_active is False
def test_c2_handle_switching_port(basic_c2_network):
"""Tests that the C2 suite will be able handle switching destination/src port."""
network: Network = basic_c2_network
network, computer_a, c2_server, computer_b, c2_beacon = setup_c2(network)
# Asserting that the c2 applications have established a c2 connection
assert c2_beacon.c2_connection_active is True
assert c2_server.c2_connection_active is True
# Assert to confirm that both the C2 server and the C2 beacon are configured correctly.
assert c2_beacon.c2_config.keep_alive_frequency is 2
assert c2_beacon.c2_config.masquerade_port is Port.HTTP
assert c2_beacon.c2_config.masquerade_protocol is IPProtocol.TCP
assert c2_server.c2_config.keep_alive_frequency is 2
assert c2_server.c2_config.masquerade_port is Port.HTTP
assert c2_server.c2_config.masquerade_protocol is IPProtocol.TCP
# Configuring the C2 Beacon.
c2_beacon.configure(
c2_server_ip_address="192.168.0.1",
keep_alive_frequency=2,
masquerade_port=Port.FTP,
masquerade_protocol=IPProtocol.TCP,
)
# Asserting that the c2 applications have established a c2 connection
assert c2_beacon.c2_connection_active is True
assert c2_server.c2_connection_active is True
# Assert to confirm that both the C2 server and the C2 beacon
# Have reconfigured their C2 settings.
assert c2_beacon.c2_config.masquerade_port is Port.FTP
assert c2_beacon.c2_config.masquerade_protocol is IPProtocol.TCP
assert c2_server.c2_config.masquerade_port is Port.FTP
assert c2_server.c2_config.masquerade_protocol is IPProtocol.TCP
def test_c2_handle_switching_frequency(basic_c2_network):
"""Tests that the C2 suite will be able handle switching keep alive frequency."""
network: Network = basic_c2_network
network, computer_a, c2_server, computer_b, c2_beacon = setup_c2(network)
# Asserting that the c2 beacon has established a c2 connection
assert c2_beacon.c2_connection_active is True
network: Network = basic_c2_network
network, computer_a, c2_server, computer_b, c2_beacon = setup_c2(network)
# Asserting that the c2 applications have established a c2 connection
assert c2_beacon.c2_connection_active is True
assert c2_server.c2_connection_active is True
# Assert to confirm that both the C2 server and the C2 beacon are configured correctly.
assert c2_beacon.c2_config.keep_alive_frequency is 2
assert c2_server.c2_config.keep_alive_frequency is 2
# Configuring the C2 Beacon.
c2_beacon.configure(c2_server_ip_address="192.168.0.1", keep_alive_frequency=10)
# Asserting that the c2 applications have established a c2 connection
assert c2_beacon.c2_connection_active is True
assert c2_server.c2_connection_active is True
# Assert to confirm that both the C2 server and the C2 beacon
# Have reconfigured their C2 settings.
assert c2_beacon.c2_config.keep_alive_frequency is 10
assert c2_server.c2_config.keep_alive_frequency is 10
# Now skipping 9 time steps to confirm keep alive inactivity
for i in range(9):
network.apply_timestep(i)
# If the keep alive reconfiguration failed then the keep alive inactivity could never reach 9
# As another keep alive would have already been sent.
assert c2_beacon.keep_alive_inactivity is 9
assert c2_server.keep_alive_inactivity is 9
network.apply_timestep(10)
assert c2_beacon.keep_alive_inactivity is 0
assert c2_server.keep_alive_inactivity is 0
def test_c2_handles_1_timestep_keep_alive(basic_c2_network):
"""Tests that the C2 suite will be able handle a C2 Beacon will a keep alive of 1 timestep."""
network: Network = basic_c2_network
network, computer_a, c2_server, computer_b, c2_beacon = setup_c2(network)
c2_beacon.configure(c2_server_ip_address="192.168.0.1", keep_alive_frequency=1)
c2_server.run()
c2_beacon.establish()
for i in range(50):
network.apply_timestep(i)
assert c2_beacon.c2_connection_active is True
assert c2_server.c2_connection_active is True
def test_c2_exfil_folder(basic_c2_network):
"""Tests that the C2 suite correctly default and setup their exfiltration_folders."""
network: Network = basic_c2_network
network, computer_a, c2_server, computer_b, c2_beacon = setup_c2(network)
c2_beacon.get_exfiltration_folder()
c2_server.get_exfiltration_folder()
assert c2_beacon.file_system.get_folder("exfiltration_folder")
assert c2_server.file_system.get_folder("exfiltration_folder")
c2_server.file_system.create_file(folder_name="test_folder", file_name="test_file")
# asserting to check that by default the c2 exfil will use "exfiltration_folder"
exfil_options = {
"username": "admin",
"password": "admin",
"target_ip_address": "192.168.0.1",
"target_folder_name": "test_folder",
"exfiltration_folder_name": None,
"target_file_name": "test_file",
}
c2_server.send_command(given_command=C2Command.DATA_EXFILTRATION, command_options=exfil_options)
assert c2_beacon.file_system.get_file(folder_name="exfiltration_folder", file_name="test_file")