Merge pull request #372 from iriusrisk/release/1.24.0

Release/1.24.0 to main

Author: dfernandezvigo

setup.py

@@ -35,6 +35,7 @@
         'dependency-injector==4.41.0',
         'google-re2==1.0',
         'xmlschema==2.5.0',
+        'word2number==1.1',
         # Do not upgrade pygraphviz unless security issues because it is heavily dependent on the underlying OS
         'pygraphviz==1.10'
     ],

sl_util/sl_util/str_utils.py

@@ -1,5 +1,6 @@
 import random
 import uuid
+from word2number import w2n
 
 
 def deterministic_uuid(source):
@@ -10,3 +11,13 @@ def deterministic_uuid(source):
 
 def get_bytes(s: str, encoding='utf-8') -> bytes:
     return bytes(s, encoding)
+
+
+def to_number(input, default_value: int = 0) -> int:
+    try:
+        return int(input)
+    except ValueError:
+        try:
+            return w2n.word_to_num(input)
+        except ValueError:
+            return default_value

sl_util/tests/unit/test_str_utils.py

@@ -1,6 +1,7 @@
-from pytest import mark
-
-from sl_util.sl_util.str_utils import deterministic_uuid
+from pytest import mark, param
+import random
+from unittest.mock import patch
+from sl_util.sl_util.str_utils import deterministic_uuid, to_number
 
 
 class TestStrUtils:
@@ -28,3 +29,50 @@ def test_deterministic_uuid_without_source(self, source):
         uuid2 = deterministic_uuid(source)
         # Then we obtain two different values
         assert uuid1 != uuid2
+
+    @mark.parametrize('source', [
+        param(random.randint(0, 100)),
+        param(str(random.randint(0, 100)))
+    ])
+    def test_to_number(self, source: any):
+        # GIVEN a random integer
+
+        # WHEN it is transformed to a number
+        result = to_number(source)
+
+        # THEN we obtain the original number
+        assert result == int(source)
+
+    @patch('sl_util.sl_util.str_utils.w2n.word_to_num', return_value=2)
+    def test_text_to_number(self, mocked_word_to_otm):
+        # GIVEN a text number
+        source = 'two'
+
+        # WHEN it is transformed to a number
+        result = to_number(source)
+
+        # THEN we obtain the number
+        assert result == 2
+
+    def test_unknown_to_number(self):
+        # GIVEN an unkown
+        source = 'unknown'
+
+        # AND a default value
+        default_value = 5
+
+        # WHEN it is transformed to a number
+        result = to_number(source, default_value)
+
+        # THEN we obtain the default value
+        assert result == 5
+
+    @mark.parametrize('source', ['sandbox', ''])
+    def test_number_conversions_to_alphanumeric(self, source):
+        # Given the source
+        # when passed an alphanumeric to function
+        number1 = to_number(source)
+        # when passed an empty string to function
+        number2 = to_number(source)
+        # Then we obtain default value 0
+        assert number1 == number2 == 0

slp_tfplan/slp_tfplan/load/tfplan_to_resource_dict.py

@@ -1,26 +1,23 @@
 from typing import Dict, List
 
 import sl_util.sl_util.secure_regex as re
+from sl_util.sl_util.str_utils import to_number
 
 
 def is_not_cloned_resource(resource: Dict) -> bool:
-    return 'index' not in resource or resource['index'] == '0' or resource['index'] == 0 or resource['index'] == 'zero'
+    return to_number(resource['index']) == 0 if 'index' in resource else True
 
 
 def get_resource_id(resource: Dict) -> str:
-    return parse_address(resource['address']) \
-        if 'index' in resource \
-        else resource['address']
+    return parse_address(resource['address']) if 'address' in resource else None
 
 
 def get_resource_name(resource: Dict, parent: str) -> str:
     return resource['name'] if not parent else f'{parent}.{resource["name"]}'
 
 
 def get_module_address(module: Dict, parent: str) -> str:
