feat(sagemaker): add processing and transform job volume CMK checks

prowler/changelog.d/sagemaker-processing-job-volume-encrypted-with-cmk.added.md

@@ -0,0 +1 @@
+`sagemaker_processing_job_volume_encrypted_with_cmk` checks that every SageMaker processing job encrypts its ML storage volume with a customer-managed KMS key (`ProcessingResources.ClusterConfig.VolumeKmsKeyId`), and reports MANUAL when the job details cannot be described

prowler/changelog.d/sagemaker-transform-job-volume-encrypted-with-cmk.added.md

@@ -0,0 +1 @@
+`sagemaker_transform_job_volume_encrypted_with_cmk` checks that every SageMaker transform job encrypts its ML storage volume with a customer-managed KMS key (`TransformResources.VolumeKmsKeyId`), and reports MANUAL when the job details cannot be described

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

@@ -0,0 +1,45 @@
+{
+  "Provider": "aws",
+  "CheckID": "sagemaker_processing_job_volume_encrypted_with_cmk",
+  "CheckTitle": "SageMaker processing job volumes use customer-managed KMS keys",
+  "CheckType": [
+    "Software and Configuration Checks/AWS Security Best Practices",
+    "Effects/Data Exposure"
+  ],
+  "ServiceName": "sagemaker",
+  "SubServiceName": "",
+  "ResourceIdTemplate": "",
+  "Severity": "high",
+  "ResourceType": "Other",
+  "ResourceGroup": "ai_ml",
+  "Description": "**Amazon SageMaker processing jobs** are assessed for **at-rest encryption** of their attached ML storage volumes using a **customer-managed KMS key**. The finding reflects whether `ProcessingResources.ClusterConfig.VolumeKmsKeyId` is set on the job.",
+  "Risk": "A processing job without `VolumeKmsKeyId` falls back to an **AWS managed key**, which cannot carry a custom key policy and whose **rotation**, **access** and **lifecycle** are outside the account owner's control. Intermediate processing data on the volume then cannot be independently revoked, and grants to that data cannot be audited or restricted through a key policy.",
+  "RelatedUrl": "",
+  "AdditionalURLs": [
+    "https://docs.aws.amazon.com/sagemaker/latest/dg/key-management.html",
+    "https://docs.aws.amazon.com/sagemaker/latest/APIReference/API_DescribeProcessingJob.html",
+    "https://docs.aws.amazon.com/sagemaker/latest/APIReference/API_ProcessingClusterConfig.html"
+  ],
+  "Remediation": {
+    "Code": {
+      "CLI": "aws sagemaker create-processing-job --processing-job-name <job_name> --role-arn <role_arn> --app-specification ImageUri=<image_uri> --processing-resources 'ClusterConfig={InstanceCount=1,InstanceType=ml.m5.xlarge,VolumeSizeInGB=30,VolumeKmsKeyId=<kms_key_id>}' --processing-output-config 'Outputs=[{OutputName=output-1,S3Output={S3Uri=s3://<bucket>/output,LocalPath=/opt/ml/processing/output,S3UploadMode=EndOfJob}}]' --processing-inputs '[{InputName=input-1,S3Input={S3Uri=s3://<bucket>/input,LocalPath=/opt/ml/processing/input,S3DataType=S3Prefix,S3InputMode=File}}]'",
+      "NativeIaC": "```yaml\n# CloudFormation: SageMaker ProcessingJob with volume CMEK\nResources:\n  <example_resource_name>:\n    Type: AWS::SageMaker::ProcessingJob\n    Properties:\n      ProcessingJobName: <example_resource_name>\n      RoleArn: <example_role_arn>\n      AppSpecification:\n        ImageUri: <example_ecr_image_uri>\n      ProcessingResources:\n        ClusterConfig:\n          InstanceCount: 1\n          InstanceType: ml.m5.xlarge\n          VolumeSizeInGB: 30\n          VolumeKmsKeyId: <example_kms_key_arn>  # Critical: encrypts the processing EBS volume with a customer-managed key\n      ProcessingOutputConfig:\n        Outputs:\n          - OutputName: output-1\n            S3Output:\n              S3Uri: s3://<example_bucket>/output\n              LocalPath: /opt/ml/processing/output\n              S3UploadMode: EndOfJob\n```",
+      "Other": "1. In the AWS console, go to SageMaker > Processing > Processing jobs\n2. Create a new processing job (existing jobs cannot change VolumeKmsKeyId)\n3. Under Resource configuration, set Volume encryption key to your customer-managed KMS key\n4. Complete required fields and create the job\n5. Verify DescribeProcessingJob shows ProcessingResources.ClusterConfig.VolumeKmsKeyId",
+      "Terraform": "```hcl\n# SageMaker Processing Job with volume CMEK\n# Note: aws_sagemaker_processing_job is not a first-class Terraform resource in all\n# providers; use the AWS CLI/SDK or CloudFormation equivalent with VolumeKmsKeyId set:\n# ProcessingResources.ClusterConfig.VolumeKmsKeyId = \"<example_kms_key_arn>\"\n```"
+    },
+    "Recommendation": {
+      "Text": "Set `VolumeKmsKeyId` on every SageMaker processing job using a **customer-managed KMS key** with a least-privilege key policy and **rotation** enabled. Existing jobs cannot change the key; recreate non-compliant jobs.",
+      "Url": "https://hub.prowler.com/check/sagemaker_processing_job_volume_encrypted_with_cmk"
+    }
+  },
+  "Categories": [
+    "encryption",
+    "gen-ai"
+  ],
+  "DependsOn": [],
+  "RelatedTo": [
+    "sagemaker_training_jobs_volume_and_output_encryption_enabled",
+    "sagemaker_transform_job_volume_encrypted_with_cmk"
+  ],
+  "Notes": "Reports MANUAL when DescribeProcessingJob fails so an unread job is never treated as compliant. VolumeKmsKeyId lives under ProcessingResources.ClusterConfig."
+}

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

