Rename Sim Actions to request

This commit is contained in:
Marek Wolan
2023-10-09 13:24:08 +01:00
parent 2722abe428
commit 5a5710c6ae
16 changed files with 266 additions and 814 deletions

File diff suppressed because one or more lines are too long

View File

@@ -11,9 +11,9 @@ from primaite import getLogger
_LOGGER = getLogger(__name__)
class ActionPermissionValidator(BaseModel):
class RequestPermissionValidator(BaseModel):
"""
Base class for action validators.
Base class for request validators.
The permissions manager is designed to be generic. So, although in the first instance the permissions
are evaluated purely on membership to AccountGroup, this class can support validating permissions based on any
@@ -22,130 +22,127 @@ class ActionPermissionValidator(BaseModel):
@abstractmethod
def __call__(self, request: List[str], context: Dict) -> bool:
"""Use the request and context paramters to decide whether the action should be permitted."""
"""Use the request and context paramters to decide whether the request should be permitted."""
pass
class AllowAllValidator(ActionPermissionValidator):
"""Always allows the action."""
class AllowAllValidator(RequestPermissionValidator):
"""Always allows the request."""
def __call__(self, request: List[str], context: Dict) -> bool:
"""Always allow the action."""
"""Always allow the request."""
return True
class Action(BaseModel):
class RequestType(BaseModel):
"""
This object stores data related to a single action.
This object stores data related to a single request type.
This includes the callable that can execute the action request, and the validator that will decide whether
the action can be performed or not.
This includes the callable that can execute the request, and the validator that will decide whether
the request can be performed or not.
"""
func: Callable[[List[str], Dict], None]
"""
``func`` is a function that accepts a request and a context dict. Typically this would be a lambda function
that invokes a class method of your SimComponent. For example if the component is a node and the action is for
that invokes a class method of your SimComponent. For example if the component is a node and the request type is for
turning it off, then the SimComponent should have a turn_off(self) method that does not need to accept any args.
Then, this Action will be given something like ``func = lambda request, context: self.turn_off()``.
Then, this request will be given something like ``func = lambda request, context: self.turn_off()``.
``func`` can also be another action manager, since ActionManager is a callable with a signature that matches what is
``func`` can also be another request manager, since RequestManager is a callable with a signature that matches what is
expected by ``func``.
"""
validator: ActionPermissionValidator = AllowAllValidator()
validator: RequestPermissionValidator = AllowAllValidator()
"""
``validator`` is an instance of `ActionPermissionValidator`. This is essentially a callable that
``validator`` is an instance of ``RequestPermissionValidator``. This is essentially a callable that
accepts `request` and `context` and returns a boolean to represent whether the permission is granted to perform
the action. The default validator will allow
the request. The default validator will allow
"""
# TODO: maybe this can be renamed to something like action selector?
# Because there are two ways it's used, to select from a list of action verbs, or to select a child object to which to
# forward the request.
class ActionManager(BaseModel):
class RequestManager(BaseModel):
"""
ActionManager is used by `SimComponent` instances to keep track of actions.
RequestManager is used by `SimComponent` instances to keep track of requests.
Its main purpose is to be a lookup from action name to action function and corresponding validation function. This
class is responsible for providing a consistent API for processing actions as well as helpful error messages.
Its main purpose is to be a lookup from request name to request function and corresponding validation function. This
class is responsible for providing a consistent API for processing requests as well as helpful error messages.
"""
actions: Dict[str, Action] = {}
"""maps action verb to an action object."""
request_types: Dict[str, RequestType] = {}
"""maps request name to an RequestType object."""
def __call__(self, request: Callable[[List[str], Dict], None], context: Dict) -> None:
"""
Process an action request.
Process an request request.
:param request: A list of strings which specify what action to take. The first string must be one of the allowed
actions, i.e. it must be a key of self.actions. The subsequent strings in the list are passed as parameters
to the action function.
:param request: A list of strings describing the request. The first string must be one of the allowed
request names, i.e. it must be a key of self.request_types. The subsequent strings in the list are passed as
parameters to the request function.
:type request: List[str]
:param context: Dictionary of additional information necessary to process or validate the request.
:type context: Dict
:raises RuntimeError: If the request parameter does not have a valid action identifier as the first item.
:raises RuntimeError: If the request parameter does not have a valid request name as the first item.
"""
action_key = request[0]
request_key = request[0]
if action_key not in self.actions:
if request_key not in self.request_types:
msg = (
f"Action request {request} could not be processed because {action_key} is not a valid action",
"within this ActionManager",
f"Request {request} could not be processed because {request_key} is not a valid request name",
"within this RequestManager",
)
_LOGGER.error(msg)
raise RuntimeError(msg)
action = self.actions[action_key]
action_options = request[1:]
request_type = self.request_types[request_key]
request_options = request[1:]
if not action.validator(action_options, context):
_LOGGER.debug(f"Action request {request} was denied due to insufficient permissions")
if not request_type.validator(request_options, context):
_LOGGER.debug(f"Request {request} was denied due to insufficient permissions")
return
action.func(action_options, context)
request_type.func(request_options, context)
def add_action(self, name: str, action: Action) -> None:
def add_request(self, name: str, request_type: RequestType) -> None:
"""
Add an action to this action manager.
Add a request type to this request manager.
:param name: The string associated to this action.
:param name: The string associated to this request.
:type name: str
:param action: Action object.
:type action: Action
:param request_type: Request type object which contains information about how to resolve request.
:type request_type: RequestType
"""
if name in self.actions:
msg = f"Attempted to register an action but the action name {name} is already taken."
if name in self.request_types:
msg = f"Attempted to register a request but the request name {name} is already taken."
_LOGGER.error(msg)
raise RuntimeError(msg)
self.actions[name] = action
self.request_types[name] = request_type
def remove_action(self, name: str) -> None:
def remove_request(self, name: str) -> None:
"""
Remove an action from this manager.
Remove a request from this manager.
:param name: name identifier of the action
:param name: name identifier of the request
:type name: str
"""
if name not in self.actions:
msg = f"Attempted to remove action {name} from action manager, but it was not registered."
if name not in self.request_types:
msg = f"Attempted to remove request {name} from request manager, but it was not registered."
_LOGGER.error(msg)
raise RuntimeError(msg)
self.actions.pop(name)
self.request_types.pop(name)
def get_action_tree(self) -> List[List[str]]:
"""Recursively generate action tree for this component."""
actions = []
for act_name, act in self.actions.items():
if isinstance(act.func, ActionManager):
sub_actions = act.func.get_action_tree()
sub_actions = [[act_name] + a for a in sub_actions]
actions.extend(sub_actions)
def get_request_types_recursively(self) -> List[List[str]]:
"""Recursively generate request tree for this component."""
requests = []
for req_name, req in self.request_types.items():
if isinstance(req.func, RequestManager):
sub_requests = req.func.get_request_types_recursively()
sub_requests = [[req_name] + a for a in sub_requests]
requests.extend(sub_requests)
else:
actions.append([act_name])
return actions
requests.append([req_name])
return requests
class SimComponent(BaseModel):
@@ -161,30 +158,30 @@ class SimComponent(BaseModel):
if not kwargs.get("uuid"):
kwargs["uuid"] = str(uuid4())
super().__init__(**kwargs)
self._action_manager: ActionManager = self._init_action_manager()
self._request_manager: RequestManager = self._init_request_manager()
self._parent: Optional["SimComponent"] = None
def _init_action_manager(self) -> ActionManager:
def _init_request_manager(self) -> RequestManager:
"""
Initialise the action manager for this component.
Initialise the request manager for this component.
When using a hierarchy of components, the child classes should call the parent class's _init_action_manager and
add additional actions on top of the existing generic ones.
When using a hierarchy of components, the child classes should call the parent class's _init_request_manager and
add additional requests on top of the existing generic ones.
Example usage for inherited classes:
..code::python
class WebBrowser(Application):
def _init_action_manager(self) -> ActionManager:
am = super()._init_action_manager() # all actions generic to any Application get initialised
am.add_action(...) # initialise any actions specific to the web browser
def _init_request_manager(self) -> RequestManager:
am = super()._init_request_manager() # all requests generic to any Application get initialised
am.add_request(...) # initialise any requests specific to the web browser
return am
:return: Actiona manager object belonging to this sim component.
:rtype: ActionManager
:return: Request manager object belonging to this sim component.
:rtype: RequestManager
"""
return ActionManager()
return RequestManager()
@abstractmethod
def describe_state(self) -> Dict:
@@ -200,27 +197,27 @@ class SimComponent(BaseModel):
}
return state
def apply_action(self, action: List[str], context: Dict = {}) -> None:
def apply_request(self, request: List[str], context: Dict = {}) -> None:
"""
Apply an action to a simulation component. Action data is passed in as a 'namespaced' list of strings.
Apply a request to a simulation component. Request data is passed in as a 'namespaced' list of strings.
If the list only has one element, the action is intended to be applied directly to this object. If the list has
multiple entries, the action is passed to the child of this object specified by the first one or two entries.
If the list only has one element, the request is intended to be applied directly to this object. If the list has
multiple entries, the request is passed to the child of this object specified by the first one or two entries.
This is essentially a namespace.
For example, ["turn_on",] is meant to apply an action of 'turn on' to this component.
For example, ["turn_on",] is meant to apply a request of 'turn on' to this component.
However, ["services", "email_client", "turn_on"] is meant to 'turn on' this component's email client service.
:param action: List describing the action to apply to this object.
:type action: List[str]
:param request: List describing the request to apply to this object.
:type request: List[str]
:param: context: Dict containing context for actions
:param: context: Dict containing context for requests
:type context: Dict
"""
if self._action_manager is None:
if self._request_manager is None:
return
self._action_manager(action, context)
self._request_manager(request, context)
def apply_timestep(self, timestep: int) -> None:
"""

