Add docs and tests
This commit is contained in:
@@ -11,3 +11,55 @@ top level, there is an object called the ``SimulationController`` _(doesn't exis
|
|||||||
and a software controller for managing software and users.
|
and a software controller for managing software and users.
|
||||||
|
|
||||||
Each node of the simulation 'tree' has responsibility for creating, deleting, and updating its direct descendants.
|
Each node of the simulation 'tree' has responsibility for creating, deleting, and updating its direct descendants.
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
Actions
|
||||||
|
=======
|
||||||
|
Agents can interact with the simulation by using actions. Actions are standardised with the
|
||||||
|
:py:class:`primaite.simulation.core.Action` class, which just holds a reference to two special functions.
|
||||||
|
|
||||||
|
1. The action function itself, it must accept a `request` parameters which is a list of strings that describe what the
|
||||||
|
action should do. It must also accept a `context` dict which can house additional information surrounding the action.
|
||||||
|
For example, the context will typically include information about which entity intiated the action.
|
||||||
|
2. A validator function. This function should return a boolean value that decides if the request is permitted or not.
|
||||||
|
It uses the same paramters as the action function.
|
||||||
|
|
||||||
|
Action Permissions
|
||||||
|
------------------
|
||||||
|
When an agent tries to perform an action on a simulation component, that action will only be executed if the request is
|
||||||
|
validated. For example, some actions can require that an agent is logged into an admin account. Each action defines its
|
||||||
|
own permissions using an instance of :py:class:`primaite.simulation.core.ActionPermissionValidator`. The below code
|
||||||
|
snippet demonstrates usage of the ``ActionPermissionValidator``.
|
||||||
|
|
||||||
|
.. code:: python
|
||||||
|
|
||||||
|
from primaite.simulator.core import Action, ActionManager, SimComponent
|
||||||
|
from primaite.simulator.domain.controller import AccountGroup, GroupMembershipValidator
|
||||||
|
|
||||||
|
class Smartphone(SimComponent):
|
||||||
|
name: str
|
||||||
|
apps = []
|
||||||
|
|
||||||
|
def __init__(self, **kwargs):
|
||||||
|
super().__init__(**kwargs)
|
||||||
|
self.action_manager = ActionManager()
|
||||||
|
|
||||||
|
self.action_manager.add_action(
|
||||||
|
"reset_factory_settings",
|
||||||
|
Action(
|
||||||
|
func = lambda request, context: self.reset_factory_settings(),
|
||||||
|
validator = GroupMembershipValidator([AccountGroup.domain_admin]),
|
||||||
|
),
|
||||||
|
)
|
||||||
|
|
||||||
|
def reset_factory_settings(self):
|
||||||
|
self.apps = []
|
||||||
|
|
||||||
|
phone = Smartphone(name="phone1")
|
||||||
|
|
||||||
|
# try to wipe the phone as a domain user, this will have no effect
|
||||||
|
phone.apply_action(["reset_factory_settings"], context={"request_source":{"groups":["domain_user"]})
|
||||||
|
|
||||||
|
# try to wipe the phone as an admin user, this will wipe the phone
|
||||||
|
phone.apply_action(["reset_factory_settings"], context={"request_source":{"groups":["domain_admin"]})
|
||||||
|
|||||||
@@ -1,6 +1,6 @@
|
|||||||
"""User account simulation."""
|
"""User account simulation."""
|
||||||
from enum import Enum
|
from enum import Enum
|
||||||
from typing import Any, Callable, Dict, List
|
from typing import Dict
|
||||||
|
|
||||||
from primaite import getLogger
|
from primaite import getLogger
|
||||||
from primaite.simulator.core import SimComponent
|
from primaite.simulator.core import SimComponent
|
||||||
@@ -49,6 +49,10 @@ class Account(SimComponent):
|
|||||||
"Account Type, currently this can be service account (used by apps) or user account."
|
"Account Type, currently this can be service account (used by apps) or user account."
|
||||||
status: AccountStatus = AccountStatus.disabled
|
status: AccountStatus = AccountStatus.disabled
|
||||||
|
|
||||||
|
def describe_state(self) -> Dict:
|
||||||
|
"""Describe state for agent observations."""
|
||||||
|
return super().describe_state()
|
||||||
|
|
||||||
def enable(self):
|
def enable(self):
|
||||||
"""Set the status to enabled."""
|
"""Set the status to enabled."""
|
||||||
self.status = AccountStatus.enabled
|
self.status = AccountStatus.enabled
|
||||||
@@ -57,8 +61,10 @@ class Account(SimComponent):
|
|||||||
"""Set the status to disabled."""
|
"""Set the status to disabled."""
|
||||||
self.status = AccountStatus.disabled
|
self.status = AccountStatus.disabled
|
||||||
|
|
||||||
def _possible_actions(self) -> Dict[str, Callable[[List[str]], None]]:
|
def log_on(self) -> None:
|
||||||
return {
|
"""TODO."""
|
||||||
"enable": self.enable,
|
self.num_logons += 1
|
||||||
"disable": self.disable,
|
|
||||||
}
|
def log_off(self) -> None:
|
||||||
|
"""TODO."""
|
||||||
|
self.num_logoffs += 1
|
||||||
|
|||||||
@@ -1,5 +1,5 @@
|
|||||||
from enum import Enum
|
from enum import Enum
|
||||||
from typing import Any, Dict, Final, List
|
from typing import Dict, Final, List, Literal, Tuple
|
||||||
|
|
||||||
from primaite.simulator.core import ActionPermissionValidator, SimComponent
|
from primaite.simulator.core import ActionPermissionValidator, SimComponent
|
||||||
from primaite.simulator.domain.account import Account, AccountType
|
from primaite.simulator.domain.account import Account, AccountType
|
||||||
@@ -7,18 +7,26 @@ from primaite.simulator.domain.account import Account, AccountType
|
|||||||
|
|
||||||
# placeholder while these objects don't yet exist
|
# placeholder while these objects don't yet exist
|
||||||
class temp_node:
|
class temp_node:
|
||||||
|
"""Placeholder for node class for type hinting purposes."""
|
||||||
|
|
||||||
pass
|
pass
|
||||||
|
|
||||||
|
|
||||||
class temp_application:
|
class temp_application:
|
||||||
|
"""Placeholder for application class for type hinting purposes."""
|
||||||
|
|
||||||
pass
|
pass
|
||||||
|
|
||||||
|
|
||||||
class temp_folder:
|
class temp_folder:
|
||||||
|
"""Placeholder for folder class for type hinting purposes."""
|
||||||
|
|
||||||
pass
|
pass
|
||||||
|
|
||||||
|
|
||||||
class temp_file:
|
class temp_file:
|
||||||
|
"""Placeholder for file class for type hinting purposes."""
|
||||||
|
|
||||||
pass
|
pass
|
||||||
|
|
||||||
|
|
||||||
@@ -59,9 +67,12 @@ class DomainController(SimComponent):
|
|||||||
accounts: List[Account] = []
|
accounts: List[Account] = []
|
||||||
groups: Final[List[AccountGroup]] = list(AccountGroup)
|
groups: Final[List[AccountGroup]] = list(AccountGroup)
|
||||||
|
|
||||||
group_membership: Dict[AccountGroup, List[Account]]
|
domain_group_membership: Dict[Literal[AccountGroup.domain_admin, AccountGroup.domain_user], List[Account]] = {}
|
||||||
|
local_group_membership: Dict[
|
||||||
|
Tuple(temp_node, Literal[AccountGroup.local_admin, AccountGroup.local_user]), List[Account]
|
||||||
|
] = {}
|
||||||
|
|
||||||
# references to non-owned objects
|
# references to non-owned objects. Not sure if all are needed here.
|
||||||
nodes: List[temp_node] = []
|
nodes: List[temp_node] = []
|
||||||
applications: List[temp_application] = []
|
applications: List[temp_application] = []
|
||||||
folders: List[temp_folder] = []
|
folders: List[temp_folder] = []
|
||||||
@@ -79,6 +90,10 @@ class DomainController(SimComponent):
|
|||||||
"""TODO."""
|
"""TODO."""
|
||||||
...
|
...
|
||||||
|
|
||||||
|
def delete_account(self, account: Account) -> None:
|
||||||
|
"""TODO."""
|
||||||
|
...
|
||||||
|
|
||||||
def rotate_all_credentials(self) -> None:
|
def rotate_all_credentials(self) -> None:
|
||||||
"""TODO."""
|
"""TODO."""
|
||||||
...
|
...
|
||||||
@@ -94,3 +109,15 @@ class DomainController(SimComponent):
|
|||||||
def remove_account_from_group(self, account: Account, group: AccountGroup) -> None:
|
def remove_account_from_group(self, account: Account, group: AccountGroup) -> None:
|
||||||
"""TODO."""
|
"""TODO."""
|
||||||
...
|
...
|
||||||
|
|
||||||
|
def check_account_permissions(self, account: Account, node: temp_node) -> List[AccountGroup]:
|
||||||
|
"""Return a list of permission groups that this account has on this node."""
|
||||||
|
...
|
||||||
|
|
||||||
|
def register_node(self, node: temp_node) -> None:
|
||||||
|
"""TODO."""
|
||||||
|
...
|
||||||
|
|
||||||
|
def deregister_node(self, node: temp_node) -> None:
|
||||||
|
"""TODO."""
|
||||||
|
...
|
||||||
|
|||||||
@@ -8,7 +8,13 @@ from primaite.simulator.domain.controller import AccountGroup, GroupMembershipVa
|
|||||||
|
|
||||||
|
|
||||||
def test_group_action_validation() -> None:
|
def test_group_action_validation() -> None:
|
||||||
"""Check that actions are denied when an unauthorised request is made."""
|
"""
|
||||||
|
Check that actions are denied when an unauthorised request is made.
|
||||||
|
|
||||||
|
This test checks the integration between SimComponent and the permissions validation system. First, we create a
|
||||||
|
basic node and folder class. We configure the node so that only admins can create a folder. Then, we try to create
|
||||||
|
a folder as both an admin user and a non-admin user.
|
||||||
|
"""
|
||||||
|
|
||||||
class Folder(SimComponent):
|
class Folder(SimComponent):
|
||||||
name: str
|
name: str
|
||||||
@@ -42,22 +48,28 @@ def test_group_action_validation() -> None:
|
|||||||
def remove_folder(self, folder: Folder) -> None:
|
def remove_folder(self, folder: Folder) -> None:
|
||||||
self.folders = [x for x in self.folders if x is not folder]
|
self.folders = [x for x in self.folders if x is not folder]
|
||||||
|
|
||||||
|
# check that the folder is created when a local admin tried to do it
|
||||||
permitted_context = {"request_source": {"agent": "BLUE", "account": "User1", "groups": ["local_admin"]}}
|
permitted_context = {"request_source": {"agent": "BLUE", "account": "User1", "groups": ["local_admin"]}}
|
||||||
|
|
||||||
my_node = Node(uuid="0000-0000-1234", name="pc")
|
my_node = Node(uuid="0000-0000-1234", name="pc")
|
||||||
my_node.apply_action(["create_folder", "memes"], context=permitted_context)
|
my_node.apply_action(["create_folder", "memes"], context=permitted_context)
|
||||||
assert len(my_node.folders) == 1
|
assert len(my_node.folders) == 1
|
||||||
assert my_node.folders[0].name == "memes"
|
assert my_node.folders[0].name == "memes"
|
||||||
|
|
||||||
|
# check that the number of folders is still 1 even after attempting to create a second one without permissions
|
||||||
invalid_context = {"request_source": {"agent": "BLUE", "account": "User1", "groups": ["local_user", "domain_user"]}}
|
invalid_context = {"request_source": {"agent": "BLUE", "account": "User1", "groups": ["local_user", "domain_user"]}}
|
||||||
|
|
||||||
my_node.apply_action(["create_folder", "memes2"], context=invalid_context)
|
my_node.apply_action(["create_folder", "memes2"], context=invalid_context)
|
||||||
assert len(my_node.folders) == 1
|
assert len(my_node.folders) == 1
|
||||||
assert my_node.folders[0].name == "memes"
|
assert my_node.folders[0].name == "memes"
|
||||||
|
|
||||||
|
|
||||||
def test_hierarchical_action_with_validation() -> None:
|
def test_hierarchical_action_with_validation() -> None:
|
||||||
"""Check that validation works with sub-objects"""
|
"""
|
||||||
|
Check that validation works with sub-objects.
|
||||||
|
|
||||||
|
This test creates a parent object (Node) and a child object (Application) which both accept actions. The node allows
|
||||||
|
action passthrough to applications. The purpose of this test is to check that after an action is passed through to
|
||||||
|
a child object, that the permission system still works as intended.
|
||||||
|
"""
|
||||||
|
|
||||||
class Application(SimComponent):
|
class Application(SimComponent):
|
||||||
name: str
|
name: str
|
||||||
@@ -100,19 +112,19 @@ def test_hierarchical_action_with_validation() -> None:
|
|||||||
return super().describe_state()
|
return super().describe_state()
|
||||||
|
|
||||||
def disable(self) -> None:
|
def disable(self) -> None:
|
||||||
self.status = "disabled"
|
self.state = "disabled"
|
||||||
|
|
||||||
def enable(self) -> None:
|
def enable(self) -> None:
|
||||||
if self.status == "disabled":
|
if self.state == "disabled":
|
||||||
self.status = "off"
|
self.state = "off"
|
||||||
|
|
||||||
def turn_on(self) -> None:
|
def turn_on(self) -> None:
|
||||||
if self.status == "off":
|
if self.state == "off":
|
||||||
self.status = "on"
|
self.state = "on"
|
||||||
|
|
||||||
def turn_off(self) -> None:
|
def turn_off(self) -> None:
|
||||||
if self.status == "on":
|
if self.state == "on":
|
||||||
self.status = "off"
|
self.state = "off"
|
||||||
|
|
||||||
class Node(SimComponent):
|
class Node(SimComponent):
|
||||||
name: str
|
name: str
|
||||||
@@ -141,7 +153,7 @@ def test_hierarchical_action_with_validation() -> None:
|
|||||||
def send_action_to_app(self, app_name: str, options: List[str], context: Dict):
|
def send_action_to_app(self, app_name: str, options: List[str], context: Dict):
|
||||||
for app in self.apps:
|
for app in self.apps:
|
||||||
if app_name == app.name:
|
if app_name == app.name:
|
||||||
app.apply_action(options)
|
app.apply_action(options, context)
|
||||||
break
|
break
|
||||||
else:
|
else:
|
||||||
msg = f"Node has no app with name {app_name}"
|
msg = f"Node has no app with name {app_name}"
|
||||||
@@ -163,9 +175,16 @@ def test_hierarchical_action_with_validation() -> None:
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
# check that a non-admin can't disable this app
|
||||||
my_node.apply_action(["apps", "Chrome", "disable"], non_admin_context)
|
my_node.apply_action(["apps", "Chrome", "disable"], non_admin_context)
|
||||||
my_node.apply_action(["apps", "Firefox", "turn_on"], non_admin_context)
|
assert my_node.apps[0].name == "Chrome" # if failure occurs on this line, the test itself is broken
|
||||||
|
assert my_node.apps[0].state == "off"
|
||||||
|
|
||||||
assert my_node.apps[0].name == "Chrome"
|
# check that a non-admin can turn this app on
|
||||||
assert my_node.apps[1].name == "Firefox"
|
my_node.apply_action(["apps", "Firefox", "turn_on"], non_admin_context)
|
||||||
assert my_node.apps[0].state == ... # TODO: finish
|
assert my_node.apps[1].name == "Firefox" # if failure occurs on this line, the test itself is broken
|
||||||
|
assert my_node.apps[1].state == "on"
|
||||||
|
|
||||||
|
# check that an admin can disable this app
|
||||||
|
my_node.apply_action(["apps", "Chrome", "disable"], admin_context)
|
||||||
|
assert my_node.apps[0].state == "disabled"
|
||||||
|
|||||||
@@ -0,0 +1,18 @@
|
|||||||
|
"""Test the account module of the simulator."""
|
||||||
|
from primaite.simulator.domain.account import Account, AccountType
|
||||||
|
|
||||||
|
|
||||||
|
def test_account_serialise():
|
||||||
|
"""Test that an account can be serialised."""
|
||||||
|
acct = Account(username="Jake", password="JakePass1!", account_type=AccountType.user)
|
||||||
|
serialised = acct.model_dump_json()
|
||||||
|
print(serialised)
|
||||||
|
|
||||||
|
|
||||||
|
def test_account_deserialise():
|
||||||
|
"""Test that an account can be deserialised."""
|
||||||
|
acct_json = (
|
||||||
|
'{"uuid":"dfb2bcaa-d3a1-48fd-af3f-c943354622b4","num_logons":0,"num_logoffs":0,"num_group_changes":0,'
|
||||||
|
'"username":"Jake","password":"JakePass1!","account_type":2,"status":2,"action_manager":null}'
|
||||||
|
)
|
||||||
|
acct = Account.model_validate_json(acct_json)
|
||||||
Reference in New Issue
Block a user