feat(aws): add Bedrock model invocation job CMK check

prowler/providers/aws/services/bedrock/bedrock_model_invocation_job_output_encrypted_with_cmk/bedrock_model_invocation_job_output_encrypted_with_cmk.metadata.json

@@ -0,0 +1,41 @@
+{
+  "Provider": "aws",
+  "CheckID": "bedrock_model_invocation_job_output_encrypted_with_cmk",
+  "CheckTitle": "Bedrock model invocation job output is encrypted with a customer-managed KMS key",
+  "CheckType": [
+    "Software and Configuration Checks/AWS Security Best Practices",
+    "Software and Configuration Checks/AWS Security Best Practices/Data Encryption"
+  ],
+  "ServiceName": "bedrock",
+  "SubServiceName": "",
+  "ResourceIdTemplate": "",
+  "Severity": "high",
+  "ResourceType": "Other",
+  "ResourceGroup": "ai_ml",
+  "Description": "Ensure that the S3 output of Amazon Bedrock model invocation jobs is configured to use a customer-managed KMS key.",
+  "Risk": "Model invocation job outputs may contain sensitive or confidential data. Without customer-managed KMS key encryption, organizations have less control over the encryption key protecting these S3 outputs.",
+  "RelatedUrl": "",
+  "AdditionalURLs": [
+    "https://docs.aws.amazon.com/bedrock/latest/APIReference/API_GetModelInvocationJob.html",
+    "https://docs.aws.amazon.com/bedrock/latest/APIReference/API_ModelInvocationJobS3OutputDataConfig.html"
+  ],
+  "Remediation": {
+    "Code": {
+      "CLI": "",
+      "NativeIaC": "",
+      "Other": "Configure the Bedrock model invocation job output S3 configuration to use a customer-managed AWS KMS key.",
+      "Terraform": ""
+    },
+    "Recommendation": {
+      "Text": "Configure the S3 output of the Bedrock model invocation job to use a customer-managed KMS key.",
+      "Url": "https://hub.prowler.com/check/bedrock_model_invocation_job_output_encrypted_with_cmk"
+    }
+  },
+  "Categories": [
+    "gen-ai",
+    "encryption"
+  ],
+  "DependsOn": [],
+  "RelatedTo": [],
+  "Notes": "The check passes when s3EncryptionKeyId is present in the model invocation job S3 output configuration. If the required job details cannot be retrieved, the finding is reported as MANUAL."
+}

prowler/providers/aws/services/bedrock/bedrock_model_invocation_job_output_encrypted_with_cmk/bedrock_model_invocation_job_output_encrypted_with_cmk.py

@@ -0,0 +1,83 @@
+from prowler.lib.check.models import Check, Check_Report_AWS
+from prowler.providers.aws.services.bedrock.bedrock_client import (
+    bedrock_client,
+)
+
+
+class bedrock_model_invocation_job_output_encrypted_with_cmk(Check):
+    """Ensure Bedrock model invocation job outputs use a customer-managed KMS
+    key."""
+
+    def execute(self) -> list[Check_Report_AWS]:
+        """Execute the check.
+
+        Returns:
+            A list of reports containing the result of the check.
+        """
+        findings = []
+
+        # If listing model invocation jobs failed in a region,
+        # we cannot determine the encryption status of jobs in that region.
+        for region, error in sorted(
+            bedrock_client.model_invocation_jobs_scan_errors.items()
+        ):
+            report = Check_Report_AWS(
+                metadata=self.metadata(),
+                resource={"region": region},
+            )
+            report.region = region
+            report.resource_id = "model-invocation-job/unknown"
+            report.resource_arn = (
+                f"arn:{bedrock_client.audited_partition}:bedrock:"
+                f"{region}:{bedrock_client.audited_account}:"
+                "model-invocation-job/unknown"
+            )
+            report.status = "MANUAL"
+            report.status_extended = (
+                f"Bedrock model invocation jobs could not be listed in region "
+                f"{region} ({error}); verify manually that every model "
+                "invocation job output uses a customer-managed KMS key."
+            )
+            findings.append(report)
+
+        # Evaluate each discovered model invocation job.
+        for job in bedrock_client.model_invocation_jobs.values():
+            report = Check_Report_AWS(
+                metadata=self.metadata(),
+                resource=job,
+            )
+
+            # If GetModelInvocationJob failed, the encryption configuration
+            # cannot be determined. Never report PASS in this situation.
+            if not job.detail_retrieved:
+                report.status = "MANUAL"
+                report.status_extended = (
+                    f"Bedrock model invocation job {job.name} "
+                    f"output encryption configuration could not be "
+                    f"retrieved in region {job.region}; verify manually "
+                    "that the S3 output uses a customer-managed KMS key."
+                )
+
+            # A present s3EncryptionKeyId indicates that the S3 output
+            # configuration specifies a KMS encryption key.
+            elif job.s3_encryption_key_id:
+                report.status = "PASS"
+                report.status_extended = (
+                    f"Bedrock model invocation job {job.name} S3 output "
+                    f"is configured with a customer-managed KMS key in "
+                    f"region {job.region}."
+                )
+
+            # Missing or empty s3EncryptionKeyId means the required
+            # customer-managed KMS key is not configured.
+            else:
+                report.status = "FAIL"
+                report.status_extended = (
+                    f"Bedrock model invocation job {job.name} S3 output "
+                    f"is not configured with a customer-managed KMS key "
+                    f"in region {job.region}."
+                )
+
+            findings.append(report)
+
+        return findings

