feat(bedrock): add checks for guardrails configuration and log encryption

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

@@ -0,0 +1,32 @@
+{
+  "Provider": "aws",
+  "CheckID": "bedrock_guardrail_prompt_attack_filter_enabled",
+  "CheckTitle": "Configure Prompt Attack Filter with the highest strength for Amazon Bedrock Guardrails.",
+  "CheckType": [],
+  "ServiceName": "bedrock",
+  "SubServiceName": "",
+  "ResourceIdTemplate": "arn:partition:bedrock:region:account-id:guardrails/resource-id",
+  "Severity": "high",
+  "ResourceType": "Other",
+  "Description": "Ensure that prompt attack filter strength is set to HIGH for Amazon Bedrock guardrails to mitigate prompt injection and bypass techniques.",
+  "Risk": "If prompt attack filter strength is not set to HIGH, Bedrock models may be more vulnerable to prompt injection attacks or jailbreak attempts, which could allow harmful or sensitive content to bypass filters and reach end users.",
+  "RelatedUrl": "https://docs.aws.amazon.com/bedrock/latest/userguide/guardrails.html",
+  "Remediation": {
+    "Code": {
+      "CLI": "aws bedrock put-guardrails-configuration --guardrails-config 'promptAttackStrength=HIGH'",
+      "NativeIaC": "",
+      "Other": "https://www.trendmicro.com/cloudoneconformity/knowledge-base/aws/Bedrock/prompt-attack-strength.html",
+      "Terraform": ""
+    },
+    "Recommendation": {
+      "Text": "Set the prompt attack filter strength to HIGH for Amazon Bedrock guardrails to prevent prompt injection attacks and ensure robust protection against content manipulation.",
+      "Url": "https://docs.aws.amazon.com/bedrock/latest/userguide/prompt-injection.html"
+    }
+  },
+  "Categories": [
+    "security"
+  ],
+  "DependsOn": [],
+  "RelatedTo": [],
+  "Notes": "Ensure that prompt attack protection is set to the highest strength to minimize the risk of prompt injection attacks."
+}

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

@@ -0,0 +1,24 @@
+from prowler.lib.check.models import Check, Check_Report_AWS
+from prowler.providers.aws.services.bedrock.bedrock_client import bedrock_client
+
+
+class bedrock_guardrail_prompt_attack_filter_enabled(Check):
+    def execute(self):
+        findings = []
+        for guardrail in bedrock_client.guardrails.values():
+            report = Check_Report_AWS(self.metadata())
+            report.region = guardrail.region
+            report.resource_id = guardrail.id
+            report.resource_arn = guardrail.arn
+            report.resource_tags = guardrail.tags
+            report.status = "PASS"
+            report.status_extended = f"Bedrock Guardrail {guardrail.name} is configured to detect and block prompt attacks with a HIGH strength."
+            if not guardrail.prompt_attack_filter_strength:
+                report.status = "FAIL"
+                report.status_extended = f"Bedrock Guardrail {guardrail.name} is not configured to block prompt attacks."
+            elif guardrail.prompt_attack_filter_strength != "HIGH":
+                report.status = "FAIL"
+                report.status_extended = f"Bedrock Guardrail {guardrail.name} is configured to block prompt attacks but with a filter strength of {guardrail.prompt_attack_filter_strength}, not HIGH."
+            findings.append(report)
+
+        return findings

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

