From 759f7b84d6d39ea50d948c2b79f32df0afd375ff Mon Sep 17 00:00:00 2001 From: Daniel Barranquero <74871504+danibarranqueroo@users.noreply.github.com> Date: Mon, 11 May 2026 17:11:49 +0200 Subject: [PATCH] feat(aws): add cloudtrail_bedrock_logging_enabled security check (#10858) --- prowler/CHANGELOG.md | 1 + ...itected_framework_security_pillar_aws.json | 1 + prowler/compliance/aws/c5_aws.json | 2 + prowler/compliance/aws/ccc_aws.json | 1 + prowler/compliance/aws/csa_ccm_4.0_aws.json | 1 + .../aws/fedramp_moderate_revision_4_aws.json | 2 + prowler/compliance/aws/ffiec_aws.json | 1 + prowler/compliance/aws/hipaa_aws.json | 2 + prowler/compliance/aws/iso27001_2022_aws.json | 1 + .../compliance/aws/kisa_isms_p_2023_aws.json | 1 + .../aws/nist_800_53_revision_5_aws.json | 3 + prowler/compliance/aws/nist_csf_2.0_aws.json | 3 + .../compliance/aws/secnumcloud_3.2_aws.json | 1 + prowler/compliance/aws/soc2_aws.json | 1 + .../__init__.py | 0 ...rail_bedrock_logging_enabled.metadata.json | 44 + .../cloudtrail_bedrock_logging_enabled.py | 213 ++++ ...cloudtrail_bedrock_logging_enabled_test.py | 1020 +++++++++++++++++ 18 files changed, 1298 insertions(+) create mode 100644 prowler/providers/aws/services/cloudtrail/cloudtrail_bedrock_logging_enabled/__init__.py create mode 100644 prowler/providers/aws/services/cloudtrail/cloudtrail_bedrock_logging_enabled/cloudtrail_bedrock_logging_enabled.metadata.json create mode 100644 prowler/providers/aws/services/cloudtrail/cloudtrail_bedrock_logging_enabled/cloudtrail_bedrock_logging_enabled.py create mode 100644 tests/providers/aws/services/cloudtrail/cloudtrail_bedrock_logging_enabled/cloudtrail_bedrock_logging_enabled_test.py diff --git a/prowler/CHANGELOG.md b/prowler/CHANGELOG.md index 8a2a4b1ac5..b505b3d569 100644 --- a/prowler/CHANGELOG.md +++ b/prowler/CHANGELOG.md @@ -7,6 +7,7 @@ All notable changes to the **Prowler SDK** are documented in this file. ### 🚀 Added - `iam_user_access_not_stale_to_sagemaker` check for aws provider with configurable `max_unused_sagemaker_access_days` (default 90) [(#11000)](https://github.com/prowler-cloud/prowler/pull/11000) +- `cloudtrail_bedrock_logging_enabled` check for AWS provider [(#10858)](https://github.com/prowler-cloud/prowler/pull/10858) --- diff --git a/prowler/compliance/aws/aws_well_architected_framework_security_pillar_aws.json b/prowler/compliance/aws/aws_well_architected_framework_security_pillar_aws.json index 7a04e4ddf7..94c50eb68a 100644 --- a/prowler/compliance/aws/aws_well_architected_framework_security_pillar_aws.json +++ b/prowler/compliance/aws/aws_well_architected_framework_security_pillar_aws.json @@ -550,6 +550,7 @@ "apigatewayv2_api_access_logging_enabled", "awslambda_function_invoke_api_operations_cloudtrail_logging_enabled", "cloudfront_distributions_logging_enabled", + "cloudtrail_bedrock_logging_enabled", "cloudtrail_cloudwatch_logging_enabled", "cloudtrail_logs_s3_bucket_access_logging_enabled", "directoryservice_directory_log_forwarding_enabled", diff --git a/prowler/compliance/aws/c5_aws.json b/prowler/compliance/aws/c5_aws.json index 60a3f1a751..8847469154 100644 --- a/prowler/compliance/aws/c5_aws.json +++ b/prowler/compliance/aws/c5_aws.json @@ -3461,6 +3461,7 @@ ], "Checks": [ "kinesis_stream_data_retention_period", + "cloudtrail_bedrock_logging_enabled", "cloudtrail_multi_region_enabled_logging_management_events" ] }, @@ -3669,6 +3670,7 @@ "awslambda_function_invoke_api_operations_cloudtrail_logging_enabled", "bedrock_model_invocation_logging_enabled", "cloudfront_distributions_logging_enabled", + "cloudtrail_bedrock_logging_enabled", "cloudtrail_cloudwatch_logging_enabled", "cloudtrail_logs_s3_bucket_access_logging_enabled", "cloudtrail_multi_region_enabled_logging_management_events", diff --git a/prowler/compliance/aws/ccc_aws.json b/prowler/compliance/aws/ccc_aws.json index ea28875c5c..d1a20ee3d0 100644 --- a/prowler/compliance/aws/ccc_aws.json +++ b/prowler/compliance/aws/ccc_aws.json @@ -1958,6 +1958,7 @@ } ], "Checks": [ + "cloudtrail_bedrock_logging_enabled", "cloudtrail_multi_region_enabled", "cloudtrail_multi_region_enabled_logging_management_events", "cloudtrail_cloudwatch_logging_enabled", diff --git a/prowler/compliance/aws/csa_ccm_4.0_aws.json b/prowler/compliance/aws/csa_ccm_4.0_aws.json index 99f91acb4d..98d87112c9 100644 --- a/prowler/compliance/aws/csa_ccm_4.0_aws.json +++ b/prowler/compliance/aws/csa_ccm_4.0_aws.json @@ -5857,6 +5857,7 @@ } ], "Checks": [ + "cloudtrail_bedrock_logging_enabled", "cloudtrail_multi_region_enabled", "cloudtrail_multi_region_enabled_logging_management_events", "cloudtrail_s3_dataevents_read_enabled", diff --git a/prowler/compliance/aws/fedramp_moderate_revision_4_aws.json b/prowler/compliance/aws/fedramp_moderate_revision_4_aws.json index a184ec3fca..c914a58b2c 100644 --- a/prowler/compliance/aws/fedramp_moderate_revision_4_aws.json +++ b/prowler/compliance/aws/fedramp_moderate_revision_4_aws.json @@ -439,6 +439,7 @@ "cloudtrail_s3_dataevents_read_enabled", "cloudtrail_s3_dataevents_write_enabled", "cloudtrail_multi_region_enabled", + "cloudtrail_bedrock_logging_enabled", "cloudtrail_cloudwatch_logging_enabled", "elbv2_logging_enabled", "elb_logging_enabled", @@ -594,6 +595,7 @@ "cloudtrail_s3_dataevents_read_enabled", "cloudtrail_s3_dataevents_write_enabled", "cloudtrail_multi_region_enabled", + "cloudtrail_bedrock_logging_enabled", "cloudtrail_cloudwatch_logging_enabled", "elbv2_logging_enabled", "elb_logging_enabled", diff --git a/prowler/compliance/aws/ffiec_aws.json b/prowler/compliance/aws/ffiec_aws.json index 697d9ee49f..23ab8953b6 100644 --- a/prowler/compliance/aws/ffiec_aws.json +++ b/prowler/compliance/aws/ffiec_aws.json @@ -119,6 +119,7 @@ ], "Checks": [ "apigateway_restapi_logging_enabled", + "cloudtrail_bedrock_logging_enabled", "cloudtrail_multi_region_enabled", "cloudtrail_s3_dataevents_read_enabled", "cloudtrail_s3_dataevents_write_enabled", diff --git a/prowler/compliance/aws/hipaa_aws.json b/prowler/compliance/aws/hipaa_aws.json index 34de052b95..036489d712 100644 --- a/prowler/compliance/aws/hipaa_aws.json +++ b/prowler/compliance/aws/hipaa_aws.json @@ -87,6 +87,7 @@ ], "Checks": [ "apigateway_restapi_logging_enabled", + "cloudtrail_bedrock_logging_enabled", "cloudtrail_multi_region_enabled", "cloudtrail_s3_dataevents_read_enabled", "cloudtrail_s3_dataevents_write_enabled", @@ -632,6 +633,7 @@ ], "Checks": [ "apigateway_restapi_logging_enabled", + "cloudtrail_bedrock_logging_enabled", "cloudtrail_multi_region_enabled", "cloudtrail_s3_dataevents_read_enabled", "cloudtrail_s3_dataevents_write_enabled", diff --git a/prowler/compliance/aws/iso27001_2022_aws.json b/prowler/compliance/aws/iso27001_2022_aws.json index 0e640e7364..d47bfcf1d1 100644 --- a/prowler/compliance/aws/iso27001_2022_aws.json +++ b/prowler/compliance/aws/iso27001_2022_aws.json @@ -1294,6 +1294,7 @@ "bedrock_model_invocation_logging_enabled", "bedrock_model_invocation_logs_encryption_enabled", "cloudfront_distributions_logging_enabled", + "cloudtrail_bedrock_logging_enabled", "cloudtrail_cloudwatch_logging_enabled", "cloudtrail_kms_encryption_enabled", "cloudtrail_log_file_validation_enabled", diff --git a/prowler/compliance/aws/kisa_isms_p_2023_aws.json b/prowler/compliance/aws/kisa_isms_p_2023_aws.json index d210cbbbd8..d172e615dd 100644 --- a/prowler/compliance/aws/kisa_isms_p_2023_aws.json +++ b/prowler/compliance/aws/kisa_isms_p_2023_aws.json @@ -2540,6 +2540,7 @@ "bedrock_model_invocation_logging_enabled", "bedrock_model_invocation_logs_encryption_enabled", "cloudfront_distributions_logging_enabled", + "cloudtrail_bedrock_logging_enabled", "cloudtrail_bucket_requires_mfa_delete", "cloudtrail_cloudwatch_logging_enabled", "cloudtrail_insights_exist", diff --git a/prowler/compliance/aws/nist_800_53_revision_5_aws.json b/prowler/compliance/aws/nist_800_53_revision_5_aws.json index 8694b69c01..858df01fd9 100644 --- a/prowler/compliance/aws/nist_800_53_revision_5_aws.json +++ b/prowler/compliance/aws/nist_800_53_revision_5_aws.json @@ -1604,6 +1604,7 @@ "cloudtrail_s3_dataevents_read_enabled", "cloudtrail_s3_dataevents_write_enabled", "cloudtrail_multi_region_enabled", + "cloudtrail_bedrock_logging_enabled", "cloudtrail_cloudwatch_logging_enabled", "elbv2_logging_enabled", "elb_logging_enabled", @@ -2162,6 +2163,7 @@ "cloudtrail_s3_dataevents_read_enabled", "cloudtrail_s3_dataevents_write_enabled", "cloudtrail_multi_region_enabled", + "cloudtrail_bedrock_logging_enabled", "cloudtrail_cloudwatch_logging_enabled", "elbv2_logging_enabled", "elb_logging_enabled", @@ -2189,6 +2191,7 @@ "cloudtrail_s3_dataevents_read_enabled", "cloudtrail_s3_dataevents_write_enabled", "cloudtrail_multi_region_enabled", + "cloudtrail_bedrock_logging_enabled", "cloudtrail_cloudwatch_logging_enabled", "elbv2_logging_enabled", "elb_logging_enabled", diff --git a/prowler/compliance/aws/nist_csf_2.0_aws.json b/prowler/compliance/aws/nist_csf_2.0_aws.json index a403932a7b..e890b08573 100644 --- a/prowler/compliance/aws/nist_csf_2.0_aws.json +++ b/prowler/compliance/aws/nist_csf_2.0_aws.json @@ -1312,6 +1312,7 @@ } ], "Checks": [ + "cloudtrail_bedrock_logging_enabled", "cloudtrail_kms_encryption_enabled", "cloudtrail_log_file_validation_enabled", "cloudtrail_logs_s3_bucket_access_logging_enabled", @@ -1475,6 +1476,7 @@ "cloudtrail_threat_detection_enumeration", "cloudtrail_threat_detection_privilege_escalation", "cloudtrail_threat_detection_llm_jacking", + "cloudtrail_bedrock_logging_enabled", "cloudtrail_cloudwatch_logging_enabled", "cloudtrail_multi_region_enabled_logging_management_events" ] @@ -1571,6 +1573,7 @@ "cloudtrail_threat_detection_llm_jacking", "cloudtrail_threat_detection_enumeration", "cloudtrail_multi_region_enabled_logging_management_events", + "cloudtrail_bedrock_logging_enabled", "cloudtrail_cloudwatch_logging_enabled", "cloudwatch_log_metric_filter_unauthorized_api_calls", "cloudwatch_log_metric_filter_authentication_failures", diff --git a/prowler/compliance/aws/secnumcloud_3.2_aws.json b/prowler/compliance/aws/secnumcloud_3.2_aws.json index 1cad58b263..1e157b47c7 100644 --- a/prowler/compliance/aws/secnumcloud_3.2_aws.json +++ b/prowler/compliance/aws/secnumcloud_3.2_aws.json @@ -818,6 +818,7 @@ } ], "Checks": [ + "cloudtrail_bedrock_logging_enabled", "cloudtrail_multi_region_enabled", "cloudtrail_multi_region_enabled_logging_management_events", "cloudtrail_s3_dataevents_read_enabled", diff --git a/prowler/compliance/aws/soc2_aws.json b/prowler/compliance/aws/soc2_aws.json index 472d20ab75..5a027d0416 100644 --- a/prowler/compliance/aws/soc2_aws.json +++ b/prowler/compliance/aws/soc2_aws.json @@ -346,6 +346,7 @@ } ], "Checks": [ + "cloudtrail_bedrock_logging_enabled", "cloudtrail_cloudwatch_logging_enabled", "cloudwatch_changes_to_network_acls_alarm_configured", "cloudwatch_changes_to_network_gateways_alarm_configured", diff --git a/prowler/providers/aws/services/cloudtrail/cloudtrail_bedrock_logging_enabled/__init__.py b/prowler/providers/aws/services/cloudtrail/cloudtrail_bedrock_logging_enabled/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/prowler/providers/aws/services/cloudtrail/cloudtrail_bedrock_logging_enabled/cloudtrail_bedrock_logging_enabled.metadata.json b/prowler/providers/aws/services/cloudtrail/cloudtrail_bedrock_logging_enabled/cloudtrail_bedrock_logging_enabled.metadata.json new file mode 100644 index 0000000000..8f27089444 --- /dev/null +++ b/prowler/providers/aws/services/cloudtrail/cloudtrail_bedrock_logging_enabled/cloudtrail_bedrock_logging_enabled.metadata.json @@ -0,0 +1,44 @@ +{ + "Provider": "aws", + "CheckID": "cloudtrail_bedrock_logging_enabled", + "CheckTitle": "CloudTrail logs Amazon Bedrock API calls for security auditing", + "CheckType": [ + "Software and Configuration Checks/AWS Security Best Practices" + ], + "ServiceName": "cloudtrail", + "SubServiceName": "", + "ResourceIdTemplate": "", + "Severity": "medium", + "ResourceType": "AwsCloudTrailTrail", + "ResourceGroup": "monitoring", + "Description": "**At least one actively logging CloudTrail trail** records **Amazon Bedrock API activity** through management events or advanced event selectors targeting Bedrock resources.\n\nThis check covers **control-plane** operations such as configuration changes through CloudTrail management events and can also cover **data-plane** Bedrock events when advanced event selectors target Bedrock resource types.", + "Risk": "Without CloudTrail logging for Bedrock control-plane operations, changes to prompts, guardrails, agents, flows, or knowledge bases can become invisible, weakening forensics and incident response. Management events do not capture `InvokeModel`; pair this control with `bedrock_model_invocation_logging_enabled` or Bedrock data event selectors for invocation visibility.", + "RelatedUrl": "", + "AdditionalURLs": [ + "https://docs.aws.amazon.com/bedrock/latest/userguide/logging-using-cloudtrail.html", + "https://docs.aws.amazon.com/awscloudtrail/latest/userguide/logging-data-events-with-cloudtrail.html" + ], + "Remediation": { + "Code": { + "CLI": "aws cloudtrail put-event-selectors --trail-name --advanced-event-selectors '[{\"Name\":\"Bedrock data events\",\"FieldSelectors\":[{\"Field\":\"eventCategory\",\"Equals\":[\"Data\"]},{\"Field\":\"resources.type\",\"Equals\":[\"AWS::Bedrock::Model\",\"AWS::Bedrock::Guardrail\",\"AWS::Bedrock::AgentAlias\",\"AWS::Bedrock::FlowAlias\",\"AWS::Bedrock::InlineAgent\",\"AWS::Bedrock::KnowledgeBase\",\"AWS::Bedrock::Prompt\"]}]}]'", + "NativeIaC": "```yaml\n# CloudFormation: enable Bedrock data event logging on an actively logging trail\nResources:\n ExampleTrail:\n Type: AWS::CloudTrail::Trail\n Properties:\n TrailName: \n S3BucketName: \n IsLogging: true\n AdvancedEventSelectors:\n - Name: Bedrock data events\n FieldSelectors:\n - Field: eventCategory\n Equals:\n - Data\n - Field: resources.type # CRITICAL: target Bedrock resources\n Equals:\n - AWS::Bedrock::Model\n - AWS::Bedrock::Guardrail\n - AWS::Bedrock::AgentAlias\n - AWS::Bedrock::FlowAlias\n - AWS::Bedrock::InlineAgent\n - AWS::Bedrock::KnowledgeBase\n - AWS::Bedrock::Prompt\n```", + "Other": "1. In the AWS Console, open CloudTrail and select a trail that is actively logging\n2. Edit the trail and enable Management events to capture Bedrock control-plane operations, or add Bedrock advanced data event selectors for data-plane visibility\n3. If using data events, select the Bedrock resource types you want to log\n4. Save changes and confirm the trail remains in logging state", + "Terraform": "```hcl\n# Terraform: enable Bedrock data event logging on an actively logging trail\nresource \"aws_cloudtrail\" \"example_resource\" {\n name = \"example_resource\"\n s3_bucket_name = \"example_resource\"\n\n advanced_event_selector {\n name = \"Bedrock data events\"\n field_selector {\n field = \"eventCategory\"\n equals = [\"Data\"]\n }\n field_selector {\n field = \"resources.type\" # CRITICAL: target Bedrock resources\n equals = [\"AWS::Bedrock::Model\", \"AWS::Bedrock::Guardrail\", \"AWS::Bedrock::AgentAlias\", \"AWS::Bedrock::FlowAlias\", \"AWS::Bedrock::InlineAgent\", \"AWS::Bedrock::KnowledgeBase\", \"AWS::Bedrock::Prompt\"]\n }\n }\n}\n```" + }, + "Recommendation": { + "Text": "Enable CloudTrail logging for Amazon Bedrock on **at least one actively logging trail**. At minimum, enable **management events** to capture Bedrock control-plane operations. For invocation-level and other data-plane visibility, add **advanced event selectors** targeting Bedrock resource types or pair this control with `bedrock_model_invocation_logging_enabled`.\n\nFor broader region coverage, pair this control with a separate multi-region CloudTrail check. Centralize logs in an encrypted bucket or CloudWatch Logs to support **defense in depth** and forensic readiness for AI workloads.", + "Url": "https://hub.prowler.com/check/cloudtrail_bedrock_logging_enabled" + } + }, + "Categories": [ + "logging", + "forensics-ready", + "gen-ai" + ], + "DependsOn": [], + "RelatedTo": [ + "cloudtrail_multi_region_enabled_logging_management_events", + "bedrock_model_invocation_logging_enabled" + ], + "Notes": "This check passes when CloudTrail captures Bedrock control-plane activity via management events or Bedrock data events via advanced selectors. It does not require multi-region coverage, and it does not by itself guarantee `InvokeModel` visibility unless Bedrock data events are selected; use `bedrock_model_invocation_logging_enabled` for model invocation logs. Additional advanced selector filters such as `eventName` or `resources.ARN` can further narrow effective coverage and should be reviewed explicitly." +} diff --git a/prowler/providers/aws/services/cloudtrail/cloudtrail_bedrock_logging_enabled/cloudtrail_bedrock_logging_enabled.py b/prowler/providers/aws/services/cloudtrail/cloudtrail_bedrock_logging_enabled/cloudtrail_bedrock_logging_enabled.py new file mode 100644 index 0000000000..d66d10f7c6 --- /dev/null +++ b/prowler/providers/aws/services/cloudtrail/cloudtrail_bedrock_logging_enabled/cloudtrail_bedrock_logging_enabled.py @@ -0,0 +1,213 @@ +from prowler.lib.check.models import Check, Check_Report_AWS +from prowler.providers.aws.services.cloudtrail.cloudtrail_client import ( + cloudtrail_client, +) +from prowler.providers.aws.services.cloudtrail.cloudtrail_service import ( + Event_Selector, +) + + +class cloudtrail_bedrock_logging_enabled(Check): + """Ensure CloudTrail is configured to log Amazon Bedrock API calls. + + This check verifies whether at least one CloudTrail trail is configured to + capture Amazon Bedrock control-plane API calls through management events or + Bedrock data events through advanced event selectors. + + - PASS: A trail logs Bedrock control-plane API calls via management events + or Bedrock data events via Bedrock-specific advanced event selectors. + - FAIL: No CloudTrail trail is configured to log Bedrock API calls. + """ + + # Bedrock resource types supported by CloudTrail advanced event selectors. + BEDROCK_RESOURCE_TYPES = frozenset( + { + "AWS::Bedrock::AgentAlias", + "AWS::Bedrock::FlowAlias", + "AWS::Bedrock::Guardrail", + "AWS::Bedrock::InlineAgent", + "AWS::Bedrock::KnowledgeBase", + "AWS::Bedrock::Model", + "AWS::Bedrock::Prompt", + } + ) + # Bedrock control-plane event sources, including Bedrock Data Automation. + BEDROCK_EVENT_SOURCES = frozenset( + { + "bedrock.amazonaws.com", + "bedrock-agent.amazonaws.com", + "bedrock-runtime.amazonaws.com", + "bedrock-agent-runtime.amazonaws.com", + "bedrock-data-automation.amazonaws.com", + "bedrock-data-automation-runtime.amazonaws.com", + } + ) + + def execute(self) -> list[Check_Report_AWS]: + """Execute the check logic. + + Returns: + A list of reports containing the result of the check. + """ + findings = [] + if cloudtrail_client.trails is not None: + for trail in cloudtrail_client.trails.values(): + if trail.is_logging: + for data_event in trail.data_events: + match_type = self._get_bedrock_match_type(data_event) + if match_type: + report = Check_Report_AWS( + metadata=self.metadata(), resource=trail + ) + report.region = trail.home_region + report.status = "PASS" + if match_type == "classic_management": + report.status_extended = ( + f"Trail {trail.name} from home region " + f"{trail.home_region} has management events " + "enabled to log Amazon Bedrock control-plane " + "API calls." + ) + elif match_type == "advanced_management": + report.status_extended = ( + f"Trail {trail.name} from home region " + f"{trail.home_region} has an advanced " + "management event selector to log Amazon " + "Bedrock control-plane API calls." + ) + else: + report.status_extended = ( + f"Trail {trail.name} from home region " + f"{trail.home_region} has an advanced data " + "event selector to log Amazon Bedrock API " + "calls." + ) + findings.append(report) + break + if not findings: + report = Check_Report_AWS( + metadata=self.metadata(), resource=cloudtrail_client.trails + ) + report.region = cloudtrail_client.region + report.resource_arn = cloudtrail_client.trail_arn_template + report.resource_id = cloudtrail_client.audited_account + report.status = "FAIL" + report.status_extended = "No CloudTrail trails are configured to log Amazon Bedrock API calls." + findings.append(report) + return findings + + def _get_bedrock_match_type(self, data_event: Event_Selector) -> str | None: + """Return the Bedrock logging match type for an event selector. + + Args: + data_event: An Event_Selector object from the trail. + + Returns: + The matching selector type, or None if the selector does not log + the Bedrock events covered by this check. + """ + if not data_event.is_advanced: + if self._logs_classic_management_events(data_event.event_selector): + return "classic_management" + return None + + field_selectors = data_event.event_selector.get("FieldSelectors", []) + if self._logs_advanced_management_events(field_selectors): + return "advanced_management" + if self._logs_advanced_bedrock_data_events(field_selectors): + return "advanced_data" + + return None + + @staticmethod + def _logs_classic_management_events(event_selector: dict) -> bool: + """Check whether a classic selector logs Bedrock control-plane events.""" + return event_selector.get( + "IncludeManagementEvents", True + ) and event_selector.get("ReadWriteType", "All") in ("All", "WriteOnly") + + def _logs_advanced_management_events(self, field_selectors: list[dict]) -> bool: + """Check whether advanced selectors log Bedrock control-plane events.""" + event_category_selectors = [ + field for field in field_selectors if field.get("Field") == "eventCategory" + ] + if not self._selectors_match_value("Management", event_category_selectors): + return False + + read_only_selectors = [ + field for field in field_selectors if field.get("Field") == "readOnly" + ] + has_read_only_restriction = bool(read_only_selectors) and not any( + self._field_selector_matches_value("false", selector) + for selector in read_only_selectors + ) + + return not has_read_only_restriction and self._logs_bedrock_management_events( + field_selectors + ) + + def _logs_advanced_bedrock_data_events(self, field_selectors: list[dict]) -> bool: + """Check whether advanced selectors log Bedrock data events.""" + event_category_selectors = [ + field for field in field_selectors if field.get("Field") == "eventCategory" + ] + if not self._selectors_match_value("Data", event_category_selectors): + return False + + resource_type_selectors = [ + field for field in field_selectors if field.get("Field") == "resources.type" + ] + return any( + self._selectors_match_value(resource_type, resource_type_selectors) + for resource_type in self.BEDROCK_RESOURCE_TYPES + ) + + def _logs_bedrock_management_events(self, field_selectors: list[dict]) -> bool: + """Check whether advanced management selectors include Bedrock sources.""" + event_source_selectors = [ + field for field in field_selectors if field.get("Field") == "eventSource" + ] + if not event_source_selectors: + return True + + return any( + self._selectors_match_value(event_source, event_source_selectors) + for event_source in self.BEDROCK_EVENT_SOURCES + ) + + def _selectors_match_value(self, value: str, selectors: list[dict]) -> bool: + """Check whether a candidate value satisfies all selectors for a field.""" + return bool(selectors) and all( + self._field_selector_matches_value(value, selector) + for selector in selectors + ) + + @staticmethod + def _field_selector_matches_value(value: str, selector: dict) -> bool: + """Evaluate a CloudTrail advanced field selector against a candidate value.""" + conditions = [] + + if "Equals" in selector: + conditions.append(value in selector["Equals"]) + if "NotEquals" in selector: + conditions.append(value not in selector["NotEquals"]) + if "StartsWith" in selector: + conditions.append( + any(value.startswith(prefix) for prefix in selector["StartsWith"]) + ) + if "NotStartsWith" in selector: + conditions.append( + all( + not value.startswith(prefix) for prefix in selector["NotStartsWith"] + ) + ) + if "EndsWith" in selector: + conditions.append( + any(value.endswith(suffix) for suffix in selector["EndsWith"]) + ) + if "NotEndsWith" in selector: + conditions.append( + all(not value.endswith(suffix) for suffix in selector["NotEndsWith"]) + ) + + return all(conditions) if conditions else True diff --git a/tests/providers/aws/services/cloudtrail/cloudtrail_bedrock_logging_enabled/cloudtrail_bedrock_logging_enabled_test.py b/tests/providers/aws/services/cloudtrail/cloudtrail_bedrock_logging_enabled/cloudtrail_bedrock_logging_enabled_test.py new file mode 100644 index 0000000000..7b0cfa6a07 --- /dev/null +++ b/tests/providers/aws/services/cloudtrail/cloudtrail_bedrock_logging_enabled/cloudtrail_bedrock_logging_enabled_test.py @@ -0,0 +1,1020 @@ +from unittest import mock + +import pytest +from boto3 import client +from moto import mock_aws + +from tests.providers.aws.utils import ( + AWS_ACCOUNT_NUMBER, + AWS_REGION_US_EAST_1, + set_mocked_aws_provider, +) + +CHECK_MODULE_PATH = "prowler.providers.aws.services.cloudtrail.cloudtrail_bedrock_logging_enabled.cloudtrail_bedrock_logging_enabled" + + +class Test_cloudtrail_bedrock_logging_enabled: + @mock_aws + def test_no_trails(self): + """Test when there are no CloudTrail trails configured.""" + from prowler.providers.aws.services.cloudtrail.cloudtrail_service import ( + Cloudtrail, + ) + + aws_provider = set_mocked_aws_provider() + + with ( + mock.patch( + "prowler.providers.common.provider.Provider.get_global_provider", + return_value=aws_provider, + ), + mock.patch( + f"{CHECK_MODULE_PATH}.cloudtrail_client", + new=Cloudtrail(aws_provider), + ), + ): + from prowler.providers.aws.services.cloudtrail.cloudtrail_bedrock_logging_enabled.cloudtrail_bedrock_logging_enabled import ( + cloudtrail_bedrock_logging_enabled, + ) + + check = cloudtrail_bedrock_logging_enabled() + result = check.execute() + + assert len(result) == 1 + assert result[0].status == "FAIL" + assert ( + result[0].status_extended + == "No CloudTrail trails are configured to log Amazon Bedrock API calls." + ) + assert result[0].resource_id == AWS_ACCOUNT_NUMBER + assert ( + result[0].resource_arn + == f"arn:aws:cloudtrail:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:trail" + ) + assert result[0].region == AWS_REGION_US_EAST_1 + + @mock_aws + def test_trail_not_logging(self): + """Test when a trail exists but is not actively logging.""" + cloudtrail_client_us = client("cloudtrail", region_name=AWS_REGION_US_EAST_1) + s3_client_us = client("s3", region_name=AWS_REGION_US_EAST_1) + trail_name = "trail_test" + bucket_name = "bucket_test" + s3_client_us.create_bucket(Bucket=bucket_name) + cloudtrail_client_us.create_trail( + Name=trail_name, S3BucketName=bucket_name, IsMultiRegionTrail=False + ) + cloudtrail_client_us.put_event_selectors( + TrailName=trail_name, + EventSelectors=[ + { + "ReadWriteType": "All", + "IncludeManagementEvents": True, + "DataResources": [ + {"Type": "AWS::S3::Object", "Values": ["arn:aws:s3"]} + ], + } + ], + ) + # Trail is not started, so is_logging remains False + + from prowler.providers.aws.services.cloudtrail.cloudtrail_service import ( + Cloudtrail, + ) + + aws_provider = set_mocked_aws_provider() + + with ( + mock.patch( + "prowler.providers.common.provider.Provider.get_global_provider", + return_value=aws_provider, + ), + mock.patch( + f"{CHECK_MODULE_PATH}.cloudtrail_client", + new=Cloudtrail(aws_provider), + ), + ): + from prowler.providers.aws.services.cloudtrail.cloudtrail_bedrock_logging_enabled.cloudtrail_bedrock_logging_enabled import ( + cloudtrail_bedrock_logging_enabled, + ) + + check = cloudtrail_bedrock_logging_enabled() + result = check.execute() + + assert len(result) == 1 + assert result[0].status == "FAIL" + assert ( + result[0].status_extended + == "No CloudTrail trails are configured to log Amazon Bedrock API calls." + ) + + @mock_aws + def test_trail_without_management_events(self): + """Test when a trail has data events but no management events enabled.""" + cloudtrail_client_us = client("cloudtrail", region_name=AWS_REGION_US_EAST_1) + s3_client_us = client("s3", region_name=AWS_REGION_US_EAST_1) + trail_name = "trail_test" + bucket_name = "bucket_test" + s3_client_us.create_bucket(Bucket=bucket_name) + cloudtrail_client_us.create_trail( + Name=trail_name, S3BucketName=bucket_name, IsMultiRegionTrail=False + ) + cloudtrail_client_us.start_logging(Name=trail_name) + cloudtrail_client_us.put_event_selectors( + TrailName=trail_name, + EventSelectors=[ + { + "ReadWriteType": "All", + "IncludeManagementEvents": False, + "DataResources": [ + {"Type": "AWS::S3::Object", "Values": ["arn:aws:s3"]} + ], + } + ], + ) + + from prowler.providers.aws.services.cloudtrail.cloudtrail_service import ( + Cloudtrail, + ) + + aws_provider = set_mocked_aws_provider() + + with ( + mock.patch( + "prowler.providers.common.provider.Provider.get_global_provider", + return_value=aws_provider, + ), + mock.patch( + f"{CHECK_MODULE_PATH}.cloudtrail_client", + new=Cloudtrail(aws_provider), + ), + ): + from prowler.providers.aws.services.cloudtrail.cloudtrail_bedrock_logging_enabled.cloudtrail_bedrock_logging_enabled import ( + cloudtrail_bedrock_logging_enabled, + ) + + check = cloudtrail_bedrock_logging_enabled() + result = check.execute() + + assert len(result) == 1 + assert result[0].status == "FAIL" + assert ( + result[0].status_extended + == "No CloudTrail trails are configured to log Amazon Bedrock API calls." + ) + + @mock_aws + def test_trail_with_classic_management_events(self): + """Test PASS when a trail has classic management events enabled.""" + cloudtrail_client_us = client("cloudtrail", region_name=AWS_REGION_US_EAST_1) + s3_client_us = client("s3", region_name=AWS_REGION_US_EAST_1) + trail_name = "trail_test" + bucket_name = "bucket_test" + s3_client_us.create_bucket(Bucket=bucket_name) + trail = cloudtrail_client_us.create_trail( + Name=trail_name, S3BucketName=bucket_name, IsMultiRegionTrail=False + ) + cloudtrail_client_us.start_logging(Name=trail_name) + cloudtrail_client_us.put_event_selectors( + TrailName=trail_name, + EventSelectors=[ + { + "ReadWriteType": "All", + "IncludeManagementEvents": True, + "DataResources": [ + {"Type": "AWS::S3::Object", "Values": ["arn:aws:s3"]} + ], + } + ], + ) + + from prowler.providers.aws.services.cloudtrail.cloudtrail_service import ( + Cloudtrail, + ) + + aws_provider = set_mocked_aws_provider() + + with ( + mock.patch( + "prowler.providers.common.provider.Provider.get_global_provider", + return_value=aws_provider, + ), + mock.patch( + f"{CHECK_MODULE_PATH}.cloudtrail_client", + new=Cloudtrail(aws_provider), + ), + ): + from prowler.providers.aws.services.cloudtrail.cloudtrail_bedrock_logging_enabled.cloudtrail_bedrock_logging_enabled import ( + cloudtrail_bedrock_logging_enabled, + ) + + check = cloudtrail_bedrock_logging_enabled() + result = check.execute() + + assert len(result) == 1 + assert result[0].status == "PASS" + assert ( + result[0].status_extended + == f"Trail {trail_name} from home region {AWS_REGION_US_EAST_1} has management events enabled to log Amazon Bedrock control-plane API calls." + ) + assert result[0].resource_id == trail_name + assert result[0].resource_arn == trail["TrailARN"] + assert result[0].region == AWS_REGION_US_EAST_1 + + @mock_aws + def test_trail_with_classic_management_events_read_only(self): + """Test FAIL when a trail has management events but ReadWriteType is ReadOnly.""" + cloudtrail_client_us = client("cloudtrail", region_name=AWS_REGION_US_EAST_1) + s3_client_us = client("s3", region_name=AWS_REGION_US_EAST_1) + trail_name = "trail_test" + bucket_name = "bucket_test" + s3_client_us.create_bucket(Bucket=bucket_name) + cloudtrail_client_us.create_trail( + Name=trail_name, S3BucketName=bucket_name, IsMultiRegionTrail=False + ) + cloudtrail_client_us.start_logging(Name=trail_name) + cloudtrail_client_us.put_event_selectors( + TrailName=trail_name, + EventSelectors=[ + { + "ReadWriteType": "ReadOnly", + "IncludeManagementEvents": True, + "DataResources": [ + {"Type": "AWS::S3::Object", "Values": ["arn:aws:s3"]} + ], + } + ], + ) + + from prowler.providers.aws.services.cloudtrail.cloudtrail_service import ( + Cloudtrail, + ) + + aws_provider = set_mocked_aws_provider() + + with ( + mock.patch( + "prowler.providers.common.provider.Provider.get_global_provider", + return_value=aws_provider, + ), + mock.patch( + f"{CHECK_MODULE_PATH}.cloudtrail_client", + new=Cloudtrail(aws_provider), + ), + ): + from prowler.providers.aws.services.cloudtrail.cloudtrail_bedrock_logging_enabled.cloudtrail_bedrock_logging_enabled import ( + cloudtrail_bedrock_logging_enabled, + ) + + check = cloudtrail_bedrock_logging_enabled() + result = check.execute() + + assert len(result) == 1 + assert result[0].status == "FAIL" + assert ( + result[0].status_extended + == "No CloudTrail trails are configured to log Amazon Bedrock API calls." + ) + + @mock_aws + def test_trail_with_advanced_management_events(self): + """Test PASS when a trail has unrestricted advanced management selectors.""" + cloudtrail_client_us = client("cloudtrail", region_name=AWS_REGION_US_EAST_1) + s3_client_us = client("s3", region_name=AWS_REGION_US_EAST_1) + trail_name = "trail_test" + bucket_name = "bucket_test" + s3_client_us.create_bucket(Bucket=bucket_name) + trail = cloudtrail_client_us.create_trail( + Name=trail_name, S3BucketName=bucket_name, IsMultiRegionTrail=False + ) + cloudtrail_client_us.start_logging(Name=trail_name) + cloudtrail_client_us.put_event_selectors( + TrailName=trail_name, + AdvancedEventSelectors=[ + { + "Name": "Management events selector", + "FieldSelectors": [ + {"Field": "eventCategory", "Equals": ["Management"]}, + ], + }, + ], + ) + + from prowler.providers.aws.services.cloudtrail.cloudtrail_service import ( + Cloudtrail, + ) + + aws_provider = set_mocked_aws_provider() + + with ( + mock.patch( + "prowler.providers.common.provider.Provider.get_global_provider", + return_value=aws_provider, + ), + mock.patch( + f"{CHECK_MODULE_PATH}.cloudtrail_client", + new=Cloudtrail(aws_provider), + ), + ): + from prowler.providers.aws.services.cloudtrail.cloudtrail_bedrock_logging_enabled.cloudtrail_bedrock_logging_enabled import ( + cloudtrail_bedrock_logging_enabled, + ) + + check = cloudtrail_bedrock_logging_enabled() + result = check.execute() + + assert len(result) == 1 + assert result[0].status == "PASS" + assert ( + result[0].status_extended + == f"Trail {trail_name} from home region {AWS_REGION_US_EAST_1} has an advanced management event selector to log Amazon Bedrock control-plane API calls." + ) + assert result[0].resource_id == trail_name + assert result[0].resource_arn == trail["TrailARN"] + assert result[0].region == AWS_REGION_US_EAST_1 + + @mock_aws + @pytest.mark.parametrize( + "event_source", + [ + pytest.param("bedrock.amazonaws.com", id="bedrock"), + pytest.param("bedrock-agent.amazonaws.com", id="bedrock-agent"), + pytest.param("bedrock-runtime.amazonaws.com", id="bedrock-runtime"), + pytest.param( + "bedrock-agent-runtime.amazonaws.com", + id="bedrock-agent-runtime", + ), + pytest.param( + "bedrock-data-automation.amazonaws.com", + id="bedrock-data-automation", + ), + pytest.param( + "bedrock-data-automation-runtime.amazonaws.com", + id="bedrock-data-automation-runtime", + ), + ], + ) + def test_trail_with_advanced_management_events_bedrock_event_sources( + self, event_source + ): + """Test PASS when advanced management events are scoped to Bedrock family sources.""" + from prowler.providers.aws.services.cloudtrail.cloudtrail_service import ( + Cloudtrail, + Event_Selector, + ) + + cloudtrail_client_us = client("cloudtrail", region_name=AWS_REGION_US_EAST_1) + s3_client_us = client("s3", region_name=AWS_REGION_US_EAST_1) + trail_name = "trail_test" + bucket_name = "bucket_test" + s3_client_us.create_bucket(Bucket=bucket_name) + trail = cloudtrail_client_us.create_trail( + Name=trail_name, S3BucketName=bucket_name, IsMultiRegionTrail=False + ) + cloudtrail_client_us.start_logging(Name=trail_name) + + aws_provider = set_mocked_aws_provider() + + with ( + mock.patch( + "prowler.providers.common.provider.Provider.get_global_provider", + return_value=aws_provider, + ), + mock.patch( + f"{CHECK_MODULE_PATH}.cloudtrail_client", + new=Cloudtrail(aws_provider), + ) as mock_cloudtrail_client, + ): + trail_arn = trail["TrailARN"] + mock_cloudtrail_client.trails[trail_arn].data_events = [ + Event_Selector( + is_advanced=True, + event_selector={ + "Name": "Bedrock management events selector", + "FieldSelectors": [ + {"Field": "eventCategory", "Equals": ["Management"]}, + {"Field": "eventSource", "Equals": [event_source]}, + ], + }, + ) + ] + + from prowler.providers.aws.services.cloudtrail.cloudtrail_bedrock_logging_enabled.cloudtrail_bedrock_logging_enabled import ( + cloudtrail_bedrock_logging_enabled, + ) + + check = cloudtrail_bedrock_logging_enabled() + result = check.execute() + + assert len(result) == 1 + assert result[0].status == "PASS" + assert ( + result[0].status_extended + == f"Trail {trail_name} from home region {AWS_REGION_US_EAST_1} has an advanced management event selector to log Amazon Bedrock control-plane API calls." + ) + assert result[0].resource_id == trail_name + assert result[0].resource_arn == trail["TrailARN"] + assert result[0].region == AWS_REGION_US_EAST_1 + + @mock_aws + def test_trail_with_advanced_bedrock_data_events(self): + """Test PASS when a trail has advanced event selectors for Bedrock resources.""" + from prowler.providers.aws.services.cloudtrail.cloudtrail_service import ( + Cloudtrail, + Event_Selector, + ) + + cloudtrail_client_us = client("cloudtrail", region_name=AWS_REGION_US_EAST_1) + s3_client_us = client("s3", region_name=AWS_REGION_US_EAST_1) + trail_name = "trail_test" + bucket_name = "bucket_test" + s3_client_us.create_bucket(Bucket=bucket_name) + trail = cloudtrail_client_us.create_trail( + Name=trail_name, S3BucketName=bucket_name, IsMultiRegionTrail=False + ) + cloudtrail_client_us.start_logging(Name=trail_name) + + aws_provider = set_mocked_aws_provider() + + with ( + mock.patch( + "prowler.providers.common.provider.Provider.get_global_provider", + return_value=aws_provider, + ), + mock.patch( + f"{CHECK_MODULE_PATH}.cloudtrail_client", + new=Cloudtrail(aws_provider), + ) as mock_cloudtrail_client, + ): + # Manually inject the Bedrock advanced event selector since moto + # does not support Bedrock resource types. + trail_arn = trail["TrailARN"] + mock_cloudtrail_client.trails[trail_arn].data_events = [ + Event_Selector( + is_advanced=True, + event_selector={ + "Name": "Bedrock data events", + "FieldSelectors": [ + {"Field": "eventCategory", "Equals": ["Data"]}, + { + "Field": "resources.type", + "Equals": ["AWS::Bedrock::Model"], + }, + ], + }, + ) + ] + + from prowler.providers.aws.services.cloudtrail.cloudtrail_bedrock_logging_enabled.cloudtrail_bedrock_logging_enabled import ( + cloudtrail_bedrock_logging_enabled, + ) + + check = cloudtrail_bedrock_logging_enabled() + result = check.execute() + + assert len(result) == 1 + assert result[0].status == "PASS" + assert ( + result[0].status_extended + == f"Trail {trail_name} from home region {AWS_REGION_US_EAST_1} has an advanced data event selector to log Amazon Bedrock API calls." + ) + assert result[0].resource_id == trail_name + assert result[0].resource_arn == trail["TrailARN"] + assert result[0].region == AWS_REGION_US_EAST_1 + + @mock_aws + def test_trail_with_advanced_bedrock_guardrail_events(self): + """Test PASS when a trail has advanced event selectors for Bedrock Guardrail resources.""" + from prowler.providers.aws.services.cloudtrail.cloudtrail_service import ( + Cloudtrail, + Event_Selector, + ) + + cloudtrail_client_us = client("cloudtrail", region_name=AWS_REGION_US_EAST_1) + s3_client_us = client("s3", region_name=AWS_REGION_US_EAST_1) + trail_name = "trail_test" + bucket_name = "bucket_test" + s3_client_us.create_bucket(Bucket=bucket_name) + trail = cloudtrail_client_us.create_trail( + Name=trail_name, S3BucketName=bucket_name, IsMultiRegionTrail=False + ) + cloudtrail_client_us.start_logging(Name=trail_name) + + aws_provider = set_mocked_aws_provider() + + with ( + mock.patch( + "prowler.providers.common.provider.Provider.get_global_provider", + return_value=aws_provider, + ), + mock.patch( + f"{CHECK_MODULE_PATH}.cloudtrail_client", + new=Cloudtrail(aws_provider), + ) as mock_cloudtrail_client, + ): + # Manually inject the Bedrock Guardrail advanced event selector + # since moto does not support Bedrock resource types. + trail_arn = trail["TrailARN"] + mock_cloudtrail_client.trails[trail_arn].data_events = [ + Event_Selector( + is_advanced=True, + event_selector={ + "Name": "Bedrock guardrail events", + "FieldSelectors": [ + {"Field": "eventCategory", "Equals": ["Data"]}, + { + "Field": "resources.type", + "Equals": ["AWS::Bedrock::Guardrail"], + }, + ], + }, + ) + ] + + from prowler.providers.aws.services.cloudtrail.cloudtrail_bedrock_logging_enabled.cloudtrail_bedrock_logging_enabled import ( + cloudtrail_bedrock_logging_enabled, + ) + + check = cloudtrail_bedrock_logging_enabled() + result = check.execute() + + assert len(result) == 1 + assert result[0].status == "PASS" + assert ( + result[0].status_extended + == f"Trail {trail_name} from home region {AWS_REGION_US_EAST_1} has an advanced data event selector to log Amazon Bedrock API calls." + ) + assert result[0].resource_id == trail_name + assert result[0].resource_arn == trail["TrailARN"] + assert result[0].region == AWS_REGION_US_EAST_1 + + @mock_aws + def test_trail_with_advanced_non_bedrock_data_events(self): + """Test FAIL when a trail has advanced event selectors for non-Bedrock resources.""" + cloudtrail_client_us = client("cloudtrail", region_name=AWS_REGION_US_EAST_1) + s3_client_us = client("s3", region_name=AWS_REGION_US_EAST_1) + trail_name = "trail_test" + bucket_name = "bucket_test" + s3_client_us.create_bucket(Bucket=bucket_name) + cloudtrail_client_us.create_trail( + Name=trail_name, S3BucketName=bucket_name, IsMultiRegionTrail=False + ) + cloudtrail_client_us.start_logging(Name=trail_name) + cloudtrail_client_us.put_event_selectors( + TrailName=trail_name, + AdvancedEventSelectors=[ + { + "Name": "S3 data events", + "FieldSelectors": [ + {"Field": "eventCategory", "Equals": ["Data"]}, + { + "Field": "resources.type", + "Equals": ["AWS::S3::Object"], + }, + ], + }, + ], + ) + + from prowler.providers.aws.services.cloudtrail.cloudtrail_service import ( + Cloudtrail, + ) + + aws_provider = set_mocked_aws_provider() + + with ( + mock.patch( + "prowler.providers.common.provider.Provider.get_global_provider", + return_value=aws_provider, + ), + mock.patch( + f"{CHECK_MODULE_PATH}.cloudtrail_client", + new=Cloudtrail(aws_provider), + ), + ): + from prowler.providers.aws.services.cloudtrail.cloudtrail_bedrock_logging_enabled.cloudtrail_bedrock_logging_enabled import ( + cloudtrail_bedrock_logging_enabled, + ) + + check = cloudtrail_bedrock_logging_enabled() + result = check.execute() + + assert len(result) == 1 + assert result[0].status == "FAIL" + assert ( + result[0].status_extended + == "No CloudTrail trails are configured to log Amazon Bedrock API calls." + ) + + @mock_aws + def test_trail_with_classic_management_events_write_only(self): + """Test PASS when a trail has management events with ReadWriteType WriteOnly.""" + cloudtrail_client_us = client("cloudtrail", region_name=AWS_REGION_US_EAST_1) + s3_client_us = client("s3", region_name=AWS_REGION_US_EAST_1) + trail_name = "trail_test" + bucket_name = "bucket_test" + s3_client_us.create_bucket(Bucket=bucket_name) + trail = cloudtrail_client_us.create_trail( + Name=trail_name, S3BucketName=bucket_name, IsMultiRegionTrail=False + ) + cloudtrail_client_us.start_logging(Name=trail_name) + cloudtrail_client_us.put_event_selectors( + TrailName=trail_name, + EventSelectors=[ + { + "ReadWriteType": "WriteOnly", + "IncludeManagementEvents": True, + "DataResources": [], + } + ], + ) + + from prowler.providers.aws.services.cloudtrail.cloudtrail_service import ( + Cloudtrail, + ) + + aws_provider = set_mocked_aws_provider() + + with ( + mock.patch( + "prowler.providers.common.provider.Provider.get_global_provider", + return_value=aws_provider, + ), + mock.patch( + f"{CHECK_MODULE_PATH}.cloudtrail_client", + new=Cloudtrail(aws_provider), + ), + ): + from prowler.providers.aws.services.cloudtrail.cloudtrail_bedrock_logging_enabled.cloudtrail_bedrock_logging_enabled import ( + cloudtrail_bedrock_logging_enabled, + ) + + check = cloudtrail_bedrock_logging_enabled() + result = check.execute() + + assert len(result) == 1 + assert result[0].status == "PASS" + assert ( + result[0].status_extended + == f"Trail {trail_name} from home region {AWS_REGION_US_EAST_1} has management events enabled to log Amazon Bedrock control-plane API calls." + ) + assert result[0].resource_id == trail_name + assert result[0].resource_arn == trail["TrailARN"] + assert result[0].region == AWS_REGION_US_EAST_1 + + @mock_aws + def test_trail_with_classic_management_events_default_read_write_type(self): + """Test PASS when a classic selector omits ReadWriteType and uses the AWS default.""" + cloudtrail_client_us = client("cloudtrail", region_name=AWS_REGION_US_EAST_1) + s3_client_us = client("s3", region_name=AWS_REGION_US_EAST_1) + trail_name = "trail_test" + bucket_name = "bucket_test" + s3_client_us.create_bucket(Bucket=bucket_name) + trail = cloudtrail_client_us.create_trail( + Name=trail_name, S3BucketName=bucket_name, IsMultiRegionTrail=False + ) + cloudtrail_client_us.start_logging(Name=trail_name) + cloudtrail_client_us.put_event_selectors( + TrailName=trail_name, + EventSelectors=[ + { + "IncludeManagementEvents": True, + "DataResources": [], + } + ], + ) + + from prowler.providers.aws.services.cloudtrail.cloudtrail_service import ( + Cloudtrail, + ) + + aws_provider = set_mocked_aws_provider() + + with ( + mock.patch( + "prowler.providers.common.provider.Provider.get_global_provider", + return_value=aws_provider, + ), + mock.patch( + f"{CHECK_MODULE_PATH}.cloudtrail_client", + new=Cloudtrail(aws_provider), + ), + ): + from prowler.providers.aws.services.cloudtrail.cloudtrail_bedrock_logging_enabled.cloudtrail_bedrock_logging_enabled import ( + cloudtrail_bedrock_logging_enabled, + ) + + check = cloudtrail_bedrock_logging_enabled() + result = check.execute() + + assert len(result) == 1 + assert result[0].status == "PASS" + assert ( + result[0].status_extended + == f"Trail {trail_name} from home region {AWS_REGION_US_EAST_1} has management events enabled to log Amazon Bedrock control-plane API calls." + ) + assert result[0].resource_id == trail_name + assert result[0].resource_arn == trail["TrailARN"] + assert result[0].region == AWS_REGION_US_EAST_1 + + @mock_aws + def test_trail_with_advanced_management_events_read_only(self): + """Test FAIL when advanced management event selector has readOnly=true restriction.""" + from prowler.providers.aws.services.cloudtrail.cloudtrail_service import ( + Cloudtrail, + Event_Selector, + ) + + cloudtrail_client_us = client("cloudtrail", region_name=AWS_REGION_US_EAST_1) + s3_client_us = client("s3", region_name=AWS_REGION_US_EAST_1) + trail_name = "trail_test" + bucket_name = "bucket_test" + s3_client_us.create_bucket(Bucket=bucket_name) + trail = cloudtrail_client_us.create_trail( + Name=trail_name, S3BucketName=bucket_name, IsMultiRegionTrail=False + ) + cloudtrail_client_us.start_logging(Name=trail_name) + + aws_provider = set_mocked_aws_provider() + + with ( + mock.patch( + "prowler.providers.common.provider.Provider.get_global_provider", + return_value=aws_provider, + ), + mock.patch( + f"{CHECK_MODULE_PATH}.cloudtrail_client", + new=Cloudtrail(aws_provider), + ) as mock_cloudtrail_client, + ): + trail_arn = trail["TrailARN"] + mock_cloudtrail_client.trails[trail_arn].data_events = [ + Event_Selector( + is_advanced=True, + event_selector={ + "Name": "Management events selector", + "FieldSelectors": [ + {"Field": "eventCategory", "Equals": ["Management"]}, + {"Field": "readOnly", "Equals": ["true"]}, + ], + }, + ) + ] + + from prowler.providers.aws.services.cloudtrail.cloudtrail_bedrock_logging_enabled.cloudtrail_bedrock_logging_enabled import ( + cloudtrail_bedrock_logging_enabled, + ) + + check = cloudtrail_bedrock_logging_enabled() + result = check.execute() + + assert len(result) == 1 + assert result[0].status == "FAIL" + assert ( + result[0].status_extended + == "No CloudTrail trails are configured to log Amazon Bedrock API calls." + ) + + @mock_aws + def test_trail_with_advanced_management_events_read_only_not_equals_false(self): + """Test FAIL when advanced management selector restricts events with NotEquals false.""" + from prowler.providers.aws.services.cloudtrail.cloudtrail_service import ( + Cloudtrail, + Event_Selector, + ) + + cloudtrail_client_us = client("cloudtrail", region_name=AWS_REGION_US_EAST_1) + s3_client_us = client("s3", region_name=AWS_REGION_US_EAST_1) + trail_name = "trail_test" + bucket_name = "bucket_test" + s3_client_us.create_bucket(Bucket=bucket_name) + trail = cloudtrail_client_us.create_trail( + Name=trail_name, S3BucketName=bucket_name, IsMultiRegionTrail=False + ) + cloudtrail_client_us.start_logging(Name=trail_name) + + aws_provider = set_mocked_aws_provider() + + with ( + mock.patch( + "prowler.providers.common.provider.Provider.get_global_provider", + return_value=aws_provider, + ), + mock.patch( + f"{CHECK_MODULE_PATH}.cloudtrail_client", + new=Cloudtrail(aws_provider), + ) as mock_cloudtrail_client, + ): + trail_arn = trail["TrailARN"] + mock_cloudtrail_client.trails[trail_arn].data_events = [ + Event_Selector( + is_advanced=True, + event_selector={ + "Name": "Management events selector", + "FieldSelectors": [ + {"Field": "eventCategory", "Equals": ["Management"]}, + {"Field": "readOnly", "NotEquals": ["false"]}, + ], + }, + ) + ] + + from prowler.providers.aws.services.cloudtrail.cloudtrail_bedrock_logging_enabled.cloudtrail_bedrock_logging_enabled import ( + cloudtrail_bedrock_logging_enabled, + ) + + check = cloudtrail_bedrock_logging_enabled() + result = check.execute() + + assert len(result) == 1 + assert result[0].status == "FAIL" + assert ( + result[0].status_extended + == "No CloudTrail trails are configured to log Amazon Bedrock API calls." + ) + + @mock_aws + def test_trail_with_advanced_management_events_other_service_event_source(self): + """Test FAIL when advanced management events are scoped to a non-Bedrock event source.""" + from prowler.providers.aws.services.cloudtrail.cloudtrail_service import ( + Cloudtrail, + Event_Selector, + ) + + cloudtrail_client_us = client("cloudtrail", region_name=AWS_REGION_US_EAST_1) + s3_client_us = client("s3", region_name=AWS_REGION_US_EAST_1) + trail_name = "trail_test" + bucket_name = "bucket_test" + s3_client_us.create_bucket(Bucket=bucket_name) + trail = cloudtrail_client_us.create_trail( + Name=trail_name, S3BucketName=bucket_name, IsMultiRegionTrail=False + ) + cloudtrail_client_us.start_logging(Name=trail_name) + + aws_provider = set_mocked_aws_provider() + + with ( + mock.patch( + "prowler.providers.common.provider.Provider.get_global_provider", + return_value=aws_provider, + ), + mock.patch( + f"{CHECK_MODULE_PATH}.cloudtrail_client", + new=Cloudtrail(aws_provider), + ) as mock_cloudtrail_client, + ): + trail_arn = trail["TrailARN"] + mock_cloudtrail_client.trails[trail_arn].data_events = [ + Event_Selector( + is_advanced=True, + event_selector={ + "Name": "EC2 management events selector", + "FieldSelectors": [ + {"Field": "eventCategory", "Equals": ["Management"]}, + { + "Field": "eventSource", + "Equals": ["ec2.amazonaws.com"], + }, + ], + }, + ) + ] + + from prowler.providers.aws.services.cloudtrail.cloudtrail_bedrock_logging_enabled.cloudtrail_bedrock_logging_enabled import ( + cloudtrail_bedrock_logging_enabled, + ) + + check = cloudtrail_bedrock_logging_enabled() + result = check.execute() + + assert len(result) == 1 + assert result[0].status == "FAIL" + assert ( + result[0].status_extended + == "No CloudTrail trails are configured to log Amazon Bedrock API calls." + ) + + @mock_aws + def test_access_denied(self): + """Test when trails are None due to access denied.""" + from prowler.providers.aws.services.cloudtrail.cloudtrail_service import ( + Cloudtrail, + ) + + aws_provider = set_mocked_aws_provider() + + with ( + mock.patch( + "prowler.providers.common.provider.Provider.get_global_provider", + return_value=aws_provider, + ), + mock.patch( + f"{CHECK_MODULE_PATH}.cloudtrail_client", + new=Cloudtrail(aws_provider), + ) as mock_cloudtrail_client, + ): + from prowler.providers.aws.services.cloudtrail.cloudtrail_bedrock_logging_enabled.cloudtrail_bedrock_logging_enabled import ( + cloudtrail_bedrock_logging_enabled, + ) + + mock_cloudtrail_client.trails = None + check = cloudtrail_bedrock_logging_enabled() + result = check.execute() + + assert len(result) == 0 + + @mock_aws + @pytest.mark.parametrize( + ("selector", "expected"), + [ + pytest.param( + {"Equals": ["bedrock.amazonaws.com"]}, + True, + id="equals-match", + ), + pytest.param( + {"Equals": ["ec2.amazonaws.com"]}, + False, + id="equals-mismatch", + ), + pytest.param( + {"NotEquals": ["ec2.amazonaws.com"]}, + True, + id="not-equals-match", + ), + pytest.param( + {"NotEquals": ["bedrock.amazonaws.com"]}, + False, + id="not-equals-mismatch", + ), + pytest.param( + {"StartsWith": ["bedrock."]}, + True, + id="starts-with-match", + ), + pytest.param( + {"StartsWith": ["ec2."]}, + False, + id="starts-with-mismatch", + ), + pytest.param( + {"NotStartsWith": ["ec2."]}, + True, + id="not-starts-with-match", + ), + pytest.param( + {"NotStartsWith": ["bedrock."]}, + False, + id="not-starts-with-mismatch", + ), + pytest.param( + {"EndsWith": [".amazonaws.com"]}, + True, + id="ends-with-match", + ), + pytest.param( + {"EndsWith": [".amazonaws.org"]}, + False, + id="ends-with-mismatch", + ), + pytest.param( + {"NotEndsWith": [".amazonaws.org"]}, + True, + id="not-ends-with-match", + ), + pytest.param( + {"NotEndsWith": [".amazonaws.com"]}, + False, + id="not-ends-with-mismatch", + ), + pytest.param({}, True, id="no-conditions"), + ], + ) + def test_field_selector_matches_value(self, selector, expected): + """Test advanced field selector operators against the Bedrock event source.""" + from prowler.providers.aws.services.cloudtrail.cloudtrail_service import ( + Cloudtrail, + ) + + aws_provider = set_mocked_aws_provider() + + with ( + mock.patch( + "prowler.providers.common.provider.Provider.get_global_provider", + return_value=aws_provider, + ), + mock.patch( + f"{CHECK_MODULE_PATH}.cloudtrail_client", + new=Cloudtrail(aws_provider), + ), + ): + from prowler.providers.aws.services.cloudtrail.cloudtrail_bedrock_logging_enabled.cloudtrail_bedrock_logging_enabled import ( + cloudtrail_bedrock_logging_enabled, + ) + + assert ( + cloudtrail_bedrock_logging_enabled._field_selector_matches_value( + "bedrock.amazonaws.com", selector + ) + is expected + )