diff --git a/prowler/providers/aws/services/ecr/ecr_repositories_scan_vulnerabilities_in_latest_image/ecr_repositories_scan_vulnerabilities_in_latest_image.py b/prowler/providers/aws/services/ecr/ecr_repositories_scan_vulnerabilities_in_latest_image/ecr_repositories_scan_vulnerabilities_in_latest_image.py index cba9f4ae3a..068203e0b7 100644 --- a/prowler/providers/aws/services/ecr/ecr_repositories_scan_vulnerabilities_in_latest_image/ecr_repositories_scan_vulnerabilities_in_latest_image.py +++ b/prowler/providers/aws/services/ecr/ecr_repositories_scan_vulnerabilities_in_latest_image/ecr_repositories_scan_vulnerabilities_in_latest_image.py @@ -15,23 +15,27 @@ class ecr_repositories_scan_vulnerabilities_in_latest_image(Check): for repository in registry.repositories: # First check if the repository has images if len(repository.images_details) > 0: - # We only want to check the latest image pushed + # We only want to check the latest image pushed that is scannable image = repository.images_details[-1] - report = Check_Report_AWS(self.metadata()) report.region = repository.region report.resource_id = repository.name report.resource_arn = repository.arn report.resource_tags = repository.tags report.status = "PASS" - report.status_extended = f"ECR repository {repository.name} has imageTag {image.latest_tag} scanned without findings." + status_extended_prefix = f"ECR repository '{repository.name}' has scanned the {image.type} container image with digest '{image.latest_digest}' and tag '{image.latest_tag}' " + report.status_extended = ( + status_extended_prefix + "without findings." + ) if not image.scan_findings_status: report.status = "FAIL" - report.status_extended = f"ECR repository {repository.name} has imageTag {image.latest_tag} without a scan." + report.status_extended = ( + status_extended_prefix + "without a scan." + ) elif image.scan_findings_status == "FAILED": report.status = "FAIL" report.status_extended = ( - f"ECR repository {repository.name} with scan status FAILED." + status_extended_prefix + "with scan status FAILED." ) elif ( image.scan_findings_status != "FAILED" @@ -42,20 +46,29 @@ class ecr_repositories_scan_vulnerabilities_in_latest_image(Check): and image.scan_findings_severity_count.critical ): report.status = "FAIL" - report.status_extended = f"ECR repository {repository.name} has imageTag {image.latest_tag} scanned with findings: CRITICAL->{image.scan_findings_severity_count.critical}." + report.status_extended = ( + status_extended_prefix + + f"with findings: CRITICAL->{image.scan_findings_severity_count.critical}." + ) elif minimum_severity == "HIGH" and ( image.scan_findings_severity_count.critical or image.scan_findings_severity_count.high ): report.status = "FAIL" - report.status_extended = f"ECR repository {repository.name} has imageTag {image.latest_tag} scanned with findings: CRITICAL->{image.scan_findings_severity_count.critical}, HIGH->{image.scan_findings_severity_count.high}." + report.status_extended = ( + status_extended_prefix + + f"with findings: CRITICAL->{image.scan_findings_severity_count.critical}, HIGH->{image.scan_findings_severity_count.high}." + ) elif minimum_severity == "MEDIUM" and ( image.scan_findings_severity_count.critical or image.scan_findings_severity_count.high or image.scan_findings_severity_count.medium ): report.status = "FAIL" - report.status_extended = f"ECR repository {repository.name} has imageTag {image.latest_tag} scanned with findings: CRITICAL->{image.scan_findings_severity_count.critical}, HIGH->{image.scan_findings_severity_count.high}, MEDIUM->{image.scan_findings_severity_count.medium}." + report.status_extended = ( + status_extended_prefix + + f"with findings: CRITICAL->{image.scan_findings_severity_count.critical}, HIGH->{image.scan_findings_severity_count.high}, MEDIUM->{image.scan_findings_severity_count.medium}." + ) findings.append(report) diff --git a/prowler/providers/aws/services/ecr/ecr_service.py b/prowler/providers/aws/services/ecr/ecr_service.py index 9c9fc60feb..36e39a6ccd 100644 --- a/prowler/providers/aws/services/ecr/ecr_service.py +++ b/prowler/providers/aws/services/ecr/ecr_service.py @@ -17,14 +17,14 @@ class ECR(AWSService): super().__init__(__class__.__name__, provider) self.registry_id = self.audited_account self.registries = {} - self.__threading_call__(self.__describe_registries_and_repositories__) - self.__threading_call__(self.__describe_repository_policies__) - self.__threading_call__(self.__get_image_details__) - self.__threading_call__(self.__get_repository_lifecycle_policy__) - self.__threading_call__(self.__get_registry_scanning_configuration__) - self.__threading_call__(self.__list_tags_for_resource__) + self.__threading_call__(self._describe_registries_and_repositories) + self.__threading_call__(self._describe_repository_policies) + self.__threading_call__(self._get_image_details) + self.__threading_call__(self._get_repository_lifecycle_policy) + self.__threading_call__(self._get_registry_scanning_configuration) + self.__threading_call__(self._list_tags_for_resource) - def __describe_registries_and_repositories__(self, regional_client): + def _describe_registries_and_repositories(self, regional_client): logger.info("ECR - Describing registries and repositories...") regional_registry_repositories = [] try: @@ -64,7 +64,7 @@ class ECR(AWSService): f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}" ) - def __describe_repository_policies__(self, regional_client): + def _describe_repository_policies(self, regional_client): logger.info("ECR - Describing repository policies...") try: if regional_client.region in self.registries: @@ -91,7 +91,7 @@ class ECR(AWSService): f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}" ) - def __get_repository_lifecycle_policy__(self, regional_client): + def _get_repository_lifecycle_policy(self, regional_client): logger.info("ECR - Getting repository lifecycle policy...") try: if regional_client.region in self.registries: @@ -119,7 +119,7 @@ class ECR(AWSService): f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}" ) - def __get_image_details__(self, regional_client): + def _get_image_details(self, regional_client): logger.info("ECR - Getting images details...") try: if regional_client.region in self.registries: @@ -139,55 +139,108 @@ class ECR(AWSService): # The following condition is required since sometimes # the AWS ECR API returns None using the iterator if image is not None: - severity_counts = None - last_scan_status = None - if "imageScanStatus" in image: - last_scan_status = image["imageScanStatus"][ - "status" - ] - - if "imageScanFindingsSummary" in image: - severity_counts = FindingSeverityCounts( - critical=0, high=0, medium=0 - ) - finding_severity_counts = image[ + artifact_media_type = image.get( + "artifactMediaType", None + ) + tags = image.get("imageTags", []) + if ECR._is_artifact_scannable( + artifact_media_type, tags + ): + severity_counts = None + last_scan_status = None + image_digest = image.get("imageDigest") + latest_tag = image.get("imageTags", ["None"])[0] + image_pushed_at = image.get("imagePushedAt") + image_scan_findings_field_name = ( "imageScanFindingsSummary" - ]["findingSeverityCounts"] - if "CRITICAL" in finding_severity_counts: - severity_counts.critical = ( - finding_severity_counts["CRITICAL"] - ) - if "HIGH" in finding_severity_counts: - severity_counts.high = ( - finding_severity_counts["HIGH"] - ) - if "MEDIUM" in finding_severity_counts: - severity_counts.medium = ( - finding_severity_counts["MEDIUM"] - ) - latest_tag = "None" - if image.get("imageTags"): - latest_tag = image["imageTags"][0] - repository.images_details.append( - ImageDetails( - latest_tag=latest_tag, - image_pushed_at=image["imagePushedAt"], - latest_digest=image["imageDigest"], - scan_findings_status=last_scan_status, - scan_findings_severity_count=severity_counts, ) - ) - # Sort the repository images by date pushed - repository.images_details.sort( - key=lambda image: image.image_pushed_at - ) + if "docker" in artifact_media_type: + type = "Docker" + elif "oci" in artifact_media_type: + type = "OCI" + else: + type = "" + + # If imageScanStatus is not present or imageScanFindingsSummary is missing, + # we need to call DescribeImageScanFindings because AWS' new version of + # basic scanning does not support imageScanFindingsSummary and imageScanStatus + # in the DescribeImages API. + if "imageScanStatus" not in image: + try: + # use "image" for scan findings to get data the same way as for an image + image = ( + client.describe_image_scan_findings( + registryId=self.registries[ + regional_client.region + ].id, + repositoryName=repository.name, + imageId={ + "imageDigest": image_digest + }, + ) + ) + image_scan_findings_field_name = ( + "imageScanFindings" + ) + except ( + client.exceptions.ImageNotFoundException + ) as error: + logger.warning( + f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}" + ) + continue + except Exception as error: + logger.error( + f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}" + ) + continue + + if "imageScanStatus" in image: + last_scan_status = image["imageScanStatus"][ + "status" + ] + + if image_scan_findings_field_name in image: + severity_counts = FindingSeverityCounts( + critical=0, high=0, medium=0 + ) + finding_severity_counts = image[ + image_scan_findings_field_name + ]["findingSeverityCounts"] + severity_counts.critical = ( + finding_severity_counts.get( + "CRITICAL", 0 + ) + ) + severity_counts.high = ( + finding_severity_counts.get("HIGH", 0) + ) + severity_counts.medium = ( + finding_severity_counts.get("MEDIUM", 0) + ) + + repository.images_details.append( + ImageDetails( + latest_tag=latest_tag, + image_pushed_at=image_pushed_at, + latest_digest=image_digest, + scan_findings_status=last_scan_status, + scan_findings_severity_count=severity_counts, + artifact_media_type=artifact_media_type, + type=type, + ) + ) + # Sort the repository images by date pushed + repository.images_details.sort( + key=lambda image: image.image_pushed_at + ) except Exception as error: logger.error( f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}" ) - def __list_tags_for_resource__(self, regional_client): + def _list_tags_for_resource(self, regional_client): logger.info("ECR - List Tags...") try: if regional_client.region in self.registries: @@ -215,7 +268,7 @@ class ECR(AWSService): f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}" ) - def __get_registry_scanning_configuration__(self, regional_client): + def _get_registry_scanning_configuration(self, regional_client): logger.info("ECR - Getting Registry Scanning Configuration...") try: if regional_client.region in self.registries: @@ -251,6 +304,44 @@ class ECR(AWSService): f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}" ) + @staticmethod + def _is_artifact_scannable(artifact_media_type: str, tags: list[str] = []) -> bool: + """ + Check if an artifact is scannable based on its media type and tags. + + Args: + artifact_media_type (str): The media type of the artifact. + tags (list): The list of tags associated with the artifact. + + Returns: + bool: True if the artifact is scannable, False otherwise. + """ + try: + if artifact_media_type is None: + return False + + # Tools like GoogleContainerTools/jib uses `application/vnd.oci.image.config.v1+json`` also for signatures, which are not scannable. + # Luckily, these are tagged with sha-.sig, so that they can still be easily recognized. + for tag in tags: + if tag.startswith("sha256-") and tag.endswith(".sig"): + return False + + scannable_artifact_media_types = [ + "application/vnd.docker.container.image.v1+json", # Docker image configuration + "application/vnd.docker.image.rootfs.diff.tar", # Docker image layer as a tar archive + "application/vnd.docker.image.rootfs.diff.tar.gzip", # Docker image layer that is compressed using gzip + "application/vnd.oci.image.config.v1+json", # OCI image configuration, but also used by GoogleContainerTools/jib for signatures + "application/vnd.oci.image.layer.v1.tar", # Uncompressed OCI image layer + "application/vnd.oci.image.layer.v1.tar+gzip", # Compressed OCI image layer + ] + + return artifact_media_type in scannable_artifact_media_types + except Exception as error: + logger.error( + f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}" + ) + return False + class FindingSeverityCounts(BaseModel): critical: int @@ -264,6 +355,8 @@ class ImageDetails(BaseModel): image_pushed_at: datetime scan_findings_status: Optional[str] scan_findings_severity_count: Optional[FindingSeverityCounts] + artifact_media_type: Optional[str] + type: str class Repository(BaseModel): diff --git a/tests/providers/aws/services/ecr/ecr_repositories_scan_vulnerabilities_in_latest_image/ecr_repositories_scan_vulnerabilities_in_latest_image_test.py b/tests/providers/aws/services/ecr/ecr_repositories_scan_vulnerabilities_in_latest_image/ecr_repositories_scan_vulnerabilities_in_latest_image_test.py index 815884a5fd..7f99502bd8 100644 --- a/tests/providers/aws/services/ecr/ecr_repositories_scan_vulnerabilities_in_latest_image/ecr_repositories_scan_vulnerabilities_in_latest_image_test.py +++ b/tests/providers/aws/services/ecr/ecr_repositories_scan_vulnerabilities_in_latest_image/ecr_repositories_scan_vulnerabilities_in_latest_image_test.py @@ -18,6 +18,11 @@ repository_arn = ( f"arn:aws:ecr:eu-west-1:{AWS_ACCOUNT_NUMBER}:repository/{repository_name}" ) latest_tag = "test-tag" +latest_digest = "test-digest" +docker_container_image_artifact_media_type = ( + "application/vnd.docker.container.image.v1+json" +) +oci_media_type = "application/vnd.oci.artifact.v1+json" repo_policy_public = { "Version": "2012-10-17", "Statement": [ @@ -32,8 +37,6 @@ repo_policy_public = { class Test_ecr_repositories_scan_vulnerabilities_in_latest_image: - # Mocked Audit Info - def test_no_registries(self): ecr_client = mock.MagicMock ecr_client.registries = {} @@ -118,7 +121,7 @@ class Test_ecr_repositories_scan_vulnerabilities_in_latest_image: result = check.execute() assert len(result) == 0 - def test_image_scaned_without_findings(self): + def test_docker_image_scaned_without_findings(self): ecr_client = mock.MagicMock ecr_client.registries = {} ecr_client.registries[AWS_REGION_EU_WEST_1] = Registry( @@ -135,12 +138,14 @@ class Test_ecr_repositories_scan_vulnerabilities_in_latest_image: images_details=[ ImageDetails( latest_tag=latest_tag, - latest_digest="test-digest", + latest_digest=latest_digest, image_pushed_at=datetime(2023, 1, 1), scan_findings_status="COMPLETE", scan_findings_severity_count=FindingSeverityCounts( critical=0, high=0, medium=0 ), + artifact_media_type=docker_container_image_artifact_media_type, + type="Docker", ), ], lifecycle_policy=None, @@ -167,10 +172,70 @@ class Test_ecr_repositories_scan_vulnerabilities_in_latest_image: assert result[0].status == "PASS" assert ( result[0].status_extended - == f"ECR repository {repository_name} has imageTag {latest_tag} scanned without findings." + == f"ECR repository '{repository_name}' has scanned the Docker container image with digest '{latest_digest}' and tag '{latest_tag}' without findings." ) assert result[0].resource_id == repository_name assert result[0].resource_arn == repository_arn + assert result[0].region == AWS_REGION_EU_WEST_1 + assert result[0].resource_tags == [] + + def test_oci_image_scaned_without_findings(self): + ecr_client = mock.MagicMock + ecr_client.registries = {} + ecr_client.registries[AWS_REGION_EU_WEST_1] = Registry( + id=AWS_ACCOUNT_NUMBER, + region=AWS_REGION_EU_WEST_1, + scan_type="BASIC", + repositories=[ + Repository( + name=repository_name, + arn=repository_arn, + region=AWS_REGION_EU_WEST_1, + scan_on_push=True, + policy=repo_policy_public, + images_details=[ + ImageDetails( + latest_tag=latest_tag, + latest_digest=latest_digest, + image_pushed_at=datetime(2023, 1, 1), + scan_findings_status="COMPLETE", + scan_findings_severity_count=FindingSeverityCounts( + critical=0, high=0, medium=0 + ), + artifact_media_type=docker_container_image_artifact_media_type, + type="OCI", + ), + ], + lifecycle_policy=None, + ) + ], + rules=[], + ) + ecr_client.audit_config = {} + + with mock.patch( + "prowler.providers.common.provider.Provider.get_global_provider", + return_value=set_mocked_aws_provider(), + ), mock.patch( + "prowler.providers.aws.services.ecr.ecr_repositories_scan_vulnerabilities_in_latest_image.ecr_repositories_scan_vulnerabilities_in_latest_image.ecr_client", + ecr_client, + ): + from prowler.providers.aws.services.ecr.ecr_repositories_scan_vulnerabilities_in_latest_image.ecr_repositories_scan_vulnerabilities_in_latest_image import ( + ecr_repositories_scan_vulnerabilities_in_latest_image, + ) + + check = ecr_repositories_scan_vulnerabilities_in_latest_image() + result = check.execute() + assert len(result) == 1 + assert result[0].status == "PASS" + assert ( + result[0].status_extended + == f"ECR repository '{repository_name}' has scanned the OCI container image with digest '{latest_digest}' and tag '{latest_tag}' without findings." + ) + assert result[0].resource_id == repository_name + assert result[0].resource_arn == repository_arn + assert result[0].region == AWS_REGION_EU_WEST_1 + assert result[0].resource_tags == [] def test_image_scanned_with_findings_default_severity_MEDIUM(self): ecr_client = mock.MagicMock @@ -189,12 +254,14 @@ class Test_ecr_repositories_scan_vulnerabilities_in_latest_image: images_details=[ ImageDetails( latest_tag=latest_tag, - latest_digest="test-digest", + latest_digest=latest_digest, image_pushed_at=datetime(2023, 1, 1), scan_findings_status="COMPLETE", scan_findings_severity_count=FindingSeverityCounts( critical=12, high=34, medium=7 ), + artifact_media_type=docker_container_image_artifact_media_type, + type="Docker", ) ], lifecycle_policy=None, @@ -225,10 +292,12 @@ class Test_ecr_repositories_scan_vulnerabilities_in_latest_image: assert result[0].status == "FAIL" assert ( result[0].status_extended - == f"ECR repository {repository_name} has imageTag {latest_tag} scanned with findings: CRITICAL->{12}, HIGH->{34}, MEDIUM->{7}." + == f"ECR repository '{repository_name}' has scanned the Docker container image with digest '{latest_digest}' and tag '{latest_tag}' with findings: CRITICAL->{12}, HIGH->{34}, MEDIUM->{7}." ) assert result[0].resource_id == repository_name assert result[0].resource_arn == repository_arn + assert result[0].region == AWS_REGION_EU_WEST_1 + assert result[0].resource_tags == [] def test_image_scanned_with_findings_default_severity_HIGH(self): ecr_client = mock.MagicMock @@ -247,12 +316,14 @@ class Test_ecr_repositories_scan_vulnerabilities_in_latest_image: images_details=[ ImageDetails( latest_tag=latest_tag, - latest_digest="test-digest", + latest_digest=latest_digest, image_pushed_at=datetime(2023, 1, 1), scan_findings_status="COMPLETE", scan_findings_severity_count=FindingSeverityCounts( critical=12, high=34, medium=7 ), + artifact_media_type=docker_container_image_artifact_media_type, + type="Docker", ) ], lifecycle_policy=None, @@ -283,10 +354,12 @@ class Test_ecr_repositories_scan_vulnerabilities_in_latest_image: assert result[0].status == "FAIL" assert ( result[0].status_extended - == f"ECR repository {repository_name} has imageTag {latest_tag} scanned with findings: CRITICAL->{12}, HIGH->{34}." + == f"ECR repository '{repository_name}' has scanned the Docker container image with digest '{latest_digest}' and tag '{latest_tag}' with findings: CRITICAL->{12}, HIGH->{34}." ) assert result[0].resource_id == repository_name assert result[0].resource_arn == repository_arn + assert result[0].region == AWS_REGION_EU_WEST_1 + assert result[0].resource_tags == [] def test_image_scanned_with_findings_default_severity_CRITICAL(self): ecr_client = mock.MagicMock @@ -305,12 +378,14 @@ class Test_ecr_repositories_scan_vulnerabilities_in_latest_image: images_details=[ ImageDetails( latest_tag=latest_tag, - latest_digest="test-digest", + latest_digest=latest_digest, image_pushed_at=datetime(2023, 1, 1), scan_findings_status="COMPLETE", scan_findings_severity_count=FindingSeverityCounts( critical=12, high=34, medium=7 ), + artifact_media_type=docker_container_image_artifact_media_type, + type="Docker", ) ], lifecycle_policy=None, @@ -341,10 +416,12 @@ class Test_ecr_repositories_scan_vulnerabilities_in_latest_image: assert result[0].status == "FAIL" assert ( result[0].status_extended - == f"ECR repository {repository_name} has imageTag {latest_tag} scanned with findings: CRITICAL->{12}." + == f"ECR repository '{repository_name}' has scanned the Docker container image with digest '{latest_digest}' and tag '{latest_tag}' with findings: CRITICAL->{12}." ) assert result[0].resource_id == repository_name assert result[0].resource_arn == repository_arn + assert result[0].region == AWS_REGION_EU_WEST_1 + assert result[0].resource_tags == [] def test_image_scanned_without_CRITICAL_findings_default_severity_CRITICAL(self): ecr_client = mock.MagicMock @@ -363,12 +440,14 @@ class Test_ecr_repositories_scan_vulnerabilities_in_latest_image: images_details=[ ImageDetails( latest_tag=latest_tag, - latest_digest="test-digest", + latest_digest=latest_digest, image_pushed_at=datetime(2023, 1, 1), scan_findings_status="COMPLETE", scan_findings_severity_count=FindingSeverityCounts( critical=0, high=34, medium=7 ), + artifact_media_type=docker_container_image_artifact_media_type, + type="Docker", ) ], lifecycle_policy=None, @@ -399,7 +478,7 @@ class Test_ecr_repositories_scan_vulnerabilities_in_latest_image: assert result[0].status == "PASS" assert ( result[0].status_extended - == f"ECR repository {repository_name} has imageTag {latest_tag} scanned without findings." + == f"ECR repository '{repository_name}' has scanned the Docker container image with digest '{latest_digest}' and tag '{latest_tag}' without findings." ) assert result[0].resource_id == repository_name assert result[0].resource_arn == repository_arn @@ -423,12 +502,14 @@ class Test_ecr_repositories_scan_vulnerabilities_in_latest_image: images_details=[ ImageDetails( latest_tag=latest_tag, - latest_digest="test-digest", + latest_digest=latest_digest, image_pushed_at=datetime(2023, 1, 1), scan_findings_status="COMPLETE", scan_findings_severity_count=FindingSeverityCounts( critical=0, high=0, medium=7 ), + artifact_media_type=docker_container_image_artifact_media_type, + type="Docker", ) ], lifecycle_policy=None, @@ -459,10 +540,12 @@ class Test_ecr_repositories_scan_vulnerabilities_in_latest_image: assert result[0].status == "PASS" assert ( result[0].status_extended - == f"ECR repository {repository_name} has imageTag {latest_tag} scanned without findings." + == f"ECR repository '{repository_name}' has scanned the Docker container image with digest '{latest_digest}' and tag '{latest_tag}' without findings." ) assert result[0].resource_id == repository_name assert result[0].resource_arn == repository_arn + assert result[0].region == AWS_REGION_EU_WEST_1 + assert result[0].resource_tags == [] def test_image_scanned_fail_scan(self): ecr_client = mock.MagicMock @@ -481,12 +564,14 @@ class Test_ecr_repositories_scan_vulnerabilities_in_latest_image: images_details=[ ImageDetails( latest_tag=latest_tag, - latest_digest="test-digest", + latest_digest=latest_digest, image_pushed_at=datetime(2023, 1, 1), scan_findings_status="FAILED", scan_findings_severity_count=FindingSeverityCounts( critical=0, high=0, medium=0 ), + artifact_media_type=docker_container_image_artifact_media_type, + type="Docker", ) ], lifecycle_policy=None, @@ -513,10 +598,12 @@ class Test_ecr_repositories_scan_vulnerabilities_in_latest_image: assert result[0].status == "FAIL" assert ( result[0].status_extended - == f"ECR repository {repository_name} with scan status FAILED." + == f"ECR repository '{repository_name}' has scanned the Docker container image with digest '{latest_digest}' and tag '{latest_tag}' with scan status FAILED." ) assert result[0].resource_id == repository_name assert result[0].resource_arn == repository_arn + assert result[0].region == AWS_REGION_EU_WEST_1 + assert result[0].resource_tags == [] def test_image_not_scanned(self): ecr_client = mock.MagicMock @@ -535,12 +622,14 @@ class Test_ecr_repositories_scan_vulnerabilities_in_latest_image: images_details=[ ImageDetails( latest_tag=latest_tag, - latest_digest="test-digest", + latest_digest=latest_digest, image_pushed_at=datetime(2023, 1, 1), scan_findings_status="", scan_findings_severity_count=FindingSeverityCounts( critical=0, high=0, medium=0 ), + artifact_media_type=docker_container_image_artifact_media_type, + type="Docker", ) ], lifecycle_policy=None, @@ -567,7 +656,9 @@ class Test_ecr_repositories_scan_vulnerabilities_in_latest_image: assert result[0].status == "FAIL" assert ( result[0].status_extended - == f"ECR repository {repository_name} has imageTag {latest_tag} without a scan." + == f"ECR repository '{repository_name}' has scanned the Docker container image with digest '{latest_digest}' and tag '{latest_tag}' without a scan." ) assert result[0].resource_id == repository_name assert result[0].resource_arn == repository_arn + assert result[0].region == AWS_REGION_EU_WEST_1 + assert result[0].resource_tags == [] diff --git a/tests/providers/aws/services/ecr/ecr_service_test.py b/tests/providers/aws/services/ecr/ecr_service_test.py index e31b651314..50bc53e564 100644 --- a/tests/providers/aws/services/ecr/ecr_service_test.py +++ b/tests/providers/aws/services/ecr/ecr_service_test.py @@ -23,6 +23,7 @@ def mock_make_api_call(self, operation_name, kwarg): if operation_name == "DescribeImages": return { "imageDetails": [ + # Scannable image #1 { "imageDigest": "sha256:d8868e50ac4c7104d2200d42f432b661b2da8c1e417ccfae217e6a1e04bb9295", "imageTags": [ @@ -35,7 +36,9 @@ def mock_make_api_call(self, operation_name, kwarg): "imageScanFindingsSummary": { "findingSeverityCounts": {"CRITICAL": 1, "HIGH": 2, "MEDIUM": 3} }, + "artifactMediaType": "application/vnd.docker.container.image.v1+json", }, + # Scannable image #2 { "imageDigest": "sha256:83251ac64627fc331584f6c498b3aba5badc01574e2c70b2499af3af16630eed", "imageTags": [ @@ -48,6 +51,64 @@ def mock_make_api_call(self, operation_name, kwarg): "imageScanFindingsSummary": { "findingSeverityCounts": {"CRITICAL": 1, "HIGH": 2, "MEDIUM": 3} }, + "artifactMediaType": "application/vnd.docker.container.image.v1+json", + }, + # Not scannable image + { + "imageDigest": "sha256:83251ac64627fc331584f6c498b3aba5badc01574e2c70b2499af3af16630eed", + "imageTags": [ + "sha256-abcdefg123456.sig", + ], + "imagePushedAt": datetime(2023, 1, 2), + "artifactMediaType": "application/vnd.docker.container.image.v1+json", + }, + # Scannable image #3 + { + "imageDigest": "sha256:33251ac64627fc331584f6c498b3aba5badc01574e2c70b2499af3af16630eed", + "imageTags": [ + "test-tag3", + ], + "imagePushedAt": datetime(2023, 1, 2), + "imageScanFindings": { + "findingSeverityCounts": {"CRITICAL": 1, "HIGH": 2, "MEDIUM": 3} + }, + "artifactMediaType": "application/vnd.docker.container.image.v1+json", + }, + # Not scannable image + { + "imageDigest": "sha256:83251ac64627fc331584f6c498b3aba5badc01574e2c70b2499af3af16630eed", + "imageTags": [ + "sha256-83251ac64627fc331584f6c498b3aba5badc01574e2c70b2499af3af16630eed.sig", + ], + "imagePushedAt": datetime(2023, 1, 2), + "imageScanStatus": { + "status": "FAILED", + }, + "artifactMediaType": "application/vnd.oci.image.config.v1+json", + }, + # Not scannable image + { + "imageDigest": "sha256:83251ac64627fc331584f6c498b3aba5badc01574e2c70b2499af3af16630eed", + "imageTags": [ + "test-tag2", + ], + "imagePushedAt": datetime(2023, 1, 2), + "imageScanStatus": { + "status": "FAILED", + }, + "artifactMediaType": "application/vnd.cncf.notary.v2.signature", + }, + # Scannable image #4 + { + "imageDigest": "sha256:43251ac64627fc331584f6c498b3aba5badc01574e2c70b2499af3af16630eed", + "imageTags": [ + "test-tag4", + ], + "imagePushedAt": datetime(2023, 1, 2), + "imageScanStatus": { + "status": "FAILED", + }, + "artifactMediaType": "application/vnd.docker.container.image.v1+json", }, ], } @@ -79,6 +140,16 @@ def mock_make_api_call(self, operation_name, kwarg): }, } + if operation_name == "DescribeImageScanFindings": + return { + "imageScanStatus": { + "status": "COMPLETE", + }, + "imageScanFindings": { + "findingSeverityCounts": {"CRITICAL": 3, "HIGH": 4, "MEDIUM": 5} + }, + } + return make_api_call(self, operation_name, kwarg) @@ -111,14 +182,14 @@ class Test_ECR_Service: assert regional_client.__class__.__name__ == "ECR" # Test ECR session - def test__get_session__(self): + def test_get_session(self): aws_provider = set_mocked_aws_provider() ecr = ECR(aws_provider) assert ecr.session.__class__.__name__ == "Session" # Test describe ECR repositories @mock_aws - def test__describe_registries_and_repositories__(self): + def test_describe_registries_and_repositories(self): ecr_client = client("ecr", region_name=AWS_REGION_EU_WEST_1) ecr_client.create_repository( repositoryName=repo_name, @@ -144,7 +215,7 @@ class Test_ECR_Service: # Test describe ECR repository policies @mock_aws - def test__describe_repository_policies__(self): + def test_describe_repository_policies(self): ecr_client = client("ecr", region_name=AWS_REGION_EU_WEST_1) ecr_client.create_repository( repositoryName=repo_name, @@ -154,43 +225,25 @@ class Test_ECR_Service: ecr = ECR(aws_provider) assert len(ecr.registries) == 1 assert len(ecr.registries[AWS_REGION_EU_WEST_1].repositories) == 1 - assert ecr.registries[AWS_REGION_EU_WEST_1].repositories[0].name == repo_name - assert ecr.registries[AWS_REGION_EU_WEST_1].repositories[0].arn == repo_arn - assert ecr.registries[AWS_REGION_EU_WEST_1].repositories[0].scan_on_push + + repository = ecr.registries[AWS_REGION_EU_WEST_1].repositories[0] + assert repository.name == repo_name + assert repository.arn == repo_arn + assert repository.scan_on_push + assert repository.policy["Statement"][0]["Sid"] == "Allow Describe Images" + assert repository.policy["Statement"][0]["Effect"] == "Allow" assert ( - ecr.registries[AWS_REGION_EU_WEST_1] - .repositories[0] - .policy["Statement"][0]["Sid"] - == "Allow Describe Images" - ) - assert ( - ecr.registries[AWS_REGION_EU_WEST_1] - .repositories[0] - .policy["Statement"][0]["Effect"] - == "Allow" - ) - assert ( - ecr.registries[AWS_REGION_EU_WEST_1] - .repositories[0] - .policy["Statement"][0]["Principal"]["AWS"][0] + repository.policy["Statement"][0]["Principal"]["AWS"][0] == f"arn:aws:iam::{AWS_ACCOUNT_NUMBER}:root" ) + assert repository.policy["Statement"][0]["Action"][0] == "ecr:DescribeImages" assert ( - ecr.registries[AWS_REGION_EU_WEST_1] - .repositories[0] - .policy["Statement"][0]["Action"][0] - == "ecr:DescribeImages" - ) - assert ( - ecr.registries[AWS_REGION_EU_WEST_1] - .repositories[0] - .policy["Statement"][0]["Action"][1] - == "ecr:DescribeRepositories" + repository.policy["Statement"][0]["Action"][1] == "ecr:DescribeRepositories" ) # Test describe ECR repository lifecycle policies @mock_aws - def test__get_lifecycle_policies__(self): + def test_get_lifecycle_policies(self): ecr_client = client("ecr", region_name=AWS_REGION_EU_WEST_1) ecr_client.create_repository( repositoryName=repo_name, @@ -207,7 +260,7 @@ class Test_ECR_Service: # Test get image details @mock_aws - def test__get_image_details__(self): + def test_get_image_details(self): ecr_client = client("ecr", region_name=AWS_REGION_EU_WEST_1) ecr_client.create_repository( repositoryName=repo_name, @@ -215,6 +268,7 @@ class Test_ECR_Service: ) aws_provider = set_mocked_aws_provider() ecr = ECR(aws_provider) + assert len(ecr.registries) == 1 assert len(ecr.registries[AWS_REGION_EU_WEST_1].repositories) == 1 assert ecr.registries[AWS_REGION_EU_WEST_1].repositories[0].name == repo_name @@ -222,19 +276,14 @@ class Test_ECR_Service: assert ecr.registries[AWS_REGION_EU_WEST_1].repositories[0].scan_on_push assert ( len(ecr.registries[AWS_REGION_EU_WEST_1].repositories[0].images_details) - == 2 + == 4 ) # First image pushed - assert ecr.registries[AWS_REGION_EU_WEST_1].repositories[0].images_details[ - 0 - ].image_pushed_at == datetime(2023, 1, 1) - assert ( - ecr.registries[AWS_REGION_EU_WEST_1] - .repositories[0] - .images_details[0] - .latest_tag - == "test-tag1" + first_image = ( + ecr.registries[AWS_REGION_EU_WEST_1].repositories[0].images_details[0] ) + assert first_image.image_pushed_at == datetime(2023, 1, 1) + assert first_image.latest_tag == "test-tag1" assert ( ecr.registries[AWS_REGION_EU_WEST_1] .repositories[0] @@ -242,85 +291,74 @@ class Test_ECR_Service: .latest_digest == "sha256:d8868e50ac4c7104d2200d42f432b661b2da8c1e417ccfae217e6a1e04bb9295" ) + assert first_image.scan_findings_status == "COMPLETE" + assert first_image.scan_findings_severity_count.critical == 1 + assert first_image.scan_findings_severity_count.high == 2 + assert first_image.scan_findings_severity_count.medium == 3 assert ( - ecr.registries[AWS_REGION_EU_WEST_1] - .repositories[0] - .images_details[0] - .scan_findings_status - == "COMPLETE" - ) - assert ( - ecr.registries[AWS_REGION_EU_WEST_1] - .repositories[0] - .images_details[0] - .scan_findings_severity_count.critical - == 1 - ) - assert ( - ecr.registries[AWS_REGION_EU_WEST_1] - .repositories[0] - .images_details[0] - .scan_findings_severity_count.high - == 2 - ) - assert ( - ecr.registries[AWS_REGION_EU_WEST_1] - .repositories[0] - .images_details[0] - .scan_findings_severity_count.medium - == 3 + first_image.artifact_media_type + == "application/vnd.docker.container.image.v1+json" ) # Second image pushed - assert ecr.registries[AWS_REGION_EU_WEST_1].repositories[0].images_details[ - 1 - ].image_pushed_at == datetime(2023, 1, 2) - assert ( - ecr.registries[AWS_REGION_EU_WEST_1] - .repositories[0] - .images_details[1] - .latest_tag - == "test-tag2" + second_image = ( + ecr.registries[AWS_REGION_EU_WEST_1].repositories[0].images_details[1] ) + assert second_image.image_pushed_at == datetime(2023, 1, 2) + assert second_image.latest_tag == "test-tag2" assert ( - ecr.registries[AWS_REGION_EU_WEST_1] - .repositories[0] - .images_details[1] - .latest_digest + second_image.latest_digest == "sha256:83251ac64627fc331584f6c498b3aba5badc01574e2c70b2499af3af16630eed" ) + assert second_image.scan_findings_status == "COMPLETE" + assert second_image.scan_findings_severity_count.critical == 1 + assert second_image.scan_findings_severity_count.high == 2 + assert second_image.scan_findings_severity_count.medium == 3 assert ( - ecr.registries[AWS_REGION_EU_WEST_1] - .repositories[0] - .images_details[1] - .scan_findings_status - == "COMPLETE" + second_image.artifact_media_type + == "application/vnd.docker.container.image.v1+json" ) - assert ( - ecr.registries[AWS_REGION_EU_WEST_1] - .repositories[0] - .images_details[1] - .scan_findings_severity_count.critical - == 1 + + # Third image pushed + third_image = ( + ecr.registries[AWS_REGION_EU_WEST_1].repositories[0].images_details[2] ) + assert third_image.image_pushed_at == datetime(2023, 1, 2) + assert third_image.latest_tag == "test-tag3" assert ( - ecr.registries[AWS_REGION_EU_WEST_1] - .repositories[0] - .images_details[1] - .scan_findings_severity_count.high - == 2 + third_image.latest_digest + == "sha256:33251ac64627fc331584f6c498b3aba5badc01574e2c70b2499af3af16630eed" ) + assert third_image.scan_findings_status == "COMPLETE" + assert third_image.scan_findings_severity_count.critical == 3 + assert third_image.scan_findings_severity_count.high == 4 + assert third_image.scan_findings_severity_count.medium == 5 assert ( - ecr.registries[AWS_REGION_EU_WEST_1] - .repositories[0] - .images_details[1] - .scan_findings_severity_count.medium - == 3 + third_image.artifact_media_type + == "application/vnd.docker.container.image.v1+json" + ) + + # Fourth image pushed + fourth_image = ( + ecr.registries[AWS_REGION_EU_WEST_1].repositories[0].images_details[3] + ) + assert fourth_image.image_pushed_at == datetime(2023, 1, 2) + assert fourth_image.latest_tag == "test-tag4" + assert ( + fourth_image.latest_digest + == "sha256:43251ac64627fc331584f6c498b3aba5badc01574e2c70b2499af3af16630eed" + ) + + assert fourth_image.scan_findings_status == "FAILED" + assert fourth_image.scan_findings_severity_count is None + assert ( + fourth_image.artifact_media_type + == "application/vnd.docker.container.image.v1+json" ) # Test get ECR Registries Scanning Configuration @mock_aws - def test__get_registry_scanning_configuration__(self): + def test_get_registry_scanning_configuration(self): aws_provider = set_mocked_aws_provider() ecr = ECR(aws_provider) assert len(ecr.registries) == 1 @@ -332,3 +370,41 @@ class Test_ECR_Service: scan_filters=[{"filter": "*", "filterType": "WILDCARD"}], ) ] + + def test_is_artifact_scannable_docker(self): + assert ECR._is_artifact_scannable( + "application/vnd.docker.container.image.v1+json" + ) + + def test_is_artifact_scannable_layer_tar(self): + assert ECR._is_artifact_scannable( + "application/vnd.docker.image.rootfs.diff.tar" + ) + + def test_is_artifact_scannable_layer_gzip(self): + assert ECR._is_artifact_scannable( + "application/vnd.docker.image.rootfs.diff.tar.gzip" + ) + + def test_is_artifact_scannable_oci(self): + assert ECR._is_artifact_scannable("application/vnd.oci.image.config.v1+json") + + def test_is_artifact_scannable_oci_tar(self): + assert ECR._is_artifact_scannable("application/vnd.oci.image.layer.v1.tar") + + def test_is_artifact_scannable_oci_compressed(self): + assert ECR._is_artifact_scannable("application/vnd.oci.image.layer.v1.tar+gzip") + + def test_is_artifact_scannable_none(self): + assert not ECR._is_artifact_scannable(None) + + def test_is_artifact_scannable_empty(self): + assert not ECR._is_artifact_scannable("") + + def test_is_artifact_scannable_non_scannable_tags(self): + assert not ECR._is_artifact_scannable("", ["sha256-abcdefg123456.sig"]) + + def test_is_artifact_scannable_scannable_tags(self): + assert ECR._is_artifact_scannable( + "application/vnd.docker.container.image.v1+json", ["abcdefg123456"] + )