@@ -0,0 +1,71 @@
+from prowler.lib.check.models import Check, Check_Report_AWS
+from prowler.providers.aws.services.sagemaker.sagemaker_client import sagemaker_client
+
+
+class sagemaker_processing_job_volume_encrypted_with_cmk(Check):
+    """Ensure SageMaker processing job volumes use a customer-managed KMS key.
+
+    Processing jobs write intermediate data to an attached ML storage volume.
+    When ``ProcessingResources.ClusterConfig.VolumeKmsKeyId`` is unset the
+    volume falls back to an AWS-managed key, which cannot carry a custom key
+    policy and whose rotation, access and lifecycle are outside the account
+    owner's control.
+
+    - PASS: ``VolumeKmsKeyId`` is present on the processing job.
+    - FAIL: ``VolumeKmsKeyId`` is absent after a successful describe.
+    - MANUAL: ``DescribeProcessingJob`` failed, or ``ListProcessingJobs`` failed
+      for a region, so encryption cannot be determined either way.
+    """
+
+    def execute(self) -> list[Check_Report_AWS]:
+        """Execute the SageMaker processing job volume CMK encryption check.
+
+        Returns:
+            One report per processing job in the inventory, plus one MANUAL
+            report per region where ``ListProcessingJobs`` failed.
+        """
+        findings = []
+
+        # Distinguishes a failed ListProcessingJobs call from a genuinely empty
+        # inventory: failed regions must not silently produce zero findings.
+        for region in sorted(sagemaker_client.processing_jobs_list_failed_regions):
+            report = Check_Report_AWS(metadata=self.metadata(), resource={})
+            report.region = region
+            report.resource_id = "sagemaker-processing-jobs"
+            report.resource_arn = (
+                f"arn:{sagemaker_client.audited_partition}:sagemaker:{region}:"
+                f"{sagemaker_client.audited_account}:processing-job"
+            )
+            report.status = "MANUAL"
+            report.status_extended = (
+                f"SageMaker processing job inventory could not be listed in "
+                f"region {region}; volume encryption cannot be verified."
+            )
+            findings.append(report)
+
+        for processing_job in sagemaker_client.sagemaker_processing_jobs:
+            report = Check_Report_AWS(
+                metadata=self.metadata(), resource=processing_job
+            )
+            if processing_job.detail_fetch_error:
+                report.status = "MANUAL"
+                report.status_extended = (
+                    f"SageMaker processing job {processing_job.name} details could "
+                    f"not be described ({processing_job.detail_fetch_error}); "
+                    "volume encryption cannot be verified."
+                )
+            elif processing_job.volume_kms_key_id:
+                report.status = "PASS"
+                report.status_extended = (
+                    f"SageMaker processing job {processing_job.name} encrypts its "
+                    f"volume with the customer-managed KMS key "
+                    f"{processing_job.volume_kms_key_id}."
+                )
+            else:
+                report.status = "FAIL"
+                report.status_extended = (
+                    f"SageMaker processing job {processing_job.name} does not "
+                    "encrypt its volume with a customer-managed KMS key."
+                )
+            findings.append(report)
+        return findings

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

