#2735 - fix up node request manager and system software

This commit is contained in:
Marek Wolan
2024-08-01 13:58:35 +01:00
parent a719b24cfc
commit 78ad95fcef
3 changed files with 33 additions and 35 deletions

View File

@@ -6,7 +6,7 @@ import secrets
from abc import ABC, abstractmethod from abc import ABC, abstractmethod
from ipaddress import IPv4Address, IPv4Network from ipaddress import IPv4Address, IPv4Network
from pathlib import Path from pathlib import Path
from typing import Any, Dict, List, Optional, TypeVar, Union from typing import Any, ClassVar, Dict, List, Optional, Type, TypeVar, Union
from prettytable import MARKDOWN, PrettyTable from prettytable import MARKDOWN, PrettyTable
from pydantic import BaseModel, Field, validate_call from pydantic import BaseModel, Field, validate_call
@@ -39,7 +39,7 @@ from primaite.simulator.system.core.software_manager import SoftwareManager
from primaite.simulator.system.core.sys_log import SysLog from primaite.simulator.system.core.sys_log import SysLog
from primaite.simulator.system.processes.process import Process from primaite.simulator.system.processes.process import Process
from primaite.simulator.system.services.service import Service 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.converters import convert_dict_enum_keys_to_enum_values
from primaite.utils.validators import IPV4Address from primaite.utils.validators import IPV4Address
@@ -897,6 +897,10 @@ class UserManager(Service):
table.add_row([user.username, user.is_admin, user.disabled]) table.add_row([user.username, user.is_admin, user.disabled])
print(table.get_string(sortby="Username")) print(table.get_string(sortby="Username"))
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: def _is_last_admin(self, username: str) -> bool:
return username in self.admins and len(self.admins) == 1 return username in self.admins and len(self.admins) == 1
@@ -1100,9 +1104,6 @@ class UserSessionManager(Service):
This class handles authentication, session management, and session timeouts for users interacting with the Node. This class handles authentication, session management, and session timeouts for users interacting with the Node.
""" """
node: Node
"""The node associated with this UserSessionManager."""
local_session: Optional[UserSession] = None local_session: Optional[UserSession] = None
"""The current local user session, if any.""" """The current local user session, if any."""
@@ -1183,7 +1184,7 @@ class UserSessionManager(Service):
if markdown: if markdown:
table.set_style(MARKDOWN) table.set_style(MARKDOWN)
table.align = "l" table.align = "l"
table.title = f"{self.node.hostname} User Sessions" table.title = f"{self.parent.hostname} User Sessions"
def _add_session_to_table(user_session: UserSession): def _add_session_to_table(user_session: UserSession):
""" """
@@ -1472,6 +1473,9 @@ class Node(SimComponent):
red_scan_countdown: int = 0 red_scan_countdown: int = 0
"Time steps until reveal to red scan is complete." "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): def __init__(self, **kwargs):
""" """
Initialize the Node with various components and managers. Initialize the Node with various components and managers.
@@ -1496,21 +1500,10 @@ class Node(SimComponent):
dns_server=kwargs.get("dns_server"), dns_server=kwargs.get("dns_server"),
) )
super().__init__(**kwargs) super().__init__(**kwargs)
self._install_system_software()
self.session_manager.node = self self.session_manager.node = self
self.session_manager.software_manager = self.software_manager self.session_manager.software_manager = self.software_manager
self.software_manager.install(UserSessionManager, node=self)
self._request_manager.add_request(
"sessions", RequestType(func=self.user_session_manager._request_manager)
) # noqa
self.software_manager.install(UserManager)
self._request_manager.add_request("accounts", RequestType(func=self.user_manager._request_manager)) # noqa
self.user_manager.add_user(username="admin", password="admin", is_admin=True, bypass_can_perform_action=True)
self._install_system_software()
@property @property
def user_manager(self) -> UserManager: def user_manager(self) -> UserManager:
"""The Nodes User Manager.""" """The Nodes User Manager."""
@@ -1767,8 +1760,6 @@ class Node(SimComponent):
"services": {svc.name: svc.describe_state() for svc in self.services.values()}, "services": {svc.name: svc.describe_state() for svc in self.services.values()},
"process": {proc.name: proc.describe_state() for proc in self.processes.values()}, "process": {proc.name: proc.describe_state() for proc in self.processes.values()},
"revealed_to_red": self.revealed_to_red, "revealed_to_red": self.revealed_to_red,
"user_manager": self.user_manager.describe_state(),
"user_session_manager": self.user_session_manager.describe_state(),
} }
) )
return state return state
@@ -2134,6 +2125,11 @@ class Node(SimComponent):
# for process_id in self.processes: # for process_id in self.processes:
# self.processes[process_id] # 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: def __contains__(self, item: Any) -> bool:
if isinstance(item, Service): if isinstance(item, Service):
return item.uuid in self.services return item.uuid in self.services

View File

@@ -5,7 +5,13 @@ from ipaddress import IPv4Address
from typing import Any, ClassVar, Dict, Optional from typing import Any, ClassVar, Dict, Optional
from primaite import getLogger 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.hardware.node_operating_state import NodeOperatingState
from primaite.simulator.network.transmission.data_link_layer import Frame from primaite.simulator.network.transmission.data_link_layer import Frame
from primaite.simulator.system.applications.application import ApplicationOperatingState from primaite.simulator.system.applications.application import ApplicationOperatingState
@@ -306,8 +312,8 @@ class HostNode(Node):
"NTPClient": NTPClient, "NTPClient": NTPClient,
"WebBrowser": WebBrowser, "WebBrowser": WebBrowser,
"NMAP": NMAP, "NMAP": NMAP,
# "UserSessionManager": UserSessionManager, "UserSessionManager": UserSessionManager,
# "UserManager": UserManager, "UserManager": UserManager,
} }
"""List of system software that is automatically installed on nodes.""" """List of system software that is automatically installed on nodes."""
@@ -340,16 +346,6 @@ class HostNode(Node):
""" """
return self.software_manager.software.get("ARP") 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)
def default_gateway_hello(self): def default_gateway_hello(self):
""" """
Sends a hello message to the default gateway to establish connectivity and resolve the gateway's MAC address. Sends a hello message to the default gateway to establish connectivity and resolve the gateway's MAC address.

View File

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