[release/1.38.0] to main

.github/workflows/qa.yml

@@ -14,8 +14,9 @@ jobs:
     name: SonarCloud Analysis
     uses: ./.github/workflows/sonar.yml
     with:
-      python-version: "3.11"
-    secrets: inherit
+      python-version: "3.12"
+    secrets:
+        SONAR_TOKEN: ${{ secrets.SONAR_TOKEN }}
   test:
     name: StartLeft Tests
     strategy:

.github/workflows/sonar.yml

@@ -40,32 +40,30 @@ jobs:
       - name: Generate coverage report
         run: coverage xml
       - name: Analyze with SonarCloud
-        # You can pin the exact commit or the version.
-        # uses: SonarSource/sonarcloud-github-action@commithas or tag
-        uses: SonarSource/sonarcloud-github-action@49e6cd3b187936a73b8280d59ffd9da69df63ec9 #v2.1.1
+        uses: SonarSource/sonarqube-scan-action@1a6d90ebcb0e6a6b1d87e37ba693fe453195ae25 #v5.3.1
         env:
-          GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}  # Needed to get PR information
-          SONAR_TOKEN: ${{ secrets.SONAR_TOKEN }}   # Generate a token on Sonarcloud.io, add it to the secrets of this repo with the name SONAR_TOKEN (Settings > Secrets > Actions > add new repository secret)
+          GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }} # Needed to get PR information
+          SONAR_TOKEN: ${{ secrets.SONAR_TOKEN }}   # SonarCloud token
+          SONAR_HOST_URL: "https://sonarcloud.io"   # Required for SonarCloud
         with:
-          # Additional arguments for the sonarcloud scanner
-          args:
+          args: >
             -Dsonar.projectKey=startleft
             -Dsonar.organization=continuumsec
-            -Dsonar.python.version=3.9,3.10,3.11
+            -Dsonar.python.version=3.10,3.11,3.12
             -Dsonar.qualitygate.wait=true
             -Dsonar.python.coverage.reportPaths=coveragereport/coverage.xml
 
-            # Args explanation
-            # Unique keys of your project and organization. You can find them in SonarCloud > Information (bottom-left menu)
-            # mandatory
-            # -Dsonar.projectKey=
-            # -Dsonar.organization=
+          # Args explanation
+          # Unique keys of your project and organization. You can find them in SonarCloud > Information (bottom-left menu)
+          # mandatory
+          # -Dsonar.projectKey=
+          # -Dsonar.organization=
 
-            # Version of supported python versions to get a more precise analysis
-            # -Dsonar.python.version=
+          # Version of supported python versions to get a more precise analysis
+          # -Dsonar.python.version=
 
-            # Flag to way for Analysis Quality Gate results, if fail the steps it will be marked as failed too.
-            # -Dsonar.qualitygate.wait=
+          # Flag to way for Analysis Quality Gate results, if fail the steps it will be marked as failed too.
+          # -Dsonar.qualitygate.wait=
 
-            # The path for coverage report to use in the SonarCloud analysis, it must be in XML format.
-            # -Dsonar.python.coverage.reportPaths=
+          # The path for coverage report to use in the SonarCloud analysis, it must be in XML format.
+          # -Dsonar.python.coverage.reportPaths=

setup.py

