diff --git a/docs/user-guide/cli/tutorials/configuration_file.mdx b/docs/user-guide/cli/tutorials/configuration_file.mdx index 90de1dbdbc..ac21f5bc6c 100644 --- a/docs/user-guide/cli/tutorials/configuration_file.mdx +++ b/docs/user-guide/cli/tutorials/configuration_file.mdx @@ -51,6 +51,7 @@ The following list includes all the AWS checks with configurable variables that | `cloudtrail_threat_detection_privilege_escalation` | `threat_detection_privilege_escalation_actions` | List of Strings | See `config.yaml` | | `cloudtrail_threat_detection_privilege_escalation` | `threat_detection_privilege_escalation_minutes` | Integer | `1440` | | `cloudtrail_threat_detection_privilege_escalation` | `threat_detection_privilege_escalation_threshold` | Float | `0.2` | +| `cloudwatch_log_group_agentcore_data_protection_policy_enabled` | `agentcore_log_group_name_prefixes` | List of Strings | See `config.yaml` | | `cloudwatch_log_group_no_secrets_in_logs` | `secrets_ignore_patterns` | List of Strings | `[]` | | `cloudwatch_log_group_retention_policy_specific_days_enabled` | `log_group_retention_days` | Integer | `365` | | `codebuild_project_no_secrets_in_variables` | `excluded_sensitive_environment_variables` | List of Strings | `[]` | diff --git a/prowler/changelog.d/cloudwatch-log-group-agentcore-data-protection-policy-enabled.added.md b/prowler/changelog.d/cloudwatch-log-group-agentcore-data-protection-policy-enabled.added.md new file mode 100644 index 0000000000..5e68d03676 --- /dev/null +++ b/prowler/changelog.d/cloudwatch-log-group-agentcore-data-protection-policy-enabled.added.md @@ -0,0 +1 @@ +`cloudwatch_log_group_agentcore_data_protection_policy_enabled` check for AWS provider, verifying that Bedrock AgentCore log groups mask sensitive data with a CloudWatch Logs data protection policy diff --git a/prowler/compliance/aws/aws_ai_security_framework_aws.json b/prowler/compliance/aws/aws_ai_security_framework_aws.json index e69acaa128..32e50d66e9 100644 --- a/prowler/compliance/aws/aws_ai_security_framework_aws.json +++ b/prowler/compliance/aws/aws_ai_security_framework_aws.json @@ -487,7 +487,8 @@ "awslambda_function_no_secrets_in_variables", "ecs_task_definitions_no_environment_secrets", "ec2_instance_secrets_user_data", - "cloudwatch_log_group_no_secrets_in_logs" + "cloudwatch_log_group_no_secrets_in_logs", + "cloudwatch_log_group_agentcore_data_protection_policy_enabled" ] }, { diff --git a/prowler/config/config.yaml b/prowler/config/config.yaml index 7dd7c2bf3d..46127c9699 100644 --- a/prowler/config/config.yaml +++ b/prowler/config/config.yaml @@ -119,6 +119,14 @@ aws: # aws.cloudwatch_log_group_retention_policy_specific_days_enabled --> by default is 365 days log_group_retention_days: 365 + # aws.cloudwatch_log_group_agentcore_data_protection_policy_enabled + # Log group name prefixes that hold Bedrock AgentCore agent telemetry. This list REPLACES + # the defaults rather than adding to them, so keep both entries below when adding your own + # or the log groups AgentCore creates itself stop being assessed. + agentcore_log_group_name_prefixes: + - "/aws/bedrock-agentcore/" + - "/aws/vendedlogs/bedrock-agentcore/" + # AWS CloudFormation Configuration # cloudformation_stack_cdktoolkit_bootstrap_version --> by default is 21 recommended_cdk_bootstrap_version: 21 diff --git a/prowler/config/schema/aws.py b/prowler/config/schema/aws.py index c0ecf9a21c..75fa5f73c0 100644 --- a/prowler/config/schema/aws.py +++ b/prowler/config/schema/aws.py @@ -267,6 +267,15 @@ class AWSProviderConfig(ProviderConfigBase): f"values accepted by the AWS API: {list(_CLOUDWATCH_RETENTION_DAYS)}." ), ) + agentcore_log_group_name_prefixes: Optional[list[str]] = Field( + default=None, + description=( + "Log group name prefixes that identify Bedrock AgentCore agent " + "telemetry. Set this when AgentCore log delivery is pointed at log " + "groups outside the service defaults; the value replaces the " + "defaults, so list them alongside any prefix of your own." + ), + ) recommended_cdk_bootstrap_version: Optional[int] = Field( default=None, ge=1, diff --git a/prowler/providers/aws/services/cloudwatch/cloudwatch_log_group_agentcore_data_protection_policy_enabled/__init__.py b/prowler/providers/aws/services/cloudwatch/cloudwatch_log_group_agentcore_data_protection_policy_enabled/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/prowler/providers/aws/services/cloudwatch/cloudwatch_log_group_agentcore_data_protection_policy_enabled/cloudwatch_log_group_agentcore_data_protection_policy_enabled.metadata.json b/prowler/providers/aws/services/cloudwatch/cloudwatch_log_group_agentcore_data_protection_policy_enabled/cloudwatch_log_group_agentcore_data_protection_policy_enabled.metadata.json new file mode 100644 index 0000000000..063474adbf --- /dev/null +++ b/prowler/providers/aws/services/cloudwatch/cloudwatch_log_group_agentcore_data_protection_policy_enabled/cloudwatch_log_group_agentcore_data_protection_policy_enabled.metadata.json @@ -0,0 +1,45 @@ +{ + "Provider": "aws", + "CheckID": "cloudwatch_log_group_agentcore_data_protection_policy_enabled", + "CheckTitle": "Bedrock AgentCore log groups have a CloudWatch Logs data protection policy activated", + "CheckType": [ + "Software and Configuration Checks/AWS Security Best Practices", + "Effects/Data Exposure", + "Sensitive Data Identifications/PII" + ], + "ServiceName": "cloudwatch", + "SubServiceName": "logs", + "ResourceIdTemplate": "arn:partition:logs:region:account-id:log-group:log-group-name", + "Severity": "medium", + "ResourceType": "Other", + "ResourceGroup": "monitoring", + "Description": "Log groups holding **Bedrock AgentCore** agent telemetry have an active CloudWatch Logs **data protection policy**, so sensitive data an agent writes to its own logs is masked at ingestion. A policy attached to the log group or inherited from an account-level policy both satisfy this.", + "Risk": "Agents log prompts, tool arguments and retrieved context. Without masking, **PII** and secrets are stored in clear text and readable by every principal holding `logs:GetLogEvents`, widening the audience for regulated data far beyond the agent and defeating filtering applied only at the model boundary.", + "RelatedUrl": "", + "AdditionalURLs": [ + "https://docs.aws.amazon.com/AmazonCloudWatch/latest/logs/mask-sensitive-log-data.html", + "https://docs.aws.amazon.com/AmazonCloudWatch/latest/logs/mask-sensitive-log-data-start.html", + "https://docs.aws.amazon.com/AmazonCloudWatch/latest/logs/mask-sensitive-log-data-start-account.html", + "https://docs.aws.amazon.com/AmazonCloudWatch/latest/logs/protect-sensitive-log-data-types.html", + "https://docs.aws.amazon.com/bedrock-agentcore/latest/devguide/observability-configure.html" + ], + "Remediation": { + "Code": { + "CLI": "aws logs put-data-protection-policy --log-group-identifier --policy-document '{\"Name\":\"agentcore-data-protection-policy\",\"Version\":\"2021-06-01\",\"Statement\":[{\"Sid\":\"audit\",\"DataIdentifier\":[\"arn:aws:dataprotection::aws:data-identifier/EmailAddress\"],\"Operation\":{\"Audit\":{\"FindingsDestination\":{}}}},{\"Sid\":\"redact\",\"DataIdentifier\":[\"arn:aws:dataprotection::aws:data-identifier/EmailAddress\"],\"Operation\":{\"Deidentify\":{\"MaskConfig\":{}}}}]}'", + "NativeIaC": "```yaml\n# CloudFormation: data protection policy on a log group you manage in the template.\n# AgentCore runtime log groups are created by the service, not by CloudFormation --\n# for those, use the CLI command or an AWS::Logs::AccountPolicy instead.\nResources:\n AgentCoreLogGroup:\n Type: AWS::Logs::LogGroup\n Properties:\n LogGroupName: \"\"\n DataProtectionPolicy:\n Name: agentcore-data-protection-policy\n Version: '2021-06-01'\n Statement:\n - Sid: audit # CRITICAL: required block, finds the sensitive terms\n DataIdentifier:\n - arn:aws:dataprotection::aws:data-identifier/EmailAddress\n Operation:\n Audit:\n FindingsDestination: {}\n - Sid: redact # CRITICAL: the Deidentify block is what actually masks the data\n DataIdentifier:\n - arn:aws:dataprotection::aws:data-identifier/EmailAddress\n Operation:\n Deidentify:\n MaskConfig: {}\n```", + "Other": "1. In the AWS console, open CloudWatch > Logs > Log groups\n2. Select the AgentCore log group (/aws/bedrock-agentcore/... or /aws/vendedlogs/bedrock-agentcore/...)\n3. Choose Actions > Create data protection policy\n4. Under Managed data identifiers, select the data types your agent can log, for example Address, EmailAddress and PhoneNumber\n5. (Optional) Choose a destination for the audit findings report\n6. Choose Activate data protection\n7. To cover every log group at once, including service-created ones, open Settings > Data protection instead and create an account-level policy", + "Terraform": "```hcl\n# Attaches a data protection policy to an existing AgentCore log group\nresource \"aws_cloudwatch_log_data_protection_policy\" \"\" {\n log_group_name = \"\"\n\n policy_document = jsonencode({\n Name = \"agentcore-data-protection-policy\"\n Version = \"2021-06-01\"\n Statement = [\n {\n Sid = \"audit\" # CRITICAL: required block, finds the sensitive terms\n DataIdentifier = [\"arn:aws:dataprotection::aws:data-identifier/EmailAddress\"]\n Operation = {\n Audit = {\n FindingsDestination = {}\n }\n }\n },\n {\n Sid = \"redact\" # CRITICAL: the Deidentify block is what actually masks the data\n DataIdentifier = [\"arn:aws:dataprotection::aws:data-identifier/EmailAddress\"]\n Operation = {\n Deidentify = {\n MaskConfig = {}\n }\n }\n }\n ]\n })\n}\n```" + }, + "Recommendation": { + "Text": "Attach a data protection policy to every AgentCore log group, choosing the managed data identifiers your agents can plausibly log, or create one account-level policy so service-created log groups are covered as they appear.\n\nMasking is *defense in depth* behind the model-boundary filters: keep restricting `logs:Unmask` so only incident responders can read the original values, and keep the log groups encrypted with a **customer-managed key**.", + "Url": "https://hub.prowler.com/check/cloudwatch_log_group_agentcore_data_protection_policy_enabled" + } + }, + "Categories": [ + "gen-ai", + "logging" + ], + "DependsOn": [], + "RelatedTo": [], + "Notes": "Scoped to log groups whose name starts with an AgentCore prefix (configurable via `agentcore_log_group_name_prefixes`, whose value REPLACES the defaults rather than adding to them). AgentCore log delivery can be pointed at an arbitrarily named log group; add those prefixes to the configuration so they are assessed. Not covered by default: gateway, built-in tool and identity spans, and runtime spans for agents created before a Region supported the per-agent span destination, are delivered to the shared `aws/spans` log group, which is account-wide rather than AgentCore-specific. Add `aws/spans` to the configuration to assess it." +} diff --git a/prowler/providers/aws/services/cloudwatch/cloudwatch_log_group_agentcore_data_protection_policy_enabled/cloudwatch_log_group_agentcore_data_protection_policy_enabled.py b/prowler/providers/aws/services/cloudwatch/cloudwatch_log_group_agentcore_data_protection_policy_enabled/cloudwatch_log_group_agentcore_data_protection_policy_enabled.py new file mode 100644 index 0000000000..339f8dad4f --- /dev/null +++ b/prowler/providers/aws/services/cloudwatch/cloudwatch_log_group_agentcore_data_protection_policy_enabled/cloudwatch_log_group_agentcore_data_protection_policy_enabled.py @@ -0,0 +1,98 @@ +from prowler.lib.check.models import Check, Check_Report_AWS +from prowler.providers.aws.services.cloudwatch.logs_client import logs_client + +# Log group name prefixes AgentCore uses for agent telemetry. The runtime creates +# /aws/bedrock-agentcore/runtimes/... itself. Log delivery for memory, gateway and built-in tool +# resources is CONFIGURED by the operator rather than defaulted -- the AgentCore devguide's +# observability-configure page is a put_delivery_source / put_delivery_destination / +# create_delivery procedure -- and it targets /aws/vendedlogs/bedrock-agentcore/... because for +# same-account delivery to a /aws/vendedlogs/ log group the log-delivery service-linked role +# grants write access implicitly, while any other destination needs an explicit resource policy +# or the delivery silently fails. So the prefix is the convention that makes delivery work, which +# is why these two and not others. +DEFAULT_AGENTCORE_LOG_GROUP_PREFIXES = [ + "/aws/bedrock-agentcore/", + "/aws/vendedlogs/bedrock-agentcore/", +] + +ACTIVATED = "ACTIVATED" +ACCOUNT_DATA_PROTECTION = "ACCOUNT_DATA_PROTECTION" + + +class cloudwatch_log_group_agentcore_data_protection_policy_enabled(Check): + """Ensure AgentCore log groups mask sensitive data with a data protection policy. + + Agents write prompts, tool arguments and retrieved context to their own log groups. A data + protection policy masks matched data at ingestion, so without one the values are stored in + clear text and readable by every principal holding logs:GetLogEvents. + + Scope: log groups whose name starts with an AgentCore prefix. AgentCore log delivery can be + pointed at an arbitrarily named log group, so the prefix list is configurable through + agentcore_log_group_name_prefixes; a configured list REPLACES the defaults rather than + extending them, and an explicitly null value falls back to the defaults. + + PASS when the log group has an ACTIVATED policy of its own, or inherits the account-level one. + FAIL when dataProtectionStatus is DELETED, ARCHIVED or DISABLED, or was never reported: all + four mean nothing is being masked at ingestion today. + MANUAL when the log group inventory could not be read, because nothing is then known about any + log group's masking. Only a denied DescribeLogGroups leaves the inventory unknown -- every + other collector failure leaves a readable, possibly partial, inventory. + """ + + def execute(self) -> list[Check_Report_AWS]: + """Execute the AgentCore log group data protection policy check. + + Returns: + A list of reports containing the result of the check: one per in-scope + AgentCore log group, or a single account-level report when the log group + inventory could not be read. + """ + findings = [] + + # An ABSENT or explicitly null value falls back to the defaults; an empty LIST does not. + # `is None` covers both a missing key and a bare `agentcore_log_group_name_prefixes:` in the + # YAML, which parses as None, and both mean "not configured". `or` additionally swallowed an + # explicitly empty list, which IS a configured value and the one way an operator can say "no + # log group is in scope" -- so the fallback overrode the operator and contradicted the + # REPLACES-the-defaults behaviour promised above. An empty tuple is meaningful downstream + # rather than degenerate: str.startswith(()) is False for every name, so nothing is selected, + # which is exactly the request. + configured_prefixes = logs_client.audit_config.get( + "agentcore_log_group_name_prefixes" + ) + prefixes = tuple( + DEFAULT_AGENTCORE_LOG_GROUP_PREFIXES + if configured_prefixes is None + else configured_prefixes + ) + + # An unreadable log group inventory must not read as compliant: without the + # inventory there is no way to tell an AgentCore log group that masks + # sensitive data from one that does not. + if logs_client.log_groups is None: + report = Check_Report_AWS(metadata=self.metadata(), resource={}) + report.status = "MANUAL" + report.status_extended = "Log groups could not be retrieved, so data protection policies for AgentCore log groups could not be verified." + report.region = logs_client.region + report.resource_id = logs_client.audited_account + report.resource_arn = logs_client.log_group_arn_template + report.resource_tags = [] + return [report] + + for log_group in logs_client.log_groups.values(): + if not log_group.name.startswith(prefixes): + continue + + report = Check_Report_AWS(metadata=self.metadata(), resource=log_group) + if log_group.data_protection_status == ACTIVATED: + report.status = "PASS" + report.status_extended = f"AgentCore log group {log_group.name} has a data protection policy activated." + elif ACCOUNT_DATA_PROTECTION in log_group.inherited_properties: + report.status = "PASS" + report.status_extended = f"AgentCore log group {log_group.name} inherits the account-level data protection policy." + else: + report.status = "FAIL" + report.status_extended = f"AgentCore log group {log_group.name} does not have an active data protection policy, so sensitive data written by the agent is not masked." + findings.append(report) + + return findings diff --git a/prowler/providers/aws/services/cloudwatch/cloudwatch_service.py b/prowler/providers/aws/services/cloudwatch/cloudwatch_service.py index 56b7ebe25c..d01ce42142 100644 --- a/prowler/providers/aws/services/cloudwatch/cloudwatch_service.py +++ b/prowler/providers/aws/services/cloudwatch/cloudwatch_service.py @@ -188,6 +188,17 @@ class Logs(AWSService): ) def _describe_log_groups(self, regional_client): + """List the log groups in a region into the complete and the analysed indexes. + + A denied DescribeLogGroups sets both indexes to None, but only while nothing has been + collected yet: that None is the state checks read as "inventory unknown", and it must stay + distinguishable from an account that genuinely has no log groups. Any other failure leaves + the indexes as they are, so a partial inventory reads as a smaller one. + + dataProtectionStatus and inheritedProperties are stored as reported. An absent + dataProtectionStatus is the API saying the log group has never had a policy, and it is kept + as None rather than a status string so a check can tell "never configured" from DISABLED. + """ logger.info("CloudWatch Logs - Describing log groups...") try: describe_log_groups_paginator = regional_client.get_paginator( @@ -215,6 +226,12 @@ class Logs(AWSService): never_expire=never_expire, kms_id=kms, creation_time=log_group.get("creationTime"), + data_protection_status=log_group.get( + "dataProtectionStatus" + ), + inherited_properties=log_group.get( + "inheritedProperties", [] + ), region=regional_client.region, ) self.all_log_groups[log_group_object.arn] = log_group_object @@ -337,6 +354,11 @@ class LogGroup(BaseModel): never_expire: bool kms_id: Optional[str] creation_time: Optional[int] = None + # None when the log group has never had a data protection policy, otherwise + # ACTIVATED, DELETED, ARCHIVED or DISABLED. + data_protection_status: Optional[str] = None + # Properties inherited from account-level settings, e.g. ACCOUNT_DATA_PROTECTION. + inherited_properties: list[str] = [] region: str log_streams: dict[str, list[str]] = ( {} diff --git a/tests/config/schema/aws_schema_test.py b/tests/config/schema/aws_schema_test.py index 838853f52a..f7c8e3b485 100644 --- a/tests/config/schema/aws_schema_test.py +++ b/tests/config/schema/aws_schema_test.py @@ -203,6 +203,53 @@ class TestAWSELBv2PQCTLSAllowedPolicies: assert _validate({"elbv2_listener_pqc_tls_allowed_policies": value}) == {} +class TestAWSAgentCoreLogGroupNamePrefixes: + def test_valid_prefix_list_round_trips(self): + prefixes = [ + "/aws/bedrock-agentcore/", + "/aws/vendedlogs/bedrock-agentcore/", + ] + + assert _validate({"agentcore_log_group_name_prefixes": prefixes}) == { + "agentcore_log_group_name_prefixes": prefixes + } + + def test_null_round_trips(self): + """A bare `agentcore_log_group_name_prefixes:` in the config file arrives as None. + + It has to survive validation rather than be dropped, because the check distinguishes it + from an empty list: None means "use the built-in defaults" while [] means "match no log + group". Dropping it would collapse the two. + """ + assert _validate({"agentcore_log_group_name_prefixes": None}) == { + "agentcore_log_group_name_prefixes": None + } + + def test_empty_list_round_trips(self): + """[] is a valid instruction, not a missing value -- see test_null_round_trips.""" + assert _validate({"agentcore_log_group_name_prefixes": []}) == { + "agentcore_log_group_name_prefixes": [] + } + + def test_key_is_exposed_in_scan_config_schema(self): + aws_properties = SCAN_CONFIG_SCHEMA["properties"]["aws"]["properties"] + + assert "agentcore_log_group_name_prefixes" in aws_properties + + @pytest.mark.parametrize( + "value", + [ + # A single prefix written without the list, which is the mistake the YAML invites. + "/aws/bedrock-agentcore/", + ["/aws/bedrock-agentcore/", 123], + {"prefix": "/aws/bedrock-agentcore/"}, + 123, + ], + ) + def test_invalid_prefix_values_are_dropped(self, value): + assert _validate({"agentcore_log_group_name_prefixes": value}) == {} + + class Test_AWS_Secrets_Ignore_Files: def test_valid_file_patterns_round_trip(self): files = ["*.deps.json", "vendor/*.js"] diff --git a/tests/providers/aws/services/cloudwatch/cloudwatch_log_group_agentcore_data_protection_policy_enabled/cloudwatch_log_group_agentcore_data_protection_policy_enabled_test.py b/tests/providers/aws/services/cloudwatch/cloudwatch_log_group_agentcore_data_protection_policy_enabled/cloudwatch_log_group_agentcore_data_protection_policy_enabled_test.py new file mode 100644 index 0000000000..2823db9ee8 --- /dev/null +++ b/tests/providers/aws/services/cloudwatch/cloudwatch_log_group_agentcore_data_protection_policy_enabled/cloudwatch_log_group_agentcore_data_protection_policy_enabled_test.py @@ -0,0 +1,333 @@ +import os +import pathlib +from unittest import mock + +import yaml +from moto import mock_aws + +from prowler.providers.aws.services.cloudwatch.cloudwatch_service import LogGroup +from tests.providers.aws.utils import ( + AWS_ACCOUNT_NUMBER, + AWS_REGION_US_EAST_1, + set_mocked_aws_provider, +) + +CHECK_MODULE = "prowler.providers.aws.services.cloudwatch.cloudwatch_log_group_agentcore_data_protection_policy_enabled.cloudwatch_log_group_agentcore_data_protection_policy_enabled" + +RUNTIME_LOG_GROUP = "/aws/bedrock-agentcore/runtimes/my_agent-1a2b3c4d5e" +VENDED_LOG_GROUP = ( + "/aws/vendedlogs/bedrock-agentcore/memory/APPLICATION_LOGS/my-memory-1a2b3c" +) + + +def log_group(name, data_protection_status=None, inherited_properties=None): + """Build a LogGroup carrying the two fields this check reads. + + Both default to the state DescribeLogGroups reports for a log group that has never had a data + protection policy: dataProtectionStatus absent, and no inherited properties. + """ + return LogGroup( + arn=f"arn:aws:logs:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:log-group:{name}:*", + name=name, + retention_days=365, + never_expire=False, + kms_id=None, + creation_time=1700000000000, + data_protection_status=data_protection_status, + inherited_properties=inherited_properties or [], + region=AWS_REGION_US_EAST_1, + ) + + +def run_check(log_groups, audit_config=None): + """Drive the check over a fixed inventory. + + The inventory is set on the service object rather than served through a + patched _make_api_call: DescribeLogGroups -> LogGroup field mapping and its + pagination are covered in cloudwatch_service_test.py, and patching a global + here made these tests sensitive to the order they run in. + """ + from prowler.providers.aws.services.cloudwatch.cloudwatch_service import Logs + + aws_provider = set_mocked_aws_provider( + [AWS_REGION_US_EAST_1], + audit_config={} if audit_config is None else audit_config, + ) + + with mock.patch( + "prowler.providers.common.provider.Provider.get_global_provider", + return_value=aws_provider, + ): + logs_client = Logs(aws_provider) + logs_client.log_groups = ( + None if log_groups is None else {group.arn: group for group in log_groups} + ) + + with mock.patch(f"{CHECK_MODULE}.logs_client", new=logs_client): + from prowler.providers.aws.services.cloudwatch.cloudwatch_log_group_agentcore_data_protection_policy_enabled.cloudwatch_log_group_agentcore_data_protection_policy_enabled import ( + cloudwatch_log_group_agentcore_data_protection_policy_enabled, + ) + + return ( + cloudwatch_log_group_agentcore_data_protection_policy_enabled().execute() + ) + + +class Test_cloudwatch_log_group_agentcore_data_protection_policy_enabled: + """Tests for the cloudwatch_log_group_agentcore_data_protection_policy_enabled check.""" + + @mock_aws + def test_no_log_groups(self): + """An account with no log groups at all must produce no findings. + + An empty inventory was read successfully, so it is not the MANUAL case; there is simply no + resource to make a claim about. + """ + assert run_check([]) == [] + + @mock_aws + def test_non_agentcore_log_group_is_out_of_scope(self): + """Log groups outside the AgentCore prefixes must produce no findings. + + Asserting a data protection policy on every log group in the account would bury the + AgentCore ones. The lookalike name is the case that matters: the prefix must be matched with + its trailing slash, or /aws/bedrock-agentcore-lookalike/ is pulled into scope. + """ + results = run_check( + [ + log_group("/aws/lambda/unrelated-function"), + log_group("aws/spans"), + log_group("/aws/bedrock-agentcore-lookalike/runtimes/x"), + ] + ) + + assert results == [] + + @mock_aws + def test_agentcore_log_group_without_data_protection_status(self): + """An AgentCore log group reporting no dataProtectionStatus must FAIL. + + dataProtectionStatus is modelled at the botocore pin, so its absence is not a parsing gap: + it is the API reporting that the log group has never had a data protection policy, which + means nothing is masked. Pins the resource identity too, since the finding has to name the + log group an operator must remediate. + """ + results = run_check([log_group(RUNTIME_LOG_GROUP)]) + + assert len(results) == 1 + assert results[0].status == "FAIL" + assert ( + results[0].status_extended + == f"AgentCore log group {RUNTIME_LOG_GROUP} does not have an active data protection policy, so sensitive data written by the agent is not masked." + ) + assert results[0].resource_id == RUNTIME_LOG_GROUP + assert ( + results[0].resource_arn + == f"arn:aws:logs:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:log-group:{RUNTIME_LOG_GROUP}:*" + ) + assert results[0].region == AWS_REGION_US_EAST_1 + + @mock_aws + def test_agentcore_log_group_with_activated_policy(self): + """ACTIVATED is the only dataProtectionStatus that must PASS: masking is running today.""" + results = run_check( + [log_group(RUNTIME_LOG_GROUP, data_protection_status="ACTIVATED")] + ) + + assert len(results) == 1 + assert results[0].status == "PASS" + assert ( + results[0].status_extended + == f"AgentCore log group {RUNTIME_LOG_GROUP} has a data protection policy activated." + ) + + @mock_aws + def test_agentcore_log_group_with_inactive_policy(self): + """DELETED, ARCHIVED and DISABLED must each FAIL, not MANUAL. + + All three were read successfully, so nothing is unknown; they mean the same thing + operationally, which is that nothing is being masked at ingestion today. Together with + ACTIVATED these are all four values the enum carries at the botocore pin. + """ + for status in ("DELETED", "ARCHIVED", "DISABLED"): + results = run_check( + [log_group(RUNTIME_LOG_GROUP, data_protection_status=status)] + ) + + assert len(results) == 1 + assert results[0].status == "FAIL" + assert ( + results[0].status_extended + == f"AgentCore log group {RUNTIME_LOG_GROUP} does not have an active data protection policy, so sensitive data written by the agent is not masked." + ) + + @mock_aws + def test_agentcore_log_group_inheriting_account_policy(self): + """A log group inheriting the account policy must PASS with no status of its own. + + An account-level policy surfaces as ACCOUNT_DATA_PROTECTION in inheritedProperties and + leaves dataProtectionStatus absent, so a check reading only dataProtectionStatus would FAIL + a log group that is fully masked. ACCOUNT_DATA_PROTECTION is the only value the + InheritedProperty enum carries at the botocore pin. + """ + results = run_check( + [ + log_group( + VENDED_LOG_GROUP, + inherited_properties=["ACCOUNT_DATA_PROTECTION"], + ) + ] + ) + + assert len(results) == 1 + assert results[0].status == "PASS" + assert ( + results[0].status_extended + == f"AgentCore log group {VENDED_LOG_GROUP} inherits the account-level data protection policy." + ) + + @mock_aws + def test_agentcore_log_groups_mixed(self): + """Multi-resource: one report per in-scope log group, with the PASS/FAIL split asserted. + + A loop that stopped at the first log group, or one that let the out-of-scope Lambda group + through, would not produce exactly these two verdicts. + """ + results = run_check( + [ + log_group(RUNTIME_LOG_GROUP), + log_group(VENDED_LOG_GROUP, data_protection_status="ACTIVATED"), + log_group("/aws/lambda/unrelated-function"), + ] + ) + + assert len(results) == 2 + assert {result.resource_id: result.status for result in results} == { + RUNTIME_LOG_GROUP: "FAIL", + VENDED_LOG_GROUP: "PASS", + } + + @mock_aws + def test_log_groups_not_retrieved_is_manual(self): + """An unreadable inventory must yield one account-level MANUAL, not PASS and not FAIL. + + PASS would assert masking never observed; FAIL would invent a finding against log groups + nothing is known about. The report is attributed to the account rather than to a log group, + because no log group was read. + """ + results = run_check(None) + + assert len(results) == 1 + assert results[0].status == "MANUAL" + assert ( + results[0].status_extended + == "Log groups could not be retrieved, so data protection policies for AgentCore log groups could not be verified." + ) + assert results[0].resource_id == AWS_ACCOUNT_NUMBER + assert results[0].region == AWS_REGION_US_EAST_1 + assert results[0].resource_tags == [] + + @mock_aws + def test_configured_prefix_brings_a_custom_log_group_into_scope(self): + """A configured prefix must put a log group outside the AWS defaults in scope and FAIL it. + + AgentCore log delivery can be pointed at an arbitrarily named log group, so an operator who + does that has no coverage until the prefix is configured. + """ + results = run_check( + [log_group("/company/agents/support-bot")], + audit_config={ + "agentcore_log_group_name_prefixes": ["/company/agents/"], + }, + ) + + assert len(results) == 1 + assert results[0].status == "FAIL" + assert results[0].resource_id == "/company/agents/support-bot" + + @mock_aws + def test_configured_prefix_replaces_the_defaults(self): + """A configured prefix list REPLACES the defaults, so a real AgentCore group drops out. + + This is the sharp edge of the setting and the reason it is pinned: an operator who adds only + their own prefix silences the check for /aws/bedrock-agentcore/ and + /aws/vendedlogs/bedrock-agentcore/, with no finding to show it happened. They must list the + defaults alongside their own. + """ + results = run_check( + [log_group(RUNTIME_LOG_GROUP)], + audit_config={ + "agentcore_log_group_name_prefixes": ["/company/agents/"], + }, + ) + + assert results == [] + + def test_shipped_config_matches_the_check_defaults(self): + """The prefixes shipped in config.yaml must equal the check's in-code defaults. + + The same two prefixes are written twice, and the shipped config wins wherever it is used, so + a prefix added only to the in-code list would be silently ignored by every operator running + the default config -- and an unmatched log group produces no finding at all, not a FAIL. This + makes that drift a failing test instead of missing coverage. + + The provider is mocked around the import because importing the check module constructs + `logs_client`, which reads the global provider's identity. Every other test here reaches that + import through `run_check`, which mocks it; this one imports the module directly for the + constant, so without the patch it is the only test in the file that cannot be selected on its + own -- 12 of 13 pass alone, and did not. + """ + aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1]) + with mock.patch( + "prowler.providers.common.provider.Provider.get_global_provider", + return_value=aws_provider, + ): + from prowler.providers.aws.services.cloudwatch.cloudwatch_log_group_agentcore_data_protection_policy_enabled.cloudwatch_log_group_agentcore_data_protection_policy_enabled import ( + DEFAULT_AGENTCORE_LOG_GROUP_PREFIXES, + ) + + repo_root = pathlib.Path(os.path.dirname(os.path.realpath(__file__))).parents[5] + shipped = yaml.safe_load( + (repo_root / "prowler" / "config" / "config.yaml").read_text() + ) + + assert ( + shipped["aws"]["agentcore_log_group_name_prefixes"] + == DEFAULT_AGENTCORE_LOG_GROUP_PREFIXES + ) + + @mock_aws + def test_explicit_null_prefixes_fall_back_to_the_defaults(self): + """An explicitly null prefix list must fall back to the defaults, not silence the check. + + A bare `agentcore_log_group_name_prefixes:` in the YAML parses as None. Passing that + straight to startswith would raise, and treating it as an empty list would skip every log + group and report nothing. + """ + results = run_check( + [log_group(RUNTIME_LOG_GROUP)], + audit_config={"agentcore_log_group_name_prefixes": None}, + ) + + assert len(results) == 1 + assert results[0].status == "FAIL" + + @mock_aws + def test_an_explicitly_empty_prefix_list_selects_no_log_group(self): + """An empty list is a CONFIGURED value and must not fall back to the defaults. + + This is the only way an operator can say "no log group is in scope for this check", and the + docstring promises a configured list REPLACES the defaults. Reading it with `or` treated `[]` + as absent and re-imposed the defaults, so the check reported on a log group the operator had + deliberately excluded, and no configuration could turn it off. + + It is the pair with the null case above that carries the assertion: null must fall back and + empty must not, and a fix that collapsed both to one behaviour would satisfy either test + alone. The distinction is only visible when both are present. + """ + results = run_check( + [log_group(RUNTIME_LOG_GROUP)], + audit_config={"agentcore_log_group_name_prefixes": []}, + ) + + assert results == [] diff --git a/tests/providers/aws/services/cloudwatch/cloudwatch_service_test.py b/tests/providers/aws/services/cloudwatch/cloudwatch_service_test.py index 3d2d53ffa0..8420d78ca0 100644 --- a/tests/providers/aws/services/cloudwatch/cloudwatch_service_test.py +++ b/tests/providers/aws/services/cloudwatch/cloudwatch_service_test.py @@ -1,5 +1,9 @@ +from unittest.mock import patch + +import botocore import pytest from boto3 import client +from botocore.exceptions import ClientError from moto import mock_aws from prowler.providers.aws.services.cloudwatch.cloudwatch_service import ( @@ -16,6 +20,8 @@ from tests.providers.aws.utils import ( set_mocked_aws_provider, ) +make_api_call = botocore.client.BaseClient._make_api_call + class Test_CloudWatch_Service: # Test CloudWatch Service @@ -226,6 +232,116 @@ class Test_CloudWatch_Service: assert logs.log_groups[arn].region == AWS_REGION_US_EAST_1 assert logs.log_groups[arn].tags == [{}] + @mock_aws + def test_describe_log_groups_data_protection_across_pages(self): + """Both data protection fields must be collected from every page of DescribeLogGroups. + + moto has no data protection support, so the response is served literally. Both pages carry + state a check reads, and each page carries a different one: a collector that stops after the + first page under-reports silently instead of failing loudly. The second page also proves + inheritedProperties survives the round trip rather than being flattened away. + """ + first_page_arn = f"arn:aws:logs:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:log-group:/aws/bedrock-agentcore/runtimes/page-one:*" + second_page_arn = f"arn:aws:logs:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:log-group:/aws/bedrock-agentcore/runtimes/page-two:*" + pages = [ + { + "logGroups": [ + { + "arn": first_page_arn, + "logGroupName": "/aws/bedrock-agentcore/runtimes/page-one", + "creationTime": 2, + "dataProtectionStatus": "DISABLED", + } + ], + "nextToken": "page-two", + }, + { + "logGroups": [ + { + "arn": second_page_arn, + "logGroupName": "/aws/bedrock-agentcore/runtimes/page-two", + "creationTime": 1, + "dataProtectionStatus": "ACTIVATED", + "inheritedProperties": ["ACCOUNT_DATA_PROTECTION"], + } + ] + }, + ] + + def mock_make_api_call(self, operation_name, kwarg): + """Serve page two once the paginator follows nextToken, page one otherwise.""" + if operation_name == "DescribeLogGroups": + return pages[1] if kwarg.get("nextToken") else pages[0] + return make_api_call(self, operation_name, kwarg) + + aws_provider = set_mocked_aws_provider( + [AWS_REGION_US_EAST_1], + expected_checks=["cloudwatch_log_group_no_secrets_in_logs"], + ) + with patch("botocore.client.BaseClient._make_api_call", new=mock_make_api_call): + logs = Logs(aws_provider) + + assert set(logs.log_groups) == {first_page_arn, second_page_arn} + assert logs.log_groups[first_page_arn].data_protection_status == "DISABLED" + assert logs.log_groups[first_page_arn].inherited_properties == [] + assert logs.log_groups[second_page_arn].data_protection_status == "ACTIVATED" + assert logs.log_groups[second_page_arn].inherited_properties == [ + "ACCOUNT_DATA_PROTECTION" + ] + + @mock_aws + def test_describe_log_groups_without_data_protection_fields(self): + """A log group reporting neither field must collect as None and an empty list. + + This is the common real response, since DescribeLogGroups omits both members for a log group + that has never had a policy. None must not become "DISABLED" and the list must not become + None, or a check cannot tell "never configured" from "switched off". + """ + logs_client = client("logs", region_name=AWS_REGION_US_EAST_1) + logs_client.create_log_group(logGroupName="/log-group/test") + + aws_provider = set_mocked_aws_provider( + expected_checks=["cloudwatch_log_group_no_secrets_in_logs"] + ) + arn = f"arn:aws:logs:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:log-group:/log-group/test:*" + logs = Logs(aws_provider) + + assert logs.log_groups[arn].data_protection_status is None + assert logs.log_groups[arn].inherited_properties == [] + + @mock_aws + def test_describe_log_groups_access_denied_leaves_inventory_unknown(self): + """A denied DescribeLogGroups must leave both indexes None, not empty dicts. + + None is the state checks read as "inventory unknown" and report MANUAL for. Collapsing to an + empty dict would be indistinguishable from an account that has no log groups, which every + log group check reads as nothing to report. + """ + + def mock_make_api_call(self, operation_name, kwarg): + """Deny DescribeLogGroups; defer every other operation to botocore.""" + if operation_name == "DescribeLogGroups": + raise ClientError( + { + "Error": { + "Code": "AccessDeniedException", + "Message": "not authorized", + } + }, + operation_name, + ) + return make_api_call(self, operation_name, kwarg) + + aws_provider = set_mocked_aws_provider( + [AWS_REGION_US_EAST_1], + expected_checks=["cloudwatch_log_group_no_secrets_in_logs"], + ) + with patch("botocore.client.BaseClient._make_api_call", new=mock_make_api_call): + logs = Logs(aws_provider) + + assert logs.log_groups is None + assert logs.all_log_groups is None + def test_log_group_limit_exposes_only_selected_resources(self): class FakeLogsClient: def __init__(self):