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

Merged PR 496: Implement User, UserManager, and UserSessionManager

Browse Source 
## Summary
*Replace this text with an explanation of what the changes are and how you implemented them. Can this impact any other parts of the codebase that we should keep in mind?*

## Test process
*How have you tested this (if applicable)?*

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

Related work items: #2735, #2778
This commit is contained in:
Christopher McCarthy
2024-08-02 14:45:10 +00:00
parent aab47289ac 71d8c6ea63
commit e275a9376b
17 changed files with 1315 additions and 34 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
.azure/azure-ci-build-pipeline.yaml
View File

@@ -102,9 +102,7 @@ stages:
version: '2.1.x'
- script: |
coverage run -m --source=primaite pytest -v -o junit_family=xunit2 --junitxml=junit/test-results.xml --cov-fail-under=80
coverage xml -o coverage.xml -i
coverage html -d htmlcov -i
python run_test_and_coverage.py
displayName: 'Run tests and code coverage'
# Run the notebooks

27
docs/source/configuration/simulation/nodes/common/common_node_attributes.rst
View File

@@ -53,3 +53,30 @@ The number of time steps required to occur in order for the node to cycle from `
Optional. Default value is ``3``.
The number of time steps required to occur in order for the node to cycle from ``ON`` to ``SHUTTING_DOWN`` and then finally ``OFF``.
``users``
---------
The list of pre-existing users that are additional to the default admin user (``username=admin``, ``password=admin``).
Additional users are configured as an array nd must contain a ``username``, ``password``, and can contain an optional
boolean ``is_admin``.
Example of adding two additional users to a node:
.. code-block:: yaml
simulation:
network:
nodes:
- hostname: client_1
type: computer
ip_address: 192.168.10.11
subnet_mask: 255.255.255.0
default_gateway: 192.168.10.1
users:
- username: jane.doe
password: '1234'
is_admin: true
- username: john.doe
password: password_1
is_admin: false

206
docs/source/simulation_components/network/base_hardware.rst
View File

@@ -97,8 +97,8 @@ Node Behaviours/Functions
- **receive_frame()**: Handles the processing of incoming network frames.
- **apply_timestep()**: Advances the state of the node according to the simulation timestep.
- **power_on()**: Initiates the node, enabling all connected Network Interfaces and starting all Services and
Applications, taking into account the `start_up_duration`.
- **power_off()**: Stops the node's operations, adhering to the `shut_down_duration`.
Applications, taking into account the ``start_up_duration``.
- **power_off()**: Stops the node's operations, adhering to the ``shut_down_duration``.
- **ping()**: Sends ICMP echo requests to a specified IP address to test connectivity.
- **has_enabled_network_interface()**: Checks if the node has any network interfaces enabled, facilitating network
communication.
@@ -109,3 +109,205 @@ Node Behaviours/Functions
The Node class handles installation of system software, network connectivity, frame processing, system logging, and
power states. It establishes baseline functionality while allowing subclassing to model specific node types like hosts,
routers, firewalls etc. The flexible architecture enables composing complex network topologies.
User, UserManager, and UserSessionManager
=========================================
The ``base.py`` module also includes essential classes for managing users and their sessions within the PrimAITE
simulation. These are the ``User``, ``UserManager``, and ``UserSessionManager`` classes. The base ``Node`` class comes
with ``UserManager``, and ``UserSessionManager`` classes pre-installed.
User Class
----------
The ``User`` class represents a user in the system. It includes attributes such as ``username``, ``password``,
``disabled``, and ``is_admin`` to define the user's credentials and status.
Example Usage
^^^^^^^^^^^^^
Creating a user:
.. code-block:: python
user = User(username="john_doe", password="12345")
UserManager Class
-----------------
The ``UserManager`` class handles user management tasks such as creating users, authenticating them, changing passwords,
and enabling or disabling user accounts. It maintains a dictionary of users and provides methods to manage them
effectively.
Example Usage
^^^^^^^^^^^^^
Creating a ``UserManager`` instance and adding a user:
.. code-block:: python
user_manager = UserManager()
user_manager.add_user(username="john_doe", password="12345")
Authenticating a user:
.. code-block:: python
user = user_manager.authenticate_user(username="john_doe", password="12345")
UserSessionManager Class
------------------------
The ``UserSessionManager`` class manages user sessions, including local and remote sessions. It handles session creation,
timeouts, and provides methods for logging users in and out.
Example Usage
^^^^^^^^^^^^^
Creating a ``UserSessionManager`` instance and logging a user in locally:
.. code-block:: python
session_manager = UserSessionManager()
session_id = session_manager.local_login(username="john_doe", password="12345")
Logging a user out:
.. code-block:: python
session_manager.local_logout()
Practical Examples
------------------
Below are unit tests which act as practical examples illustrating how to use the ``User``, ``UserManager``, and
``UserSessionManager`` classes within the context of a client-server network simulation.
Setting up a Client-Server Network
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
.. code-block:: python
from typing import Tuple
from uuid import uuid4
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
@pytest.fixture(scope="function")
def client_server_network() -> Tuple[Computer, Server, Network]:
network = Network()
client = Computer(
hostname="client",
ip_address="192.168.1.2",
subnet_mask="255.255.255.0",
default_gateway="192.168.1.1",
start_up_duration=0,
)
client.power_on()
server = Server(
hostname="server",
ip_address="192.168.1.3",
subnet_mask="255.255.255.0",
default_gateway="192.168.1.1",
start_up_duration=0,
)
server.power_on()
network.connect(client.network_interface[1], server.network_interface[1])
return client, server, network
Local Login Success
^^^^^^^^^^^^^^^^^^^
.. code-block:: python
def test_local_login_success(client_server_network):
client, server, network = client_server_network
assert not client.user_session_manager.local_user_logged_in
client.user_session_manager.local_login(username="admin", password="admin")
assert client.user_session_manager.local_user_logged_in
Local Login Failure
^^^^^^^^^^^^^^^^^^^
.. code-block:: python
def test_local_login_failure(client_server_network):
client, server, network = client_server_network
assert not client.user_session_manager.local_user_logged_in
client.user_session_manager.local_login(username="jane.doe", password="12345")
assert not client.user_session_manager.local_user_logged_in
Adding a New User and Successful Local Login
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
.. code-block:: python
def test_new_user_local_login_success(client_server_network):
client, server, network = client_server_network
assert not client.user_session_manager.local_user_logged_in
client.user_manager.add_user(username="jane.doe", password="12345")
client.user_session_manager.local_login(username="jane.doe", password="12345")
assert client.user_session_manager.local_user_logged_in
Clearing Previous Login on New Local Login
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
.. code-block:: python
def test_new_local_login_clears_previous_login(client_server_network):
client, server, network = client_server_network
assert not client.user_session_manager.local_user_logged_in
current_session_id = client.user_session_manager.local_login(username="admin", password="admin")
assert client.user_session_manager.local_user_logged_in
assert client.user_session_manager.local_session.user.username == "admin"
client.user_manager.add_user(username="jane.doe", password="12345")
new_session_id = client.user_session_manager.local_login(username="jane.doe", password="12345")
assert client.user_session_manager.local_user_logged_in
assert client.user_session_manager.local_session.user.username == "jane.doe"
assert new_session_id != current_session_id
Persistent Login for the Same User
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
.. code-block:: python
def test_new_local_login_attempt_same_uses_persists(client_server_network):
client, server, network = client_server_network
assert not client.user_session_manager.local_user_logged_in
current_session_id = client.user_session_manager.local_login(username="admin", password="admin")
assert client.user_session_manager.local_user_logged_in
assert client.user_session_manager.local_session.user.username == "admin"
new_session_id = client.user_session_manager.local_login(username="admin", password="admin")
assert client.user_session_manager.local_user_logged_in
assert client.user_session_manager.local_session.user.username == "admin"
assert new_session_id == current_session_id

22
run_test_and_coverage.py Normal file
View File

@@ -0,0 +1,22 @@
# © Crown-owned copyright 2024, Defence Science and Technology Laboratory UK
import subprocess
import sys
from typing import Any
def run_command(command: Any):
"""Runs a command and returns the exit code."""
result = subprocess.run(command, shell=True)
if result.returncode != 0:
sys.exit(result.returncode)
# Run pytest with coverage
run_command(
"coverage run -m --source=primaite pytest -v -o junit_family=xunit2 "
"--junitxml=junit/test-results.xml --cov-fail-under=80"
)
# Generate coverage reports if tests passed
run_command("coverage xml -o coverage.xml -i")
run_command("coverage html -d htmlcov -i")

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

@@ -18,7 +18,7 @@ from primaite.game.agent.scripted_agents.tap001 import TAP001
from primaite.game.science import graph_has_cycle, topological_sort
from primaite.simulator import SIM_OUTPUT
from primaite.simulator.network.airspace import AirSpaceFrequency
from primaite.simulator.network.hardware.base import NetworkInterface, NodeOperatingState
from primaite.simulator.network.hardware.base import NetworkInterface, NodeOperatingState, UserManager
from primaite.simulator.network.hardware.nodes.host.computer import Computer
from primaite.simulator.network.hardware.nodes.host.host_node import NIC
from primaite.simulator.network.hardware.nodes.host.server import Printer, Server
@@ -269,6 +269,7 @@ class PrimaiteGame:
for node_cfg in nodes_cfg:
n_type = node_cfg["type"]
new_node = None
if n_type == "computer":
new_node = Computer(
hostname=node_cfg["hostname"],
@@ -318,6 +319,11 @@ class PrimaiteGame:
msg = f"invalid node type {n_type} in config"
_LOGGER.error(msg)
raise ValueError(msg)
if "users" in node_cfg and new_node.software_manager.software.get("UserManager"):
user_manager: UserManager = new_node.software_manager.software["UserManager"] # noqa
for user_cfg in node_cfg["users"]:
user_manager.add_user(**user_cfg, bypass_can_perform_action=True)
if "services" in node_cfg:
for service_cfg in node_cfg["services"]:
new_service = None

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

@@ -6,10 +6,10 @@ import secrets
from abc import ABC, abstractmethod
from ipaddress import IPv4Address, IPv4Network
from pathlib import Path
from typing import Any, ClassVar, Dict, Optional, TypeVar, Union
from typing import Any, ClassVar, Dict, List, Optional, Type, TypeVar, Union
from prettytable import MARKDOWN, PrettyTable
from pydantic import BaseModel, Field
from pydantic import BaseModel, Field, validate_call
from primaite import getLogger
from primaite.exceptions import NetworkError
@@ -22,6 +22,7 @@ from primaite.simulator.network.hardware.node_operating_state import NodeOperati
from primaite.simulator.network.nmne import NMNEConfig
from primaite.simulator.network.transmission.data_link_layer import Frame
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
from primaite.simulator.system.core.packet_capture import PacketCapture
from primaite.simulator.system.core.session_manager import SessionManager
@@ -29,7 +30,7 @@ from primaite.simulator.system.core.software_manager import SoftwareManager
from primaite.simulator.system.core.sys_log import SysLog
from primaite.simulator.system.processes.process import Process
from primaite.simulator.system.services.service import Service
from primaite.simulator.system.software import IOSoftware
from primaite.simulator.system.software import IOSoftware, Software
from primaite.utils.converters import convert_dict_enum_keys_to_enum_values
from primaite.utils.validators import IPV4Address
@@ -788,6 +789,650 @@ class Link(SimComponent):
self.current_load = 0.0
class User(SimComponent):
"""
Represents a user in the PrimAITE system.
:ivar username: The username of the user
:ivar password: The password of the user
:ivar disabled: Boolean flag indicating whether the user is disabled
:ivar is_admin: Boolean flag indicating whether the user has admin privileges
"""
username: str
"""The username of the user"""
password: str
"""The password of the user"""
disabled: bool = False
"""Boolean flag indicating whether the user is disabled"""
is_admin: bool = False
"""Boolean flag indicating whether the user has admin privileges"""
num_of_logins: int = 0
"""Counts the number of the User has logged in"""
def describe_state(self) -> Dict:
"""
Returns a dictionary representing the current state of the user.
:return: A dict containing the state of the user
"""
return self.model_dump()
class UserManager(Service):
"""
Manages users within the PrimAITE system, handling creation, authentication, and administration.
:param users: A dictionary of all users by their usernames
:param admins: A dictionary of admin users by their usernames
:param disabled_admins: A dictionary of currently disabled admin users by their usernames
"""
users: Dict[str, User] = {}
def __init__(self, **kwargs):
"""
Initializes a UserManager instanc.
:param username: The username for the default admin user
:param password: The password for the default admin user
"""
kwargs["name"] = "UserManager"
kwargs["port"] = Port.NONE
kwargs["protocol"] = IPProtocol.NONE
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()
# todo add doc about requeest schemas
rm.add_request(
"change_password",
RequestType(
func=lambda request, context: RequestResponse.from_bool(
self.change_user_password(username=request[0], current_password=request[1], new_password=request[2])
)
),
)
return rm
def describe_state(self) -> Dict:
"""
Returns the state of the UserManager along with the number of users and admins.
:return: A dict containing detailed state information
"""
state = super().describe_state()
state.update({"total_users": len(self.users), "total_admins": len(self.admins) + len(self.disabled_admins)})
state["users"] = {k: v.describe_state() for k, v in self.users.items()}
return state
def show(self, markdown: bool = False):
"""
Display the Users.
:param markdown: Whether to display the table in Markdown format or not. Default is `False`.
"""
table = PrettyTable(["Username", "Admin", "Disabled"])
if markdown:
table.set_style(MARKDOWN)
table.align = "l"
table.title = f"{self.sys_log.hostname} User Manager"
for user in self.users.values():
table.add_row([user.username, user.is_admin, user.disabled])
print(table.get_string(sortby="Username"))
@property
def non_admins(self) -> Dict[str, User]:
"""
Returns a dictionary of all enabled non-admin users.
:return: A dictionary with usernames as keys and User objects as values for non-admin, non-disabled users.
"""
return {k: v for k, v in self.users.items() if not v.is_admin and not v.disabled}
@property
def disabled_non_admins(self) -> Dict[str, User]:
"""
Returns a dictionary of all disabled non-admin users.
:return: A dictionary with usernames as keys and User objects as values for non-admin, disabled users.
"""
return {k: v for k, v in self.users.items() if not v.is_admin and v.disabled}
@property
def admins(self) -> Dict[str, User]:
"""
Returns a dictionary of all enabled admin users.
:return: A dictionary with usernames as keys and User objects as values for admin, non-disabled users.
"""
return {k: v for k, v in self.users.items() if v.is_admin and not v.disabled}
@property
def disabled_admins(self) -> Dict[str, User]:
"""
Returns a dictionary of all disabled admin users.
:return: A dictionary with usernames as keys and User objects as values for admin, disabled users.
"""
return {k: v for k, v in self.users.items() if v.is_admin and v.disabled}
def install(self) -> None:
"""Setup default user during first-time installation."""
self.add_user(username="admin", password="admin", is_admin=True, bypass_can_perform_action=True)
def _is_last_admin(self, username: str) -> bool:
return username in self.admins and len(self.admins) == 1
def add_user(
self, username: str, password: str, is_admin: bool = False, bypass_can_perform_action: bool = False
) -> bool:
"""
Adds a new user to the system.
:param username: The username for the new user
:param password: The password for the new user
:param is_admin: Flag indicating if the new user is an admin
:return: True if user was successfully added, False otherwise
"""
if not bypass_can_perform_action and not self._can_perform_action():
return False
if username in self.users:
self.sys_log.info(f"{self.name}: Failed to create new user {username} as this user name already exists")
return False
user = User(username=username, password=password, is_admin=is_admin)
self.users[username] = user
self.sys_log.info(f"{self.name}: Added new {'admin' if is_admin else 'user'}: {username}")
return True
def authenticate_user(self, username: str, password: str) -> Optional[User]:
"""
Authenticates a user's login attempt.
:param username: The username of the user trying to log in
:param password: The password provided by the user
:return: The User object if authentication is successful, None otherwise
"""
if not self._can_perform_action():
return None
user = self.users.get(username)
if user and not user.disabled and user.password == password:
self.sys_log.info(f"{self.name}: User authenticated: {username}")
return user
self.sys_log.info(f"{self.name}: Authentication failed for: {username}")
return None
def change_user_password(self, username: str, current_password: str, new_password: str) -> bool:
"""
Changes a user's password.
:param username: The username of the user changing their password
:param current_password: The current password of the user
:param new_password: The new password for the user
:return: True if the password was changed successfully, False otherwise
"""
if not self._can_perform_action():
return False
user = self.users.get(username)
if user and user.password == current_password:
user.password = new_password
self.sys_log.info(f"{self.name}: Password changed for {username}")
return True
self.sys_log.info(f"{self.name}: Password change failed for {username}")
return False
def disable_user(self, username: str) -> bool:
"""
Disables a user account, preventing them from logging in.
:param username: The username of the user to disable
:return: True if the user was disabled successfully, False otherwise
"""
if not self._can_perform_action():
return False
if username in self.users and not self.users[username].disabled:
if self._is_last_admin(username):
self.sys_log.info(f"{self.name}: Cannot disable User {username} as they are the only enabled admin")
return False
self.users[username].disabled = True
self.sys_log.info(f"{self.name}: User disabled: {username}")
return True
self.sys_log.info(f"{self.name}: Failed to disable user: {username}")
return False
def enable_user(self, username: str) -> bool:
"""
Enables a previously disabled user account.
:param username: The username of the user to enable
:return: True if the user was enabled successfully, False otherwise
"""
if username in self.users and self.users[username].disabled:
self.users[username].disabled = False
self.sys_log.info(f"{self.name}: User enabled: {username}")
return True
self.sys_log.info(f"{self.name}: Failed to enable user: {username}")
return False
class UserSession(SimComponent):
"""
Represents a user session on the Node.
This class manages the state of a user session, including the user, session start, last active step,
and end step. It also indicates whether the session is local.
:ivar user: The user associated with this session.
:ivar start_step: The timestep when the session was started.
:ivar last_active_step: The last timestep when the session was active.
:ivar end_step: The timestep when the session ended, if applicable.
:ivar local: Indicates if the session is local. Defaults to True.
"""
user: User
"""The user associated with this session."""
start_step: int
"""The timestep when the session was started."""
last_active_step: int
"""The last timestep when the session was active."""
end_step: Optional[int] = None
"""The timestep when the session ended, if applicable."""
local: bool = True
"""Indicates if the session is a local session or a remote session. Defaults to True as a local session."""
@classmethod
def create(cls, user: User, timestep: int) -> UserSession:
"""
Creates a new instance of UserSession.
This class method initialises a user session with the given user and timestep.
:param user: The user associated with this session.
:param timestep: The timestep when the session is created.
:return: An instance of UserSession.
"""
user.num_of_logins += 1
return UserSession(user=user, start_step=timestep, last_active_step=timestep)
def describe_state(self) -> Dict:
"""
Describes the current state of the user session.
:return: A dictionary representing the state of the user session.
"""
return self.model_dump()
class RemoteUserSession(UserSession):
"""
Represents a remote user session on the Node.
This class extends the UserSession class to include additional attributes and methods specific to remote sessions.
:ivar remote_ip_address: The IP address of the remote user.
:ivar local: Indicates that this is not a local session. Always set to False.
"""
remote_ip_address: IPV4Address
"""The IP address of the remote user."""
local: bool = False
"""Indicates that this is not a local session. Always set to False."""
@classmethod
def create(cls, user: User, timestep: int, remote_ip_address: IPV4Address) -> RemoteUserSession: # noqa
"""
Creates a new instance of RemoteUserSession.
This class method initialises a remote user session with the given user, timestep, and remote IP address.
:param user: The user associated with this session.
:param timestep: The timestep when the session is created.
:param remote_ip_address: The IP address of the remote user.
:return: An instance of RemoteUserSession.
"""
return RemoteUserSession(
user=user, start_step=timestep, last_active_step=timestep, remote_ip_address=remote_ip_address
)
def describe_state(self) -> Dict:
"""
Describes the current state of the remote user session.
This method extends the base describe_state method to include the remote IP address.
:return: A dictionary representing the state of the remote user session.
"""
state = super().describe_state()
state["remote_ip_address"] = str(self.remote_ip_address)
return state
class UserSessionManager(Service):
"""
Manages user sessions on a Node, including local and remote sessions.
This class handles authentication, session management, and session timeouts for users interacting with the Node.
"""
local_session: Optional[UserSession] = None
"""The current local user session, if any."""
remote_sessions: Dict[str, RemoteUserSession] = {}
"""A dictionary of active remote user sessions."""
historic_sessions: List[UserSession] = Field(default_factory=list)
"""A list of historic user sessions."""
local_session_timeout_steps: int = 30
"""The number of steps before a local session times out due to inactivity."""
remote_session_timeout_steps: int = 5
"""The number of steps before a remote session times out due to inactivity."""
max_remote_sessions: int = 3
"""The maximum number of concurrent remote sessions allowed."""
current_timestep: int = 0
"""The current timestep in the simulation."""
def __init__(self, **kwargs):
"""
Initializes a UserSessionManager instance.
:param username: The username for the default admin user
:param password: The password for the default admin user
"""
kwargs["name"] = "UserSessionManager"
kwargs["port"] = Port.NONE
kwargs["protocol"] = IPProtocol.NONE
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()
# todo add doc about requeest schemas
rm.add_request(
"remote_login",
RequestType(
func=lambda request, context: RequestResponse.from_bool(
self.remote_login(username=request[0], password=request[1], remote_ip_address=request[2])
)
),
)
rm.add_request(
"remote_logout",
RequestType(
func=lambda request, context: RequestResponse.from_bool(
self.remote_logout(remote_session_id=request[0])
)
),
)
return rm
def show(self, markdown: bool = False, include_session_id: bool = False, include_historic: bool = False):
"""
Displays a table of the user sessions on the Node.
:param markdown: Whether to display the table in markdown format.
:param include_session_id: Whether to include session IDs in the table.
:param include_historic: Whether to include historic sessions in the table.
"""
headers = ["Session ID", "Username", "Type", "Remote IP", "Start Step", "Step Last Active", "End Step"]
if not include_session_id:
headers = headers[1:]
table = PrettyTable(headers)
if markdown:
table.set_style(MARKDOWN)
table.align = "l"
table.title = f"{self.parent.hostname} User Sessions"
def _add_session_to_table(user_session: UserSession):
"""
Adds a user session to the table for display.
This helper function determines whether the session is local or remote and formats the session data
accordingly. It then adds the session data to the table.
:param user_session: The user session to add to the table.
"""
session_type = "local"
remote_ip = ""
if isinstance(user_session, RemoteUserSession):
session_type = "remote"
remote_ip = str(user_session.remote_ip_address)
data = [
user_session.uuid,
user_session.user.username,
session_type,
remote_ip,
user_session.start_step,
user_session.last_active_step,
user_session.end_step if user_session.end_step else "",
]
if not include_session_id:
data = data[1:]
table.add_row(data)
if self.local_session is not None:
_add_session_to_table(self.local_session)
for user_session in self.remote_sessions.values():
_add_session_to_table(user_session)
if include_historic:
for user_session in self.historic_sessions:
_add_session_to_table(user_session)
print(table.get_string(sortby="Step Last Active", reversesort=True))
def describe_state(self) -> Dict:
"""
Describes the current state of the UserSessionManager.
:return: A dictionary representing the current state.
"""
state = super().describe_state()
state["active_remote_logins"] = len(self.remote_sessions)
return state
@property
def _user_manager(self) -> UserManager:
"""
Returns the UserManager instance.
:return: The UserManager instance.
"""
return self.software_manager.software["UserManager"] # noqa
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
if self.local_session:
if self.local_session.last_active_step + self.local_session_timeout_steps <= timestep:
self._timeout_session(self.local_session)
def _timeout_session(self, session: UserSession) -> None:
"""
Handles session timeout logic.
:param session: The session to be timed out.
"""
session.end_step = self.current_timestep
session_identity = session.user.username
if session.local:
self.local_session = None
session_type = "Local"
else:
self.remote_sessions.pop(session.uuid)
session_type = "Remote"
session_identity = f"{session_identity} {session.remote_ip_address}"
self.sys_log.info(f"{self.name}: {session_type} {session_identity} session timeout due to inactivity")
@property
def remote_session_limit_reached(self) -> bool:
"""
Checks if the maximum number of remote sessions has been reached.
:return: True if the limit is reached, otherwise False.
"""
return len(self.remote_sessions) >= self.max_remote_sessions
def validate_remote_session_uuid(self, remote_session_id: str) -> bool:
"""
Validates if a given remote session ID exists.
:param remote_session_id: The remote session ID to validate.
:return: True if the session ID exists, otherwise False.
"""
return remote_session_id in self.remote_sessions
def _login(
self, username: str, password: str, local: bool = True, remote_ip_address: Optional[IPv4Address] = None
) -> Optional[str]:
"""
Logs a user in either locally or remotely.
:param username: The username of the account.
:param password: The password of the account.
:param local: Whether the login is local or remote.
:param remote_ip_address: The remote IP address for remote login.
:return: The session ID if login is successful, otherwise None.
"""
if not self._can_perform_action():
return None
user = self._user_manager.authenticate_user(username=username, password=password)
if not user:
self.sys_log.info(f"{self.name}: Incorrect username or password")
return None
session_id = None
if local:
create_new_session = True
if self.local_session:
if self.local_session.user != user:
# logout the current user
self.local_logout()
else:
# not required as existing logged-in user attempting to re-login
create_new_session = False
if create_new_session:
self.local_session = UserSession.create(user=user, timestep=self.current_timestep)
session_id = self.local_session.uuid
else:
if not self.remote_session_limit_reached:
remote_session = RemoteUserSession.create(
user=user, timestep=self.current_timestep, remote_ip_address=remote_ip_address
)
session_id = remote_session.uuid
self.remote_sessions[session_id] = remote_session
self.sys_log.info(f"{self.name}: User {user.username} logged in")
return session_id
def local_login(self, username: str, password: str) -> Optional[str]:
"""
Logs a user in locally.
:param username: The username of the account.
:param password: The password of the account.
:return: The session ID if login is successful, otherwise None.
"""
return self._login(username=username, password=password, local=True)
@validate_call()
def remote_login(self, username: str, password: str, remote_ip_address: IPV4Address) -> Optional[str]:
"""
Logs a user in remotely.
:param username: The username of the account.
:param password: The password of the account.
:param remote_ip_address: The remote IP address for the remote login.
:return: The session ID if login is successful, otherwise None.
"""
return self._login(username=username, password=password, local=False, remote_ip_address=remote_ip_address)
def _logout(self, local: bool = True, remote_session_id: Optional[str] = None) -> bool:
"""
Logs a user out either locally or remotely.
:param local: Whether the logout is local or remote.
:param remote_session_id: The remote session ID for remote logout.
:return: True if logout successful, otherwise False.
"""
if not self._can_perform_action():
return False
session = None
if local and self.local_session:
session = self.local_session
session.end_step = self.current_timestep
self.local_session = None
if not local and remote_session_id:
session = self.remote_sessions.pop(remote_session_id)
if session:
self.historic_sessions.append(session)
self.sys_log.info(f"{self.name}: User {session.user.username} logged out")
return True
return False
def local_logout(self) -> bool:
"""
Logs out the current local user.
:return: True if logout successful, otherwise False.
"""
return self._logout(local=True)
def remote_logout(self, remote_session_id: str) -> bool:
"""
Logs out a remote user by session ID.
:param remote_session_id: The remote session ID.
:return: True if logout successful, otherwise False.
"""
return self._logout(local=False, remote_session_id=remote_session_id)
@property
def local_user_logged_in(self) -> bool:
"""
Checks if a local user is currently logged in.
:return: True if a local user is logged in, otherwise False.
"""
return self.local_session is not None
class Node(SimComponent):
"""
A basic Node class that represents a node on the network.
@@ -855,11 +1500,14 @@ class Node(SimComponent):
red_scan_countdown: int = 0
"Time steps until reveal to red scan is complete."
SYSTEM_SOFTWARE: ClassVar[Dict[str, Type[Software]]] = {}
"Base system software that must be preinstalled."
def __init__(self, **kwargs):
"""
Initialize the Node with various components and managers.
This method initializes the ARP cache, ICMP handler, session manager, and software manager if they are not
This method initialises the ARP cache, ICMP handler, session manager, and software manager if they are not
provided.
"""
if not kwargs.get("sys_log"):
@@ -879,9 +1527,40 @@ class Node(SimComponent):
dns_server=kwargs.get("dns_server"),
)
super().__init__(**kwargs)
self._install_system_software()
self.session_manager.node = self
self.session_manager.software_manager = self.software_manager
self._install_system_software()
@property
def user_manager(self) -> Optional[UserManager]:
"""The Nodes User Manager."""
return self.software_manager.software.get("UserManager") # noqa
@property
def user_session_manager(self) -> Optional[UserSessionManager]:
"""The Nodes User Session Manager."""
return self.software_manager.software.get("UserSessionManager") # noqa
def local_login(self, username: str, password: str) -> Optional[str]:
"""
Attempt to log in to the node uas a local user.
This method attempts to authenticate a local user with the given username and password. If successful, it
returns a session token. If authentication fails, it returns None.
:param username: The username of the account attempting to log in.
:param password: The password of the account attempting to log in.
:return: A session token if the login is successful, otherwise None.
"""
return self.user_session_manager.local_login(username, password)
def local_logout(self) -> None:
"""
Log out the current local user from the node.
This method ends the current local user's session and invalidates the session token.
"""
return self.user_session_manager.local_logout()
def ip_is_network_interface(self, ip_address: IPv4Address, enabled_only: bool = False) -> bool:
"""
@@ -936,7 +1615,7 @@ class Node(SimComponent):
@property
def fail_message(self) -> str:
"""Message that is reported when a request is rejected by this validator."""
return f"Cannot perform request on node '{self.node.hostname}' because it is not turned on."
return f"Cannot perform request on node '{self.node.hostname}' because it is not powered on."
class _NodeIsOffValidator(RequestPermissionValidator):
"""
@@ -1085,10 +1764,6 @@ class Node(SimComponent):
return rm
def _install_system_software(self):
"""Install System Software - software that is usually provided with the OS."""
pass
def describe_state(self) -> Dict:
"""
Produce a dictionary describing the current state of this object.
@@ -1477,6 +2152,11 @@ class Node(SimComponent):
# for process_id in self.processes:
# self.processes[process_id]
def _install_system_software(self) -> None:
"""Preinstall required software."""
for _, software_class in self.SYSTEM_SOFTWARE.items():
self.software_manager.install(software_class)
def __contains__(self, item: Any) -> bool:
if isinstance(item, Service):
return item.uuid in self.services

22
src/primaite/simulator/network/hardware/nodes/host/host_node.py
View File

@@ -5,7 +5,13 @@ from ipaddress import IPv4Address
from typing import Any, ClassVar, Dict, Optional
from primaite import getLogger
from primaite.simulator.network.hardware.base import IPWiredNetworkInterface, Link, Node
from primaite.simulator.network.hardware.base import (
IPWiredNetworkInterface,
Link,
Node,
UserManager,
UserSessionManager,
)
from primaite.simulator.network.hardware.node_operating_state import NodeOperatingState
from primaite.simulator.network.transmission.data_link_layer import Frame
from primaite.simulator.system.applications.application import ApplicationOperatingState
@@ -306,6 +312,8 @@ class HostNode(Node):
"NTPClient": NTPClient,
"WebBrowser": WebBrowser,
"NMAP": NMAP,
"UserSessionManager": UserSessionManager,
"UserManager": UserManager,
}
"""List of system software that is automatically installed on nodes."""
@@ -338,18 +346,6 @@ class HostNode(Node):
"""
return self.software_manager.software.get("ARP")
def _install_system_software(self):
"""
Installs the system software and network services typically found on an operating system.
This method equips the host with essential network services and applications, preparing it for various
network-related tasks and operations.
"""
for _, software_class in self.SYSTEM_SOFTWARE.items():
self.software_manager.install(software_class)
super()._install_system_software()
def default_gateway_hello(self):
"""
Sends a hello message to the default gateway to establish connectivity and resolve the gateway's MAC address.

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

@@ -4,14 +4,14 @@ from __future__ import annotations
import secrets
from enum import Enum
from ipaddress import IPv4Address, IPv4Network
from typing import Any, Dict, List, Optional, Tuple, Union
from typing import Any, ClassVar, Dict, List, Optional, Tuple, Union
from prettytable import MARKDOWN, PrettyTable
from pydantic import validate_call
from primaite.interface.request import RequestResponse
from primaite.simulator.core import RequestManager, RequestType, SimComponent
from primaite.simulator.network.hardware.base import IPWiredNetworkInterface
from primaite.simulator.network.hardware.base import IPWiredNetworkInterface, UserManager, UserSessionManager
from primaite.simulator.network.hardware.node_operating_state import NodeOperatingState
from primaite.simulator.network.hardware.nodes.network.network_node import NetworkNode
from primaite.simulator.network.protocols.arp import ARPPacket
@@ -1200,6 +1200,11 @@ class Router(NetworkNode):
RouteTable, RouterARP, and RouterICMP services.
"""
SYSTEM_SOFTWARE: ClassVar[Dict] = {
"UserSessionManager": UserSessionManager,
"UserManager": UserManager,
}
num_ports: int
network_interfaces: Dict[str, RouterInterface] = {}
"The Router Interfaces on the node."
@@ -1235,6 +1240,7 @@ class Router(NetworkNode):
resolution within the network. These services are crucial for the router's operation, enabling it to manage
network traffic efficiently.
"""
super()._install_system_software()
self.software_manager.install(RouterICMP)
icmp: RouterICMP = self.software_manager.icmp # noqa
icmp.router = self

3
src/primaite/simulator/network/hardware/nodes/network/switch.py
View File

@@ -108,6 +108,9 @@ class Switch(NetworkNode):
for i in range(1, self.num_ports + 1):
self.connect_nic(SwitchPort())
def _install_system_software(self):
pass
def show(self, markdown: bool = False):
"""
Prints a table of the SwitchPorts on the Switch.

8
src/primaite/simulator/system/core/software_manager.py
View File

@@ -103,7 +103,7 @@ class SoftwareManager:
return True
return False
def install(self, software_class: Type[IOSoftware]):
def install(self, software_class: Type[IOSoftware], **install_kwargs):
"""
Install an Application or Service.
@@ -113,7 +113,11 @@ class SoftwareManager:
self.sys_log.warning(f"Cannot install {software_class} as it is already installed")
return
software = software_class(
software_manager=self, sys_log=self.sys_log, file_system=self.file_system, dns_server=self.dns_server
software_manager=self,
sys_log=self.sys_log,
file_system=self.file_system,
dns_server=self.dns_server,
**install_kwargs,
)
software.parent = self.node
if isinstance(software, Application):

1
src/primaite/simulator/system/services/access/__init__.py Normal file
View File

@@ -0,0 +1 @@
# © Crown-owned copyright 2024, Defence Science and Technology Laboratory UK

1
src/primaite/simulator/system/services/access/user_manager.py Normal file
View File

@@ -0,0 +1 @@
# © Crown-owned copyright 2024, Defence Science and Technology Laboratory UK

1
src/primaite/simulator/system/services/access/user_session_manager.py Normal file
View File

@@ -0,0 +1 @@
# © Crown-owned copyright 2024, Defence Science and Technology Laboratory UK

2
src/primaite/simulator/system/software.py
View File

@@ -291,7 +291,7 @@ class IOSoftware(Software):
"""
if self.software_manager and self.software_manager.node.operating_state != NodeOperatingState.ON:
self.software_manager.node.sys_log.error(
f"{self.name} Error: {self.software_manager.node.hostname} is not online."
f"{self.name} Error: {self.software_manager.node.hostname} is not powered on."
)
return False
return True

34
tests/assets/configs/basic_node_with_users.yaml Normal file
View File

@@ -0,0 +1,34 @@
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
protocols:
- ICMP
- UDP
simulation:
network:
nodes:
- hostname: client_1
type: computer
ip_address: 192.168.10.11
subnet_mask: 255.255.255.0
default_gateway: 192.168.10.1
users:
- username: jane.doe
password: '1234'
is_admin: true
- username: john.doe
password: password_1
is_admin: false

26
tests/integration_tests/network/test_users_creation_from_config.py Normal file
View File

@@ -0,0 +1,26 @@
# © Crown-owned copyright 2024, Defence Science and Technology Laboratory UK
import yaml
from primaite.game.game import PrimaiteGame
from primaite.simulator.network.hardware.base import UserManager
from tests import TEST_ASSETS_ROOT
def test_users_from_config():
config_path = TEST_ASSETS_ROOT / "configs" / "basic_node_with_users.yaml"
with open(config_path, "r") as f:
config_dict = yaml.safe_load(f)
network = PrimaiteGame.from_config(cfg=config_dict).simulation.network
client_1 = network.get_node_by_hostname("client_1")
user_manager: UserManager = client_1.software_manager.software["UserManager"]
assert len(user_manager.users) == 3
assert user_manager.users["jane.doe"].password == "1234"
assert user_manager.users["jane.doe"].is_admin
assert user_manager.users["john.doe"].password == "password_1"
assert not user_manager.users["john.doe"].is_admin

274
tests/integration_tests/system/test_user_session_manager_logins.py Normal file
View File

@@ -0,0 +1,274 @@
# © Crown-owned copyright 2024, Defence Science and Technology Laboratory UK
from typing import Tuple
from uuid import uuid4
import pytest
from primaite.simulator.network.container import Network
from primaite.simulator.network.hardware.base import User
from primaite.simulator.network.hardware.nodes.host.computer import Computer
from primaite.simulator.network.hardware.nodes.host.server import Server
@pytest.fixture(scope="function")
def client_server_network() -> Tuple[Computer, Server, Network]:
network = Network()
client = Computer(
hostname="client",
ip_address="192.168.1.2",
subnet_mask="255.255.255.0",
default_gateway="192.168.1.1",
start_up_duration=0,
)
client.power_on()
server = Server(
hostname="server",
ip_address="192.168.1.3",
subnet_mask="255.255.255.0",
default_gateway="192.168.1.1",
start_up_duration=0,
)
server.power_on()
network.connect(client.network_interface[1], server.network_interface[1])
return client, server, network
def test_local_login_success(client_server_network):
client, server, network = client_server_network
assert not client.user_session_manager.local_user_logged_in
client.user_session_manager.local_login(username="admin", password="admin")
assert client.user_session_manager.local_user_logged_in
def test_login_count_increases(client_server_network):
client, server, network = client_server_network
admin_user: User = client.user_manager.users["admin"]
assert admin_user.num_of_logins == 0
client.user_session_manager.local_login(username="admin", password="admin")
assert admin_user.num_of_logins == 1
client.user_session_manager.local_login(username="admin", password="admin")
# shouldn't change as user is already logged in
assert admin_user.num_of_logins == 1
client.user_session_manager.local_logout()
client.user_session_manager.local_login(username="admin", password="admin")
assert admin_user.num_of_logins == 2
def test_local_login_failure(client_server_network):
client, server, network = client_server_network
assert not client.user_session_manager.local_user_logged_in
client.user_session_manager.local_login(username="jane.doe", password="12345")
assert not client.user_session_manager.local_user_logged_in
def test_new_user_local_login_success(client_server_network):
client, server, network = client_server_network
assert not client.user_session_manager.local_user_logged_in
client.user_manager.add_user(username="jane.doe", password="12345")
client.user_session_manager.local_login(username="jane.doe", password="12345")
assert client.user_session_manager.local_user_logged_in
def test_new_local_login_clears_previous_login(client_server_network):
client, server, network = client_server_network
assert not client.user_session_manager.local_user_logged_in
current_session_id = client.user_session_manager.local_login(username="admin", password="admin")
assert client.user_session_manager.local_user_logged_in
assert client.user_session_manager.local_session.user.username == "admin"
client.user_manager.add_user(username="jane.doe", password="12345")
new_session_id = client.user_session_manager.local_login(username="jane.doe", password="12345")
assert client.user_session_manager.local_user_logged_in
assert client.user_session_manager.local_session.user.username == "jane.doe"
assert new_session_id != current_session_id
def test_new_local_login_attempt_same_uses_persists(client_server_network):
client, server, network = client_server_network
assert not client.user_session_manager.local_user_logged_in
current_session_id = client.user_session_manager.local_login(username="admin", password="admin")
assert client.user_session_manager.local_user_logged_in
assert client.user_session_manager.local_session.user.username == "admin"
new_session_id = client.user_session_manager.local_login(username="admin", password="admin")
assert client.user_session_manager.local_user_logged_in
assert client.user_session_manager.local_session.user.username == "admin"
assert new_session_id == current_session_id
def test_remote_login_success(client_server_network):
# partial test for now until we get the terminal application in so that amn actual remote connection can be made
client, server, network = client_server_network
assert not server.user_session_manager.remote_sessions
remote_session_id = server.user_session_manager.remote_login(
username="admin", password="admin", remote_ip_address="192.168.1.10"
)
assert server.user_session_manager.validate_remote_session_uuid(remote_session_id)
server.user_session_manager.remote_logout(remote_session_id)
assert not server.user_session_manager.validate_remote_session_uuid(remote_session_id)
def test_remote_login_failure(client_server_network):
# partial test for now until we get the terminal application in so that amn actual remote connection can be made
client, server, network = client_server_network
assert not server.user_session_manager.remote_sessions
remote_session_id = server.user_session_manager.remote_login(
username="jane.doe", password="12345", remote_ip_address="192.168.1.10"
)
assert not server.user_session_manager.validate_remote_session_uuid(remote_session_id)
def test_new_user_remote_login_success(client_server_network):
client, server, network = client_server_network
server.user_manager.add_user(username="jane.doe", password="12345")
remote_session_id = server.user_session_manager.remote_login(
username="jane.doe", password="12345", remote_ip_address="192.168.1.10"
)
assert server.user_session_manager.validate_remote_session_uuid(remote_session_id)
server.user_session_manager.remote_logout(remote_session_id)
assert not server.user_session_manager.validate_remote_session_uuid(remote_session_id)
def test_max_remote_sessions_same_user(client_server_network):
client, server, network = client_server_network
remote_session_ids = [
server.user_session_manager.remote_login(username="admin", password="admin", remote_ip_address="192.168.1.10")
for _ in range(server.user_session_manager.max_remote_sessions)
]
assert all([server.user_session_manager.validate_remote_session_uuid(id) for id in remote_session_ids])
def test_max_remote_sessions_different_users(client_server_network):
client, server, network = client_server_network
remote_session_ids = []
for i in range(server.user_session_manager.max_remote_sessions):
username = str(uuid4())
password = "12345"
server.user_manager.add_user(username=username, password=password)
remote_session_ids.append(
server.user_session_manager.remote_login(
username=username, password=password, remote_ip_address="192.168.1.10"
)
)
assert all([server.user_session_manager.validate_remote_session_uuid(id) for id in remote_session_ids])
def test_max_remote_sessions_limit_reached(client_server_network):
client, server, network = client_server_network
remote_session_ids = [
server.user_session_manager.remote_login(username="admin", password="admin", remote_ip_address="192.168.1.10")
for _ in range(server.user_session_manager.max_remote_sessions)
]
assert all([server.user_session_manager.validate_remote_session_uuid(id) for id in remote_session_ids])
assert len(server.user_session_manager.remote_sessions) == server.user_session_manager.max_remote_sessions
fourth_attempt_session_id = server.user_session_manager.remote_login(
username="admin", password="admin", remote_ip_address="192.168.1.10"
)
assert not server.user_session_manager.validate_remote_session_uuid(fourth_attempt_session_id)
assert all([server.user_session_manager.validate_remote_session_uuid(id) for id in remote_session_ids])
def test_single_remote_logout_others_persist(client_server_network):
client, server, network = client_server_network
server.user_manager.add_user(username="jane.doe", password="12345")
server.user_manager.add_user(username="john.doe", password="12345")
admin_session_id = server.user_session_manager.remote_login(
username="admin", password="admin", remote_ip_address="192.168.1.10"
)
jane_session_id = server.user_session_manager.remote_login(
username="jane.doe", password="12345", remote_ip_address="192.168.1.10"
)
john_session_id = server.user_session_manager.remote_login(
username="john.doe", password="12345", remote_ip_address="192.168.1.10"
)
server.user_session_manager.remote_logout(admin_session_id)
assert not server.user_session_manager.validate_remote_session_uuid(admin_session_id)
assert server.user_session_manager.validate_remote_session_uuid(jane_session_id)
assert server.user_session_manager.validate_remote_session_uuid(john_session_id)
server.user_session_manager.remote_logout(jane_session_id)
assert not server.user_session_manager.validate_remote_session_uuid(admin_session_id)
assert not server.user_session_manager.validate_remote_session_uuid(jane_session_id)
assert server.user_session_manager.validate_remote_session_uuid(john_session_id)
server.user_session_manager.remote_logout(john_session_id)
assert not server.user_session_manager.validate_remote_session_uuid(admin_session_id)
assert not server.user_session_manager.validate_remote_session_uuid(jane_session_id)
assert not server.user_session_manager.validate_remote_session_uuid(john_session_id)