@@ -16,14 +16,14 @@
     include_package_data=True,
     python_requires='>= 3.10, < 3.13',
     install_requires=[
-        'pyyaml==6.0.1',
+        'pyyaml==6.0.3',
         'jsonschema==4.19.0',
         'deepmerge==1.1.0',
         'jmespath==1.0.1',
         'python-hcl2==4.3.2',
         'requests==2.32.4',
-        'fastapi>=0.116.1,<0.117.0',
-        "python-multipart==0.0.19",
+        'fastapi>=0.120.4,<0.121.0',
+        "python-multipart==0.0.20",
         'click==8.1.7',
         'uvicorn==0.23.2',
         'vsdx==0.5.19',

sl_util/sl_util/secure_regex.py

@@ -23,3 +23,7 @@ def split(pattern, text, maxsplit=0, options=None):
 
 def compile(pattern, options=None):
     return re2.compile(pattern, options)
+
+
+def search(pattern, string, options=None):
+    return re2.search(pattern, string, options)

sl_util/sl_util/str_utils.py

@@ -1,7 +1,11 @@
+import html
 import random
 import uuid
+
 from word2number import w2n
 
+from sl_util.sl_util import secure_regex as re
+
 
 def deterministic_uuid(source):
     if source:
@@ -22,5 +26,19 @@ def to_number(input, default_value: int = 0) -> int:
         except ValueError:
             return default_value
 
+
 def truncate(s: str, max_length: int) -> str:
-    return s[:max_length] if s else s
+    return s[:max_length] if s else s
+
+
+def remove_html_tags_and_entities(s: str) -> str:
+    if s is None:
+        return ''
+
+    pattern_tags = re.compile(r'<\s*/?\s*[a-zA-Z]+.*?>')
+    no_html = re.sub(pattern_tags, ' ', s).strip() if s else s
+
+    pattern_spaces = re.compile(r'\s+')
+    no_spaces = re.sub(pattern_spaces, ' ', no_html) if no_html else no_html
+
+    return html.unescape(no_spaces).replace('\xa0', ' ').strip()

sl_util/tests/unit/test_secure_regex_wrapper.py

@@ -54,3 +54,8 @@ def test_find_all(self, expression, value, expected):
     ])
     def test_split(self, expression, value, expected):
         assert sre.findall(expression, value) == expected
+
+
+    def test_search(self):
+        assert sre.search(r"match\d+.*match\d{1}", "match1 and match2") is not None
+        assert sre.search(r"match\d+.*match\d{1}", "matchA not found") is None

sl_util/tests/unit/test_str_utils.py

@@ -1,7 +1,9 @@
-from pytest import mark, param
 import random
 from unittest.mock import patch
-from sl_util.sl_util.str_utils import deterministic_uuid, to_number
+
+from pytest import mark, param
+
+from sl_util.sl_util.str_utils import deterministic_uuid, to_number, remove_html_tags_and_entities
 
 
 class TestStrUtils:
@@ -76,3 +78,26 @@ def test_number_conversions_to_alphanumeric(self, source):
         number2 = to_number(source)
         # Then we obtain default value 0
         assert number1 == number2 == 0
+
+    @mark.parametrize('source, expected', [
+        param('<a href="http://example.com">Link</a>', 'Link', id='only link tag'),
+        param('<p>This is an <b>AWS</b> component.</p>', 'This is an AWS component.', id='with nested tags'),
+        param('<div><h1>DDBB</h1> <p>Postgres SQL</p></div>', 'DDBB Postgres SQL', id='with multiple nested tags'),
+        param('< p>This is an <b >AWS</b > component.< /p > <a href="http://example.com">Link</a>',
+              'This is an AWS component. Link', id='with tags and link'),
+        param('<p></p>Void tag', 'Void tag', id='void tag'),
+        param('IN < http & https', 'IN < http & https', id='with lt and ampersand'),
+        param('OUT > socket & https', 'OUT > socket & https', id='with gt and ampersand'),
+        param(' 2 < 3 socket > </3  https> <&udp> <=tcp>', '2 < 3 socket > </3 https> <&udp> <=tcp>', id='with non html gt and lt'),
+        param('No HTML tags here.', 'No HTML tags here.', id='without html tags'),
+        param('HTML&nbsp;entities&nbsp;&lt;&gt;&amp;&pound;&euro;&copy;', 'HTML entities <>&£€©', id='with html entities'),
+        param('', '', id='empty string'),
+        param(None, '', id='null value')
+    ])
+    def test_remove_html_tags_and_entities(self, source, expected):
+        # GIVEN a string with html tags
+        # WHEN removing html tags
+        result = remove_html_tags_and_entities(source)
+
+        # THEN we obtain the expected string
+        assert result == expected

sl_util/tests/util/file_utils.py

@@ -15,3 +15,11 @@ def get_upload_file(source: str) -> UploadFile:
     tmp_file.seek(0)
 
     return UploadFile(filename=os.path.split(source)[1], file=tmp_file)
+
+
+def generate_temporary_file(size_in_bytes: int, filename: str = "temp.txt") -> bytes:
+    temporary_file = SpooledTemporaryFile()
+    temporary_file.write(b'0' * size_in_bytes)
+    temporary_file.seek(0)
+
+    return UploadFile(filename=filename, file=temporary_file).file.read()

slp_cft/tests/integration/test_cft_processor.py

