feat(check): new check ecr_registry_scan_images_on_push_enabled (#2237)

prowler/providers/aws/services/ecr/ecr_registry_scan_images_on_push_enabled/ecr_registry_scan_images_on_push_enabled.metadata.json

@@ -0,0 +1,33 @@
+{
+  "Provider": "aws",
+  "CheckID": "ecr_registry_scan_images_on_push_enabled",
+  "CheckTitle": "Check if ECR Registry has scan on push enabled",
+  "CheckType": [
+    "Identify",
+    "Vulnerability, patch, and version management"
+  ],
+  "ServiceName": "ecr",
+  "SubServiceName": "",
+  "ResourceIdTemplate": "arn:partition:ecr:region:account-id:resource-id",
+  "Severity": "medium",
+  "ResourceType": "AwsEcrRegistry",
+  "Description": "Check if ECR Registry has scan on push enabled",
+  "Risk": "Amazon ECR image scanning helps in identifying software vulnerabilities in your container images. Amazon ECR uses the Common Vulnerabilities and Exposures (CVEs) database from the open-source Clair project and provides a list of scan findings. ",
+  "RelatedUrl": "",
+  "Remediation": {
+    "Code": {
+      "CLI": "aws ecr put-registry-scanning-configuration --rules 'scanFrequency=SCAN_ON_PUSH,repositoryFilters=[{filter=string,filterType=WILDCARD}]'",
+      "NativeIaC": "",
+      "Other": "",
+      "Terraform": ""
+    },
+    "Recommendation": {
+      "Text": "Enable ECR image scanning and review the scan findings for information about the security of the container images that are being deployed.",
+      "Url": "https://docs.aws.amazon.com/AmazonECR/latest/userguide/image-scanning.html"
+    }
+  },
+  "Categories": [],
+  "DependsOn": [],
+  "RelatedTo": [],
+  "Notes": ""
+}

prowler/providers/aws/services/ecr/ecr_registry_scan_images_on_push_enabled/ecr_registry_scan_images_on_push_enabled.py

@@ -0,0 +1,28 @@
+from prowler.lib.check.models import Check, Check_Report_AWS
+from prowler.providers.aws.services.ecr.ecr_client import ecr_client
+
+
+class ecr_registry_scan_images_on_push_enabled(Check):
+    def execute(self):
+        findings = []
+        for registry in ecr_client.registries:
+            report = Check_Report_AWS(self.metadata())
+            report.region = registry.region
+            report.resource_id = registry.id
+            report.resource_tags = registry.tags
+            report.status = "FAIL"
+            report.status_extended = f"ECR registry {registry.id} has {registry.scan_type} scanning without scan on push"
+            if registry.rules:
+                report.status = "PASS"
+                report.status_extended = f"ECR registry {registry.id} has {registry.scan_type} scan with scan on push"
+                filters = True
+                for rule in registry.rules:
+                    if not rule.scan_filters or "'*'" in str(rule.scan_filters):
+                        filters = False
+                if filters:
+                    report.status = "FAIL"
+                    report.status_extended = f"ECR registry {registry.id} has {registry.scan_type} scanning with scan on push but with repository filters"
+
+            findings.append(report)
+
+        return findings

prowler/providers/aws/services/ecr/ecr_repositories_scan_images_on_push_enabled/ecr_repositories_scan_images_on_push_enabled.metadata.json

@@ -1,7 +1,7 @@
 {
   "Provider": "aws",
   "CheckID": "ecr_repositories_scan_images_on_push_enabled",
-  "CheckTitle": "Check if ECR image scan on push is enabled",
+  "CheckTitle": "[DEPRECATED] Check if ECR image scan on push is enabled",
   "CheckType": [
     "Identify",
     "Vulnerability, patch, and version management"
@@ -11,7 +11,7 @@
   "ResourceIdTemplate": "arn:partition:service:region:account-id:resource-id",
   "Severity": "medium",
   "ResourceType": "AwsEcrRepository",
-  "Description": "Check if ECR image scan on push is enabled",
+  "Description": "[DEPRECATED] Check if ECR image scan on push is enabled",
   "Risk": "Amazon ECR image scanning helps in identifying software vulnerabilities in your container images. Amazon ECR uses the Common Vulnerabilities and Exposures (CVEs) database from the open-source Clair project and provides a list of scan findings. ",
   "RelatedUrl": "",
   "Remediation": {

prowler/providers/aws/services/ecr/ecr_service.py

@@ -17,10 +17,12 @@ class ECR:
         self.audit_resources = audit_info.audit_resources
         self.regional_clients = generate_regional_clients(self.service, audit_info)
         self.repositories = []
+        self.registries = []
         self.__threading_call__(self.__describe_repositories__)
         self.__describe_repository_policies__()
         self.__get_image_details__()
         self.__get_repository_lifecycle_policy__()
+        self.__threading_call__(self.__get_registry_scanning_configuration__)
         self.__list_tags_for_resource__()
 
     def __get_session__(self):
@@ -163,6 +165,33 @@ class ECR:
                 f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
             )
 
+    def __get_registry_scanning_configuration__(self, regional_client):
+        logger.info("ECR - Getting Registry Scanning Configuration...")
+        try:
+            response = regional_client.get_registry_scanning_configuration()
+            rules = []
+            for rule in response.get("scanningConfiguration").get("rules", []):
+                rules.append(
+                    ScanningRule(
+                        scan_frequency=rule.get("scanFrequency"),
+                        scan_filters=rule.get("repositoryFilters"),
+                    )
+                )
+            self.registries.append(
+                Registry(
+                    id=response.get("registryId", ""),
+                    scan_type=response.get("scanningConfiguration").get(
+                        "scanType", "BASIC"
+                    ),
+                    region=regional_client.region,
+                    rules=rules,
+                )
+            )
+        except Exception as error:
+            logger.error(
+                f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
+            )
+
 
 class FindingSeverityCounts(BaseModel):
     critical: int
@@ -186,3 +215,16 @@ class Repository(BaseModel):
     images_details: Optional[list[ImageDetails]]
     lyfecicle_policy: Optional[str]
     tags: Optional[list] = []
+
+
+class ScanningRule(BaseModel):
+    scan_frequency: str
+    scan_filters: list[dict]
+
+
+class Registry(BaseModel):
+    id: str
+    region: str
+    scan_type: str
+    rules: list[ScanningRule]
+    tags: Optional[list] = []

tests/providers/aws/services/ecr/ecr_registry_scan_images_on_push_enabled/ecr_registry_scan_images_on_push_enabled_test.py

@@ -0,0 +1,123 @@
+from re import search
+from unittest import mock
+
+from prowler.providers.aws.services.ecr.ecr_service import Registry, ScanningRule
+
+# Mock Test Region
+AWS_REGION = "eu-west-1"
+AWS_ACCOUNT_NUMBER = "123456789012"
+
+
+class Test_ecr_registry_scan_images_on_push_enabled:
+    def test_no_registries(self):
+        ecr_client = mock.MagicMock
+        ecr_client.registries = []
+
+        with mock.patch(
+            "prowler.providers.aws.services.ecr.ecr_service.ECR",
+            ecr_client,
+        ):
+            from prowler.providers.aws.services.ecr.ecr_registry_scan_images_on_push_enabled.ecr_registry_scan_images_on_push_enabled import (
+                ecr_registry_scan_images_on_push_enabled,
+            )
+
+            check = ecr_registry_scan_images_on_push_enabled()
+            result = check.execute()
+            assert len(result) == 0
+
+    def test_scan_on_push_enabled(self):
+        ecr_client = mock.MagicMock
+        ecr_client.registries = []
+        ecr_client.registries.append(
+            Registry(
+                id=AWS_ACCOUNT_NUMBER,
+                region=AWS_REGION,
+                scan_type="BASIC",
+                rules=[
+                    ScanningRule(
+                        scan_frequency="SCAN_ON_PUSH",
+                        scan_filters=[{"filter": "*", "filterType": "WILDCARD"}],
+                    )
+                ],
+            )
+        )
+
+        with mock.patch(
+            "prowler.providers.aws.services.ecr.ecr_service.ECR",
+            ecr_client,
+        ):
+            from prowler.providers.aws.services.ecr.ecr_registry_scan_images_on_push_enabled.ecr_registry_scan_images_on_push_enabled import (
+                ecr_registry_scan_images_on_push_enabled,
+            )
+
+            check = ecr_registry_scan_images_on_push_enabled()
+            result = check.execute()
+            assert len(result) == 1
+            assert result[0].status == "PASS"
+            assert search("with scan on push", result[0].status_extended)
+            assert result[0].resource_id == AWS_ACCOUNT_NUMBER
+            assert result[0].region == AWS_REGION
+
+    def test_scan_on_push_enabled_with_filters(self):
+        ecr_client = mock.MagicMock
+        ecr_client.registries = []
+        ecr_client.registries.append(
+            Registry(
+                id=AWS_ACCOUNT_NUMBER,
+                region=AWS_REGION,
+                scan_type="BASIC",
+                rules=[
+                    ScanningRule(
+                        scan_frequency="SCAN_ON_PUSH",
+                        scan_filters=[{"filter": "test", "filterType": "WILDCARD"}],
+                    )
+                ],
+            )
+        )
+
+        with mock.patch(
+            "prowler.providers.aws.services.ecr.ecr_service.ECR",
+            ecr_client,
+        ):
+            from prowler.providers.aws.services.ecr.ecr_registry_scan_images_on_push_enabled.ecr_registry_scan_images_on_push_enabled import (
+                ecr_registry_scan_images_on_push_enabled,
+            )
+
+            check = ecr_registry_scan_images_on_push_enabled()
+            result = check.execute()
+            assert len(result) == 1
+            assert result[0].status == "FAIL"
+            assert search(
+                "scanning with scan on push but with repository filters",
+                result[0].status_extended,
+            )
+            assert result[0].resource_id == AWS_ACCOUNT_NUMBER
+            assert result[0].region == AWS_REGION
+
+    def test_scan_on_push_disabled(self):
+        ecr_client = mock.MagicMock
+        ecr_client.registries = []
+        ecr_client.registries.append(
+            Registry(
+                id=AWS_ACCOUNT_NUMBER,
+                region=AWS_REGION,
+                scan_type="BASIC",
+                rules=[],
+            )
+        )
+
+        with mock.patch(
+            "prowler.providers.aws.services.ecr.ecr_service.ECR",
+            ecr_client,
+        ):
+            from prowler.providers.aws.services.ecr.ecr_registry_scan_images_on_push_enabled.ecr_registry_scan_images_on_push_enabled import (
+                ecr_registry_scan_images_on_push_enabled,
+            )
+
+            check = ecr_registry_scan_images_on_push_enabled()
+            result = check.execute()
+            assert len(result) == 1
+            assert result[0].status == "FAIL"
+            assert search("scanning without scan on push", result[0].status_extended)
+            assert result[0].resource_id == AWS_ACCOUNT_NUMBER
+            assert result[0].region == AWS_REGION

tests/providers/aws/services/ecr/ecr_service_test.py

@@ -5,7 +5,7 @@ from boto3 import client, session
 from moto import mock_ecr
 
 from prowler.providers.aws.lib.audit_info.models import AWS_Audit_Info
-from prowler.providers.aws.services.ecr.ecr_service import ECR
+from prowler.providers.aws.services.ecr.ecr_service import ECR, ScanningRule
 
 AWS_ACCOUNT_NUMBER = "123456789012"
 AWS_REGION = "eu-west-1"
@@ -53,6 +53,21 @@ def mock_make_api_call(self, operation_name, kwarg):
             "repositoryName": "string",
             "lifecyclePolicyText": "test-policy",
         }
+    if operation_name == "GetRegistryScanningConfiguration":
+        return {
+            "registryId": AWS_ACCOUNT_NUMBER,
+            "scanningConfiguration": {
+                "scanType": "BASIC",
+                "rules": [
+                    {
+                        "scanFrequency": "SCAN_ON_PUSH",
+                        "repositoryFilters": [
+                            {"filter": "*", "filterType": "WILDCARD"},
+                        ],
+                    },
+                ],
+            },
+        }
     return make_api_call(self, operation_name, kwarg)
 
 
@@ -218,3 +233,18 @@ class Test_ECR_Service:
         )
         assert not ecr.repositories[0].images_details[1].scan_findings_status
         assert not ecr.repositories[0].images_details[1].scan_findings_severity_count
+
+    # Test get ECR Registries
+    @mock_ecr
+    def test__get_registry_scanning_configuration__(self):
+        audit_info = self.set_mocked_audit_info()
+        ecr = ECR(audit_info)
+        assert len(ecr.registries) == 1
+        assert ecr.registries[0].id == AWS_ACCOUNT_NUMBER
+        assert ecr.registries[0].scan_type == "BASIC"
+        assert ecr.registries[0].rules == [
+            ScanningRule(
+                scan_frequency="SCAN_ON_PUSH",
+                scan_filters=[{"filter": "*", "filterType": "WILDCARD"}],
+            )
+        ]