-    if 'address' in module:
-        module_address = parse_address(module['address'])
-        return f'{parent}.{module_address}' if parent else module_address
+    return parse_address(module['address']) if 'address' in module else parent
 
 
 def parse_address(address: str) -> str:

slp_tfplan/slp_tfplan/parse/tfplan_parser.py

@@ -2,6 +2,7 @@
 
 from networkx import DiGraph
 
+from sl_util.sl_util.iterations_utils import remove_duplicates
 from slp_base import ProviderParser, OTMBuildingError
 from slp_tfplan.slp_tfplan.load.launch_templates_loader import LaunchTemplatesLoader
 from slp_tfplan.slp_tfplan.load.security_groups_loader import SecurityGroupsLoader
@@ -46,6 +47,7 @@ def build_otm(self):
             self.__calculate_dataflows()
             self.__calculate_attack_surface()
             self.__calculate_singletons()
+            self.__remove_duplicates()
 
         except Exception as e:
             logger.error(f'{e}')
@@ -77,3 +79,6 @@ def __calculate_attack_surface(self):
 
     def __calculate_singletons(self):
         SingletonTransformer(self.otm).transform()
+
+    def __remove_duplicates(self):
+        self.otm.components = remove_duplicates(self.otm.components)

slp_tfplan/tests/unit/load/test_tfplan_loader.py

@@ -12,7 +12,8 @@
 from slp_tfplan.slp_tfplan.load.tfplan_loader import TFPlanLoader
 from slp_tfplan.tests.resources import test_resource_paths
 from slp_tfplan.tests.util.asserts import assert_resource_values
-from slp_tfplan.tests.util.builders import build_tfplan, generate_resources, generate_child_modules
+from slp_tfplan.tests.util.builders import build_tfplan, generate_resources, generate_child_modules, \
+    generate_child_modules_instances
 
 INVALID_YAML = test_resource_paths.invalid_yaml
 TF_FILE_YAML_EXCEPTION = JSONDecodeError('HLC2 cannot be processed as JSON', doc='sample-doc', pos=0)
@@ -73,7 +74,7 @@ def test_load_no_modules(self, yaml_mock):
 
         for i, resource in enumerate(resources):
             i += 1
-            assert resource['resource_id'] == f'r{i}-addr'
+            assert resource['resource_id'] == f'r{i}-type.r{i}-name'
             assert resource['resource_type'] == f'r{i}-type'
             assert resource['resource_name'] == f'r{i}-name'
 
@@ -102,7 +103,7 @@ def test_load_only_modules(self, yaml_mock):
             for child_index in range(1, 3):
                 resource = resources[resource_index]
 
-                assert resource['resource_id'] == f'r{child_index}-addr'
+                assert resource['resource_id'] == f'r{child_index}-type.r{child_index}-name'
                 assert resource['resource_type'] == f'r{child_index}-type'
                 assert resource['resource_name'] == f'{module_address}.r{child_index}-name'
 
@@ -130,9 +131,9 @@ def test_load_nested_modules(self, yaml_mock):
         # AND resource_id, resource_type, resource_name and resource_properties are right
         assert len(resources) == 1
 
-        assert resource['resource_id'] == 'r1-addr'
+        assert resource['resource_id'] == 'r1-type.r1-name'
         assert resource['resource_type'] == 'r1-type'
-        assert resource['resource_name'] == 'cm1-addr.cm1-addr.r1-name'
+        assert resource['resource_name'] == 'cm1-addr.r1-name'
 
         assert_resource_values(resource['resource_values'])
 
@@ -155,15 +156,15 @@ def test_load_complex_structure(self, yaml_mock):
         # AND resource_type, resource_name and resource_properties from top level are right
         resource = resources[0]
 
-        assert resource['resource_id'] == 'r1-addr'
+        assert resource['resource_id'] == 'r1-type.r1-name'
         assert resource['resource_type'] == 'r1-type'
         assert resource['resource_name'] == 'r1-name'
 
         assert_resource_values(resource['resource_values'])
 
         # AND resource_type, resource_name and resource_properties from child modules are right
         resource = resources[1]