@@ -2,13 +2,16 @@
 
 from sl_util.sl_util.file_utils import get_byte_data
 from slp_base.slp_base.errors import OTMBuildingError, MappingFileNotValidError, IacFileNotValidError, \
-    LoadingIacFileError
+    LoadingIacFileError, ErrorCode
+from slp_base.slp_base.mapping import MAX_SIZE as MAPPING_MAX_SIZE, MIN_SIZE as MAPPING_MIN_SIZE
 from slp_base.tests.util.otm import validate_and_compare_otm, validate_and_compare
 from slp_cft import CloudformationProcessor
 from slp_cft.tests.resources import test_resource_paths
 from slp_cft.tests.resources.test_resource_paths import expected_orphan_component_is_not_mapped, \
     cft_components_with_trustzones_of_same_type_otm, cloudformation_minimal_content_otm
 from slp_cft.tests.utility import excluded_regex
+from sl_util.tests.util.file_utils import generate_temporary_file
+from slp_cft.slp_cft.validate.cft_validator import MAX_SIZE as FILE_MAX_SIZE, MIN_SIZE as FILE_MIN_SIZE
 
 SAMPLE_ID = 'id'
 SAMPLE_NAME = 'name'
@@ -17,8 +20,16 @@
 SAMPLE_SINGLE_VALID_CFT_FILE = test_resource_paths.cloudformation_single_file
 SAMPLE_VALID_MAPPING_FILE_IR = test_resource_paths.cloudformation_mapping_iriusrisk
 SAMPLE_MAPPING_FILE_WITHOUT_REF = test_resource_paths.cloudformation_mapping_without_ref
+SAMPLE_DEFAULT_OLD_MAPPING = test_resource_paths.cloudformation_old_default_mapping
+SAMPLE_DEFAULT_NEW_MAPPING = test_resource_paths.cloudformation_new_default_mapping
+SAMPLE_MAPPING_WITHOUT_TRUSTZONE_TYPE = test_resource_paths.cloudformation_mapping_valid_without_trustzone_type
+SAMPLE_CLOUDFORMATION_MAPPING_ALL_FUNCTIONS = test_resource_paths.cloudformation_mapping_all_functions
 SAMPLE_NETWORKS_CFT_FILE = test_resource_paths.cloudformation_networks_file
 SAMPLE_RESOURCES_CFT_FILE = test_resource_paths.cloudformation_resources_file
+SAMPLE_RESOURCES_INVALID_CFT_FILE = test_resource_paths.cloudformation_resources_invalid
+SAMPLE_REACT_CORS_SPA_STACK = test_resource_paths.cloudformation_react_cors_spa_stack
+SAMPLE_CLOUDFORMATION_ALL_FUNCTIONS = test_resource_paths.cloudformation_all_functions
+SAMPLE_CLOUDFORMATION_TEST = test_resource_paths.cloudformation_test
 SAMPLE_REF_DEFAULT_JSON = test_resource_paths.cloudformation_with_ref_function_and_default_property_json
 SAMPLE_REF_DEFAULT_YAML = test_resource_paths.cloudformation_with_ref_function_and_default_property_yaml
 SAMPLE_REF_WITHOUT_DEFAULT_JSON = test_resource_paths.cloudformation_with_ref_function_and_without_default_property_json
@@ -486,7 +497,7 @@ def test_invalid_cloudformation_file(self, cloudformation_file):
         mapping_file = [get_byte_data(SAMPLE_VALID_MAPPING_FILE)]
 
         # WHEN creating OTM project from IaC file
-        # THEN raises OTMBuildingError
+        # THEN raises IacFileNotValidError
         with pytest.raises(IacFileNotValidError):
             CloudformationProcessor(SAMPLE_ID, SAMPLE_NAME, cloudformation_file, mapping_file).process()
 
@@ -522,7 +533,7 @@ def test_run_empty_multiple_iac_files(self):
         # GIVEN a request without any iac_file key
         mapping_file = get_byte_data(SAMPLE_VALID_MAPPING_FILE_IR)
         # WHEN the method CloudformationProcessor::process is invoked
-        # THEN an RequestValidationError is raised
+        # THEN an LoadingIacFileError is raised
         with pytest.raises(LoadingIacFileError):
             CloudformationProcessor('multiple-files', 'multiple-files', [], mapping_file).process()
 