View File

@@ -1,7 +1,7 @@
from enum import Enum
from typing import Dict, Final, List, Literal, Tuple
from primaite.simulator.core import Action, ActionManager, ActionPermissionValidator, SimComponent
from primaite.simulator.core import RequestManager, RequestPermissionValidator, RequestType, SimComponent
from primaite.simulator.domain.account import Account, AccountType
@@ -43,10 +43,10 @@ class AccountGroup(Enum):
"For full access"
class GroupMembershipValidator(ActionPermissionValidator):
class GroupMembershipValidator(RequestPermissionValidator):
"""Permit actions based on group membership."""
allowed_groups:List[AccountGroup]
allowed_groups: List[AccountGroup]
def __call__(self, request: List[str], context: Dict) -> bool:
"""Permit the action if the request comes from an account which belongs to the right group."""
@@ -79,14 +79,14 @@ class DomainController(SimComponent):
def __init__(self, **kwargs):
super().__init__(**kwargs)
def _init_action_manager(self) -> ActionManager:
am = super()._init_action_manager()
def _init_request_manager(self) -> RequestManager:
am = super()._init_request_manager()
# Action 'account' matches requests like:
# ['account', '<account-uuid>', *account_action]
am.add_action(
am.add_request(
"account",
Action(
func=lambda request, context: self.accounts[request.pop(0)].apply_action(request, context),
RequestType(
func=lambda request, context: self.accounts[request.pop(0)].apply_request(request, context),
validator=GroupMembershipValidator(allowed_groups=[AccountGroup.DOMAIN_ADMIN]),
),
)

View File

@@ -9,7 +9,7 @@ from typing import Dict, Optional
from prettytable import MARKDOWN, PrettyTable
from primaite import getLogger
from primaite.simulator.core import Action, ActionManager, SimComponent
from primaite.simulator.core import RequestManager, RequestType, SimComponent
from primaite.simulator.file_system.file_type import FileType, get_file_type_from_extension
from primaite.simulator.system.core.sys_log import SysLog
@@ -94,14 +94,14 @@ class FileSystem(SimComponent):
if not self.folders:
self.create_folder("root")
def _init_action_manager(self) -> ActionManager:
am = super()._init_action_manager()
def _init_request_manager(self) -> RequestManager:
am = super()._init_request_manager()
self._folder_action_manager = ActionManager()
am.add_action("folder", Action(func=self._folder_action_manager))
self._folder_request_manager = RequestManager()
am.add_request("folder", RequestType(func=self._folder_request_manager))
self._file_action_manager = ActionManager()
am.add_action("file", Action(func=self._file_action_manager))
self._file_request_manager = RequestManager()
am.add_request("file", RequestType(func=self._file_request_manager))
return am
@@ -165,7 +165,7 @@ class FileSystem(SimComponent):
self.folders[folder.uuid] = folder
self._folders_by_name[folder.name] = folder
self.sys_log.info(f"Created folder /{folder.name}")
self._folder_action_manager.add_action(folder.uuid, Action(func=folder._action_manager))
self._folder_request_manager.add_request(folder.uuid, RequestType(func=folder._request_manager))
return folder
def delete_folder(self, folder_name: str):
@@ -184,7 +184,7 @@ class FileSystem(SimComponent):
self.folders.pop(folder.uuid)
self._folders_by_name.pop(folder.name)
self.sys_log.info(f"Deleted folder /{folder.name} and its contents")
self._folder_action_manager.remove_action(folder.uuid)
self._folder_request_manager.remove_request(folder.uuid)
else:
_LOGGER.debug(f"Cannot delete folder as it does not exist: {folder_name}")
@@ -226,7 +226,7 @@ class FileSystem(SimComponent):
)
folder.add_file(file)
self.sys_log.info(f"Created file /{file.path}")
self._file_action_manager.add_action(file.uuid, Action(func=file._action_manager))
self._file_request_manager.add_request(file.uuid, RequestType(func=file._request_manager))
return file
def get_file(self, folder_name: str, file_name: str) -> Optional[File]:
@@ -254,7 +254,7 @@ class FileSystem(SimComponent):
file = folder.get_file(file_name)
if file:
folder.remove_file(file)
self._file_action_manager.remove_action(file.uuid)
self._file_request_manager.remove_request(file.uuid)
self.sys_log.info(f"Deleted file /{file.path}")
def move_file(self, src_folder_name: str, src_file_name: str, dst_folder_name: str):
@@ -332,15 +332,15 @@ class Folder(FileSystemItemABC):
is_quarantined: bool = False
"Flag that marks the folder as quarantined if true."
def _init_action_manager(self) -> ActionManager:
am = super()._init_action_manager()
def _init_request_manager(self) -> RequestManager:
am = super()._init_request_manager()
am.add_action("scan", Action(func=lambda request, context: ...)) # TODO implement action
am.add_action("checkhash", Action(func=lambda request, context: ...)) # TODO implement action
am.add_action("repair", Action(func=lambda request, context: ...)) # TODO implement action
am.add_action("restore", Action(func=lambda request, context: ...)) # TODO implement action
am.add_action("delete", Action(func=lambda request, context: ...)) # TODO implement action
am.add_action("corrupt", Action(func=lambda request, context: ...)) # TODO implement action
am.add_request("scan", RequestType(func=lambda request, context: ...)) # TODO implement request
am.add_request("checkhash", RequestType(func=lambda request, context: ...)) # TODO implement request
am.add_request("repair", RequestType(func=lambda request, context: ...)) # TODO implement request
am.add_request("restore", RequestType(func=lambda request, context: ...)) # TODO implement request
am.add_request("delete", RequestType(func=lambda request, context: ...)) # TODO implement request
am.add_request("corrupt", RequestType(func=lambda request, context: ...)) # TODO implement request
return am
@@ -509,15 +509,15 @@ class File(FileSystemItemABC):
with open(self.sim_path, mode="a"):
pass
def _init_action_manager(self) -> ActionManager:
am = super()._init_action_manager()
def _init_request_manager(self) -> RequestManager:
am = super()._init_request_manager()
am.add_action("scan", Action(func=lambda request, context: ...)) # TODO implement action
am.add_action("checkhash", Action(func=lambda request, context: ...)) # TODO implement action
am.add_action("delete", Action(func=lambda request, context: ...)) # TODO implement action
am.add_action("repair", Action(func=lambda request, context: ...)) # TODO implement action
am.add_action("restore", Action(func=lambda request, context: ...)) # TODO implement action
am.add_action("corrupt", Action(func=lambda request, context: ...)) # TODO implement action
am.add_request("scan", RequestType(func=lambda request, context: ...)) # TODO implement request
am.add_request("checkhash", RequestType(func=lambda request, context: ...)) # TODO implement request
am.add_request("delete", RequestType(func=lambda request, context: ...)) # TODO implement request
am.add_request("repair", RequestType(func=lambda request, context: ...)) # TODO implement request
am.add_request("restore", RequestType(func=lambda request, context: ...)) # TODO implement request
am.add_request("corrupt", RequestType(func=lambda request, context: ...)) # TODO implement request
return am

View File

@@ -6,7 +6,7 @@ from networkx import MultiGraph
from prettytable import MARKDOWN, PrettyTable
from primaite import getLogger
from primaite.simulator.core import Action, ActionManager, SimComponent
from primaite.simulator.core import RequestManager, RequestType, SimComponent
from primaite.simulator.network.hardware.base import Link, NIC, Node, SwitchPort
from primaite.simulator.network.hardware.nodes.computer import Computer
from primaite.simulator.network.hardware.nodes.router import Router
@@ -37,18 +37,18 @@ class Network(SimComponent):
Initialise the network.
Constructs the network and sets up its initial state including
the action manager and an empty MultiGraph for topology representation.
the request manager and an empty MultiGraph for topology representation.
"""
super().__init__(**kwargs)
self._nx_graph = MultiGraph()
def _init_action_manager(self) -> ActionManager:
am = super()._init_action_manager()
self._node_action_manager = ActionManager()
am.add_action(
def _init_request_manager(self) -> RequestManager:
am = super()._init_request_manager()
self._node_request_manager = RequestManager()
am.add_request(
"node",
Action(func=self._node_action_manager),
RequestType(func=self._node_request_manager),
)
return am
@@ -182,7 +182,7 @@ class Network(SimComponent):
node.parent = self
self._nx_graph.add_node(node.hostname)
_LOGGER.info(f"Added node {node.uuid} to Network {self.uuid}")
self._node_action_manager.add_action(name=node.uuid, action=Action(func=node._action_manager))
self._node_request_manager.add_request(name=node.uuid, request_type=RequestType(func=node._request_manager))
def get_node_by_hostname(self, hostname: str) -> Optional[Node]:
"""
@@ -216,7 +216,7 @@ class Network(SimComponent):
break
node.parent = None
_LOGGER.info(f"Removed node {node.uuid} from network {self.uuid}")
self._node_action_manager.remove_action(name=node.uuid)
self._node_request_manager.remove_request(name=node.uuid)
def connect(self, endpoint_a: Union[NIC, SwitchPort], endpoint_b: Union[NIC, SwitchPort], **kwargs) -> None:
"""

View File

@@ -12,7 +12,7 @@ from prettytable import MARKDOWN, PrettyTable
from primaite import getLogger
from primaite.exceptions import NetworkError
from primaite.simulator import SIM_OUTPUT
from primaite.simulator.core import Action, ActionManager, SimComponent
from primaite.simulator.core import RequestManager, RequestType, SimComponent
from primaite.simulator.domain.account import Account
from primaite.simulator.file_system.file_system import FileSystem
from primaite.simulator.network.protocols.arp import ARPEntry, ARPPacket
@@ -144,11 +144,11 @@ class NIC(SimComponent):
)
return state
def _init_action_manager(self) -> ActionManager:
am = super()._init_action_manager()
def _init_request_manager(self) -> RequestManager:
am = super()._init_request_manager()
am.add_action("enable", Action(func=lambda request, context: self.enable()))
am.add_action("disable", Action(func=lambda request, context: self.disable()))
am.add_request("enable", RequestType(func=lambda request, context: self.enable()))
am.add_request("disable", RequestType(func=lambda request, context: self.disable()))
return am
@@ -946,31 +946,31 @@ class Node(SimComponent):
self.session_manager.software_manager = self.software_manager
self._install_system_software()
def _init_action_manager(self) -> ActionManager:
def _init_request_manager(self) -> RequestManager:
# TODO: I see that this code is really confusing and hard to read right now... I think some of these things will
# need a better name and better documentation.
am = super()._init_action_manager()
# since there are potentially many services, create an action manager that can map service name
self._service_action_manager = ActionManager()
am.add_action("service", Action(func=self._service_action_manager))
self._nic_action_manager = ActionManager()
am.add_action("nic", Action(func=self._nic_action_manager))
am = super()._init_request_manager()
# since there are potentially many services, create an request manager that can map service name
self._service_request_manager = RequestManager()
am.add_request("service", RequestType(func=self._service_request_manager))
self._nic_request_manager = RequestManager()
am.add_request("nic", RequestType(func=self._nic_request_manager))
am.add_action("file_system", Action(func=self.file_system._action_manager))
am.add_request("file_system", RequestType(func=self.file_system._request_manager))
# currently we don't have any applications nor processes, so these will be empty
self._process_action_manager = ActionManager()
am.add_action("process", Action(func=self._process_action_manager))
self._application_action_manager = ActionManager()
am.add_action("application", Action(func=self._application_action_manager))
self._process_request_manager = RequestManager()
am.add_request("process", RequestType(func=self._process_request_manager))
self._application_request_manager = RequestManager()
am.add_request("application", RequestType(func=self._application_request_manager))
am.add_action("scan", Action(func=lambda request, context: ...)) # TODO implement OS scan
am.add_request("scan", RequestType(func=lambda request, context: ...)) # TODO implement OS scan
am.add_action("shutdown", Action(func=lambda request, context: self.power_off()))
am.add_action("startup", Action(func=lambda request, context: self.power_on()))
am.add_action("reset", Action(func=lambda request, context: ...)) # TODO implement node reset
am.add_action("logon", Action(func=lambda request, context: ...)) # TODO implement logon action
am.add_action("logoff", Action(func=lambda request, context: ...)) # TODO implement logoff action
am.add_request("shutdown", RequestType(func=lambda request, context: self.power_off()))
am.add_request("startup", RequestType(func=lambda request, context: self.power_on()))
am.add_request("reset", RequestType(func=lambda request, context: ...)) # TODO implement node reset
am.add_request("logon", RequestType(func=lambda request, context: ...)) # TODO implement logon request
am.add_request("logoff", RequestType(func=lambda request, context: ...)) # TODO implement logoff request
return am
@@ -1071,7 +1071,7 @@ class Node(SimComponent):
self.sys_log.info(f"Connected NIC {nic}")
if self.operating_state == NodeOperatingState.ON:
nic.enable()
self._nic_action_manager.add_action(nic.uuid, Action(func=nic._action_manager))
self._nic_request_manager.add_request(nic.uuid, RequestType(func=nic._request_manager))
else:
msg = f"Cannot connect NIC {nic} as it is already connected"
self.sys_log.logger.error(msg)
@@ -1096,7 +1096,7 @@ class Node(SimComponent):
nic.parent = None
nic.disable()
self.sys_log.info(f"Disconnected NIC {nic}")
self._nic_action_manager.remove_action(nic.uuid)
self._nic_request_manager.remove_request(nic.uuid)
else:
msg = f"Cannot disconnect NIC {nic} as it is not connected"
self.sys_log.logger.error(msg)
@@ -1194,7 +1194,7 @@ class Node(SimComponent):
service.install() # Perform any additional setup, such as creating files for this service on the node.
self.sys_log.info(f"Installed service {service.name}")
_LOGGER.info(f"Added service {service.uuid} to node {self.uuid}")
self._service_action_manager.add_action(service.uuid, Action(func=service._action_manager))
self._service_request_manager.add_request(service.uuid, RequestType(func=service._request_manager))
def uninstall_service(self, service: Service) -> None:
"""Uninstall and completely remove service from this node.
@@ -1210,7 +1210,7 @@ class Node(SimComponent):
service.parent = None
self.sys_log.info(f"Uninstalled service {service.name}")
_LOGGER.info(f"Removed service {service.uuid} from node {self.uuid}")
self._service_action_manager.remove_action(service.uuid)
self._service_request_manager.remove_request(service.uuid)
def __contains__(self, item: Any) -> bool:
if isinstance(item, Service):

View File

@@ -7,7 +7,7 @@ from typing import Dict, List, Optional, Tuple, Union
from prettytable import MARKDOWN, PrettyTable
from primaite.simulator.core import Action, ActionManager, SimComponent
from primaite.simulator.core import RequestManager, RequestType, SimComponent
from primaite.simulator.network.hardware.base import ARPCache, ICMP, NIC, Node
from primaite.simulator.network.transmission.data_link_layer import EthernetHeader, Frame
from primaite.simulator.network.transmission.network_layer import ICMPPacket, ICMPType, IPPacket, IPProtocol
@@ -43,7 +43,7 @@ class ACLRule(SimComponent):
def __str__(self) -> str:
rule_strings = []
for key, value in self.model_dump(exclude={"uuid", "action_manager"}).items():
for key, value in self.model_dump(exclude={"uuid", "request_manager"}).items():
if value is None:
value = "ANY"
if isinstance(value, Enum):
@@ -87,8 +87,8 @@ class AccessControlList(SimComponent):
super().__init__(**kwargs)
self._acl = [None] * (self.max_acl_rules - 1)
def _init_action_manager(self) -> ActionManager:
am = super()._init_action_manager()
def _init_request_manager(self) -> RequestManager:
am = super()._init_request_manager()
# When the request reaches this action, it should now contain solely positional args for the 'add_rule' action.
# POSITIONAL ARGUMENTS:
@@ -99,9 +99,9 @@ class AccessControlList(SimComponent):
# 4: destination ip address (str castable to IPV4Address (e.g. '10.10.1.2'))
# 5: destination port (str name of a Port (e.g. "HTTP"))
# 6: position (int)
am.add_action(
am.add_request(
"add_rule",
Action(
RequestType(
func=lambda request, context: self.add_rule(
ACLAction[request[0]],
IPProtocol[request[1]],
@@ -114,7 +114,7 @@ class AccessControlList(SimComponent):
),
)
am.add_action("remove_rule", Action(func=lambda request, context: self.remove_rule(int(request[0]))))
am.add_request("remove_rule", RequestType(func=lambda request, context: self.remove_rule(int(request[0]))))
return am
def describe_state(self) -> Dict:
@@ -626,9 +626,9 @@ class Router(Node):
self.arp.nics = self.nics
self.icmp.arp = self.arp
def _init_action_manager(self) -> ActionManager:
am = super()._init_action_manager()
am.add_action("acl", Action(func=self.acl._action_manager))
def _init_request_manager(self) -> RequestManager:
am = super()._init_request_manager()
am.add_request("acl", RequestType(func=self.acl._request_manager))
return am
def _get_port_of_nic(self, target_nic: NIC) -> Optional[int]:

View File

@@ -1,6 +1,6 @@
from typing import Dict
from primaite.simulator.core import Action, ActionManager, AllowAllValidator, SimComponent
from primaite.simulator.core import RequestManager, RequestType, SimComponent
from primaite.simulator.domain.controller import DomainController
from primaite.simulator.network.container import Network
@@ -21,12 +21,12 @@ class Simulation(SimComponent):
super().__init__(**kwargs)
def _init_action_manager(self) -> ActionManager:
am = super()._init_action_manager()
# pass through network actions to the network objects
am.add_action("network", Action(func=self.network._action_manager))
# pass through domain actions to the domain object
am.add_action("domain", Action(func=self.domain._action_manager))
def _init_request_manager(self) -> RequestManager:
am = super()._init_request_manager()
# pass through network requests to the network objects
am.add_request("network", RequestType(func=self.network._request_manager))
# pass through domain requests to the domain object
am.add_request("domain", RequestType(func=self.domain._request_manager))
return am
def describe_state(self) -> Dict:

View File

@@ -2,7 +2,7 @@ from enum import Enum
from typing import Any, Dict, Optional
from primaite import getLogger
from primaite.simulator.core import Action, ActionManager
from primaite.simulator.core import RequestManager, RequestType
from primaite.simulator.system.software import IOSoftware
_LOGGER = getLogger(__name__)
@@ -39,15 +39,15 @@ class Service(IOSoftware):
_restart_countdown: Optional[int] = None
"If currently restarting, how many timesteps remain until the restart is finished."
def _init_action_manager(self) -> ActionManager:
am = super()._init_action_manager()
am.add_action("stop", Action(func=lambda request, context: self.stop()))
am.add_action("start", Action(func=lambda request, context: self.start()))
am.add_action("pause", Action(func=lambda request, context: self.pause()))
am.add_action("resume", Action(func=lambda request, context: self.resume()))
am.add_action("restart", Action(func=lambda request, context: self.restart()))
am.add_action("disable", Action(func=lambda request, context: self.disable()))
am.add_action("enable", Action(func=lambda request, context: self.enable()))
def _init_request_manager(self) -> RequestManager:
am = super()._init_request_manager()
am.add_request("stop", RequestType(func=lambda request, context: self.stop()))
am.add_request("start", RequestType(func=lambda request, context: self.start()))
am.add_request("pause", RequestType(func=lambda request, context: self.pause()))
am.add_request("resume", RequestType(func=lambda request, context: self.resume()))
am.add_request("restart", RequestType(func=lambda request, context: self.restart()))
am.add_request("disable", RequestType(func=lambda request, context: self.disable()))
am.add_request("enable", RequestType(func=lambda request, context: self.enable()))
return am
def describe_state(self) -> Dict:

View File

@@ -2,7 +2,7 @@ from abc import abstractmethod
from enum import Enum
from typing import Any, Dict, Optional
from primaite.simulator.core import Action, ActionManager, SimComponent
from primaite.simulator.core import RequestManager, RequestType, SimComponent
from primaite.simulator.file_system.file_system import FileSystem, Folder
from primaite.simulator.network.transmission.transport_layer import Port
from primaite.simulator.system.core.sys_log import SysLog
@@ -85,15 +85,15 @@ class Software(SimComponent):
folder: Optional[Folder] = None
"The folder on the file system the Software uses."
def _init_action_manager(self) -> ActionManager:
am = super()._init_action_manager()
am.add_action(
def _init_request_manager(self) -> RequestManager:
am = super()._init_request_manager()
am.add_request(
"compromise",
Action(
RequestType(
func=lambda request, context: self.set_health_state(SoftwareHealthState.COMPROMISED),
),
)
am.add_action("scan", Action(func=lambda request, context: self.scan()))
am.add_request("scan", RequestType(func=lambda request, context: self.scan()))
return am
@abstractmethod