Move init action manager function to the top
This commit is contained in:
@@ -142,20 +142,6 @@ class SimComponent(BaseModel):
|
|||||||
self._action_manager: ActionManager = self._init_action_manager()
|
self._action_manager: ActionManager = self._init_action_manager()
|
||||||
self._parent: Optional["SimComponent"] = None
|
self._parent: Optional["SimComponent"] = None
|
||||||
|
|
||||||
@abstractmethod
|
|
||||||
def describe_state(self) -> Dict:
|
|
||||||
"""
|
|
||||||
Return a dictionary describing the state of this object and any objects managed by it.
|
|
||||||
|
|
||||||
This is similar to pydantic ``model_dump()``, but it only outputs information about the objects owned by this
|
|
||||||
object. If there are objects referenced by this object that are owned by something else, it is not included in
|
|
||||||
this output.
|
|
||||||
"""
|
|
||||||
state = {
|
|
||||||
"uuid": self.uuid,
|
|
||||||
}
|
|
||||||
return state
|
|
||||||
|
|
||||||
def _init_action_manager(self) -> ActionManager:
|
def _init_action_manager(self) -> ActionManager:
|
||||||
"""
|
"""
|
||||||
Initialise the action manager for this component.
|
Initialise the action manager for this component.
|
||||||
@@ -178,6 +164,20 @@ class SimComponent(BaseModel):
|
|||||||
"""
|
"""
|
||||||
return ActionManager()
|
return ActionManager()
|
||||||
|
|
||||||
|
@abstractmethod
|
||||||
|
def describe_state(self) -> Dict:
|
||||||
|
"""
|
||||||
|
Return a dictionary describing the state of this object and any objects managed by it.
|
||||||
|
|
||||||
|
This is similar to pydantic ``model_dump()``, but it only outputs information about the objects owned by this
|
||||||
|
object. If there are objects referenced by this object that are owned by something else, it is not included in
|
||||||
|
this output.
|
||||||
|
"""
|
||||||
|
state = {
|
||||||
|
"uuid": self.uuid,
|
||||||
|
}
|
||||||
|
return state
|
||||||
|
|
||||||
def apply_action(self, action: List[str], context: Dict = {}) -> None:
|
def apply_action(self, action: List[str], context: Dict = {}) -> None:
|
||||||
"""
|
"""
|
||||||
Apply an action to a simulation component. Action data is passed in as a 'namespaced' list of strings.
|
Apply an action to a simulation component. Action data is passed in as a 'namespaced' list of strings.
|
||||||
|
|||||||
@@ -85,19 +85,6 @@ class DomainController(SimComponent):
|
|||||||
def __init__(self, **kwargs):
|
def __init__(self, **kwargs):
|
||||||
super().__init__(**kwargs)
|
super().__init__(**kwargs)
|
||||||
|
|
||||||
def describe_state(self) -> Dict:
|
|
||||||
"""
|
|
||||||
Produce a dictionary describing the current state of this object.
|
|
||||||
|
|
||||||
Please see :py:meth:`primaite.simulator.core.SimComponent.describe_state` for a more detailed explanation.
|
|
||||||
|
|
||||||
:return: Current state of this object and child objects.
|
|
||||||
:rtype: Dict
|
|
||||||
"""
|
|
||||||
state = super().describe_state()
|
|
||||||
state.update({"accounts": {uuid: acct.describe_state() for uuid, acct in self.accounts.items()}})
|
|
||||||
return state
|
|
||||||
|
|
||||||
def _init_action_manager(self) -> ActionManager:
|
def _init_action_manager(self) -> ActionManager:
|
||||||
am = super()._init_action_manager()
|
am = super()._init_action_manager()
|
||||||
# Action 'account' matches requests like:
|
# Action 'account' matches requests like:
|
||||||
@@ -111,6 +98,19 @@ class DomainController(SimComponent):
|
|||||||
)
|
)
|
||||||
return am
|
return am
|
||||||
|
|
||||||
|
def describe_state(self) -> Dict:
|
||||||
|
"""
|
||||||
|
Produce a dictionary describing the current state of this object.
|
||||||
|
|
||||||
|
Please see :py:meth:`primaite.simulator.core.SimComponent.describe_state` for a more detailed explanation.
|
||||||
|
|
||||||
|
:return: Current state of this object and child objects.
|
||||||
|
:rtype: Dict
|
||||||
|
"""
|
||||||
|
state = super().describe_state()
|
||||||
|
state.update({"accounts": {uuid: acct.describe_state() for uuid, acct in self.accounts.items()}})
|
||||||
|
return state
|
||||||
|
|
||||||
def _register_account(self, account: Account) -> None:
|
def _register_account(self, account: Account) -> None:
|
||||||
"""TODO."""
|
"""TODO."""
|
||||||
...
|
...
|
||||||
|
|||||||
@@ -17,6 +17,18 @@ class Network(SimComponent):
|
|||||||
"""Initialise the network."""
|
"""Initialise the network."""
|
||||||
super().__init__(**kwargs)
|
super().__init__(**kwargs)
|
||||||
|
|
||||||
|
def _init_action_manager(self) -> ActionManager:
|
||||||
|
am = super()._init_action_manager()
|
||||||
|
|
||||||
|
am.add_action(
|
||||||
|
"node",
|
||||||
|
Action(
|
||||||
|
func=lambda request, context: self.nodes[request.pop(0)].apply_action(request, context),
|
||||||
|
validator=AllowAllValidator(),
|
||||||
|
),
|
||||||
|
)
|
||||||
|
return am
|
||||||
|
|
||||||
def describe_state(self) -> Dict:
|
def describe_state(self) -> Dict:
|
||||||
"""
|
"""
|
||||||
Produce a dictionary describing the current state of this object.
|
Produce a dictionary describing the current state of this object.
|
||||||
@@ -35,18 +47,6 @@ class Network(SimComponent):
|
|||||||
)
|
)
|
||||||
return state
|
return state
|
||||||
|
|
||||||
def _init_action_manager(self) -> ActionManager:
|
|
||||||
am = super()._init_action_manager()
|
|
||||||
|
|
||||||
am.add_action(
|
|
||||||
"node",
|
|
||||||
Action(
|
|
||||||
func=lambda request, context: self.nodes[request.pop(0)].apply_action(request, context),
|
|
||||||
validator=AllowAllValidator(),
|
|
||||||
),
|
|
||||||
)
|
|
||||||
return am
|
|
||||||
|
|
||||||
def add_node(self, node: Node) -> None:
|
def add_node(self, node: Node) -> None:
|
||||||
"""
|
"""
|
||||||
Add an existing node to the network.
|
Add an existing node to the network.
|
||||||
|
|||||||
@@ -21,24 +21,6 @@ class Simulation(SimComponent):
|
|||||||
|
|
||||||
super().__init__(**kwargs)
|
super().__init__(**kwargs)
|
||||||
|
|
||||||
def describe_state(self) -> Dict:
|
|
||||||
"""
|
|
||||||
Produce a dictionary describing the current state of this object.
|
|
||||||
|
|
||||||
Please see :py:meth:`primaite.simulator.core.SimComponent.describe_state` for a more detailed explanation.
|
|
||||||
|
|
||||||
:return: Current state of this object and child objects.
|
|
||||||
:rtype: Dict
|
|
||||||
"""
|
|
||||||
state = super().describe_state()
|
|
||||||
state.update(
|
|
||||||
{
|
|
||||||
"network": self.network.describe_state(),
|
|
||||||
"domain": self.domain.describe_state(),
|
|
||||||
}
|
|
||||||
)
|
|
||||||
return state
|
|
||||||
|
|
||||||
def _init_action_manager(self) -> ActionManager:
|
def _init_action_manager(self) -> ActionManager:
|
||||||
am = super()._init_action_manager()
|
am = super()._init_action_manager()
|
||||||
# pass through network actions to the network objects
|
# pass through network actions to the network objects
|
||||||
@@ -56,3 +38,21 @@ class Simulation(SimComponent):
|
|||||||
),
|
),
|
||||||
)
|
)
|
||||||
return am
|
return am
|
||||||
|
|
||||||
|
def describe_state(self) -> Dict:
|
||||||
|
"""
|
||||||
|
Produce a dictionary describing the current state of this object.
|
||||||
|
|
||||||
|
Please see :py:meth:`primaite.simulator.core.SimComponent.describe_state` for a more detailed explanation.
|
||||||
|
|
||||||
|
:return: Current state of this object and child objects.
|
||||||
|
:rtype: Dict
|
||||||
|
"""
|
||||||
|
state = super().describe_state()
|
||||||
|
state.update(
|
||||||
|
{
|
||||||
|
"network": self.network.describe_state(),
|
||||||
|
"domain": self.domain.describe_state(),
|
||||||
|
}
|
||||||
|
)
|
||||||
|
return state
|
||||||
|
|||||||
@@ -33,6 +33,17 @@ class Service(IOSoftware):
|
|||||||
operating_state: ServiceOperatingState
|
operating_state: ServiceOperatingState
|
||||||
"The current operating state of the Service."
|
"The current operating state of the Service."
|
||||||
|
|
||||||
|
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()))
|
||||||
|
return am
|
||||||
|
|
||||||
@abstractmethod
|
@abstractmethod
|
||||||
def describe_state(self) -> Dict:
|
def describe_state(self) -> Dict:
|
||||||
"""
|
"""
|
||||||
@@ -47,17 +58,6 @@ class Service(IOSoftware):
|
|||||||
state.update({"operating_state": self.operating_state.name})
|
state.update({"operating_state": self.operating_state.name})
|
||||||
return state
|
return state
|
||||||
|
|
||||||
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()))
|
|
||||||
return am
|
|
||||||
|
|
||||||
def reset_component_for_episode(self, episode: int):
|
def reset_component_for_episode(self, episode: int):
|
||||||
"""
|
"""
|
||||||
Resets the Service component for a new episode.
|
Resets the Service component for a new episode.
|
||||||
|
|||||||
@@ -75,6 +75,17 @@ class Software(SimComponent):
|
|||||||
revealed_to_red: bool = False
|
revealed_to_red: bool = False
|
||||||
"Indicates if the software has been revealed to red agent, defaults is False."
|
"Indicates if the software has been revealed to red agent, defaults is False."
|
||||||
|
|
||||||
|
def _init_action_manager(self) -> ActionManager:
|
||||||
|
am = super()._init_action_manager()
|
||||||
|
am.add_action(
|
||||||
|
"compromise",
|
||||||
|
Action(
|
||||||
|
func=lambda request, context: self.set_health_state(SoftwareHealthState.COMPROMISED),
|
||||||
|
),
|
||||||
|
)
|
||||||
|
am.add_action("scan", Action(func=lambda request, context: self.scan()))
|
||||||
|
return am
|
||||||
|
|
||||||
@abstractmethod
|
@abstractmethod
|
||||||
def describe_state(self) -> Dict:
|
def describe_state(self) -> Dict:
|
||||||
"""
|
"""
|
||||||
@@ -98,17 +109,6 @@ class Software(SimComponent):
|
|||||||
)
|
)
|
||||||
return state
|
return state
|
||||||
|
|
||||||
def _init_action_manager(self) -> ActionManager:
|
|
||||||
am = super()._init_action_manager()
|
|
||||||
am.add_action(
|
|
||||||
"compromise",
|
|
||||||
Action(
|
|
||||||
func=lambda request, context: self.set_health_state(SoftwareHealthState.COMPROMISED),
|
|
||||||
),
|
|
||||||
)
|
|
||||||
am.add_action("scan", Action(func=lambda request, context: self.scan()))
|
|
||||||
return am
|
|
||||||
|
|
||||||
def reset_component_for_episode(self, episode: int):
|
def reset_component_for_episode(self, episode: int):
|
||||||
"""
|
"""
|
||||||
Resets the software component for a new episode.
|
Resets the software component for a new episode.
|
||||||
|
|||||||
Reference in New Issue
Block a user