feat(aws/sagemaker): add sagemaker_clarify_exists check (#11211)

Signed-off-by: Oleksandr Sanin <alexaaander.sanin@gmail.com>
Signed-off-by: Oleksandr Yizchak Sanin <alexaaander.sanin@gmail.com>
Co-authored-by: Daniel Barranquero <danielbo2001@gmail.com>

prowler/CHANGELOG.md

@@ -8,6 +8,7 @@ All notable changes to the **Prowler SDK** are documented in this file.
 
 - `securityhub_delegated_admin_enabled_all_regions` check for AWS provider, verifying that Security Hub has a delegated administrator, is active in all opted-in regions, and has organization auto-enable on [(#11259)](https://github.com/prowler-cloud/prowler/pull/11259)
 - `config_delegated_admin_and_org_aggregator_all_regions` check for AWS provider, verifying that AWS Config has a delegated administrator and an organization aggregator covering all AWS regions [(#11259)](https://github.com/prowler-cloud/prowler/pull/11259)
+- `sagemaker_clarify_exists` check for AWS provider [(#11211)](https://github.com/prowler-cloud/prowler/pull/11211)
 
 ---
 

prowler/providers/aws/services/sagemaker/sagemaker_clarify_exists/sagemaker_clarify_exists.metadata.json

@@ -0,0 +1,39 @@
+{
+  "Provider": "aws",
+  "CheckID": "sagemaker_clarify_exists",
+  "CheckTitle": "Amazon SageMaker Clarify processing jobs exist in the region",
+  "CheckType": [
+    "Software and Configuration Checks/AWS Security Best Practices"
+  ],
+  "ServiceName": "sagemaker",
+  "SubServiceName": "",
+  "ResourceIdTemplate": "",
+  "Severity": "low",
+  "ResourceType": "Other",
+  "ResourceGroup": "ai_ml",
+  "Description": "**SageMaker Clarify** provides bias detection and model explainability for ML workloads.\n\nThis check verifies that at least one SageMaker processing job using the AWS-managed Clarify container image exists in each successfully scanned region. The absence of Clarify jobs indicates that responsible-AI controls such as bias detection and explainability are not in place.",
+  "Risk": "Without **SageMaker Clarify** processing jobs, ML models may be deployed without bias analysis or explainability reports. This can lead to:\n- **Regulatory non-compliance** with AI governance frameworks\n- **Undetected bias** in model predictions affecting protected groups\n- **Lack of accountability** for ML model decisions in production",
+  "RelatedUrl": "",
+  "AdditionalURLs": [
+    "https://docs.aws.amazon.com/sagemaker/latest/dg/clarify-configure-processing-jobs.html",
+    "https://docs.aws.amazon.com/sagemaker/latest/dg-ecr-paths/sagemaker-algo-docker-registry-paths.html"
+  ],
+  "Remediation": {
+    "Code": {
+      "CLI": "aws sagemaker create-processing-job --processing-job-name clarify-bias-check --app-specification ImageUri=<clarify-image-uri> --role-arn <role-arn> --processing-resources 'ClusterConfig={InstanceCount=1,InstanceType=ml.m5.xlarge,VolumeSizeInGB=20}'",
+      "NativeIaC": "",
+      "Other": "1. Open the AWS Console and go to Amazon SageMaker\n2. Navigate to Processing > Processing jobs\n3. Click Create processing job\n4. Select the SageMaker Clarify container image for your region\n5. Configure input/output paths and the analysis configuration\n6. Click Create processing job",
+      "Terraform": ""
+    },
+    "Recommendation": {
+      "Text": "Create SageMaker Clarify processing jobs to evaluate models for bias and explainability before deployment. Integrate Clarify into your ML pipeline to ensure responsible AI practices.",
+      "Url": "https://hub.prowler.com/check/sagemaker_clarify_exists"
+    }
+  },
+  "Categories": [
+    "gen-ai"
+  ],
+  "DependsOn": [],
+  "RelatedTo": [],
+  "Notes": "Results are generated per scanned region. Regions where `ListProcessingJobs` cannot be queried are omitted from the findings."
+}

prowler/providers/aws/services/sagemaker/sagemaker_clarify_exists/sagemaker_clarify_exists.py

@@ -0,0 +1,54 @@
+from prowler.lib.check.models import Check, Check_Report_AWS
+from prowler.providers.aws.services.sagemaker.sagemaker_client import sagemaker_client
+
+
+class sagemaker_clarify_exists(Check):
+    """Check whether at least one SageMaker Clarify processing job exists per region.
+
+    A region is reported only when ListProcessingJobs succeeded for it; regions
+    where the API call failed (e.g. AccessDenied, unsupported region) are
+    skipped at the service layer and produce no finding.
+
+    - PASS: At least one processing job uses the AWS-managed Clarify container
+      image in the region (one finding per job).
+    - FAIL: No processing job uses the Clarify container image in the region
+      (one finding per region).
+    """
+
+    def execute(self) -> list[Check_Report_AWS]:
+        """Execute the SageMaker Clarify exists check.
+
+        Returns:
+            A list of reports containing the result of the check.
+        """
+        findings = []
+        for region in sorted(sagemaker_client.processing_jobs_scanned_regions):
+            clarify_jobs = sorted(
+                (
+                    job
+                    for job in sagemaker_client.sagemaker_processing_jobs
+                    if job.region == region
+                    and job.image_uri
+                    and "sagemaker-clarify-processing" in job.image_uri
+                ),
+                key=lambda job: job.name,
+            )
+
+            if clarify_jobs:
+                for job in clarify_jobs:
+                    report = Check_Report_AWS(metadata=self.metadata(), resource=job)
+                    report.status = "PASS"
+                    report.status_extended = f"SageMaker Clarify processing job {job.name} exists in region {region}."
+                    findings.append(report)
+            else:
+                report = Check_Report_AWS(metadata=self.metadata(), resource={})
+                report.region = region
+                report.resource_id = "sagemaker-clarify"
+                report.resource_arn = f"arn:{sagemaker_client.audited_partition}:sagemaker:{region}:{sagemaker_client.audited_account}:processing-job"
+                report.status = "FAIL"
+                report.status_extended = (
+                    f"No SageMaker Clarify processing jobs found in region {region}."
+                )
+                findings.append(report)
+
+        return findings

prowler/providers/aws/services/sagemaker/sagemaker_service.py

@@ -15,6 +15,8 @@ class SageMaker(AWSService):
         self.sagemaker_notebook_instances = []
         self.sagemaker_models = []
         self.sagemaker_training_jobs = []
+        self.sagemaker_processing_jobs = []
+        self.processing_jobs_scanned_regions = set()
         self.sagemaker_domains = []
         self.endpoint_configs = {}
         self.sagemaker_model_registries = []
@@ -24,6 +26,7 @@ class SageMaker(AWSService):
         self.__threading_call__(self._list_notebook_instances)
         self.__threading_call__(self._list_models)
         self.__threading_call__(self._list_training_jobs)
+        self.__threading_call__(self._list_processing_jobs)
         self.__threading_call__(self._list_endpoint_configs)
         self.__threading_call__(self._list_domains)
         self.__threading_call__(self._list_model_package_groups)
@@ -37,6 +40,9 @@ class SageMaker(AWSService):
         self.__threading_call__(
             self._describe_training_job, self.sagemaker_training_jobs
         )
+        self.__threading_call__(
+            self._describe_processing_job, self.sagemaker_processing_jobs
+        )
         self.__threading_call__(
             self._describe_endpoint_config, list(self.endpoint_configs.values())
         )
@@ -51,6 +57,9 @@ class SageMaker(AWSService):
         self.__threading_call__(
             self._list_tags_for_resource, self.sagemaker_training_jobs
         )
+        self.__threading_call__(
+            self._list_tags_for_resource, self.sagemaker_processing_jobs
+        )
         self.__threading_call__(
             self._list_tags_for_resource, list(self.endpoint_configs.values())
         )
@@ -128,6 +137,66 @@ class SageMaker(AWSService):
                 f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
             )
 
+    def _list_processing_jobs(self, regional_client):
+        """List SageMaker processing jobs in a region.
+
+        Populates ``self.sagemaker_processing_jobs`` with `ProcessingJob`
+        entries and adds ``regional_client.region`` to
+        ``self.processing_jobs_scanned_regions`` once pagination succeeds, so
+        regions where ``ListProcessingJobs`` fails are skipped by checks that
+        consume that set.
+
+        Args:
+            regional_client: Regional SageMaker boto3 client.
+        """
+        logger.info("SageMaker - listing processing jobs...")
+        try:
+            list_processing_jobs_paginator = regional_client.get_paginator(
+                "list_processing_jobs"
+            )
+            for page in list_processing_jobs_paginator.paginate():
+                for processing_job in page["ProcessingJobSummaries"]:
+                    if not self.audit_resources or (
+                        is_resource_filtered(
+                            processing_job["ProcessingJobArn"], self.audit_resources
+                        )
+                    ):
+                        self.sagemaker_processing_jobs.append(
+                            ProcessingJob(
+                                name=processing_job["ProcessingJobName"],
+                                region=regional_client.region,
+                                arn=processing_job["ProcessingJobArn"],
+                            )
+                        )
+            self.processing_jobs_scanned_regions.add(regional_client.region)
+        except Exception as error:
+            logger.error(
+                f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
+            )
+
+    def _describe_processing_job(self, processing_job):
+        """Describe a SageMaker processing job and enrich its image metadata.
+
+        Reads ``AppSpecification.ImageUri`` from ``DescribeProcessingJob`` and
+        stores it on ``processing_job.image_uri``. Errors are logged and
+        swallowed so a failure in one job does not abort the scan.
+
+        Args:
+            processing_job: ProcessingJob model to enrich in-place.
+        """
+        logger.info("SageMaker - describing processing job...")
+        try:
+            regional_client = self.regional_clients[processing_job.region]
+            describe_processing_job = regional_client.describe_processing_job(
+                ProcessingJobName=processing_job.name
+            )
+            app_spec = describe_processing_job.get("AppSpecification", {})
+            processing_job.image_uri = app_spec.get("ImageUri")
+        except Exception as error:
+            logger.error(
+                f"{processing_job.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
+            )
+
     def _describe_notebook_instance(self, notebook_instance):
         logger.info("SageMaker - describing notebook instances...")
         try:
@@ -451,6 +520,25 @@ class TrainingJob(BaseModel):
     tags: Optional[list] = []
 
 
+class ProcessingJob(BaseModel):
+    """Represents a SageMaker processing job.
+
+    Attributes:
+        name: Processing job name.
+        region: AWS region where the job lives.
+        arn: Processing job ARN.
+        image_uri: Container image URI from `AppSpecification.ImageUri`,
+            populated by `_describe_processing_job`.
+        tags: Resource tags, populated by `_list_tags_for_resource`.
+    """
+
+    name: str
+    region: str
+    arn: str
+    image_uri: Optional[str] = None
+    tags: Optional[list] = []
+
+
 class ProductionVariant(BaseModel):
     name: str
     initial_instance_count: int

tests/providers/aws/services/sagemaker/sagemaker_clarify_exists/sagemaker_clarify_exists_test.py

@@ -0,0 +1,247 @@
+from unittest import mock
+
+from prowler.providers.aws.services.sagemaker.sagemaker_service import ProcessingJob
+from tests.providers.aws.utils import (
+    AWS_ACCOUNT_NUMBER,
+    AWS_REGION_EU_WEST_1,
+    AWS_REGION_US_EAST_1,
+    set_mocked_aws_provider,
+)
+
+CLARIFY_IMAGE_URI = f"{AWS_ACCOUNT_NUMBER}.dkr.ecr.{AWS_REGION_US_EAST_1}.amazonaws.com/sagemaker-clarify-processing:1.0"
+NON_CLARIFY_IMAGE_URI = f"{AWS_ACCOUNT_NUMBER}.dkr.ecr.{AWS_REGION_US_EAST_1}.amazonaws.com/sagemaker-xgboost:1.0"
+CUSTOM_CLARIFY_IMAGE_URI = f"{AWS_ACCOUNT_NUMBER}.dkr.ecr.{AWS_REGION_US_EAST_1}.amazonaws.com/my-clarify-thing:1.0"
+PROCESSING_JOB_ARN = f"arn:aws:sagemaker:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:processing-job/clarify-job"
+
+
+class Test_sagemaker_clarify_exists:
+    def test_no_processing_jobs_no_scanned_regions(self):
+        sagemaker_client = mock.MagicMock
+        sagemaker_client.sagemaker_processing_jobs = []
+        sagemaker_client.processing_jobs_scanned_regions = set()
+
+        aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
+
+        with (
+            mock.patch(
+                "prowler.providers.common.provider.Provider.get_global_provider",
+                return_value=aws_provider,
+            ),
+            mock.patch(
+                "prowler.providers.aws.services.sagemaker.sagemaker_clarify_exists.sagemaker_clarify_exists.sagemaker_client",
+                sagemaker_client,
+            ),
+        ):
+            from prowler.providers.aws.services.sagemaker.sagemaker_clarify_exists.sagemaker_clarify_exists import (
+                sagemaker_clarify_exists,
+            )
+
+            check = sagemaker_clarify_exists()
+            result = check.execute()
+            assert len(result) == 0
+
+    def test_no_processing_jobs_region_scanned(self):
+        sagemaker_client = mock.MagicMock
+        sagemaker_client.sagemaker_processing_jobs = []
+        sagemaker_client.processing_jobs_scanned_regions = {AWS_REGION_US_EAST_1}
+        sagemaker_client.audited_partition = "aws"
+        sagemaker_client.audited_account = AWS_ACCOUNT_NUMBER
+
+        aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
+
+        with (
+            mock.patch(
+                "prowler.providers.common.provider.Provider.get_global_provider",
+                return_value=aws_provider,
+            ),
+            mock.patch(
+                "prowler.providers.aws.services.sagemaker.sagemaker_clarify_exists.sagemaker_clarify_exists.sagemaker_client",
+                sagemaker_client,
+            ),
+        ):
+            from prowler.providers.aws.services.sagemaker.sagemaker_clarify_exists.sagemaker_clarify_exists import (
+                sagemaker_clarify_exists,
+            )
+
+            check = sagemaker_clarify_exists()
+            result = check.execute()
+            assert len(result) == 1
+            assert result[0].status == "FAIL"
+            assert (
+                result[0].status_extended
+                == f"No SageMaker Clarify processing jobs found in region {AWS_REGION_US_EAST_1}."
+            )
+            assert result[0].resource_id == "sagemaker-clarify"
+
+    def test_non_clarify_processing_job(self):
+        sagemaker_client = mock.MagicMock
+        sagemaker_client.sagemaker_processing_jobs = [
+            ProcessingJob(
+                name="xgboost-job",
+                arn=f"arn:aws:sagemaker:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:processing-job/xgboost-job",
+                region=AWS_REGION_US_EAST_1,
+                image_uri=NON_CLARIFY_IMAGE_URI,
+            )
+        ]
+        sagemaker_client.processing_jobs_scanned_regions = {AWS_REGION_US_EAST_1}
+        sagemaker_client.audited_partition = "aws"
+        sagemaker_client.audited_account = AWS_ACCOUNT_NUMBER
+
+        aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
+
+        with (
+            mock.patch(
+                "prowler.providers.common.provider.Provider.get_global_provider",
+                return_value=aws_provider,
+            ),
+            mock.patch(
+                "prowler.providers.aws.services.sagemaker.sagemaker_clarify_exists.sagemaker_clarify_exists.sagemaker_client",
+                sagemaker_client,
+            ),
+        ):
+            from prowler.providers.aws.services.sagemaker.sagemaker_clarify_exists.sagemaker_clarify_exists import (
+                sagemaker_clarify_exists,
+            )
+
+            check = sagemaker_clarify_exists()
+            result = check.execute()
+            assert len(result) == 1
+            assert result[0].status == "FAIL"
+            assert (
+                result[0].status_extended
+                == f"No SageMaker Clarify processing jobs found in region {AWS_REGION_US_EAST_1}."
+            )
+
+    def test_custom_image_with_clarify_in_name_does_not_match(self):
+        sagemaker_client = mock.MagicMock
+        sagemaker_client.sagemaker_processing_jobs = [
+            ProcessingJob(
+                name="my-clarify-thing-job",
+                arn=f"arn:aws:sagemaker:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:processing-job/my-clarify-thing-job",
+                region=AWS_REGION_US_EAST_1,
+                image_uri=CUSTOM_CLARIFY_IMAGE_URI,
+            )
+        ]
+        sagemaker_client.processing_jobs_scanned_regions = {AWS_REGION_US_EAST_1}
+        sagemaker_client.audited_partition = "aws"
+        sagemaker_client.audited_account = AWS_ACCOUNT_NUMBER
+
+        aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
+
+        with (
+            mock.patch(
+                "prowler.providers.common.provider.Provider.get_global_provider",
+                return_value=aws_provider,
+            ),
+            mock.patch(
+                "prowler.providers.aws.services.sagemaker.sagemaker_clarify_exists.sagemaker_clarify_exists.sagemaker_client",
+                sagemaker_client,
+            ),
+        ):
+            from prowler.providers.aws.services.sagemaker.sagemaker_clarify_exists.sagemaker_clarify_exists import (
+                sagemaker_clarify_exists,
+            )
+
+            check = sagemaker_clarify_exists()
+            result = check.execute()
+            assert len(result) == 1
+            assert result[0].status == "FAIL"
+            assert (
+                result[0].status_extended
+                == f"No SageMaker Clarify processing jobs found in region {AWS_REGION_US_EAST_1}."
+            )
+
+    def test_clarify_processing_job_exists(self):
+        sagemaker_client = mock.MagicMock
+        sagemaker_client.sagemaker_processing_jobs = [
+            ProcessingJob(
+                name="clarify-job",
+                arn=PROCESSING_JOB_ARN,
+                region=AWS_REGION_US_EAST_1,
+                image_uri=CLARIFY_IMAGE_URI,
+            )
+        ]
+        sagemaker_client.processing_jobs_scanned_regions = {AWS_REGION_US_EAST_1}
+
+        aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
+
+        with (
+            mock.patch(
+                "prowler.providers.common.provider.Provider.get_global_provider",
+                return_value=aws_provider,
+            ),
+            mock.patch(
+                "prowler.providers.aws.services.sagemaker.sagemaker_clarify_exists.sagemaker_clarify_exists.sagemaker_client",
+                sagemaker_client,
+            ),
+        ):
+            from prowler.providers.aws.services.sagemaker.sagemaker_clarify_exists.sagemaker_clarify_exists import (
+                sagemaker_clarify_exists,
+            )
+
+            check = sagemaker_clarify_exists()
+            result = check.execute()
+            assert len(result) == 1
+            assert result[0].status == "PASS"
+            assert (
+                result[0].status_extended
+                == f"SageMaker Clarify processing job clarify-job exists in region {AWS_REGION_US_EAST_1}."
+            )
+            assert result[0].resource_id == "clarify-job"
+            assert result[0].resource_arn == PROCESSING_JOB_ARN
+
+    def test_mixed_regions(self):
+        sagemaker_client = mock.MagicMock
+        sagemaker_client.sagemaker_processing_jobs = [
+            ProcessingJob(
+                name="clarify-job",
+                arn=PROCESSING_JOB_ARN,
+                region=AWS_REGION_US_EAST_1,
+                image_uri=CLARIFY_IMAGE_URI,
+            )
+        ]
+        sagemaker_client.processing_jobs_scanned_regions = {
+            AWS_REGION_US_EAST_1,
+            AWS_REGION_EU_WEST_1,
+        }
+        sagemaker_client.audited_partition = "aws"
+        sagemaker_client.audited_account = AWS_ACCOUNT_NUMBER
+
+        aws_provider = set_mocked_aws_provider(
+            [AWS_REGION_US_EAST_1, AWS_REGION_EU_WEST_1]
+        )
+
+        with (
+            mock.patch(
+                "prowler.providers.common.provider.Provider.get_global_provider",
+                return_value=aws_provider,
+            ),
+            mock.patch(
+                "prowler.providers.aws.services.sagemaker.sagemaker_clarify_exists.sagemaker_clarify_exists.sagemaker_client",
+                sagemaker_client,
+            ),
+        ):
+            from prowler.providers.aws.services.sagemaker.sagemaker_clarify_exists.sagemaker_clarify_exists import (
+                sagemaker_clarify_exists,
+            )
+
+            check = sagemaker_clarify_exists()
+            result = check.execute()
+
+            assert len(result) == 2
+
+            results_by_region = {r.region: r for r in result}
+
+            us_result = results_by_region[AWS_REGION_US_EAST_1]
+            assert us_result.status == "PASS"
+            assert (
+                us_result.status_extended
+                == f"SageMaker Clarify processing job clarify-job exists in region {AWS_REGION_US_EAST_1}."
+            )
+
+            eu_result = results_by_region[AWS_REGION_EU_WEST_1]
+            assert eu_result.status == "FAIL"
+            assert (
+                eu_result.status_extended
+                == f"No SageMaker Clarify processing jobs found in region {AWS_REGION_EU_WEST_1}."
+            )

tests/providers/aws/services/sagemaker/sagemaker_service_test.py

@@ -396,13 +396,13 @@ class Test_SageMaker_Service:
                 sagemaker_service = SageMaker(audit_info)
 
                 # Check that __threading_call__ was called for _list_tags_for_resource
-                # (one for each resource type: models, notebooks, training jobs, endpoint configs, domains)
+                # (one for each resource type: models, notebooks, training jobs, processing jobs, endpoint configs, domains)
                 tag_calls = [
                     c
                     for c in mock_threading_call.call_args_list
                     if c[0][0] == sagemaker_service._list_tags_for_resource
                 ]
-                assert len(tag_calls) == 5
+                assert len(tag_calls) == 6
 
     # Test SageMaker list model package groups
     def test_list_model_package_groups(self):