@@ -0,0 +1,32 @@
+{
+  "Provider": "aws",
+  "CheckID": "bedrock_guardrail_sensitive_information_filter_enabled",
+  "CheckTitle": "Configure Sensitive Information Filters for Amazon Bedrock Guardrails.",
+  "CheckType": [],
+  "ServiceName": "bedrock",
+  "SubServiceName": "",
+  "ResourceIdTemplate": "arn:partition:bedrock:region:account-id:guardrails/resource-id",
+  "Severity": "high",
+  "ResourceType": "Other",
+  "Description": "Ensure that sensitive information filters are enabled for Amazon Bedrock guardrails to prevent the leakage of sensitive data such as personally identifiable information (PII), financial data, or confidential corporate information.",
+  "Risk": "If sensitive information filters are not enabled, Bedrock models may inadvertently generate or expose confidential or sensitive information in responses, leading to data breaches, regulatory violations, or reputational damage.",
+  "RelatedUrl": "https://docs.aws.amazon.com/bedrock/latest/userguide/guardrails.html",
+  "Remediation": {
+    "Code": {
+      "CLI": "aws bedrock put-guardrails-configuration --guardrails-config 'sensitiveInformationFilter=true'",
+      "NativeIaC": "",
+      "Other": "https://www.trendmicro.com/cloudoneconformity/knowledge-base/aws/Bedrock/guardrails-with-pii-mask-block.html",
+      "Terraform": ""
+    },
+    "Recommendation": {
+      "Text": "Enable sensitive information filters for Amazon Bedrock guardrails to prevent the exposure of sensitive or confidential information.",
+      "Url": "https://docs.aws.amazon.com/bedrock/latest/userguide/guardrails-sensitive-filters.html"
+    }
+  },
+  "Categories": [
+    "security"
+  ],
+  "DependsOn": [],
+  "RelatedTo": [],
+  "Notes": ""
+}

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

@@ -0,0 +1,21 @@
+from prowler.lib.check.models import Check, Check_Report_AWS
+from prowler.providers.aws.services.bedrock.bedrock_client import bedrock_client
+
+
+class bedrock_guardrail_sensitive_information_filter_enabled(Check):
+    def execute(self):
+        findings = []
+        for guardrail in bedrock_client.guardrails.values():
+            report = Check_Report_AWS(self.metadata())
+            report.region = guardrail.region
+            report.resource_id = guardrail.id
+            report.resource_arn = guardrail.arn
+            report.resource_tags = guardrail.tags
+            report.status = "PASS"
+            report.status_extended = f"Bedrock Guardrail {guardrail.name} is blocking or masking sensitive information."
+            if not guardrail.sensitive_information_filter:
+                report.status = "FAIL"
+                report.status_extended = f"Bedrock Guardrail {guardrail.name} is not configured to block or mask sensitive information."
+            findings.append(report)
+
+        return findings

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

@@ -0,0 +1,34 @@
+{
+  "Provider": "aws",
+  "CheckID": "bedrock_model_invocation_logs_encryption_enabled",
+  "CheckTitle": "Ensure that Amazon Bedrock model invocation logs are encrypted with KMS.",
+  "CheckType": [],
+  "ServiceName": "bedrock",
+  "SubServiceName": "",
+  "ResourceIdTemplate": "arn:partition:bedrock:region:account-id:model/resource-id",
+  "Severity": "high",
+  "ResourceType": "Other",
+  "Description": "Ensure that Amazon Bedrock model invocation logs are encrypted using AWS KMS to protect sensitive data in the request and response logs for all model invocations.",
+  "Risk": "If Amazon Bedrock model invocation logs are not encrypted, sensitive data such as prompts, responses, and token usage could be exposed to unauthorized parties. This may lead to data breaches, security vulnerabilities, or unintended use of sensitive information.",
+  "RelatedUrl": "https://docs.aws.amazon.com/bedrock/latest/userguide/data-protection.html",
+  "Remediation": {
+    "Code": {
+      "CLI": "",
+      "NativeIaC": "",
+      "Other": "",
+      "Terraform": ""
+    },
+    "Recommendation": {
+      "Text": "Ensure that model invocation logs for Amazon Bedrock are encrypted using AWS KMS to prevent unauthorized access to sensitive log data.",
+      "Url": "hhttps://docs.aws.amazon.com/bedrock/latest/userguide/data-protection.html"
+    }
+  },
+  "Categories": [
+    "encryption",
+    "logging",
+    "security"
+  ],
+  "DependsOn": [],
+  "RelatedTo": [],
+  "Notes": ""
+}

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

