Merge pull request #409 from iriusrisk/release/1.29.0

[release/1.29.0] to main

docs/startleft-processors/iac/tfplan/Terraform-Plan-how-to-create-a-mapping-file.md

@@ -86,25 +86,29 @@ A mapping list must be defined in the `components` section to find and configure
   - label: aws_cloudwatch_metric_alarm
     type: cloudwatch
     $singleton: true
+    $category: CloudWatch
 ```
 
 !!! note ""
 
     This configuration maps all the available components of type `aws_cloudwatch_metric_alarm` to a 
-    **unique component** of type `cloudwatch`
+    **unique component** of type `cloudwatch`. 
+    The `$category` is used to name the group of components in the Threat Model.
 
 #### Mapping by a Regex
 
 ```yaml
   - label: {$regex: ^aws_api_gateway_\w*$}
     type: api-gateway
     $singleton: true
+    $category: API Gateway
 ```
 
 !!! note ""
 
     This configuration maps all the components whose type matches the regex `^aws_api_gateway\w*$`.
     It may be used along `$singleton` to create a **unique component** of type `api-gateway`
+    The `$category` is used to name the group of components in the Threat Model.
 
 ### Mapping Configuration
 

setup.py

@@ -21,7 +21,7 @@
         'jmespath==1.0.1',
         'python-hcl2==4.3.2',
         'requests==2.32.3',
-        'fastapi==0.109.2',
+        'fastapi>=0.115.2,<0.116.0',
         'python-multipart==0.0.7',
         'click==8.1.7',
         'uvicorn==0.23.2',

slp_tfplan/resources/schemas/iac_tfplan_mapping_schema.json

@@ -59,6 +59,9 @@
             },
             "$singleton":{
                "type":"boolean"
+            },
+            "$category":{
+               "type":"string"
             }
          },
          "required":[

slp_tfplan/slp_tfplan/graph/relationships_extractor.py

@@ -1,11 +1,13 @@
+import logging
 from typing import Union, List
 
 import networkx as nx
 from networkx import DiGraph
 
-from slp_tfplan.slp_tfplan.objects.tfplan_objects import TFPlanComponent
 from slp_tfplan.slp_tfplan.load.tfplan_to_resource_dict import remove_name_prefix
+from slp_tfplan.slp_tfplan.objects.tfplan_objects import TFPlanComponent
 
+logger = logging.getLogger(__name__)
 
 class RelationshipsExtractor:
 
@@ -31,7 +33,7 @@ def get_closest_resources(self, source_component: TFPlanComponent, target_candid
             if not linked_resources or path_size < min_size:
                 min_size = path_size
                 linked_resources = [target_candidate.id]
-            elif path_size == min_size:
+            elif path_size == min_size and target_candidate.id not in linked_resources:
                 linked_resources.append(target_candidate.id)
 
         return linked_resources
@@ -48,7 +50,8 @@ def __get_shortest_valid_path(self, source_label: str, target_label: str) -> Uni
 
         shortest_path = None
         if nx.has_path(self.graph, source_node, target_node):
-            for path in nx.all_simple_paths(self.graph, source=source_node, target=target_node):
+            paths = self.__get_all_simple_paths(source_node, target_node)
+            for path in paths:
                 if self.__is_straight_path(path):
                     if not shortest_path or len(shortest_path) > len(path):
                         shortest_path = path
@@ -65,3 +68,17 @@ def __is_straight_path(self, path: list) -> bool:
 
     def __nodes_to_labels(self, path: []) -> set:
         return set(filter(lambda x: x is not None, map(lambda p: remove_name_prefix(self.nodes_labels[p]), path)))
+
+    def __get_all_simple_paths(self, source_node, target_node):
+        hop_limit = 100
+        for max_hops in range(1, hop_limit):
+            paths = list(nx.all_simple_paths(self.graph, source_node, target_node, cutoff=max_hops))
+            if paths:
+                return paths
+        logger.warning(f"Reached limit of {hop_limit} hops to find a path. "
+                       f"Path not found between {source_node} and {target_node} "
+                       f"with less than {hop_limit} hops.")
+        return []
+
+
+

slp_tfplan/slp_tfplan/map/mapping.py

@@ -22,7 +22,8 @@ def type(self) -> str:
     @property
     def configuration(self) -> dict:
         return {
-            '$singleton': self.__component.get('$singleton', False)
+            '$singleton': self.__component.get('$singleton', False),
+            '$category': self.__component.get('$category', None),
         }
 
     def __str__(self) -> str:

slp_tfplan/slp_tfplan/objects/tfplan_objects.py

@@ -1,5 +1,5 @@
 from enum import Enum
-from typing import List, Dict, Union
+from typing import List, Dict, Union, Optional
 
 from otm.otm.entity.component import Component
 from otm.otm.entity.dataflow import Dataflow
@@ -39,6 +39,10 @@ def __init__(self,
     def is_singleton(self) -> bool:
         return self.configuration.get('$singleton', False)
 
+    @property
+    def category(self) -> Optional[str]:
+        return self.configuration.get('$category', None)
+
     def __eq__(self, other):
         """Overrides the default implementation"""
         if isinstance(other, TFPlanComponent):

slp_tfplan/slp_tfplan/transformers/children_calculator.py

@@ -18,4 +18,4 @@ def _calculate_component_parents(self, component: TFPlanComponent) -> [str]:
 
         return self._find_parent_by_closest_relationship(
             component,
-            self._get_parent_candidates(PARENTS_TYPES_BY_CHILDREN_TYPE[component.tf_type]))
+            self._find_components_by_type(PARENTS_TYPES_BY_CHILDREN_TYPE[component.tf_type]))

slp_tfplan/slp_tfplan/transformers/hierarchy_calculator.py

@@ -36,7 +36,7 @@ def set_component_parent(component: TFPlanComponent, parent_id: str):
     component.parent = parent_id
 
 
-def _find_parent_candidates_by_type(components: [], parent_type: str) -> []:
+def _find_components_by_type(components: [], parent_type: str) -> []:
     return list(filter(lambda c: __extract_type(c.id) == parent_type, components))
 
 
@@ -79,10 +79,10 @@ def _calculate_component_parents(self, component: TFPlanComponent) -> [str]:
     def _find_parent_by_closest_relationship(self, component: TFPlanComponent, parent_candidates: []):
         return self.relationships_extractor.get_closest_resources(component, parent_candidates)
 
-    def _get_parent_candidates(self, parent_types: []):
-        parent_candidates = []
+    def _find_components_by_type(self, _types: []):
+        components = []
 
-        for parent_type in parent_types:
-            parent_candidates.extend(_find_parent_candidates_by_type(self.otm.components, parent_type))
+        for _type in _types:
+            components.extend(_find_components_by_type(self.otm.components, _type))
 
-        return parent_candidates
+        return components

slp_tfplan/slp_tfplan/transformers/parent_calculator.py

@@ -3,16 +3,18 @@
 from slp_tfplan.slp_tfplan.objects.tfplan_objects import TFPlanComponent, TFPlanOTM
 from slp_tfplan.slp_tfplan.transformers.hierarchy_calculator import HierarchyCalculator
 
-PARENT_TYPES = ['aws_subnet', 'aws_vpc', 'azurerm_subnet', 'azurerm_virtual_network']
-
+PARENT_TYPES = {
+   'subnets': ['aws_subnet', 'azurerm_subnet'],
+   'vpcs': ['aws_vpc', 'azurerm_virtual_network']
+}
 
 class ParentCalculator(HierarchyCalculator):
 
     def __init__(self, otm: TFPlanOTM, graph: DiGraph):
         super().__init__(otm, graph)
-        self.parent_candidates = self._get_parent_candidates(PARENT_TYPES)
+        self.subnet_candidates = self._find_components_by_type(PARENT_TYPES['subnets'])
+        self.vpc_candidates = self._find_components_by_type(PARENT_TYPES['vpcs'])
 
     def _calculate_component_parents(self, component: TFPlanComponent) -> [str]:
-        return self._find_parent_by_closest_relationship(component, self.parent_candidates)
-
-
+       return self._find_parent_by_closest_relationship(component, self.subnet_candidates) or \
+                self._find_parent_by_closest_relationship(component, self.vpc_candidates)

slp_tfplan/slp_tfplan/transformers/singleton_transformer.py

@@ -26,6 +26,14 @@ def _find_equivalent_dataflows(dataflow: Dataflow, dataflows: List[Dataflow]) ->
     return equivalent_dataflows
 
 
+def _are_sibling(component, sibling):
+    if component.category and sibling.category:
+        return component.category == sibling.category
+    if not component.category and not sibling.category:
+        return component.type == sibling.type
+    return False
+
+
 def _are_equivalent_dataflows(dataflow_1: Dataflow, dataflow_2: Dataflow) -> bool:
     is_same_dataflow = (dataflow_1.source_node == dataflow_2.source_node
                     and dataflow_1.destination_node == dataflow_2.destination_node)
@@ -58,14 +66,16 @@ def _merge_dataflows(origin_dataflow: Dataflow, dataflows: List[Dataflow]) -> Da
 
     return origin_dataflow
 
+def __build_singleton_name(component: TFPlanComponent):
+    return component.category or f"{component.type} (grouped)"
 
 def _build_singleton_component(otm_components: List[TFPlanComponent]) -> TFPlanComponent:
     tags = list(set(itertools.chain.from_iterable([c.tags or [] for c in otm_components])))
     configuration = _merge_component_configurations(otm_components)
     component_id = otm_components[0].id
     return TFPlanComponent(
         component_id=component_id,
-        name=f"{otm_components[0].type} (grouped)",
+        name=__build_singleton_name(otm_components[0]),
         component_type=otm_components[0].type,
         parent=otm_components[0].parent,
         parent_type=otm_components[0].parent_type,
@@ -91,7 +101,7 @@ def transform(self):
     def __populate_singleton_component_relations(self):
         for component in self.otm_components:
             if component.is_singleton and self.__is_not_parent(component):
-                sibling_components = self.__find_siblings_components(component.type, component.parent)
+                sibling_components = self.__find_siblings_components(component)
                 if len(sibling_components) > 1:
                     self.singleton_component_relations[component.id] = \
                         self.singleton_component_relations.get(sibling_components[0].id) \
@@ -100,20 +110,19 @@ def __populate_singleton_component_relations(self):
     def __is_not_parent(self, component: TFPlanComponent):
         return not any(c.parent == component.id for c in self.otm_components)
 
-    def __find_siblings_components(self, component_type: str, parent_id: str):
+    def __find_siblings_components(self, component: TFPlanComponent):
         """
         Returns all the component marked as singleton with the given type and parent identifier
-        :param component_type: Type of the component
-        :param parent_id:  Identifier of the parent component
+        :param component: The component type to search
         :return: A list with all the related components
         """
         found_components = []
-        for component in self.otm_components:
-            if (component.is_singleton
-                    and self.__is_not_parent(component)
-                    and component.type == component_type
-                    and component.parent == parent_id):
-                found_components.append(component)
+        for sibling in self.otm_components:
+            if (sibling.is_singleton
+                    and self.__is_not_parent(sibling)
+                    and _are_sibling(component, sibling)
+                    and sibling.parent == component.parent):
+                found_components.append(sibling)
 
         return found_components