@@ -541,22 +552,6 @@ def test_security_group_configuration(self, source):
         assert len(otm.components) == 1
         assert otm.components[0].parent == 'f0ba7722-39b6-4c81-8290-a30a248bb8d9'
 
-    def test_multiple_stack_plus_s3_ec2(self):
-        # GIVEN the file with multiple Subnet AWS::EC2::Instance different configurations
-        cloudformation_file = get_byte_data(test_resource_paths.multiple_stack_plus_s3_ec2)
-        # AND a valid iac mappings file
-        mapping_file = [get_byte_data(SAMPLE_VALID_MAPPING_FILE)]
-
-        # WHEN processing
-        otm = CloudformationProcessor(SAMPLE_ID, SAMPLE_NAME, [cloudformation_file], mapping_file).process()
-
-        assert len(otm.components) == 9
-        publicSubnet1Id = [component for component in otm.components if component.name == 'PublicSubnet1'][0].id
-        assert publicSubnet1Id
-        ec2WithWrongParent = [component for component in otm.components if
-                              component.type == 'ec2' and component.parent != publicSubnet1Id]
-        assert len(ec2WithWrongParent) == 0
-
     def test_parsing_cft_json_file_with_ref(self):
         # GIVEN a cloudformation JSON  file
         cloudformation_file = get_byte_data(SAMPLE_REF_DEFAULT_JSON)
@@ -687,3 +682,130 @@ def test_components_with_trustzones_of_same_type(self):
         # THEN the result should be the expected
         result, expected = validate_and_compare(otm, cft_components_with_trustzones_of_same_type_otm, None)
         assert result == expected