@@ -0,0 +1,48 @@
+from prowler.lib.check.models import Check, Check_Report_AWS
+from prowler.providers.aws.services.bedrock.bedrock_client import bedrock_client
+from prowler.providers.aws.services.cloudwatch.logs_client import logs_client
+from prowler.providers.aws.services.s3.s3_client import s3_client
+
+
+class bedrock_model_invocation_logs_encryption_enabled(Check):
+    def execute(self):
+        findings = []
+        for region, logging in bedrock_client.logging_configurations.items():
+            if logging.enabled:
+                s3_encryption = True
+                cloudwatch_encryption = True
+                report = Check_Report_AWS(self.metadata())
+                report.region = region
+                report.resource_id = bedrock_client.audited_account
+                report.resource_arn = bedrock_client.audited_account_arn
+                report.status = "PASS"
+                report.status_extended = "Bedrock Model Invocation logs are encrypted."
+                if logging.s3_bucket:
+                    bucket_arn = (
+                        f"arn:{s3_client.audited_partition}:s3:::{logging.s3_bucket}"
+                    )
+                    if (
+                        bucket_arn in s3_client.buckets
+                        and not s3_client.buckets[bucket_arn].encryption
+                    ):
+                        s3_encryption = False
+                if logging.cloudwatch_log_group:
+                    log_group_arn = f"arn:{logs_client.audited_partition}:logs:{region}:{logs_client.audited_account}:log-group:{logging.cloudwatch_log_group}"
+                    if (
+                        log_group_arn in logs_client.log_groups
+                        and not logs_client.log_groups[log_group_arn].kms_id
+                    ):
+                        cloudwatch_encryption = False
+                if not s3_encryption and not cloudwatch_encryption:
+                    report.status = "FAIL"
+                    report.status_extended = f"Bedrock Model Invocation logs are not encrypted in S3 bucket: {logging.s3_bucket} and CloudWatch Log Group: {logging.cloudwatch_log_group}."
+                elif not s3_encryption:
+                    report.status = "FAIL"
+                    report.status_extended = f"Bedrock Model Invocation logs are not encrypted in S3 bucket: {logging.s3_bucket}."
+                elif not cloudwatch_encryption:
+                    report.status = "FAIL"
+                    report.status_extended = f"Bedrock Model Invocation logs are not encrypted in CloudWatch Log Group: {logging.cloudwatch_log_group}."
+
+                findings.append(report)
+
+        return findings

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

@@ -11,7 +11,11 @@
         # Call AWSService's __init__
         super().__init__(__class__.__name__, provider)
         self.logging_configurations = {}
+        self.guardrails = {}
         self.__threading_call__(self._get_model_invocation_logging_configuration)
+        self.__threading_call__(self._list_guardrails)
+        self.__threading_call__(self._get_guardrail, self.guardrails.values())
+        self.__threading_call__(self._list_tags_for_resource, self.guardrails.values())
 
     def _get_model_invocation_logging_configuration(self, regional_client):
         logger.info("Bedrock - Getting Model Invocation Logging Configuration...")
@@ -40,8 +44,65 @@
                 f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
             )
 
+    def _list_guardrails(self, regional_client):
+        logger.info("Bedrock - Listing Guardrails...")
+        try:
+            for guardrail in regional_client.list_guardrails().get("guardrails", []):
+                self.guardrails[guardrail["arn"]] = Guardrail(
+                    id=guardrail["id"],
+                    name=guardrail["name"],
+                    arn=guardrail["arn"],
+                    region=regional_client.region,
+                )
+        except Exception as error:
+            logger.error(
+                f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
+            )
+
+    def _get_guardrail(self, guardrail):
+        logger.info("Bedrock - Getting Guardrail...")
+        try:
+            guardrail_info = self.regional_clients[guardrail.region].get_guardrail(
+                guardrailIdentifier=guardrail.id
+            )
+            guardrail.sensitive_information_filter = (
+                "sensitiveInformationPolicy" in guardrail_info
+            )
+            for filter in guardrail_info.get("contentPolicy", {}).get("filters", []):
+                if filter.get("type") == "PROMPT_ATTACK":
+                    guardrail.prompt_attack_filter_strength = filter.get(
+                        "inputStrength", "NONE"
+                    )
+        except Exception as error:
+            logger.error(
+                f"{guardrail.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
+            )
+
+    def _list_tags_for_resource(self, guardrail):
+        logger.info("Bedrock - Listing Tags for Resource...")
+        try:
+            guardrail.tags = (
+                self.regional_clients[guardrail.region]
+                .list_tags_for_resource(resourceARN=guardrail.arn)
+                .get("tags", [])
+            )
+        except Exception as error:
+            logger.error(
+                f"{guardrail.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
+            )
+
 
 class LoggingConfiguration(BaseModel):
     enabled: bool = False
     cloudwatch_log_group: Optional[str] = None
     s3_bucket: Optional[str] = None
