[feature/OPT-908] to dev (#325)

* OPT-908 Refactor resource data and extractors

* OPT-908 Create inbound connections by Module: security-group

---------

Co-authored-by: David Antolín <99404665+dantolin-iriusrisk@users.noreply.github.com>
Co-authored-by: PacoCid <117292868+PacoCid@users.noreply.github.com>

slp_tfplan/slp_tfplan/load/resource_data_extractors.py

@@ -3,9 +3,6 @@
 
 # TODO Consider migrating these functions to use jmespath
 
-def _get_resource_properties_root(resource: {}, path: Literal['expressions', 'planned_values']) -> Dict:
-    return resource['resource_properties'].get(path, {})
-
 
 def _get_referenced_resources(references: List[str]):
     valid_references = []
@@ -17,12 +14,7 @@ def _get_referenced_resources(references: List[str]):
     return valid_references
 
 
-def _get_value_from(resource: Dict, root: Literal['expressions', 'planned_values'], path: List[str]):
-    expressions = _get_resource_properties_root(resource, root)
-    if not expressions:
-        return []
-
-    source = expressions
+def _get_value_from_path(source: Dict, path: List[str]):
     for i, element in enumerate(path):
         source = source.get(element)
         if not source:
@@ -37,16 +29,18 @@ def _get_value_from(resource: Dict, root: Literal['expressions', 'planned_values
     return source
 
 
-def _get_value_from_planned_values(resource: Dict, path: List[str]):
-    return _get_value_from(resource, 'planned_values', path)
+def _get_from_values(resource: Dict, path: List[str]):
+    expressions_path = ['resource_values'] + path
+    return _get_value_from_path(resource, expressions_path)
 
 
-def _get_value_from_expressions(resource: Dict, path: List[str]):
-    return _get_value_from(resource, 'expressions', path)
+def _get_from_expressions(resource: Dict, path: List[str]):
+    expressions_path = ['resource_configuration', 'expressions'] + path
+    return _get_value_from_path(resource, expressions_path)
 
 
 def _get_references_from_expressions(resource: Dict, path: List[str]) -> List[str]:
-    source = _get_value_from_expressions(resource, path)
+    source = _get_from_expressions(resource, path)
     if not source or isinstance(source, str):
         return []
 
@@ -68,10 +62,6 @@ def security_groups_ids_from_type_property(resource: {}, cidr_type: Literal['ing
     return _get_references_from_expressions(resource, [cidr_type, 'references'])
 
 
-def cidr_from_type_property(resource: {}, cidr_type: Literal['ingress', 'egress']) -> List[dict]:
-    return _get_value_from_planned_values(resource, [cidr_type])
-
-
 def source_security_group_id_from_rule(resource: {}) -> str:
     sources_list = _get_references_from_expressions(resource, ['source_security_group_id', 'references'])
     if sources_list:
@@ -84,25 +74,29 @@ def security_group_id_from_rule(resource: {}) -> str:
         return sg_ids[0]
 
 
+def cidr_from_type_property(resource: {}, cidr_type: Literal['ingress', 'egress']) -> List[dict]:
+    return _get_from_values(resource, [cidr_type])
+
+
 def description_from_rule(resource: {}) -> str:
-    return _get_value_from_planned_values(resource, ['description'])
+    return _get_from_values(resource, ['description'])
 
 
 def protocol_from_rule(resource: {}) -> str:
-    return _get_value_from_planned_values(resource, ['protocol'])
+    return _get_from_values(resource, ['protocol'])
 
 
 def from_port_from_rule(resource: {}) -> str:
-    return _get_value_from_planned_values(resource, ['from_port'])
+    return _get_from_values(resource, ['from_port'])
 
 
 def to_port_from_rule(resource: {}) -> str:
-    return _get_value_from_planned_values(resource, ['to_port'])
+    return _get_from_values(resource, ['to_port'])
 
 
 def cidr_blocks_from_rule(resource: {}) -> List[str]:
-    return _get_value_from_planned_values(resource, ['cidr_blocks'])
+    return _get_from_values(resource, ['cidr_blocks'])
 
 
 def security_group_rule_type(resource: {}) -> str:
-    return _get_value_from_expressions(resource, ['type', 'constant_value'])
+    return _get_from_values(resource, ['type'])

slp_tfplan/slp_tfplan/load/security_groups_loader.py

@@ -1,9 +1,13 @@
 from typing import List, Dict, Union
 
+from networkx import DiGraph
+
+from slp_tfplan.slp_tfplan.graph.relationships_extractor import RelationshipsExtractor
 from slp_tfplan.slp_tfplan.load.resource_data_extractors import security_group_id_from_rule, \
     description_from_rule, protocol_from_rule, from_port_from_rule, to_port_from_rule, cidr_blocks_from_rule, \
     cidr_from_type_property, source_security_group_id_from_rule, \
     security_group_rule_type, security_groups_ids_from_type_property
+from slp_tfplan.slp_tfplan.matcher.sg_and_sgrules_matcher import SGAndSGRulesMatcher
 from slp_tfplan.slp_tfplan.objects.tfplan_objects import SecurityGroup, TFPlanOTM, SecurityGroupCIDR, \
     SecurityGroupCIDRType
 
@@ -17,7 +21,8 @@ def _get_security_group_rules(resources: List[Dict]) -> List[Dict[str, str]]:
         return []
 
     return list(map(lambda sg_rule:
-                    {'security_group_id': security_group_id_from_rule(sg_rule),
+                    {'resource_id': sg_rule.get('resource_id'),
+                     'security_group_id': security_group_id_from_rule(sg_rule),
                      'description': description_from_rule(sg_rule),
                      'protocol': protocol_from_rule(sg_rule),
                      'from_port': from_port_from_rule(sg_rule),
@@ -28,53 +33,48 @@ def _get_security_group_rules(resources: List[Dict]) -> List[Dict[str, str]]:
                     sg_rules))
 
 
-def _get_sg_rules_by_sg_id(sg_id: str, rule_type: SecurityGroupCIDRType, sg_rules: List[Dict[str, str]]
-                           ) -> List[Dict[str, str]]:
-    return list(filter(
-        lambda sg_rule:
-        sg_rule['security_group_id'] == sg_id and sg_rule['type'] == rule_type.value,
-        sg_rules))
-
-
-def _source_security_group_ids_of_type_from_rule(security_group: Dict, rule_type: SecurityGroupCIDRType,
-                                                 sg_rules: List[Dict[str, str]]):
-    sg_rules_by_sg_id = _get_sg_rules_by_sg_id(security_group['resource_id'], rule_type, sg_rules)
-    return list(filter(lambda sg_id: sg_id is not None,
-                       list(set([r['source_security_group_id'] for r in sg_rules_by_sg_id]))))
-
+def _is_valid_cidr_object(sg_rule: Union[str, Dict[str, str]]) -> bool:
+    if isinstance(sg_rule, str):
+        return False
+    return 'cidr_blocks' in sg_rule
 
-def _cidr_of_type_from_rule(security_group: Dict, rule_type: SecurityGroupCIDRType,
-                            sg_rules: List[Dict[str, str]]):
-    sg_rules_by_sg_id = _get_sg_rules_by_sg_id(security_group['resource_id'], rule_type, sg_rules)
 
-    return list(filter(lambda sg_rule: sg_rule.get('cidr_blocks', None) is not None, sg_rules_by_sg_id))
+def _filter_sg_rule_by_type(sg_rules: List[Dict], rule_type: SecurityGroupCIDRType):
+    return list(filter(lambda sg_rule: sg_rule['type'] == rule_type.value, sg_rules))
 
 
-def _get_sgs_of_type(security_group: Dict, sg_type: SecurityGroupCIDRType,
-                     sg_rules: List[Dict[str, str]]) -> List[str]:
+def _get_sgs_of_type(security_group: Dict, related_sg_rules: List[Dict],
+                     sg_type: SecurityGroupCIDRType) -> List[str]:
     return security_groups_ids_from_type_property(security_group, sg_type.value) or \
-           _source_security_group_ids_of_type_from_rule(security_group, sg_type, sg_rules)
+           _get_sgs_of_type_from_rule(related_sg_rules, sg_type)
 
 
-def __is_valid_cidr_object(sg_rule: Union[str, Dict[str, str]]) -> bool:
-    if isinstance(sg_rule, str):
-        return False
-    return 'cidr_blocks' in sg_rule
+def _get_sgs_of_type_from_rule(related_sg_rules: List[Dict], rule_type: SecurityGroupCIDRType):
+    related_sg_rules_filtered = _filter_sg_rule_by_type(related_sg_rules, rule_type)
+
+    return list(filter(lambda sg_id: sg_id is not None,
+                       list(set([r['source_security_group_id'] for r in related_sg_rules_filtered]))))
 
 
-def _get_cidr_of_type(security_group: Dict, cidr_type: SecurityGroupCIDRType,
-                      sg_rules: List[Dict[str, str]]) -> List[SecurityGroupCIDR]:
+def _get_cidr_of_type(security_group: Dict, related_sg_rules: List[Dict],
+                      cidr_type: SecurityGroupCIDRType) -> List[SecurityGroupCIDR]:
     sg_cidr = cidr_from_type_property(security_group, cidr_type.value) or \
-              _cidr_of_type_from_rule(security_group, cidr_type, sg_rules)
+              _get_cidr_of_type_from_rule(related_sg_rules, cidr_type)
 
-    sg_cidr = list(filter(lambda cidr: __is_valid_cidr_object(cidr), sg_cidr))
+    sg_cidr = list(filter(lambda cidr: _is_valid_cidr_object(cidr), sg_cidr))
 
     if not sg_cidr:
         return []
 
     return list(map(lambda cidr: SecurityGroupCIDRLoader(cidr, cidr_type).load(), sg_cidr))
 
 
+def _get_cidr_of_type_from_rule(related_sg_rules: List[Dict], rule_type: SecurityGroupCIDRType):
+    related_sg_rules_filtered = _filter_sg_rule_by_type(related_sg_rules, rule_type)
+
+    return list(filter(lambda sg_rule: sg_rule.get('cidr_blocks', None) is not None, related_sg_rules_filtered))
+
+
 class SecurityGroupCIDRLoader:
 
     def __init__(self, security_group_cidr: dict, cidr_type: SecurityGroupCIDRType):
@@ -100,24 +100,28 @@ def load(self):
 
 class SecurityGroupsLoader:
 
-    def __init__(self, otm: TFPlanOTM, tfplan: {}):
+    def __init__(self, otm: TFPlanOTM, tfplan: {}, graph: DiGraph):
         self.otm = otm
 
         self._resources = tfplan['resource']
-        self._sg_rules: List[Dict[str, str]] = []
+        self._sg_rules: List[Dict[str, str]] = _get_security_group_rules(self._resources)
+
+        self._relationships_extractor = RelationshipsExtractor(
+            mapped_resources_ids=self.otm.mapped_resources_ids,
+            graph=graph)
 
     def load(self):
-        self._sg_rules = _get_security_group_rules(self._resources)
         for resource in self._resources:
             if resource['resource_type'] in SECURITY_GROUPS_TYPES:
                 self.otm.security_groups.append(self.__build_security_group(resource))
 
     def __build_security_group(self, resource: {}) -> SecurityGroup:
+        related_sg_rules = SGAndSGRulesMatcher(resource, self._sg_rules, self._relationships_extractor).match()
         return SecurityGroup(
             security_group_id=resource['resource_id'],
             name=resource['resource_name'],
-            ingress_sgs=_get_sgs_of_type(resource, SecurityGroupCIDRType.INGRESS, self._sg_rules),
-            egress_sgs=_get_sgs_of_type(resource, SecurityGroupCIDRType.EGRESS, self._sg_rules),
-            ingress_cidr=_get_cidr_of_type(resource, SecurityGroupCIDRType.INGRESS, self._sg_rules),
-            egress_cidr=_get_cidr_of_type(resource, SecurityGroupCIDRType.EGRESS, self._sg_rules),
+            ingress_sgs=_get_sgs_of_type(resource, related_sg_rules, SecurityGroupCIDRType.INGRESS),
+            egress_sgs=_get_sgs_of_type(resource, related_sg_rules, SecurityGroupCIDRType.EGRESS),
+            ingress_cidr=_get_cidr_of_type(resource, related_sg_rules, SecurityGroupCIDRType.INGRESS),
+            egress_cidr=_get_cidr_of_type(resource, related_sg_rules, SecurityGroupCIDRType.EGRESS),
         )

slp_tfplan/slp_tfplan/load/tfplan_to_resource_dict.py

@@ -3,18 +3,6 @@
 import sl_util.sl_util.secure_regex as re
 
 
-def map_resource_properties(resource: Dict) -> {}:
-    return {
-        'resource_mode': resource['mode'],
-        'resource_provider_name': resource['provider_name'],
-        'resource_schema_version': resource['schema_version'],
-        'resource_address': resource['address'],
-        # Sensitive and usual values may be overlapped
-        **resource.get('sensitive_values', {}),
-        **resource.get('values', {})
-    }
-
-
 def is_not_cloned_resource(resource: Dict) -> bool:
     return 'index' not in resource or resource['index'] == '0' or resource['index'] == 'zero'
 
@@ -78,16 +66,14 @@ def __map_resource(self, resource: Dict, parent: str = None) -> Dict:
             'resource_id': get_resource_id(resource),
             'resource_name': get_resource_name(resource, parent),
             'resource_type': resource['type'],
-            'resource_properties': self.__get_resource_properties(resource)
+            'resource_values': resource.get('values', {}),
+            'resource_configuration': {
+                'expressions': self.__get_resource_configuration_expressions(resource)
+            }
         }
 
-    def __get_resource_properties(self, resource: Dict) -> Dict:
-        _resource_configuration = self.__get_resource_configuration(resource['address'])
-        _resource_properties = map_resource_properties(resource)
-        if _resource_configuration:
-            _resource_configuration["planned_values"] = _resource_properties
-
-        return _resource_configuration or _resource_properties
+    def __get_resource_configuration_expressions(self, resource: Dict) -> Dict:
+        return self.__get_resource_configuration(resource).get('expressions', {})
 
-    def __get_resource_configuration(self, resource_address: str) -> Dict:
-        return next(filter(lambda r: r['address'] == resource_address, self.resources_configuration), None)
+    def __get_resource_configuration(self, resource: Dict) -> Dict:
+        return next(filter(lambda r: r['address'] == resource['address'], self.resources_configuration), {})

slp_tfplan/slp_tfplan/matcher/__init__.py

@@ -1,11 +1,13 @@
 from slp_tfplan.slp_tfplan.matcher.components_and_sgs_matcher import ComponentsAndSGsMatcher
 from slp_tfplan.slp_tfplan.matcher.resource_matcher import ResourcesMatcherContainer
+from slp_tfplan.slp_tfplan.matcher.sg_and_sgrules_matcher import SGAndSGRulesMatcher
 from slp_tfplan.slp_tfplan.matcher.sgs_matcher import SGsMatcher
 from slp_tfplan.slp_tfplan.matcher.strategies.match_strategy import MatchStrategyContainer
 
 MatchStrategyContainer().wire(packages=[__name__])
 
 ResourcesMatcherContainer().wire(modules=[
     ComponentsAndSGsMatcher.__module__,
-    SGsMatcher.__module__
+    SGsMatcher.__module__,
+    SGAndSGRulesMatcher.__module__
 ])

slp_tfplan/slp_tfplan/matcher/resource_matcher.py

@@ -54,10 +54,15 @@ class ResourcesMatcherContainer(DeclarativeContainer):
 
     sgs_matcher = providers.Singleton(
         ResourceMatcher,
-        strategies=MatchStrategyContainer.sg_match_strategies
+        strategies=MatchStrategyContainer.sg_sg_match_strategies
     )
 
     component_sg_matcher = providers.Singleton(
         ResourceMatcher,
         strategies=MatchStrategyContainer.component_sg_match_strategies
     )
+
+    sg_rule_matcher = providers.Singleton(
+        ResourceMatcher,
+        strategies=MatchStrategyContainer.sg_sg_rule_match_strategies
+    )

slp_tfplan/slp_tfplan/matcher/sg_and_sgrules_matcher.py

@@ -0,0 +1,36 @@
+from typing import Dict, List
+
+from dependency_injector.wiring import inject, Provide
+
+from slp_tfplan.slp_tfplan.graph.relationships_extractor import RelationshipsExtractor
+from slp_tfplan.slp_tfplan.matcher.resource_matcher import ResourceMatcher, ResourcesMatcherContainer
+
+
+class SGAndSGRulesMatcher:
+    """
+    This class is responsible for matching security groups and security groups rules.
+    """
+    @inject
+    def __init__(self,
+                 security_group: Dict, security_group_rules: List[Dict],
+                 relationships_extractor: RelationshipsExtractor,
+                 sg_rule_matcher: ResourceMatcher = Provide[ResourcesMatcherContainer.sg_rule_matcher]):
+        # Data structures
+        self._security_group: Dict = security_group
+        self._security_group_rules: List[Dict] = security_group_rules
+        self._relationships_extractor: RelationshipsExtractor = relationships_extractor
+
+        # Injected dependencies
+        self._are_related = sg_rule_matcher.are_related
+
+    def match(self) -> List[Dict]:
+        """
+        Returns a list of security group rules related with the security group.
+        :return: List of security group rules
+        """
+        related_sg_rule = []
+        for sg_rule in self._security_group_rules:
+            if self._are_related(self._security_group, sg_rule, relationships_extractor=self._relationships_extractor):
+                related_sg_rule.append(sg_rule)
+
+        return related_sg_rule

slp_tfplan/slp_tfplan/matcher/strategies/match_strategy.py

@@ -37,5 +37,6 @@ class MatchStrategyContainer(DeclarativeContainer):
     Here are defined a list of instances for each of these groups.
     """
 
-    sg_match_strategies = List()
+    sg_sg_match_strategies = List()
+    sg_sg_rule_match_strategies = List()
     component_sg_match_strategies = List()

slp_tfplan/slp_tfplan/matcher/strategies/security_group_match_strategies.py → slp_tfplan/slp_tfplan/matcher/strategies/sg_sg_match_strategies.py

@@ -3,25 +3,26 @@
 from slp_tfplan.slp_tfplan.objects.tfplan_objects import SecurityGroup
 
 
-
-@register(MatchStrategyContainer.sg_match_strategies)
+@register(MatchStrategyContainer.sg_sg_match_strategies)
 class SecurityGroupByConfigurationStrategy(MatchStrategy):
     """
     Two Security Groups SG1 and SG2 are related if the ID of SG1 is in the ingress_ids of the SG2 or if the
     ID of the SG2 is in the egress_ids of the SG1.
     """
+
     def are_related(self, source_security_group: SecurityGroup, target_security_group: SecurityGroup,
                     **kwargs) -> bool:
         return source_security_group.id in target_security_group.ingress_sgs \
-            or target_security_group.id in source_security_group.egress_sgs
+               or target_security_group.id in source_security_group.egress_sgs
 
 
-@register(MatchStrategyContainer.sg_match_strategies)
+@register(MatchStrategyContainer.sg_sg_match_strategies)
 class SecurityGroupByGraphStrategy(MatchStrategy):
     """
     Two Security Groups are related if there is a straight relationship between them in the tfgraph. This means that
     there is a relationships between a SG1 and a SG2 with no mapped components in the middle.
     """
+
     def are_related(self, source_security_group: SecurityGroup, target_security_group: SecurityGroup,
                     **kwargs) -> bool:
         return kwargs['relationships_extractor'].exist_valid_path(target_security_group.id, source_security_group.id)