-        assert resource['resource_id'] == 'r1-addr'
+        assert resource['resource_id'] == 'r1-type.r1-name'
         assert resource['resource_type'] == 'r1-type'
         assert resource['resource_name'] == 'cm1-addr.r1-name'
 
@@ -193,7 +194,7 @@ def test_load_resources_same_name(self, yaml_mock):
         assert len(resources) == 1
 
         # AND The duplicated resource is unified and the index is no present in name or id
-        assert resources[0]['resource_id'] == 'r1-addr'
+        assert resources[0]['resource_id'] == 'r1-type.r1-name'
         assert resources[0]['resource_name'] == 'cm1-addr.r1-name'
 
     @patch('json.loads')
@@ -230,9 +231,85 @@ def test_load_modules_same_name(self, yaml_mock):
         assert len(resources) == 1
 
         # AND The duplicated resource is unified and the index is not present in name or id
-        assert resources[0]['resource_id'] == 'cm1-addr.r1-addr'
+        assert resources[0]['resource_id'] == 'cm1-addr.r1-type.r1-name'
+        assert resources[0]['resource_name'] == 'cm1-addr.r1-name'
+
+    @patch('json.loads')
+    def test_load_modules_instances(self, yaml_mock):
+        # GIVEN a valid plain Terraform Plan file with only modules
+        yaml_mock.side_effect = [build_tfplan(
+            child_modules=generate_child_modules_instances(module_count=2, resource_count=2))]
+
+        # WHEN TFPlanLoader::load is invoked
+        tfplan_loader = TFPlanLoader(sources=[b'MOCKED', b'MOCKED'])
+        tfplan_loader.load()
+
+        # THEN TF contents are loaded in TfplanLoader.terraform
+        assert tfplan_loader.terraform
+        resources = tfplan_loader.terraform['resource']
+        assert len(resources) == 2
+
+        # AND resource_id, resource_type, resource_name and resource_properties are right
+        resource_index = 0
+        for _ in range(1, 2):
+
+            module_address = 'cm-addr'
+
+            for child_index in range(1, 3):
+                resource = resources[resource_index]
+
+                assert (resource['resource_id'] == f'{module_address}.r{child_index}-type.r{child_index}-name')
+                assert resource['resource_type'] == f'r{child_index}-type'
+                assert resource['resource_name'] == f'{module_address}.r{child_index}-name'
+
+                assert_resource_values(resource['resource_values'])
+
+                resource_index += 1
+
+    @patch('json.loads')
+    def test_load_modules_mixed(self, yaml_mock):
+        # GIVEN a valid plain Terraform Plan file with two instances of a module and two resources each
+        mixed_modules = generate_child_modules(module_count=1, resource_count=1)
+        module_instances = generate_child_modules_instances(module_count=2, resource_count=2)
+        mixed_modules.append(module_instances[0])
+        mixed_modules.append(module_instances[1])
+
+        tfplan = build_tfplan(child_modules=mixed_modules)
+
+        # GIVEN a valid plain Terraform Plan file with only modules
+        yaml_mock.side_effect = [tfplan]
+
+        # WHEN TFPlanLoader::load is invoked
+        tfplan_loader = TFPlanLoader(sources=[b'MOCKED', b'MOCKED'])
+        tfplan_loader.load()
+
+        # THEN TF contents are loaded in TfplanLoader.terraform
+        assert tfplan_loader.terraform
+        resources = tfplan_loader.terraform['resource']
+        assert len(resources) == 3
+
+        # AND resource_id, resource_type, resource_name and resource_properties are right
+
+        assert resources[0]['resource_id'] == 'r1-type.r1-name'
         assert resources[0]['resource_name'] == 'cm1-addr.r1-name'
 
+        resource_index = 1
+
+        for _ in range(1, 2):
+
+            module_address = 'cm-addr'
+
+            for child_index in range(1, 2):
+                resource = resources[resource_index]
+
+                assert (resource['resource_id'] == f'{module_address}.r{child_index}-type.r{child_index}-name')
+                assert resource['resource_type'] == f'r{child_index}-type'
+                assert resource['resource_name'] == f'{module_address}.r{child_index}-name'
+
+                assert_resource_values(resource['resource_values'])
+
+                resource_index += 1
+
     @patch('json.loads')
     def test_load_no_resources(self, yaml_mock):
         # GIVEN a valid Terraform Plan file with no resources