+
+
+class Guardrail(BaseModel):
+    id: str
+    name: str
+    arn: str
+    region: str
+    tags: Optional[list] = []
+    sensitive_information_filter: bool = False
+    prompt_attack_filter_strength: Optional[str]

prowler/providers/aws/services/cloudwatch/cloudwatch_log_group_kms_encryption_enabled/cloudwatch_log_group_kms_encryption_enabled.py

@@ -6,7 +6,7 @@ class cloudwatch_log_group_kms_encryption_enabled(Check):
     def execute(self):
         findings = []
         if logs_client.log_groups:
-            for log_group in logs_client.log_groups:
+            for log_group in logs_client.log_groups.values():
                 report = Check_Report_AWS(self.metadata())
                 report.region = log_group.region
                 report.resource_id = log_group.name

prowler/providers/aws/services/cloudwatch/cloudwatch_log_group_no_secrets_in_logs/cloudwatch_log_group_no_secrets_in_logs.py

@@ -15,7 +15,7 @@ def execute(self):
             secrets_ignore_patterns = logs_client.audit_config.get(
                 "secrets_ignore_patterns", []
             )
-            for log_group in logs_client.log_groups:
+            for log_group in logs_client.log_groups.values():
                 report = Check_Report_AWS(self.metadata())
                 report.status = "PASS"
                 report.status_extended = (

prowler/providers/aws/services/cloudwatch/cloudwatch_log_group_retention_policy_specific_days_enabled/cloudwatch_log_group_retention_policy_specific_days_enabled.py

@@ -11,7 +11,7 @@ def execute(self):
             "log_group_retention_days", 365
         )
         if logs_client.log_groups:
-            for log_group in logs_client.log_groups:
+            for log_group in logs_client.log_groups.values():
                 report = Check_Report_AWS(self.metadata())
                 report.region = log_group.region
                 report.resource_id = log_group.name

prowler/providers/aws/services/cloudwatch/cloudwatch_service.py

@@ -82,7 +82,7 @@
         # Call AWSService's __init__
         super().__init__(__class__.__name__, provider)
         self.log_group_arn_template = f"arn:{self.audited_partition}:logs:{self.region}:{self.audited_account}:log-group"
-        self.log_groups = []
+        self.log_groups = {}
         self.__threading_call__(self._describe_log_groups)
         self.metric_filters = []
         self.__threading_call__(self._describe_metric_filters)
@@ -95,7 +95,9 @@
                     1000  # The threshold for number of events to return per log group.
                 )
                 self.__threading_call__(self._get_log_events)
-            self.__threading_call__(self._list_tags_for_resource, self.log_groups)
+            self.__threading_call__(
+                self._list_tags_for_resource, self.log_groups.values()
+            )
 
     def _describe_metric_filters(self, regional_client):
         logger.info("CloudWatch Logs - Describing metric filters...")
@@ -113,7 +115,7 @@
                             self.metric_filters = []
 
                         log_group = None
-                        for lg in self.log_groups:
+                        for lg in self.log_groups.values():
                             if lg.name == filter["logGroupName"]:
                                 log_group = lg
                                 break
@@ -162,16 +164,14 @@
                             never_expire = True
                             retention_days = 9999
                         if self.log_groups is None:
-                            self.log_groups = []
-                        self.log_groups.append(
-                            LogGroup(
-                                arn=log_group["arn"],
-                                name=log_group["logGroupName"],
-                                retention_days=retention_days,
-                                never_expire=never_expire,
-                                kms_id=kms,
-                                region=regional_client.region,
-                            )
+                            self.log_groups = {}
+                        self.log_groups[log_group["arn"]] = LogGroup(
+                            arn=log_group["arn"],
+                            name=log_group["logGroupName"],
+                            retention_days=retention_days,
+                            never_expire=never_expire,
+                            kms_id=kms,
+                            region=regional_client.region,
                         )
         except ClientError as error:
             if error.response["Error"]["Code"] == "AccessDeniedException":
@@ -192,7 +192,7 @@
     def _get_log_events(self, regional_client):
         regional_log_groups = [
             log_group
-            for log_group in self.log_groups
+            for log_group in self.log_groups.values()
             if log_group.region == regional_client.region
         ]
         total_log_groups = len(regional_log_groups)