+
+    def test_multiple_stack_plus_s3_ec2(self):
+        # GIVEN the file with multiple Subnet AWS::EC2::Instance different configurations
+        cloudformation_file = get_byte_data(test_resource_paths.multiple_stack_plus_s3_ec2)
+        # AND a valid iac mappings file
+        mapping_file = get_byte_data(SAMPLE_VALID_MAPPING_FILE)
+
+        # WHEN processing
+        otm = CloudformationProcessor(SAMPLE_ID, SAMPLE_NAME, [cloudformation_file], [mapping_file]).process()
+
+        assert len(otm.components) == 9
+        publicSubnet1Id = [component for component in otm.components if component.name == 'PublicSubnet1'][0].id
+        assert publicSubnet1Id
+        ec2WithWrongParent = [component for component in otm.components if
+                              component.type == 'ec2' and component.parent != publicSubnet1Id]
+        assert len(ec2WithWrongParent) == 0
+
+    def test_improve_parsing_problems_built_in_functions(self):
+        # GIVEN a cloudformation file with built-in functions
+        cloudformation_file = get_byte_data(SAMPLE_REACT_CORS_SPA_STACK)
+        # AND a valid iac mappings file
+        mapping_file = get_byte_data(SAMPLE_DEFAULT_OLD_MAPPING)
+
+        # WHEN processing
+        otm = CloudformationProcessor(SAMPLE_ID, SAMPLE_NAME, [cloudformation_file], [mapping_file]).process()
+
+        assert len(otm.trustzones) == 1
+        assert len(otm.dataflows) == 1
+        assert len(otm.components) == 4
+
+    def test_checking_jmespath_functions(self):
+        # GIVEN a cloudformation file with all JMESPath functions
+        cloudformation_file = get_byte_data(SAMPLE_CLOUDFORMATION_ALL_FUNCTIONS)
+        # AND a valid iac mappings file
+        mapping_file = get_byte_data(SAMPLE_CLOUDFORMATION_MAPPING_ALL_FUNCTIONS)
+
+        # WHEN processing
+        otm = CloudformationProcessor(SAMPLE_ID, SAMPLE_NAME, [cloudformation_file], [mapping_file]).process()
+
+        assert len(otm.trustzones) == 1
+        assert len(otm.dataflows) == 0
+        assert len(otm.components) == 5
+
+    def test_not_present_parents(self):
+        # GIVEN a cloudformation file with all JMESPath functions
+        cloudformation_file = get_byte_data(SAMPLE_CLOUDFORMATION_TEST)
+        # AND a valid iac mappings file
+        mapping_file = get_byte_data(SAMPLE_DEFAULT_NEW_MAPPING)
+
+        # WHEN processing
+        otm = CloudformationProcessor(SAMPLE_ID, SAMPLE_NAME, [cloudformation_file], [mapping_file]).process()
+
+        assert len(otm.trustzones) == 1
+        assert len(otm.dataflows) == 0
+        assert len(otm.components) == 4
+
+    def test_invalid_resources_mapping_file(self):
+        # GIVEN a valid CFT file with altsource resources
+        cloudformation_file = get_byte_data(SAMPLE_VALID_CFT_FILE)
+
+        # AND a invalid format CFT mapping file
+        mapping_file = get_byte_data(SAMPLE_RESOURCES_INVALID_CFT_FILE)
+
+        # WHEN the CFT file is processed
+        # THEN an MappingFileNotValidError is raised
+        with pytest.raises(MappingFileNotValidError) as error:
+            CloudformationProcessor(SAMPLE_ID, SAMPLE_NAME, [cloudformation_file], [mapping_file]).process()
+
+        # AND the error details are correct
+        assert ErrorCode.MAPPING_FILE_NOT_VALID == error.value.error_code
+        assert 'Mapping files are not valid' == error.value.title
+        assert 'Mapping file does not comply with the schema' == error.value.detail
+        assert "'trustzones' is a required property" == error.value.message
+
+    @pytest.mark.parametrize('cft_file_size', [FILE_MAX_SIZE + 1, FILE_MIN_SIZE - 1])
+    def test_min_max_cloudformation_file_sizes(self, cft_file_size):
+        # GIVEN a max file size limit and a valid CFT file
+        max_file_size_allowed_in_bytes = 1024 * 1024
+        cloudformation_file = generate_temporary_file(cft_file_size, "test_max_size.txt")
+
+        # AND a valid CFT mapping file
+        mapping_file = get_byte_data(SAMPLE_VALID_MAPPING_FILE)
+
+        # WHEN the CFT file is processed
+        # THEN an IacFileNotValidError is raised
+        with pytest.raises(IacFileNotValidError) as error:
+            CloudformationProcessor(SAMPLE_ID, SAMPLE_NAME, [cloudformation_file], [mapping_file]).process()
+
+        # AND the error details are correct
+        assert ErrorCode.IAC_NOT_VALID == error.value.error_code
+        assert 'CloudFormation file is not valid' == error.value.title
+        assert 'Provided iac_file is not valid. Invalid size' == error.value.detail
+        assert 'Provided iac_file is not valid. Invalid size' == error.value.message
+
+    @pytest.mark.parametrize('mapping_file_size', [MAPPING_MAX_SIZE + 1, MAPPING_MIN_SIZE - 1])
+    def test_min_max_mapping_file_sizes(self, mapping_file_size):
+        # GIVEN a valid CFT file with altsource resources
+        cloudformation_file = get_byte_data(SAMPLE_VALID_CFT_FILE)
+
+        # AND a invalid size CFT mapping file
+        mapping_file = generate_temporary_file(mapping_file_size, "test_mapping_sizes.txt")
+
+        # WHEN the CFT file is processed
+        # THEN an MappingFileNotValidError is raised
+        with pytest.raises(MappingFileNotValidError) as error:
+            CloudformationProcessor(SAMPLE_ID, SAMPLE_NAME, [cloudformation_file], [mapping_file]).process()
+
+        # AND the error details are correct
+        assert ErrorCode.MAPPING_FILE_NOT_VALID == error.value.error_code
+        assert 'Mapping files are not valid' == error.value.title
+        assert 'Mapping files are not valid. Invalid size' == error.value.detail
+        assert 'Mapping files are not valid. Invalid size' == error.value.message
+
+    def test_mapping_trustzone_no_type(self):
+        # GIVEN a valid CFT file with some resources
+        cloudformation_file = get_byte_data(test_resource_paths.cloudformation_for_security_group_tests_json)
+
+        # AND a valid CFT mapping file
+        mapping_file = get_byte_data(SAMPLE_MAPPING_WITHOUT_TRUSTZONE_TYPE)
+
+        # WHEN the CFT file is processed
+        otm = CloudformationProcessor(SAMPLE_ID, SAMPLE_NAME, [cloudformation_file], [mapping_file]).process()
+
+        # THEN the number of TZs, components and dataflows are right
+        assert len(otm.trustzones) == 2
+        assert len(otm.components) == 22
+        assert len(otm.dataflows) == 22