slp_tfplan/tests/util/builders.py

@@ -107,7 +107,8 @@ def build_security_group_mock(id: str,
 
 def build_security_group_cidr_mock(cidr_blocks: List[str], description: str = None, from_port: int = None,
                                    to_port: int = None, protocol: str = None):
-    return Mock(cidr_blocks=cidr_blocks, description=description, type=SecurityGroupCIDRType.INGRESS, from_port=from_port, to_port=to_port,
+    return Mock(cidr_blocks=cidr_blocks, description=description, type=SecurityGroupCIDRType.INGRESS,
+                from_port=from_port, to_port=to_port,
                 protocol=protocol)
 
 
@@ -142,11 +143,14 @@ def build_tfplan(resources: List[Dict] = None, child_modules: List[Dict] = None)
 def generate_resources(resource_count: int, module_child: bool = False) -> List[Dict]:
     resources = []
     for i in range(1, resource_count + 1):
+        resource_name = f'r{i}-name'
+        resource_type = f'r{i}-type'
+
         resource = {
-            'address': f'r{i}-addr',
+            'address': f'{resource_type}.{resource_name}',
             'mode': 'managed',
-            'type': f'r{i}-type',
-            'name': f'r{i}-name',
+            'type': resource_type,
+            'name': resource_name,
             'provider_name': 'registry.terraform.io/hashicorp/aws',
             'schema_version': 0,
             'values': {
@@ -167,6 +171,35 @@ def generate_resources(resource_count: int, module_child: bool = False) -> List[
     return resources
 
 
+def generate_resources_for_module(resource_count: int, module_name: str) -> List[Dict]:
+    resources = []
+    for i in range(1, resource_count + 1):
+        resource_name = f'r{i}-name'
+        resource_type = f'r{i}-type'
+        resource_address = f'{module_name}.{resource_type}.{resource_name}'
+
+        resource = {
+            'address': resource_address,
+            'mode': 'managed',
+            'type': resource_type,
+            'name': resource_name,
+            'provider_name': 'registry.terraform.io/hashicorp/aws',
+            'schema_version': 0,
+            'values': {
+                'val1': 'value1',
+                'val2': 'value2',
+            },
+            'sensitive_values': {
+                'senval1': 'value1',
+                'senval2': 'value2',
+            }
+        }
+
+        resources.append(resource)
+
+    return resources
+
+
 def generate_child_modules(module_count: int,
                            child_modules: List[Dict] = None,
                            resource_count: int = None) -> List[Dict]:
@@ -187,6 +220,27 @@ def generate_child_modules(module_count: int,
     return modules
 
 
+def generate_child_modules_instances(module_count: int,
+                                     child_modules: List[Dict] = None,
+                                     resource_count: int = None) -> List[Dict]:
+    modules = []
+    for i in range(1, module_count + 1):
+        module_name = f'cm-addr[instance-{chr(96 + i)}]'
+        module = {
+            'address': module_name,
+        }
+
+        if child_modules:
+            module['child_modules'] = child_modules
+
+        if resource_count:
+            module['resources'] = generate_resources_for_module(resource_count, module_name)
+
+        modules.append(module)
+
+    return modules
+
+
 ###########
 # TFGRAPH #
 ###########

startleft/startleft/_version/local_scheme.py

@@ -7,7 +7,7 @@ def choose_strategy_by_branch(branch_name: str) -> callable:
     :param branch_name: The name of the branch for which the version is being calculated
     :return: The callable for the version strategy calculation
     """
-    if branch_name == 'main' or 'release/' in branch_name:
+    if branch_name == 'main' or 'release/' in branch_name or 'support/' in branch_name:
         return _no_local_version_strategy
     else:
         return _node_strategy