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

#2710 - Pre-commit run ahead of raising PR

Browse Source
This commit is contained in:
Charlie Crane
2024-07-15 08:22:18 +01:00
parent 2eb36149b2
commit 32c2ea0b10
3 changed files with 41 additions and 32 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
src/primaite/simulator/system/services/database/database_service.py
View File

@@ -19,9 +19,9 @@ _LOGGER = getLogger(__name__)
class DatabaseService(Service): class DatabaseService(Service):
""" """
A class for simulating a generic SQL Server service. A class for simulating a generic SQL Server service.
This class inherits from the `Service` class and provides methods to simulate a SQL database. This class inherits from the `Service` class and provides methods to simulate a SQL database.
""" """
password: Optional[str] = None password: Optional[str] = None

28
src/primaite/simulator/system/services/terminal/terminal.py
View File

@@ -124,7 +124,6 @@ class Terminal(Service):
if self.operating_state != ServiceOperatingState.RUNNING: if self.operating_state != ServiceOperatingState.RUNNING:
self.sys_log.warning("Cannot process login as service is not running") self.sys_log.warning("Cannot process login as service is not running")
return False return False
user_account = f"Username: placeholder, Password: placeholder"
if self.connection_uuid in self.user_connections: if self.connection_uuid in self.user_connections:
self.sys_log.debug("User authentication passed") self.sys_log.debug("User authentication passed")
return True return True
@@ -133,9 +132,9 @@ class Terminal(Service):
# TODO: Refactor with UserManager changes to provide correct credentials and validate. # TODO: Refactor with UserManager changes to provide correct credentials and validate.
transport_message = SSHTransportMessage.SSH_MSG_USERAUTH_REQUEST transport_message = SSHTransportMessage.SSH_MSG_USERAUTH_REQUEST
connection_message = SSHConnectionMessage.SSH_MSG_CHANNEL_OPEN connection_message = SSHConnectionMessage.SSH_MSG_CHANNEL_OPEN
payload: SSHPacket = SSHPacket(payload="login", payload: SSHPacket = SSHPacket(
transport_message=transport_message, payload="login", transport_message=transport_message, connection_message=connection_message
connection_message=connection_message) )
self.sys_log.debug(f"Sending login request to {dest_ip_address}") self.sys_log.debug(f"Sending login request to {dest_ip_address}")
self.send(payload=payload, dest_ip_address=dest_ip_address) self.send(payload=payload, dest_ip_address=dest_ip_address)
@@ -158,8 +157,11 @@ class Terminal(Service):
self.sys_log.info(f"{self.name}: Connect request for ID: {self.connection_uuid} authorised") self.sys_log.info(f"{self.name}: Connect request for ID: {self.connection_uuid} authorised")
transport_message = SSHTransportMessage.SSH_MSG_USERAUTH_SUCCESS transport_message = SSHTransportMessage.SSH_MSG_USERAUTH_SUCCESS
connection_message = SSHConnectionMessage.SSH_MSG_CHANNEL_OPEN_CONFIRMATION connection_message = SSHConnectionMessage.SSH_MSG_CHANNEL_OPEN_CONFIRMATION
new_connection = TerminalClientConnection(parent_node = self.software_manager.node, new_connection = TerminalClientConnection(
connection_id=self.connection_uuid, dest_ip_address=dest_ip_address) parent_node=self.software_manager.node,
connection_id=self.connection_uuid,
dest_ip_address=dest_ip_address,
)
self.user_connections[self.connection_uuid] = new_connection self.user_connections[self.connection_uuid] = new_connection
self.is_connected = True self.is_connected = True
@@ -178,7 +180,7 @@ class Terminal(Service):
return False return False
if self.operating_state != ServiceOperatingState.RUNNING: if self.operating_state != ServiceOperatingState.RUNNING:
self.sys_log.warning(f"Cannot process message as not running") self.sys_log.warning("Cannot process message as not running")
return False return False
self.sys_log.debug(f"Received payload: {payload} from session: {session_id}") self.sys_log.debug(f"Received payload: {payload} from session: {session_id}")
@@ -212,7 +214,7 @@ class Terminal(Service):
"""Remote login to terminal via SSH.""" """Remote login to terminal via SSH."""
if not user_account: if not user_account:
# TODO: Generic hardcoded info, will need to be updated with UserManager. # TODO: Generic hardcoded info, will need to be updated with UserManager.
user_account = f"Username: placeholder, Password: placeholder" user_account = "Username: placeholder, Password: placeholder"
# something like self.user_manager.get_user_details ? # something like self.user_manager.get_user_details ?
# Implement SSHPacket class # Implement SSHPacket class
@@ -263,15 +265,13 @@ class Terminal(Service):
software_manager.send_payload_to_session_manager( software_manager.send_payload_to_session_manager(
payload={"type": "disconnect", "connection_id": self.connection_uuid}, payload={"type": "disconnect", "connection_id": self.connection_uuid},
dest_ip_address=dest_ip_address, dest_ip_address=dest_ip_address,
dest_port=self.port dest_port=self.port,
) )
connection = self.user_connections.pop(self.connection_uuid) connection = self.user_connections.pop(self.connection_uuid)
connection.is_active = False connection.is_active = False
self.sys_log.info( self.sys_log.info(f"{self.name}: Disconnected {self.connection_uuid}")
f"{self.name}: Disconnected {self.connection_uuid}"
)
return True return True
def send( def send(
@@ -289,6 +289,4 @@ class Terminal(Service):
self.sys_log.warning(f"Cannot send commands when Operating state is {self.operating_state}!") self.sys_log.warning(f"Cannot send commands when Operating state is {self.operating_state}!")
return False return False
self.sys_log.debug(f"Sending payload: {payload}") self.sys_log.debug(f"Sending payload: {payload}")
return super().send( return super().send(payload=payload, dest_ip_address=dest_ip_address, dest_port=self.port)
payload=payload, dest_ip_address=dest_ip_address, dest_port=self.port
)

31
tests/unit_tests/_primaite/_simulator/_system/_services/test_terminal.py
View File

@@ -11,14 +11,18 @@ from primaite.simulator.system.services.service import ServiceOperatingState
from primaite.simulator.system.services.terminal.terminal import Terminal from primaite.simulator.system.services.terminal.terminal import Terminal
from primaite.simulator.system.software import SoftwareHealthState from primaite.simulator.system.software import SoftwareHealthState
@pytest.fixture(scope="function") @pytest.fixture(scope="function")
def terminal_on_computer() -> Tuple[Terminal, Computer]: def terminal_on_computer() -> Tuple[Terminal, Computer]:
computer: Computer = Computer(hostname="node_a", ip_address="192.168.0.10", subnet_mask="255.255.255.0", start_up_duration=0) computer: Computer = Computer(
hostname="node_a", ip_address="192.168.0.10", subnet_mask="255.255.255.0", start_up_duration=0
)
computer.power_on() computer.power_on()
terminal: Terminal = computer.software_manager.software.get("Terminal") terminal: Terminal = computer.software_manager.software.get("Terminal")
return [terminal, computer] return [terminal, computer]
@pytest.fixture(scope="function") @pytest.fixture(scope="function")
def basic_network() -> Network: def basic_network() -> Network:
network = Network() network = Network()
@@ -37,6 +41,7 @@ def test_terminal_creation(terminal_on_computer):
terminal, computer = terminal_on_computer terminal, computer = terminal_on_computer
terminal.describe_state() terminal.describe_state()
def test_terminal_install_default(): def test_terminal_install_default():
"""Terminal should be auto installed onto Nodes""" """Terminal should be auto installed onto Nodes"""
computer = Computer(hostname="node_a", ip_address="192.168.0.10", subnet_mask="255.255.255.0", start_up_duration=0) computer = Computer(hostname="node_a", ip_address="192.168.0.10", subnet_mask="255.255.255.0", start_up_duration=0)
@@ -44,22 +49,25 @@ def test_terminal_install_default():
assert computer.software_manager.software.get("Terminal") assert computer.software_manager.software.get("Terminal")
def test_terminal_not_on_switch(): def test_terminal_not_on_switch():
"""Ensure terminal does not auto-install to switch""" """Ensure terminal does not auto-install to switch"""
test_switch = Switch(hostname="Test") test_switch = Switch(hostname="Test")
assert not test_switch.software_manager.software.get("Terminal") assert not test_switch.software_manager.software.get("Terminal")
def test_terminal_send(basic_network): def test_terminal_send(basic_network):
"""Check that Terminal can send """ """Check that Terminal can send"""
network: Network = basic_network network: Network = basic_network
computer_a: Computer = network.get_node_by_hostname("node_a") computer_a: Computer = network.get_node_by_hostname("node_a")
terminal_a: Terminal = computer_a.software_manager.software.get("Terminal") terminal_a: Terminal = computer_a.software_manager.software.get("Terminal")
payload: SSHPacket = SSHPacket(payload="Test_Payload", payload: SSHPacket = SSHPacket(
transport_message=SSHTransportMessage.SSH_MSG_SERVICE_REQUEST, payload="Test_Payload",
connection_message=SSHConnectionMessage.SSH_MSG_CHANNEL_OPEN) transport_message=SSHTransportMessage.SSH_MSG_SERVICE_REQUEST,
connection_message=SSHConnectionMessage.SSH_MSG_CHANNEL_OPEN,
)
assert terminal_a.send(payload=payload, dest_ip_address="192.168.0.11") assert terminal_a.send(payload=payload, dest_ip_address="192.168.0.11")
@@ -91,6 +99,7 @@ def test_terminal_disconnect(basic_network):
assert terminal.is_connected is False assert terminal.is_connected is False
def test_terminal_ignores_when_off(basic_network): def test_terminal_ignores_when_off(basic_network):
"""Terminal should ignore commands when not running""" """Terminal should ignore commands when not running"""
network: Network = basic_network network: Network = basic_network
@@ -99,14 +108,16 @@ def test_terminal_ignores_when_off(basic_network):
computer_b: Computer = network.get_node_by_hostname("node_b") computer_b: Computer = network.get_node_by_hostname("node_b")
terminal_a.login(dest_ip_address="192.168.0.11") # login to computer_b terminal_a.login(dest_ip_address="192.168.0.11") # login to computer_b
assert terminal_a.is_connected is True assert terminal_a.is_connected is True
terminal_a.operating_state = ServiceOperatingState.STOPPED terminal_a.operating_state = ServiceOperatingState.STOPPED
payload: SSHPacket = SSHPacket(payload="Test_Payload", payload: SSHPacket = SSHPacket(
transport_message=SSHTransportMessage.SSH_MSG_SERVICE_REQUEST, payload="Test_Payload",
connection_message=SSHConnectionMessage.SSH_MSG_CHANNEL_DATA) transport_message=SSHTransportMessage.SSH_MSG_SERVICE_REQUEST,
connection_message=SSHConnectionMessage.SSH_MSG_CHANNEL_DATA,
)
assert not terminal_a.send(payload=payload, dest_ip_address="192.168.0.11") assert not terminal_a.send(payload=payload, dest_ip_address="192.168.0.11")