prowler/providers/aws/services/bedrock/bedrock_service.py

@@ -16,6 +16,8 @@ def __init__(self, provider):
         self.guardrails = {}
         self.guardrails_scanned_regions = set()
         self.guardrails_scan_errors = {}
+        self.model_invocation_jobs = {}
+        self.model_invocation_jobs_scan_errors = {}
         self.custom_models = {}
         self.custom_models_scan_errors = {}
         self.__threading_call__(self._get_model_invocation_logging_configuration)
@@ -24,6 +26,10 @@ def __init__(self, provider):
         self.__threading_call__(self._list_tags_for_resource, self.guardrails.values())
         self.__threading_call__(self._list_custom_models)
         self.__threading_call__(self._get_custom_model, self.custom_models.values())
+        self.__threading_call__(self._list_model_invocation_jobs)
+        self.__threading_call__(
+            self._get_model_invocation_job, self.model_invocation_jobs.values()
+        )
 
     def _get_model_invocation_logging_arn_template(self, region):
         return (
@@ -59,6 +65,62 @@ def _get_model_invocation_logging_configuration(self, regional_client):
                 f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
             )
 
+    def _list_model_invocation_jobs(self, regional_client):
+        """List Bedrock model invocation jobs in a region."""
+        logger.info("Bedrock - Listing Model Invocation Jobs...")
+        try:
+            paginator = regional_client.get_paginator("list_model_invocation_jobs")
+            for page in paginator.paginate():
+                for job in page.get("invocationJobSummaries", []):
+                    job_arn = job.get("jobArn", "")
+                    if job_arn and (
+                        not self.audit_resources
+                        or is_resource_filtered(job_arn, self.audit_resources)
+                    ):
+                        self.model_invocation_jobs[job_arn] = ModelInvocationJob(
+                            name=job.get("jobName", ""),
+                            arn=job_arn,
+                            region=regional_client.region,
+                        )
+        except ClientError as error:
+            code = error.response["Error"].get("Code", error.__class__.__name__)
+            if code != "ValidationException":
+                self.model_invocation_jobs_scan_errors[regional_client.region] = code
+            logger.error(
+                f"{regional_client.region} -- "
+                f"{error.__class__.__name__}"
+                f"[{error.__traceback__.tb_lineno}]: {error}"
+            )
+        except Exception as error:
+            self.model_invocation_jobs_scan_errors[regional_client.region] = (
+                error.__class__.__name__
+            )
+            logger.error(
+                f"{regional_client.region} -- "
+                f"{error.__class__.__name__}"
+                f"[{error.__traceback__.tb_lineno}]: {error}"
+            )
+
+    def _get_model_invocation_job(self, job):
+        """Fetch S3 output encryption for a model invocation job."""
+        logger.info("Bedrock - Getting Model Invocation Job...")
+        try:
+            job_info = self.regional_clients[job.region].get_model_invocation_job(
+                jobIdentifier=job.arn
+            )
+            job.s3_encryption_key_id = (
+                job_info.get("outputDataConfig", {})
+                .get("s3OutputDataConfig", {})
+                .get("s3EncryptionKeyId")
+            )
+            job.detail_retrieved = True
+        except Exception as error:
+            logger.error(
+                f"{job.region} -- "
+                f"{error.__class__.__name__}"
+                f"[{error.__traceback__.tb_lineno}]: {error}"
+            )
+
     def _list_guardrails(self, regional_client):
         """List the guardrails in a region."""
         logger.info("Bedrock - Listing Guardrails...")
@@ -240,6 +302,16 @@ class CustomModel(BaseModel):
     detail_retrieved: bool = False
 
 
+class ModelInvocationJob(BaseModel):
+    """Model representing a Bedrock model invocation job."""
+
+    name: str
+    arn: str
+    region: str
+    s3_encryption_key_id: Optional[str] = None
+    detail_retrieved: bool = False
+
+
 class BedrockAgent(AWSService):
     """Bedrock Agent service class for managing agents and prompts."""