ben.carrasco/prowler
1
0
Fork 0
mirror of https://github.com/prowler-cloud/prowler.git synced 2026-07-20 19:11:54 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(aws): only check artifacts that can be scanned for vulnerabilities by ecr_repositories_scan_vulnerabilities_in_latest_image (#4507)

Browse Source 
Co-authored-by: Pepe Fagoaga <pepe@prowler.com>
This commit is contained in:
Kay Agahd
2024-07-31 14:27:26 +01:00
committed by GitHub
parent 563ddb3707
commit 26a5ffaf82
4 changed files with 456 additions and 183 deletions
Download Patch File Download Diff File Expand all files Collapse all files
prowler/providers/aws/services/ecr/ecr_repositories_scan_vulnerabilities_in_latest_image/ecr_repositories_scan_vulnerabilities_in_latest_image.py
+21 -8
View File
@@ -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)
prowler/providers/aws/services/ecr/ecr_service.py
+145 -52
View File
@@ -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-<HASH-CODE>.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):
tests/providers/aws/services/ecr/ecr_repositories_scan_vulnerabilities_in_latest_image/ecr_repositories_scan_vulnerabilities_in_latest_image_test.py
+110 -19
View File
@@ -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 == []
tests/providers/aws/services/ecr/ecr_service_test.py
+180 -104
View File
@@ -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"]
)