@@ -18,6 +18,10 @@ def __init__(self, provider):
         self.sagemaker_training_jobs = []
         self.sagemaker_processing_jobs = []
         self.processing_jobs_scanned_regions = set()
+        self.processing_jobs_list_failed_regions = set()
+        self.sagemaker_transform_jobs = []
+        self.transform_jobs_scanned_regions = set()
+        self.transform_jobs_list_failed_regions = set()
         self.sagemaker_domains = []
         self.endpoint_configs = {}
         self.sagemaker_model_registries = []
@@ -28,6 +32,7 @@ def __init__(self, provider):
         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_transform_jobs)
         self.__threading_call__(self._list_endpoint_configs)
         self.__threading_call__(self._list_domains)
         self.__threading_call__(self._list_model_package_groups)
@@ -49,6 +54,9 @@ def __init__(self, provider):
         self.__threading_call__(
             self._describe_processing_job, self.sagemaker_processing_jobs
         )
+        self.__threading_call__(
+            self._describe_transform_job, self.sagemaker_transform_jobs
+        )
         self.__threading_call__(
             self._describe_endpoint_config, list(self.endpoint_configs.values())
         )
@@ -66,6 +74,9 @@ def __init__(self, provider):
         self.__threading_call__(
             self._list_tags_for_resource, self.sagemaker_processing_jobs
         )
+        self.__threading_call__(
+            self._list_tags_for_resource, self.sagemaker_transform_jobs
+        )
         self.__threading_call__(
             self._list_tags_for_resource, list(self.endpoint_configs.values())
         )
@@ -148,9 +159,10 @@ def _list_processing_jobs(self, regional_client):
 
         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.
+        ``self.processing_jobs_scanned_regions`` once pagination succeeds.
+        Regions where ``ListProcessingJobs`` fails are recorded in
+        ``self.processing_jobs_list_failed_regions`` so checks can emit MANUAL
+        instead of treating a failed inventory as empty.
 
         Args:
             regional_client: Regional SageMaker boto3 client.
@@ -176,16 +188,19 @@ def _list_processing_jobs(self, regional_client):
                         )
             self.processing_jobs_scanned_regions.add(regional_client.region)
         except Exception as error:
+            self.processing_jobs_list_failed_regions.add(regional_client.region)
             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.
