DynaMAL Implementation

.github/CODEOWNERS

@@ -3,3 +3,5 @@
 
 # Owners for a specific folder
 maltoolbox/language/compiler/ @nkakouros
+maltoolbox/language/compiler/mal_compiler.py @nkakouros @sandorstormen
+maltoolbox/language/language_graph_model_effect.py @sandorstormen

.gitignore

@@ -66,4 +66,8 @@ nosetests.xml
 !examples/resources/*
 
 # uv lock file
-uv.lock
+uv.lock
+
+# Claude
+.claude
+*.local.*

maltoolbox/language/compiler/mal_analyzer.py

@@ -584,12 +584,12 @@ def _read_steps(self, asset: str, parents: List) -> None:
                             or attackStep_node_reaches.child_by_field_name(
                                 'operator'
                             ).text.decode()
-                            == '->'
+                            in ('->', 'A>', 'R>')
                         ):
                             # Valid step
                             current_steps.append(attackStep)
                         else:
-                            # Step was defined using '+>', but there is nothing to inherit from
+                            # Step was defined using '+>', '+A>', or '+R>', but there is nothing to inherit from
                             error_msg = f"Cannot inherit attack step '{attackStep}' without previous definition"
                             self._raise_analyzer_exception(error_msg)
                     else:

maltoolbox/language/compiler/mal_compiler.py

@@ -462,11 +462,6 @@ def visit_attack_step(self, cursor: TreeCursor) -> tuple[str, dict]:
             requires = self.visit(cursor)
             go_to_sibling(cursor)
 
-        reaches = None
-        if assert_node(cursor.node).type == 'reaching':
-            reaches = self.visit(cursor)
-            go_to_sibling(cursor)
-
         ret = {
             'name': name,
             'meta': meta,
@@ -477,9 +472,23 @@ def visit_attack_step(self, cursor: TreeCursor) -> tuple[str, dict]:
             'risk': risk,
             'ttc': ttc,
             'requires': requires,
-            'reaches': reaches,
+            'reaches': None,
+            'append_reaches': None,
+            'remove_reaches': None
         }
 
+        reaching_types = ('reaching', 'append_reaching', 'remove_reaching')
+
+        while assert_node(cursor.node).type in reaching_types:
+            if assert_node(cursor.node).type == 'reaching':
+                ret['reaches'] = self.visit(cursor)
+            elif assert_node(cursor.node).type == 'append_reaching':
+                ret['append_reaches'] = self.visit(cursor)
+            elif assert_node(cursor.node).type == 'remove_reaching':
+                ret['remove_reaches'] = self.visit(cursor)
+            if not go_to_sibling(cursor):
+                break
+
         return ('step', ret)
 
     def visit_detector(self, cursor: TreeCursor):
@@ -797,9 +806,76 @@ def visit_reaching(self, cursor: TreeCursor):
             ret['stepExpressions'].append(self.visit(cursor))
 
         return ret
+
+    def visit_append_reaching(self, cursor: TreeCursor):
+        #################################################
+        # ( '+A>' | 'A>' ) (reaches) ( ',' (reaches) )* #
+        #################################################
+
+        ret: dict[str, Any] = {}
+
+        # Get type of reaches
+        ret['overrides'] = assert_node(cursor.node).text == b'A>'
+        go_to_sibling(cursor)
+
+        # Visit the steps
+        ret['stepExpressions'] = [self.visit(cursor)]
+
+        while go_to_sibling(cursor):  # check if we have a ','
+            go_to_sibling(cursor)  # ignore the ','
+            ret['stepExpressions'].append(self.visit(cursor))
+
+        return ret
+
+    def visit_remove_reaching(self, cursor: TreeCursor):
+        #################################################
+        # ( '+R>' | 'R>' ) (reaches) ( ',' (reaches) )* #
+        #################################################
+
+        ret: dict[str, Any] = {}
+
+        # Get type of reaches
+        ret['overrides'] = assert_node(cursor.node).text == b'R>'
+        go_to_sibling(cursor)
+
+        # Visit the steps
+        ret['stepExpressions'] = [self.visit(cursor)]
+
+        while go_to_sibling(cursor):  # check if we have a ','
+            go_to_sibling(cursor)  # ignore the ','
+            ret['stepExpressions'].append(self.visit(cursor))
+
+        return ret
+
+    def visit_dyn_sentence(self, cursor: TreeCursor):
+        #########################################
+        # (base) '/' (target) ( '^' (target) )* #
+        #########################################
+
+        base = self.visit(cursor)
+        go_to_sibling(cursor)
+        assert assert_node(cursor.node).type == 'dynamic_separator', 'Expected dynamic separator in dynamic sentence'
+        go_to_sibling(cursor)
+        targets = [self.visit(cursor)]
+        while go_to_sibling(cursor):
+            assert assert_node(cursor.node).type == 'hold', 'Expected hold between targets in dynamic sentence'
+            go_to_sibling(cursor)
+            targets.append(self.visit(cursor))
+
+        return {"type": "dyn_sentence", "base": base, "targets": targets}
+
+    def visit_assoc_and_asset_expr(self, cursor: TreeCursor):
+        if assert_node(cursor.node).type == 'assoc_op':
+            assert cursor.goto_next_sibling(), "Expected next sibling after assoc_op"
+            result = {"type": "assoc_op", "operand": self.visit(cursor)}
+            assert result["operand"] is not None, "Expected operand in assoc_and_asset_expr"
+        else:
+            result = self.visit(cursor)
+        return result
 
     def visit_asset_expr(self, cursor: TreeCursor):
-        return self._visit_inline_asset_expr(cursor)
+        ret = self._visit_inline_asset_expr(cursor)
+        return ret
 
     def _visit_inline_asset_expr(self, cursor: TreeCursor):
         #############################################################################################################################################

maltoolbox/language/expression_chain.py

@@ -32,6 +32,7 @@ class ExprType(str, Enum):
     FIELD = "field"
     TRANSITIVE = "transitive"
     SUBTYPE = "subType"
+    ASSOC_OP = "assoc_op"
 
     def is_binary(self) -> bool:
         return self in {
@@ -74,6 +75,10 @@ def _validate(self) -> None:
         if self.type == ExprType.SUBTYPE:
             if not self.sub_link or not self.subtype:
                 raise ValueError("SUBTYPE requires sub_link and subtype")
+
+        if self.type == ExprType.ASSOC_OP:
+            if not self.sub_link:
+                raise ValueError("ASSOC_OP requires sub_link")
 
     def to_dict(self) -> dict:
         if self.type.is_binary():
@@ -87,6 +92,9 @@ def to_dict(self) -> dict:
 
         if self.type == ExprType.SUBTYPE:
             return self._subtype_to_dict()
+
+        if self.type == ExprType.ASSOC_OP:
+            return self._assoc_op_to_dict()
 
         raise LanguageGraphAssociationError(
             f"Unknown expressions chain element {self.type}"
@@ -116,6 +124,13 @@ def _subtype_to_dict(self) -> dict:
             "expression": self.sub_link.to_dict(),
             "type": self.type.value,
         }
+
+    def _assoc_op_to_dict(self) -> dict:
+        assert self.sub_link, "ASSOC_OP expression requires sub_link"
+        return {
+                "operand": self.sub_link.to_dict(),
+                "type": self.type.value
+            }
 
     def _field_to_dict(self) -> dict:
         asset_type = self._resolve_field_asset_type()

maltoolbox/language/language_graph_attack_step.py

@@ -6,11 +6,12 @@
 from dataclasses import dataclass, field
 from functools import cached_property
 
+
 if TYPE_CHECKING:
     from maltoolbox.language.expression_chain import ExpressionsChain
     from maltoolbox.language.language_graph_asset import LanguageGraphAsset
     from maltoolbox.language.language_graph_detector import LanguageGraphDetector
-
+    from maltoolbox.language.language_graph_model_effect import LanguageGraphModelEffect
 
 @dataclass
 class LanguageGraphAttackStep:
@@ -26,6 +27,8 @@ class LanguageGraphAttackStep:
     own_children: dict[
         LanguageGraphAttackStep, list[ExpressionsChain | None]
     ] = field(default_factory=dict)
+    own_additive_model_effects: list[LanguageGraphModelEffect] = field(default_factory=list)
+    own_subtractive_model_effects: list[LanguageGraphModelEffect] = field(default_factory=list)
     own_parents: dict[
         LanguageGraphAttackStep, list[ExpressionsChain | None]
     ] = field(default_factory=dict)

maltoolbox/language/language_graph_builder.py

@@ -9,6 +9,7 @@
 from maltoolbox.language.language_graph_asset import LanguageGraphAsset
 from maltoolbox.language.language_graph_assoc import LanguageGraphAssociation, LanguageGraphAssociationField, link_association_to_assets
 from maltoolbox.language.language_graph_attack_step import LanguageGraphAttackStep
+from maltoolbox.language.language_graph_model_effect import build_model_effect
 from maltoolbox.language.step_expression_processor import process_step_expression, resolve_variable, reverse_expr_chain
 
 
@@ -200,7 +201,7 @@ def _connect_attack_steps(
     lang_spec: dict[str, Any],
     attack_step_dicts: dict[str, dict]
 ) -> None:
-    """Connect attack steps based on the 'reaches' and 'requires' expressions in the language specification."""
+    """Connect attack steps based on the 'reaches', 'append_reaches', 'remove_reaches' and 'requires' expressions in the language specification."""
 
     for asset in assets.values():
         for step in asset.attack_steps.values():
@@ -244,6 +245,37 @@ def _connect_attack_steps(
                             f'Failed to find existence step requirement for:\n{expr}'
                         )
                     step.own_requires.append(chain)
+
+            # append_reaches = attack_step_dict.get('append_reaches') or {}
+            # for append_expression in append_reaches.get('stepExpressions', []):
+            #     base_asset, base_expr, base_targets = proccess_dyn_sentence(assets, step.asset, append_expression, lang_spec)
+
+def _connect_model_effects(
+    assets: dict[str, LanguageGraphAsset],
+    lang_spec: dict[str, Any],
+    attack_step_dicts: dict[str, dict]
+) -> None:
+    """Connect model effects to attack steps"""
+
+    for asset in assets.values():   
+        for step in asset.attack_steps.values():
+            logger.debug('Determining children for attack step %s', step.name)
+
+            if step.full_name not in attack_step_dicts:
+                # Skip steps that are not in the lang spec (inherited steps)
+                continue
+
+            attack_step_dict = attack_step_dicts[step.full_name]
+
+            append_reaches = attack_step_dict.get('append_reaches') or {}
+            for expr in append_reaches.get('stepExpressions', []):
+                model_effect = build_model_effect(assets, step.asset, expr, lang_spec)
+                step.own_additive_model_effects.append(model_effect)
+
+            remove_reaches = attack_step_dict.get('remove_reaches') or {}
+            for expr in remove_reaches.get('stepExpressions', []):
+                model_effect = build_model_effect(assets, step.asset, expr, lang_spec)
+                step.own_subtractive_model_effects.append(model_effect)
 
 
 def generate_attack_steps(assets: dict[str, LanguageGraphAsset], lang_spec: dict) -> None:
@@ -257,6 +289,7 @@ def generate_attack_steps(assets: dict[str, LanguageGraphAsset], lang_spec: dict
     2. Inherit attack steps from super-assets, respecting overrides.
     3. Link attack steps via 'reaches' and evaluate 'exist'/'notExist'
     requirements.
+    4. Link model effects via 'append_reaches' and 'remove_reaches'.
 
     Args:
         assets (dict): Mapping of asset names to asset objects.
@@ -271,6 +304,7 @@ def generate_attack_steps(assets: dict[str, LanguageGraphAsset], lang_spec: dict
     attack_step_dicts = _create_lg_attack_step_nodes(assets, lang_spec)
     _inherit_attack_steps(assets)
     _connect_attack_steps(assets, lang_spec, attack_step_dicts)
+    _connect_model_effects(assets, lang_spec, attack_step_dicts)
 
 
 def create_associations_for_assets(

maltoolbox/language/language_graph_model_effect.py

@@ -0,0 +1,103 @@
+"""LanguageGraphAttackStep functionality
+- Represents a step (type) defined in a MAL language
+"""
+from __future__ import annotations
+from copy import deepcopy
+from typing import TYPE_CHECKING, NamedTuple, Optional
+from dataclasses import dataclass
+
+from maltoolbox.language.expression_chain import ExprType
+from maltoolbox.language.step_expression_processor import process_assoc_op_step_expression, process_step_expression
+
+if TYPE_CHECKING:
+    from maltoolbox.language.expression_chain import ExpressionsChain
+    from maltoolbox.language.language_graph_asset import LanguageGraphAsset
+
+
+class AssocTraversal(NamedTuple):
+    field_name: str
+    asset_filter: Optional[LanguageGraphAsset]
+    quantity_filter: Optional[int]
+
+SELF_TRAVERSAL = AssocTraversal(field_name="self", asset_filter=None, quantity_filter=None)
+
+class DynTarget(NamedTuple):
+    """A target in a dynamic sentence"""
+    assoc_op: bool  # Decides if we are adding/removing assets or connections to assets
+    assoc_traversal: list[AssocTraversal | tuple[AssocTraversal, AssocTraversal]]
+
+@dataclass
+class LanguageGraphModelEffect:
+    """An effect on the model instance triggered by compromising the attack step."""
+
+    base: list[AssocTraversal | tuple[AssocTraversal, AssocTraversal]]
+    targets: list[DynTarget]  # Assumes ^ between targets
+
+
+def build_model_effect(
+    assets: dict[str, LanguageGraphAsset],
+    target_asset: LanguageGraphAsset,
+    step_expression: dict,
+    lang_spec
+    ) -> LanguageGraphModelEffect:
+    """Build a model effect from a step expression"""
+
+    assert step_expression["type"] == "dyn_sentence", "Only dynamic sentences can be used to build model effects"
+
+    def build_assoc_traversal(expr_chain: ExpressionsChain | None) -> list[AssocTraversal]:
+        """Create a list of AssocTraversal from an expression chain"""
+
+        if expr_chain is None:
+            return []
+        elif expr_chain.type == ExprType.COLLECT:
+            left = build_assoc_traversal(expr_chain.left_link)
+            right = build_assoc_traversal(expr_chain.right_link)
+            return left + right
+        elif expr_chain.type == ExprType.DIFFERENCE:
+            raise NotImplementedError()
+        elif expr_chain.type == ExprType.FIELD:
+            assert expr_chain.fieldname is not None, "Fieldname must be set for FIELD expression"
+            return [AssocTraversal(field_name=expr_chain.fieldname, asset_filter=None, quantity_filter=None)]
+        elif expr_chain.type == ExprType.INTERSECTION:
+            raise NotImplementedError()
+        elif expr_chain.type == ExprType.SUBTYPE:
+            ret = build_assoc_traversal(expr_chain.sub_link)
+            ret[-1] = ret[-1]._replace(asset_filter=expr_chain.subtype)
+            return ret
+        elif expr_chain.type == ExprType.TRANSITIVE:
+            raise NotImplementedError()
+        elif expr_chain.type == ExprType.UNION:
+            left = build_assoc_traversal(expr_chain.left_link)
+            right = build_assoc_traversal(expr_chain.right_link)
+            return (left, right)
+
+        raise ValueError(f"Unexpected expression chain type: {expr_chain.type}")
+
+    base_expr = step_expression["base"]
+    target_exprs = step_expression["targets"]
+
+    base_target_asset, base_expr_chain, base_step = process_step_expression(assets, target_asset, None, base_expr, lang_spec)
+    assert not base_step, "Base can not refer to an attack step in a dynamic sentence"
+    base = build_assoc_traversal(base_expr_chain)
+
+    targets = []
+    for target_expr in target_exprs:
+        process_function = process_assoc_op_step_expression if target_expr["type"] == "assoc_op" else process_step_expression
+        try:
+            _, target_expr_chain, target_step = process_function(assets, target_asset, None, target_expr, lang_spec)
+        except LookupError:
+            _, target_expr_chain, target_step = process_function(assets, base_target_asset, None, target_expr, lang_spec)
+        assert not target_step, "Targets can not refer to an attack step in a dynamic sentence"
+        if target_expr_chain:
+            if target_expr_chain.type == ExprType.ASSOC_OP:
+                assoc_op = True
+                target_expr_chain = target_expr_chain.sub_link
+            else:
+                assoc_op = False
+            assoc_traversal = build_assoc_traversal(target_expr_chain)
+        else:
+            assoc_op = False
+            assoc_traversal = [deepcopy(SELF_TRAVERSAL)]
+        targets.append(DynTarget(assoc_op=assoc_op, assoc_traversal=assoc_traversal))
+
+    return LanguageGraphModelEffect(base=base, targets=targets)