diff --git a/maltoolbox/attackgraph/analyzers/apriori.py b/maltoolbox/attackgraph/analyzers/apriori.py index 98c3c402..74a2381e 100644 --- a/maltoolbox/attackgraph/analyzers/apriori.py +++ b/maltoolbox/attackgraph/analyzers/apriori.py @@ -178,13 +178,12 @@ def prune_unviable_and_unnecessary_nodes(graph: AttackGraph) -> None: graph - the attack graph for which we wish to remove the the nodes which are not viable or necessary. """ - logger.debug( - 'Prune unviable and unnecessary nodes from the attack graph.') + logger.debug('Prune unviable and unnecessary nodes from the attack graph.') nodes_to_remove = set() for node in graph.nodes.values(): if node.type in ('or', 'and') and \ - (not node.is_viable or not node.is_necessary): + not node.is_viable or not node.is_necessary: nodes_to_remove.add(node) # Do the removal separatly so we don't remove diff --git a/maltoolbox/attackgraph/attackgraph.py b/maltoolbox/attackgraph/attackgraph.py index 6295142a..37cfabd9 100644 --- a/maltoolbox/attackgraph/attackgraph.py +++ b/maltoolbox/attackgraph/attackgraph.py @@ -2,12 +2,12 @@ MAL-Toolbox Attack Graph Module """ from __future__ import annotations -import copy import logging import json import sys import zipfile +from copy import deepcopy from itertools import chain from typing import TYPE_CHECKING @@ -20,10 +20,11 @@ from ..model import Model from ..language import (LanguageGraph, ExpressionsChain, LanguageGraphAttackStep, disaggregate_attack_step_full_name) -from ..file_utils import ( +from ..utils import ( load_dict_from_json_file, load_dict_from_yaml_file, - save_dict_to_file + save_dict_to_file, + LookupDict, ) @@ -82,11 +83,10 @@ def create_attack_graph( class AttackGraph(): """Graph representation of attack steps""" def __init__(self, lang_graph, model: Optional[Model] = None): - self.nodes: dict[int, AttackGraphNode] = {} + self.nodes: LookupDict[int, AttackGraphNode] = LookupDict() self.attackers: dict[int, Attacker] = {} # Dictionaries used in optimization to get nodes and attackers by id # or full name faster - self._full_name_to_node: dict[str, AttackGraphNode] = {} self.model = model self.lang_graph = lang_graph @@ -123,38 +123,24 @@ def __deepcopy__(self, memo): return memo[id(self)] copied_attackgraph = AttackGraph(self.lang_graph) - copied_attackgraph.model = self.model - copied_attackgraph.nodes = {} + memo[id(self)] = copied_attackgraph - # Deep copy nodes - for node_id, node in self.nodes.items(): - copied_node = copy.deepcopy(node, memo) - copied_attackgraph.nodes[node_id] = copied_node + copied_attackgraph.model = self.model + copied_attackgraph.nodes = deepcopy(self.nodes, memo) + copied_attackgraph.attackers = deepcopy(self.attackers, memo) - # Re-link node references for node in self.nodes.values(): + nid = id(node) if node.parents: - memo[id(node)].parents = copy.deepcopy(node.parents, memo) + memo[nid].parents = deepcopy(node.parents, memo) if node.children: - memo[id(node)].children = copy.deepcopy(node.children, memo) - - # Deep copy attackers - for attacker_id, attacker in self.attackers.items(): - copied_attacker = copy.deepcopy(attacker, memo) - copied_attackgraph.attackers[attacker_id] = copied_attacker - - # Re-link attacker references - for node in self.nodes.values(): + memo[nid].children = deepcopy(node.children, memo) if node.compromised_by: - memo[id(node)].compromised_by = copy.deepcopy( - node.compromised_by, memo) + memo[nid].compromised_by = deepcopy(node.compromised_by, memo) - # Copy lookup dicts - copied_attackgraph._full_name_to_node = \ - copy.deepcopy(self._full_name_to_node, memo) + copied_attackgraph.attackers = deepcopy(self.attackers, memo) - # Copy counters copied_attackgraph.next_node_id = self.next_node_id copied_attackgraph.next_attacker_id = self.next_attacker_id @@ -189,8 +175,9 @@ def _from_dict( # Recreate asset links if model is available. node_asset = None if model and 'asset' in node_dict: - node_asset = model.get_asset_by_name(node_dict['asset']) - if node_asset is None: + try: + node_asset = model.assets.fetch("name", node_dict['asset']) + except KeyError: msg = ('Failed to find asset with name "%s"' ' when loading from attack graph dict') logger.error(msg, node_dict["asset"]) @@ -293,21 +280,6 @@ def load_from_file( return cls._from_dict(serialized_attack_graph, lang_graph, model = model) - def get_node_by_full_name(self, full_name: str) -> Optional[AttackGraphNode]: - """ - Return the attack node that matches the full name provided. - - Arguments: - full_name - the full name of the attack graph node we are looking - for - - Return: - The attack step node that matches the given full name. - """ - - logger.debug(f'Looking up node with full name "%s"', full_name) - return self._full_name_to_node.get(full_name) - def attach_attackers(self) -> None: """ Create attackers and their entry point nodes and attach them to the @@ -340,7 +312,7 @@ def attach_attackers(self) -> None: for (asset, attack_steps) in attacker_info.entry_points: for attack_step in attack_steps: full_name = asset.name + ':' + attack_step - ag_node = self.get_node_by_full_name(full_name) + ag_node = self.nodes.fetch("full_name", full_name) if not ag_node: logger.warning( 'Failed to find attacker entry point ' @@ -620,8 +592,8 @@ def _generate_graph(self) -> None: if target_asset is not None: target_node_full_name = target_asset.name + \ ':' + target_attack_step.name - target_node = self.get_node_by_full_name( - target_node_full_name) + target_node = self.nodes.fetch("full_name", + target_node_full_name) if target_node is None: msg = ('Failed to find target node ' '"%s" to link with for attack ' @@ -665,7 +637,7 @@ def regenerate_graph(self) -> None: the MAL language specification provided at initialization. """ - self.nodes = {} + self.nodes = LookupDict() self.attackers = {} self._generate_graph() @@ -713,7 +685,6 @@ def add_node( node_id )) - node = AttackGraphNode( node_id = node_id, lg_attack_step = lg_attack_step, @@ -722,8 +693,7 @@ def add_node( existence_status = existence_status ) - self.nodes[node_id] = node - self._full_name_to_node[node.full_name] = node + self.nodes[node.id] = node return node @@ -740,10 +710,7 @@ def remove_node(self, node: AttackGraphNode) -> None: for parent in node.parents: parent.children.remove(node) - if not isinstance(node.id, int): - raise ValueError(f'Invalid node id.') - del self.nodes[node.id] - del self._full_name_to_node[node.full_name] + self.nodes.pop(node.id, None) def add_attacker( self, diff --git a/maltoolbox/attackgraph/node.py b/maltoolbox/attackgraph/node.py index 1d9466f1..aa4dccdd 100644 --- a/maltoolbox/attackgraph/node.py +++ b/maltoolbox/attackgraph/node.py @@ -118,6 +118,8 @@ def __deepcopy__(self, memo) -> AttackGraphNode: return copied_node + def __hash__(self): + return hash(self.full_name) def is_compromised(self) -> bool: """ diff --git a/maltoolbox/file_utils.py b/maltoolbox/file_utils.py deleted file mode 100644 index 476cd732..00000000 --- a/maltoolbox/file_utils.py +++ /dev/null @@ -1,62 +0,0 @@ -"""Utily functions for file handling""" - -import json -import yaml - -def save_dict_to_json_file(filename: str, serialized_object: dict) -> None: - """Save serialized object to a json file. - - Arguments: - filename - the name of the output file - data - dict to output as json - """ - - with open(filename, 'w', encoding='utf-8') as f: - json.dump(serialized_object, f, indent=4) - - -def save_dict_to_yaml_file(filename: str, serialized_object: dict) -> None: - """Save serialized object to a yaml file. - - Arguments: - filename - the name of the output file - data - dict to output as yaml - """ - - class NoAliasSafeDumper(yaml.SafeDumper): - def ignore_aliases(self, data): - return True - - with open(filename, 'w', encoding='utf-8') as f: - yaml.dump(serialized_object, f, Dumper=NoAliasSafeDumper) - - -def load_dict_from_yaml_file(filename: str) -> dict: - """Open json file and read as dict""" - with open(filename, 'r', encoding='utf-8') as file: - object_dict = yaml.safe_load(file) - return object_dict - - -def load_dict_from_json_file(filename: str) -> dict: - """Open yaml file and read as dict""" - with open(filename, 'r', encoding='utf-8') as file: - object_dict = json.loads(file.read()) - return object_dict - - -def save_dict_to_file(filename: str, dictionary: dict) -> None: - """Save serialized object to json or yaml file - depending on file extension. - - Arguments: - filename - the name of the output file - dictionary - the dict to save to the file - """ - - if filename.endswith(('.yml', '.yaml')): - save_dict_to_yaml_file(filename, dictionary) - elif filename.endswith('.json'): - save_dict_to_json_file(filename, dictionary) - else: - raise ValueError('Unknown file extension, expected json/yml/yaml') diff --git a/maltoolbox/language/languagegraph.py b/maltoolbox/language/languagegraph.py index 2127ab65..3559d8eb 100644 --- a/maltoolbox/language/languagegraph.py +++ b/maltoolbox/language/languagegraph.py @@ -12,10 +12,6 @@ from functools import cached_property from typing import Any, Optional -from maltoolbox.file_utils import ( - load_dict_from_yaml_file, load_dict_from_json_file, - save_dict_to_file -) from .compiler import MalCompiler from ..exceptions import ( LanguageGraphAssociationError, @@ -23,6 +19,10 @@ LanguageGraphException, LanguageGraphSuperAssetNotFoundError ) +from ..utils import ( + load_dict_from_yaml_file, load_dict_from_json_file, + save_dict_to_file +) logger = logging.getLogger(__name__) diff --git a/maltoolbox/model.py b/maltoolbox/model.py index c8d54b91..e3d34db9 100644 --- a/maltoolbox/model.py +++ b/maltoolbox/model.py @@ -9,12 +9,14 @@ from typing import TYPE_CHECKING import math -from .file_utils import ( +from .utils import ( load_dict_from_json_file, load_dict_from_yaml_file, - save_dict_to_file + save_dict_to_file, + LookupDict, ) + from . import __version__ from .exceptions import ModelException @@ -146,8 +148,7 @@ def __init__( ): self.name = name - self.assets: dict[int, ModelAsset] = {} - self._name_to_asset:dict[str, ModelAsset] = {} # optimization + self.assets: LookupDict[int, ModelAsset] = LookupDict() self.attackers: list[AttackerAttachment] = [] self.lang_graph = lang_graph self.maltoolbox_version: str = mt_version @@ -224,7 +225,6 @@ def add_asset( 'Add "%s"(%d) to model "%s".', name, asset_id, self.name ) self.assets[asset_id] = asset - self._name_to_asset[name] = asset return asset @@ -262,7 +262,6 @@ def remove_asset(self, asset: ModelAsset) -> None: attacker.entry_points.remove(entry_point_tuple) del self.assets[asset.id] - del self._name_to_asset[asset.name] def add_attacker( @@ -288,44 +287,6 @@ def add_attacker( self.attackers.append(attacker) - def get_asset_by_id( - self, asset_id: int - ) -> Optional[ModelAsset]: - """ - Find an asset in the model based on its id. - - Arguments: - asset_id - the id of the asset we are looking for - - Return: - An asset matching the id if it exists in the model. - """ - logger.debug( - 'Get asset with id %d from model "%s".', - asset_id, self.name - ) - return self.assets.get(asset_id, None) - - - def get_asset_by_name( - self, asset_name: str - ) -> Optional[ModelAsset]: - """ - Find an asset in the model based on its name. - - Arguments: - asset_name - the name of the asset we are looking for - - Return: - An asset matching the name if it exists in the model. - """ - logger.debug( - 'Get asset with name "%s" from model "%s".', - asset_name, self.name - ) - return self._name_to_asset.get(asset_name, None) - - def get_attacker_by_id( self, attacker_id: int ) -> Optional[AttackerAttachment]: @@ -471,9 +432,9 @@ def _from_dict( attacker.entry_points = [] for asset_name, entry_points_dict in \ attackers_info[attacker_id]['entry_points'].items(): - target_asset = model.get_asset_by_id( - entry_points_dict['asset_id']) - if target_asset is None: + try: + target_asset = model.assets[entry_points_dict['asset_id']] + except KeyError: raise LookupError( 'Asset "%s"(%d) is not part of model "%s".' % ( asset_name, diff --git a/maltoolbox/translators/updater.py b/maltoolbox/translators/updater.py index 1d8e983b..ba21b9f4 100644 --- a/maltoolbox/translators/updater.py +++ b/maltoolbox/translators/updater.py @@ -6,7 +6,7 @@ import logging from ..model import Model from ..language import LanguageGraph -from ..file_utils import load_dict_from_json_file, load_dict_from_yaml_file +from ..utils import load_dict_from_json_file, load_dict_from_yaml_file logger = logging.getLogger(__name__) diff --git a/maltoolbox/utils.py b/maltoolbox/utils.py new file mode 100644 index 00000000..9006b30e --- /dev/null +++ b/maltoolbox/utils.py @@ -0,0 +1,187 @@ +"""Various generic classes and methods.""" +import json +import operator +from typing import Any, Dict, TypeVar + +import yaml + +K = TypeVar("K") +V = TypeVar("V") + + +class LookupDict(Dict[K, V]): + """ + A dict subclass that provides lookup functionality based on object attributes. + + Maintains an internal index to optimize repeated lookups based on a specific key. + """ + + def __init__(self, *args, **kwargs): + """ + Initialize the LookupDict with optional initial dictionary data. + + Arguments should be passes similar to `dict()`. + + Args: + *args: Positional arguments passed to the base dictionary. + **kwargs: Keyword arguments passed to the base dictionary. + """ + super().__init__(*args, **kwargs) + self._indices: dict[str, dict] = {} + + def __setitem__(self, key, value): + """ + Set an item in the dictionary and clear all cached indices. + + Args: + key (K): The key under which the value is stored. + value (V): The value to store. + """ + super().__setitem__(key, value) + + for key in self._indices: + self._indices[key][getattr(value, key)] = value + + def __delitem__(self, key): + """ + Delete an item from the dictionary and clear all cached indices. + + Args: + key (K): The key to delete. + """ + item = self[key] + + super().__delitem__(key) + + for index in self._indices: + del self._indices[index][getattr(item, index)] + + def lookup(self, key: str, value: Any, op: str = "eq") -> set[Any]: + """ + Search in the dictionary based on an attribute and a comparison operation. + + Args: + key (str): The attribute name to look up within the stored values. + value (Any): The value to compare against. + op (str, optional): The comparison operator (default: "eq"). + Supported operators include "eq", "in", and others from the `operator` + module. + + Returns: + set[Any]: A set of matching values from the dictionary. + + Raises: + ValueError: If an invalid operator is provided. + """ + if key not in self._indices: + self._indices[key] = {getattr(v, key): set() for v in self.values()} + for v in self.values(): + if not (vkey := getattr(v, key)) in self._indices[key].keys(): + self._indices[key][vkey] = set() + self._indices[key][vkey].add(v) + + if op == "eq": + return self._indices[key].get(value, set()) + + if op == "in": + def oper(a, b): return operator.contains(b, a) + else: + try: + oper = getattr(operator, op) + except AttributeError: + raise ValueError(f"Invalid operator {op}") + + ret = set() + for i in filter(lambda k: oper(k, value), self._indices[key]): + ret.update(self._indices[key][i]) + + return ret + + def fetch(self, key: str, value: Any, op: str = "eq") -> Any: + """ + Fetch a single matching item from the dictionary. + + If multiple items match, an exception is raised. If you expect + multiple matches use `lookup()` instead. + + Args: + key (str): The attribute name to look up within the stored values. + value (Any): The value to compare against. + op (str, optional): The comparison operator (default: "eq"). + Supported operators include "eq", "in", and others from the `operator` + module. + + Returns: + Any: The matching item if one is found, otherwise `None`. + + Raises: + ValueError: If multiple matching items are found. + """ + ret = self.lookup(key, value, op) + + if len(ret) > 1: + raise ValueError(f"Multiple items match {key} ~ {op} ~ {value}") + + try: + return next(iter(ret)) + except KeyError: + return None + + +def save_dict_to_json_file(filename: str, serialized_object: dict) -> None: + """Save serialized object to a json file. + + Arguments: + filename - the name of the output file + data - dict to output as json + """ + + with open(filename, 'w', encoding='utf-8') as f: + json.dump(serialized_object, f, indent=4) + + +def save_dict_to_yaml_file(filename: str, serialized_object: dict) -> None: + """Save serialized object to a yaml file. + + Arguments: + filename - the name of the output file + data - dict to output as yaml + """ + + class NoAliasSafeDumper(yaml.SafeDumper): + def ignore_aliases(self, data): + return True + + with open(filename, 'w', encoding='utf-8') as f: + yaml.dump(serialized_object, f, Dumper=NoAliasSafeDumper) + + +def load_dict_from_yaml_file(filename: str) -> dict: + """Open json file and read as dict""" + with open(filename, 'r', encoding='utf-8') as file: + object_dict = yaml.safe_load(file) + return object_dict + + +def load_dict_from_json_file(filename: str) -> dict: + """Open yaml file and read as dict""" + with open(filename, 'r', encoding='utf-8') as file: + object_dict = json.loads(file.read()) + return object_dict + + +def save_dict_to_file(filename: str, dictionary: dict) -> None: + """Save serialized object to json or yaml file + depending on file extension. + + Arguments: + filename - the name of the output file + dictionary - the dict to save to the file + """ + + if filename.endswith(('.yml', '.yaml')): + save_dict_to_yaml_file(filename, dictionary) + elif filename.endswith('.json'): + save_dict_to_json_file(filename, dictionary) + else: + raise ValueError('Unknown file extension, expected json/yml/yaml') diff --git a/tests/attackgraph/test_attackgraph.py b/tests/attackgraph/test_attackgraph.py index 2669cf67..89e508e4 100644 --- a/tests/attackgraph/test_attackgraph.py +++ b/tests/attackgraph/test_attackgraph.py @@ -13,6 +13,7 @@ create_attack_graph ) from maltoolbox.model import Model +from maltoolbox.utils import LookupDict def test_attackgraph_init(corelang_lang_graph, model): @@ -177,7 +178,7 @@ def attackgraph_save_and_load_json_yml_model_given( # Make sure node was added to lookup dict with correct id / name assert node.id is not None assert loaded_attackgraph.nodes[node.id] == node - assert loaded_attackgraph.get_node_by_full_name(node.full_name) == node + assert loaded_attackgraph.nodes.fetch("full_name", node.full_name) == node for loaded_attacker in loaded_attackgraph.attackers.values(): if not isinstance(loaded_attacker.id, int): @@ -214,10 +215,10 @@ def test_attackgraph_save_and_load_json_yml_model_given_with_attackers( def test_attackgraph_attach_attackers(example_attackgraph: AttackGraph): """Make sure attackers are properly attached to graph""" - app1_ncu = example_attackgraph.get_node_by_full_name( + app1_ncu = example_attackgraph.nodes.fetch("full_name", 'Application 1:networkConnectUninspected' ) - app1_auv = example_attackgraph.get_node_by_full_name( + app1_auv = example_attackgraph.nodes.fetch("full_name", 'Application 1:attemptUseVulnerability' ) @@ -250,7 +251,7 @@ def test_attackgraph_generate_graph(example_attackgraph: AttackGraph): # TODO: Add test cases with defense steps # Empty the attack graph - example_attackgraph.nodes = {} + example_attackgraph.nodes = LookupDict() example_attackgraph.attackers = {} # Generate the attack graph again @@ -400,9 +401,6 @@ def test_attackgraph_deepcopy(example_attackgraph: AttackGraph): assert list(copied_attackgraph.attackers.keys()) \ == list(example_attackgraph.attackers.keys()) - assert list(copied_attackgraph._full_name_to_node.keys()) \ - == list(example_attackgraph._full_name_to_node.keys()) - assert id(copied_attackgraph.model) == id(example_attackgraph.model) assert len(copied_attackgraph.nodes) \ @@ -542,17 +540,17 @@ def test_attackgraph_subtype(): lang_graph=test_lang_graph, model=test_model ) - ba_1_base_step1 = test_attack_graph.get_node_by_full_name( + ba_1_base_step1 = test_attack_graph.nodes.fetch("full_name", 'BaseAsset 1:base_step1') - ba_1_base_step2 = test_attack_graph.get_node_by_full_name( + ba_1_base_step2 = test_attack_graph.nodes.fetch("full_name", 'BaseAsset 1:base_step2') - sa_1_base_step1 = test_attack_graph.get_node_by_full_name( + sa_1_base_step1 = test_attack_graph.nodes.fetch("full_name", 'SubAsset 1:base_step1') - sa_1_base_step2 = test_attack_graph.get_node_by_full_name( + sa_1_base_step2 = test_attack_graph.nodes.fetch("full_name", 'SubAsset 1:base_step2') - sa_1_subasset_step1 = test_attack_graph.get_node_by_full_name( + sa_1_subasset_step1 = test_attack_graph.nodes.fetch("full_name", 'SubAsset 1:subasset_step1') - oa_1_other_step1 = test_attack_graph.get_node_by_full_name( + oa_1_other_step1 = test_attack_graph.nodes.fetch("full_name", 'OtherAsset 1:other_step1') assert ba_1_base_step1 in oa_1_other_step1.children @@ -593,25 +591,25 @@ def test_attackgraph_setops(): model=test_model ) - assetA1_opsA = test_attack_graph.get_node_by_full_name( + assetA1_opsA = test_attack_graph.nodes.fetch("full_name", 'SetOpsAssetA 1:testStepSetOpsA') - assetB1_opsB1 = test_attack_graph.get_node_by_full_name( + assetB1_opsB1 = test_attack_graph.nodes.fetch("full_name", 'SetOpsAssetB 1:testStepSetOpsB1') - assetB1_opsB2 = test_attack_graph.get_node_by_full_name( + assetB1_opsB2 = test_attack_graph.nodes.fetch("full_name", 'SetOpsAssetB 1:testStepSetOpsB2') - assetB1_opsB3 = test_attack_graph.get_node_by_full_name( + assetB1_opsB3 = test_attack_graph.nodes.fetch("full_name", 'SetOpsAssetB 1:testStepSetOpsB3') - assetB2_opsB1 = test_attack_graph.get_node_by_full_name( + assetB2_opsB1 = test_attack_graph.nodes.fetch("full_name", 'SetOpsAssetB 2:testStepSetOpsB1') - assetB2_opsB2 = test_attack_graph.get_node_by_full_name( + assetB2_opsB2 = test_attack_graph.nodes.fetch("full_name", 'SetOpsAssetB 2:testStepSetOpsB2') - assetB2_opsB3 = test_attack_graph.get_node_by_full_name( + assetB2_opsB3 = test_attack_graph.nodes.fetch("full_name", 'SetOpsAssetB 2:testStepSetOpsB3') - assetB3_opsB1 = test_attack_graph.get_node_by_full_name( + assetB3_opsB1 = test_attack_graph.nodes.fetch("full_name", 'SetOpsAssetB 3:testStepSetOpsB1') - assetB3_opsB2 = test_attack_graph.get_node_by_full_name( + assetB3_opsB2 = test_attack_graph.nodes.fetch("full_name", 'SetOpsAssetB 3:testStepSetOpsB2') - assetB3_opsB3 = test_attack_graph.get_node_by_full_name( + assetB3_opsB3 = test_attack_graph.nodes.fetch("full_name", 'SetOpsAssetB 3:testStepSetOpsB3') assert assetB1_opsB1 in assetA1_opsA.children @@ -659,17 +657,17 @@ def test_attackgraph_transitive(): model=test_model ) - asset1_test_step = test_attack_graph.get_node_by_full_name( + asset1_test_step = test_attack_graph.nodes.fetch("full_name", 'TestAsset 1:test_step') - asset2_test_step = test_attack_graph.get_node_by_full_name( + asset2_test_step = test_attack_graph.nodes.fetch("full_name", 'TestAsset 2:test_step') - asset3_test_step = test_attack_graph.get_node_by_full_name( + asset3_test_step = test_attack_graph.nodes.fetch("full_name", 'TestAsset 3:test_step') - asset4_test_step = test_attack_graph.get_node_by_full_name( + asset4_test_step = test_attack_graph.nodes.fetch("full_name", 'TestAsset 4:test_step') - asset5_test_step = test_attack_graph.get_node_by_full_name( + asset5_test_step = test_attack_graph.nodes.fetch("full_name", 'TestAsset 5:test_step') - asset6_test_step = test_attack_graph.get_node_by_full_name( + asset6_test_step = test_attack_graph.nodes.fetch("full_name", 'TestAsset 6:test_step') assert asset1_test_step in asset1_test_step.children @@ -745,13 +743,13 @@ def test_attackgraph_transitive_advanced(): model=test_model ) - asset1_test_step = test_attack_graph.get_node_by_full_name( + asset1_test_step = test_attack_graph.nodes.fetch("full_name", 'TestAsset 1:test_step') - asset2_test_step = test_attack_graph.get_node_by_full_name( + asset2_test_step = test_attack_graph.nodes.fetch("full_name", 'TestAsset 2:test_step') - asset3_test_step = test_attack_graph.get_node_by_full_name( + asset3_test_step = test_attack_graph.nodes.fetch("full_name", 'TestAsset 3:test_step') - asset4_test_step = test_attack_graph.get_node_by_full_name( + asset4_test_step = test_attack_graph.nodes.fetch("full_name", 'TestAsset 4:test_step') assert asset1_test_step in asset1_test_step.children @@ -781,7 +779,7 @@ def check_parent_child_relationship( ag: AttackGraph, parent_fn: str, children_fns: list[str] ): - parent = ag.get_node_by_full_name(parent_fn) + parent = ag.nodes.fetch("full_name", parent_fn) assert parent, f"Could not find node {parent_fn}" # Verify that parent has given children @@ -789,7 +787,7 @@ def check_parent_child_relationship( # Verify that child has given parent for child_fn in children_fns: - child = ag.get_node_by_full_name(child_fn) + child = ag.nodes.fetch("full_name", child_fn) assert child, f"Could not find child by full name {child_fn}" assert parent_fn in [p.full_name for p in child.parents] diff --git a/tests/test_model.py b/tests/test_model.py index df98ca7a..2c1c7be9 100644 --- a/tests/test_model.py +++ b/tests/test_model.py @@ -473,34 +473,6 @@ def test_model_add_attacker(model: Model): assert attacker2.name == f'Attacker:{attacker2_id}' -def test_model_get_asset_by_id(model: Model): - """Make sure correct asset is returned or None - if no asset with that ID exists""" - - # Create and add 2 applications - asset1 = model.add_asset(asset_type = 'Application') - asset2 = model.add_asset(asset_type = 'Application') - - # Correct assets removed and None if asset with that not exists - assert model.get_asset_by_id(asset1.id) == asset1 - assert model.get_asset_by_id(asset2.id) == asset2 - assert model.get_asset_by_id(1337) is None - - -def test_model_get_asset_by_name(model: Model): - """Make sure correct asset is returned or None - if no asset with that name exists""" - - # Create and add 2 applications - asset1 = model.add_asset(asset_type = 'Application') - asset2 = model.add_asset(asset_type = 'Application') - - # Correct assets removed and None if asset with that name not exists - assert model.get_asset_by_name(asset1.name) == asset1 - assert model.get_asset_by_name(asset2.name) == asset2 - assert model.get_asset_by_name("Program 3") is None - - def test_model_get_attacker_by_id(model: Model): """Make sure correct attacker is returned of None if no attacker with that ID exists"""