+        """Describe a SageMaker processing job and enrich inventory fields.
 
-        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.
+        Reads ``AppSpecification.ImageUri`` and
+        ``ProcessingResources.ClusterConfig.VolumeKmsKeyId`` from
+        ``DescribeProcessingJob``. On API failure, sets
+        ``processing_job.detail_fetch_error`` so checks can emit MANUAL rather
+        than asserting compliance from an unread response.
 
         Args:
             processing_job: ProcessingJob model to enrich in-place.
@@ -198,7 +213,12 @@ def _describe_processing_job(self, processing_job):
             )
             app_spec = describe_processing_job.get("AppSpecification", {})
             processing_job.image_uri = app_spec.get("ImageUri")
+            cluster_config = describe_processing_job.get(
+                "ProcessingResources", {}
+            ).get("ClusterConfig", {})
+            processing_job.volume_kms_key_id = cluster_config.get("VolumeKmsKeyId")
         except Exception as error:
+            processing_job.detail_fetch_error = error.__class__.__name__
             logger.error(
                 f"{processing_job.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
             )
@@ -436,6 +456,71 @@ def _list_model_package_groups(self, regional_client):
             )
         )
 
+
+    def _list_transform_jobs(self, regional_client):
+        """List SageMaker transform jobs in a region.
+
+        Populates ``self.sagemaker_transform_jobs`` with `TransformJob`
+        entries and adds ``regional_client.region`` to
+        ``self.transform_jobs_scanned_regions`` once pagination succeeds.
+        Regions where ``ListTransformJobs`` fails are recorded in
+        ``self.transform_jobs_list_failed_regions`` so checks can emit MANUAL
+        instead of treating a failed inventory as empty.
+
+        Args:
+            regional_client: Regional SageMaker boto3 client.
+        """
+        logger.info("SageMaker - listing transform jobs...")
+        try:
+            list_transform_jobs_paginator = regional_client.get_paginator(
+                "list_transform_jobs"
+            )
+            for page in list_transform_jobs_paginator.paginate():
+                for transform_job in page["TransformJobSummaries"]:
+                    if not self.audit_resources or (
+                        is_resource_filtered(
+                            transform_job["TransformJobArn"], self.audit_resources
+                        )
+                    ):
+                        self.sagemaker_transform_jobs.append(
+                            TransformJob(
+                                name=transform_job["TransformJobName"],
+                                region=regional_client.region,
+                                arn=transform_job["TransformJobArn"],
+                            )
+                        )
+            self.transform_jobs_scanned_regions.add(regional_client.region)
+        except Exception as error:
+            self.transform_jobs_list_failed_regions.add(regional_client.region)
+            logger.error(
+                f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
+            )
+
+    def _describe_transform_job(self, transform_job):
+        """Describe a SageMaker transform job and retain volume KMS key state.
+
+        Reads ``TransformResources.VolumeKmsKeyId`` from
+        ``DescribeTransformJob``. On API failure, sets
+        ``transform_job.detail_fetch_error`` so checks can emit MANUAL rather
+        than asserting compliance from an unread response.
+
+        Args:
+            transform_job: TransformJob model to enrich in-place.
+        """
+        logger.info("SageMaker - describing transform job...")
+        try:
+            regional_client = self.regional_clients[transform_job.region]
+            describe_transform_job = regional_client.describe_transform_job(
+                TransformJobName=transform_job.name
+            )
+            transform_resources = describe_transform_job.get("TransformResources", {})
+            transform_job.volume_kms_key_id = transform_resources.get("VolumeKmsKeyId")
+        except Exception as error:
+            transform_job.detail_fetch_error = error.__class__.__name__
+            logger.error(
+                f"{transform_job.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
+            )
+
     def _list_tags_for_resource(self, resource):
         """
         Lists tags for a specific SageMaker resource.
@@ -630,13 +715,44 @@ class ProcessingJob(BaseModel):
         arn: Processing job ARN.
         image_uri: Container image URI from `AppSpecification.ImageUri`,
             populated by `_describe_processing_job`.
+        volume_kms_key_id: KMS key ID from
+            `ProcessingResources.ClusterConfig.VolumeKmsKeyId`.
+        detail_fetch_error: Exception class name when DescribeProcessingJob
+            failed. Distinguishes "no VolumeKmsKeyId" (None + no error) from
+            "details could not be described" (None + error). Checks that assert
+            volume encryption should emit MANUAL when this is set.
         tags: Resource tags, populated by `_list_tags_for_resource`.
     """
 
     name: str
     region: str
     arn: str
     image_uri: Optional[str] = None
+    volume_kms_key_id: Optional[str] = None
+    detail_fetch_error: Optional[str] = None
+    tags: Optional[list] = []
+
+
+class TransformJob(BaseModel):
+    """Represents a SageMaker batch transform job.
+
+    Attributes:
+        name: Transform job name.
+        region: AWS region where the job lives.
+        arn: Transform job ARN.
+        volume_kms_key_id: KMS key ID from `TransformResources.VolumeKmsKeyId`.
+        detail_fetch_error: Exception class name when DescribeTransformJob
+            failed. Distinguishes "no VolumeKmsKeyId" (None + no error) from
+            "details could not be described" (None + error). Checks that assert
+            volume encryption should emit MANUAL when this is set.
+        tags: Resource tags, populated by `_list_tags_for_resource`.
+    """
+
+    name: str
+    region: str
+    arn: str
+    volume_kms_key_id: Optional[str] = None
+    detail_fetch_error: Optional[str] = None
     tags: Optional[list] = []