From f39c92b8f83b392ea79f065587bdc4f07459776e Mon Sep 17 00:00:00 2001 From: Jonathan Nguyen <87379568+tamg-aws@users.noreply.github.com> Date: Fri, 21 Aug 2026 04:50:34 -0500 Subject: [PATCH] feat(bedrock): add model artifact and guardrail grounding security checks for the AWS provider (#12459) Co-authored-by: Hugo P.Brito --- .../bedrock-agent-arn-partition.fixed.md | 1 + .../bedrock-model-artifact-checks.added.md | 1 + .../aws/aws_ai_security_framework_aws.json | 22 +- .../__init__.py | 0 ...ole_not_shared_across_agents.metadata.json | 42 ++ ...ock_agent_role_not_shared_across_agents.py | 164 +++++ .../__init__.py | 0 ...tom_model_encrypted_with_cmk.metadata.json | 43 ++ ...bedrock_custom_model_encrypted_with_cmk.py | 54 ++ .../__init__.py | 0 ...ual_grounding_filter_enabled.metadata.json | 41 ++ ...ail_contextual_grounding_filter_enabled.py | 136 ++++ .../__init__.py | 0 ...edge_base_encrypted_with_cmk.metadata.json | 43 ++ ...drock_knowledge_base_encrypted_with_cmk.py | 80 +++ .../aws/services/bedrock/bedrock_service.py | 378 ++++++++++- ...gent_role_not_shared_across_agents_test.py | 633 ++++++++++++++++++ ..._not_shared_across_agents_versions_test.py | 556 +++++++++++++++ ...ck_custom_model_encrypted_with_cmk_test.py | 287 ++++++++ ...ontextual_grounding_filter_enabled_test.py | 466 +++++++++++++ ..._knowledge_base_encrypted_with_cmk_test.py | 338 ++++++++++ .../services/bedrock/bedrock_service_test.py | 37 +- 22 files changed, 3289 insertions(+), 33 deletions(-) create mode 100644 prowler/changelog.d/bedrock-agent-arn-partition.fixed.md create mode 100644 prowler/changelog.d/bedrock-model-artifact-checks.added.md create mode 100644 prowler/providers/aws/services/bedrock/bedrock_agent_role_not_shared_across_agents/__init__.py create mode 100644 prowler/providers/aws/services/bedrock/bedrock_agent_role_not_shared_across_agents/bedrock_agent_role_not_shared_across_agents.metadata.json create mode 100644 prowler/providers/aws/services/bedrock/bedrock_agent_role_not_shared_across_agents/bedrock_agent_role_not_shared_across_agents.py create mode 100644 prowler/providers/aws/services/bedrock/bedrock_custom_model_encrypted_with_cmk/__init__.py create mode 100644 prowler/providers/aws/services/bedrock/bedrock_custom_model_encrypted_with_cmk/bedrock_custom_model_encrypted_with_cmk.metadata.json create mode 100644 prowler/providers/aws/services/bedrock/bedrock_custom_model_encrypted_with_cmk/bedrock_custom_model_encrypted_with_cmk.py create mode 100644 prowler/providers/aws/services/bedrock/bedrock_guardrail_contextual_grounding_filter_enabled/__init__.py create mode 100644 prowler/providers/aws/services/bedrock/bedrock_guardrail_contextual_grounding_filter_enabled/bedrock_guardrail_contextual_grounding_filter_enabled.metadata.json create mode 100644 prowler/providers/aws/services/bedrock/bedrock_guardrail_contextual_grounding_filter_enabled/bedrock_guardrail_contextual_grounding_filter_enabled.py create mode 100644 prowler/providers/aws/services/bedrock/bedrock_knowledge_base_encrypted_with_cmk/__init__.py create mode 100644 prowler/providers/aws/services/bedrock/bedrock_knowledge_base_encrypted_with_cmk/bedrock_knowledge_base_encrypted_with_cmk.metadata.json create mode 100644 prowler/providers/aws/services/bedrock/bedrock_knowledge_base_encrypted_with_cmk/bedrock_knowledge_base_encrypted_with_cmk.py create mode 100644 tests/providers/aws/services/bedrock/bedrock_agent_role_not_shared_across_agents/bedrock_agent_role_not_shared_across_agents_test.py create mode 100644 tests/providers/aws/services/bedrock/bedrock_agent_role_not_shared_across_agents/bedrock_agent_role_not_shared_across_agents_versions_test.py create mode 100644 tests/providers/aws/services/bedrock/bedrock_custom_model_encrypted_with_cmk/bedrock_custom_model_encrypted_with_cmk_test.py create mode 100644 tests/providers/aws/services/bedrock/bedrock_guardrail_contextual_grounding_filter_enabled/bedrock_guardrail_contextual_grounding_filter_enabled_test.py create mode 100644 tests/providers/aws/services/bedrock/bedrock_knowledge_base_encrypted_with_cmk/bedrock_knowledge_base_encrypted_with_cmk_test.py diff --git a/prowler/changelog.d/bedrock-agent-arn-partition.fixed.md b/prowler/changelog.d/bedrock-agent-arn-partition.fixed.md new file mode 100644 index 0000000000..96d4424272 --- /dev/null +++ b/prowler/changelog.d/bedrock-agent-arn-partition.fixed.md @@ -0,0 +1 @@ +Bedrock Agent ARNs are now built from the audited partition instead of a hardcoded `arn:aws:`, so findings in GovCloud and China carry a resolvable ARN and `--resource-arn` scoping matches agents in those partitions. diff --git a/prowler/changelog.d/bedrock-model-artifact-checks.added.md b/prowler/changelog.d/bedrock-model-artifact-checks.added.md new file mode 100644 index 0000000000..43d08aaf79 --- /dev/null +++ b/prowler/changelog.d/bedrock-model-artifact-checks.added.md @@ -0,0 +1 @@ +`bedrock_guardrail_contextual_grounding_filter_enabled`, `bedrock_custom_model_encrypted_with_cmk`, `bedrock_knowledge_base_encrypted_with_cmk` and `bedrock_agent_role_not_shared_across_agents` are four new AWS Bedrock checks covering guardrail contextual grounding, custom model encryption, knowledge-base data-source encryption, and non-shared agent execution roles. diff --git a/prowler/compliance/aws/aws_ai_security_framework_aws.json b/prowler/compliance/aws/aws_ai_security_framework_aws.json index f20baf4735..fa8c9d83a7 100644 --- a/prowler/compliance/aws/aws_ai_security_framework_aws.json +++ b/prowler/compliance/aws/aws_ai_security_framework_aws.json @@ -110,7 +110,9 @@ "eks_cluster_kms_cmk_encryption_in_secrets_enabled", "dynamodb_tables_kms_cmk_encryption_enabled", "sns_topics_kms_encryption_at_rest_enabled", - "sqs_queues_server_side_encryption_enabled" + "sqs_queues_server_side_encryption_enabled", + "bedrock_custom_model_encrypted_with_cmk", + "bedrock_knowledge_base_encrypted_with_cmk" ] }, { @@ -638,10 +640,12 @@ "Section": "AI Application Security", "SubSection": "Output Validation", "Service": "bedrock", - "Type": "Manual" + "Type": "Automated" } ], - "Checks": [] + "Checks": [ + "bedrock_guardrail_contextual_grounding_filter_enabled" + ] }, { "Id": "AISF-AI-10", @@ -652,10 +656,12 @@ "Section": "AI Application Security", "SubSection": "RAG Security", "Service": "bedrock", - "Type": "Manual" + "Type": "Automated" } ], - "Checks": [] + "Checks": [ + "bedrock_knowledge_base_encrypted_with_cmk" + ] }, { "Id": "AISF-AI-11", @@ -680,10 +686,12 @@ "Section": "AI Application Security", "SubSection": "Agent Governance", "Service": "bedrock", - "Type": "Manual" + "Type": "Automated" } ], - "Checks": [] + "Checks": [ + "bedrock_agent_role_not_shared_across_agents" + ] }, { "Id": "AISF-AGENT-02", diff --git a/prowler/providers/aws/services/bedrock/bedrock_agent_role_not_shared_across_agents/__init__.py b/prowler/providers/aws/services/bedrock/bedrock_agent_role_not_shared_across_agents/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/prowler/providers/aws/services/bedrock/bedrock_agent_role_not_shared_across_agents/bedrock_agent_role_not_shared_across_agents.metadata.json b/prowler/providers/aws/services/bedrock/bedrock_agent_role_not_shared_across_agents/bedrock_agent_role_not_shared_across_agents.metadata.json new file mode 100644 index 0000000000..00581d65df --- /dev/null +++ b/prowler/providers/aws/services/bedrock/bedrock_agent_role_not_shared_across_agents/bedrock_agent_role_not_shared_across_agents.metadata.json @@ -0,0 +1,42 @@ +{ + "Provider": "aws", + "CheckID": "bedrock_agent_role_not_shared_across_agents", + "CheckTitle": "Bedrock Agent has a dedicated execution role", + "CheckType": [ + "Software and Configuration Checks/AWS Security Best Practices" + ], + "ServiceName": "bedrock", + "SubServiceName": "", + "ResourceIdTemplate": "", + "Severity": "high", + "ResourceType": "Other", + "ResourceGroup": "ai_ml", + "Description": "Every **Bedrock Agent** assumes the role in its `agentResourceRoleArn`. That role must belong to exactly one agent, so no agent inherits another's permissions.", + "Risk": "A **shared execution role** gives every agent using it the union of all their permissions, so a low-trust agent reaching an untrusted tool holds the rights granted for a high-trust one, turning a single **prompt injection** into access it was never scoped for.\n\nIt also destroys **attribution**: **CloudTrail** records the role session, so an action cannot be traced to one agent.", + "RelatedUrl": "", + "AdditionalURLs": [ + "https://docs.aws.amazon.com/bedrock/latest/userguide/agents-permissions.html", + "https://docs.aws.amazon.com/bedrock/latest/APIReference/API_agent_UpdateAgent.html" + ], + "Remediation": { + "Code": { + "CLI": "", + "NativeIaC": "```yaml\nResources:\n ExampleResource:\n Type: AWS::Bedrock::Agent\n Properties:\n AgentName: example_resource\n AgentResourceRoleArn: arn:aws:iam::123456789012:role/AmazonBedrockExecutionRoleForAgents_example_resource # Critical: one role per agent, never shared\n```", + "Other": "1. Open the AWS Console and go to Amazon Bedrock\n2. Select **Agents**, then an agent sharing a role\n3. Create a new IAM role trusted by bedrock.amazonaws.com holding only that agent's permissions\n4. Scope its trust policy with aws:SourceArn set to the agent ARN\n5. Edit the agent to use the new role, keeping its other settings, then prepare the agent to cut a version\n6. Open **Aliases** and point every alias at the new version, since an alias left on the old version keeps invoking the shared role\n7. Repeat until each agent has its own role, then remove the unused permissions from the old role", + "Terraform": "```hcl\nresource \"aws_bedrockagent_agent\" \"example_resource\" {\n agent_name = \"example_resource\"\n agent_resource_role_arn = aws_iam_role.example_resource_agent.arn # Critical: one role per agent, never shared\n foundation_model = \"example-model-id\"\n}\n```" + }, + "Recommendation": { + "Text": "Give each Bedrock Agent its own execution role scoped to that agent's ARN with aws:SourceArn. Repoint the draft with UpdateAgent, passing every field it already has because omitted ones are not preserved, then PrepareAgent and UpdateAgentAlias to move each alias onto the new version: an alias left on the old version keeps invoking the shared role.", + "Url": "https://hub.prowler.com/check/bedrock_agent_role_not_shared_across_agents" + } + }, + "Categories": [ + "gen-ai", + "identity-access" + ], + "DependsOn": [], + "RelatedTo": [ + "bedrock_agent_role_least_privilege" + ], + "Notes": "Reports one finding per AGENT, not per role, so a role shared by three agents produces three findings each naming the other two. Sharing is counted across the whole account inventory, so a partial scan can under-report. An agent whose role ARN could not be retrieved from GetAgent returns MANUAL rather than PASS, because an unknown role can be neither confirmed nor ruled out as shared, and it is also excluded from every other agent's share count." +} diff --git a/prowler/providers/aws/services/bedrock/bedrock_agent_role_not_shared_across_agents/bedrock_agent_role_not_shared_across_agents.py b/prowler/providers/aws/services/bedrock/bedrock_agent_role_not_shared_across_agents/bedrock_agent_role_not_shared_across_agents.py new file mode 100644 index 0000000000..55fe930e1b --- /dev/null +++ b/prowler/providers/aws/services/bedrock/bedrock_agent_role_not_shared_across_agents/bedrock_agent_role_not_shared_across_agents.py @@ -0,0 +1,164 @@ +from prowler.lib.check.models import Check, Check_Report_AWS +from prowler.providers.aws.services.bedrock.bedrock_agent_client import ( + bedrock_agent_client, +) + +SHARED_ROLE_AGENT_COUNT = 2 + + +class bedrock_agent_role_not_shared_across_agents(Check): + """Ensure each Bedrock Agent has a dedicated execution role. + + A shared `agentResourceRoleArn` means every agent using it inherits the + union of all their permissions, so the least-privileged agent in the set + still holds the rights of the most privileged one. It also destroys + attribution: CloudTrail records the role session, so an action taken with + that role cannot be tied back to a single agent. + + Every role an agent can run under counts, not only the working draft's. + GetAgent returns the draft, but an agent version is an immutable snapshot + that keeps the role it was cut with, and an alias routes invocations at a + specific version — so a deployed version can still hold a role the draft no + longer has, and sharing it is the same exposure. Only versions an alias + routes to are considered, since a version no alias points at cannot be + invoked. An alias routing at DRAFT resolves to the draft role already read. + + Sharing is judged against the whole account inventory, so the verdict is only + as complete as that inventory. Dedication cannot be asserted while any part of + the picture is missing — an unlisted Region, an agent whose own role could + not be read, or an agent whose deployed versions could not be listed may hold + the same role. A role already seen on two agents is shared whatever else is + missing, so FAIL survives an incomplete inventory. + + A scan scoped with ``--resource-arn`` narrows which agents are REPORTED on, not + which agents count towards sharing: the role index is built from the complete + account inventory, so selecting one of two agents that share a role still FAILs. + + - PASS: No role this agent holds is used by any other agent, every Region's + agent inventory was listed, every discovered agent's role was readable, and + every agent's deployed versions were listed. + - FAIL: Two or more agents hold the same role, on the draft or on a deployed + version; the other agents holding it are named in the message, and the + version is named when the sharing is through one. + - MANUAL: This agent's own execution role could not be retrieved from + GetAgent; or its roles look dedicated but another agent's role or deployed + versions are unknown, or a Region could not be listed; or ListAgents failed + for a Region, which therefore contributed no agents at all. + """ + + def execute(self) -> list[Check_Report_AWS]: + """Execute the check logic. + + Returns: + A list of reports containing the result of the check. + """ + findings = [] + + incomplete_regions = sorted(bedrock_agent_client.agents_scan_errors) + for region, error in sorted(bedrock_agent_client.agents_scan_errors.items()): + report = Check_Report_AWS( + metadata=self.metadata(), resource={"region": region} + ) + report.region = region + report.resource_id = "agent/unknown" + report.resource_arn = f"arn:{bedrock_agent_client.audited_partition}:bedrock:{region}:{bedrock_agent_client.audited_account}:agent/unknown" + report.status = "MANUAL" + report.status_extended = f"Bedrock Agents could not be listed in region {region} ({error}); verify manually that no execution role is shared between agents." + findings.append(report) + + # Keyed on agent ARN, not name, so two same-named agents still count + # twice. An agent is indexed under every role it holds, because a + # deployed version keeps the role it was cut with: sharing through a + # version is the same exposure as sharing through the draft. Indexed over + # all_agents, not the filtered agents; findings are emitted from the + # filtered set below. + agents_by_role = {} + unresolved_agents = [] + for agent in bedrock_agent_client.all_agents.values(): + entry = (agent.arn, agent.name or agent.id) + for role_arn in self._roles_held_by(agent): + if entry not in agents_by_role.setdefault(role_arn, []): + agents_by_role[role_arn].append(entry) + # Both gaps are recorded, not just the first: an agent can have an + # unreadable draft role and an unlistable version inventory at once, + # and each independently keeps another agent from being called + # dedicated. + if not agent.detail_retrieved or not agent.role_arn: + unresolved_agents.append( + f"the execution role of {agent.name or agent.id}" + ) + if not agent.versions_listed: + unresolved_agents.append( + f"deployed versions of {agent.name or agent.id}" + ) + + # Dedication can only be asserted from a complete picture: an unlisted + # Region, an agent whose role could not be read, or an agent whose + # deployed versions could not be listed may hold the same role. + incomplete = sorted(unresolved_agents) + [ + f"agents in region {region}" for region in incomplete_regions + ] + + for agent in bedrock_agent_client.agents.values(): + report = Check_Report_AWS(metadata=self.metadata(), resource=agent) + name = agent.name or agent.id + + if not agent.detail_retrieved or not agent.role_arn: + report.status = "MANUAL" + report.status_extended = f"Bedrock Agent {name} execution role could not be retrieved in region {agent.region}; verify manually that no other agent shares it." + findings.append(report) + continue + + # Any role this agent holds, on the draft or on a deployed version, + # is a finding when another agent holds it too. Reported on the + # lowest-sorted shared role for determinism across scans. + shared_roles = sorted( + role_arn + for role_arn in self._roles_held_by(agent) + if len(agents_by_role.get(role_arn, [])) >= SHARED_ROLE_AGENT_COUNT + ) + if shared_roles: + role_arn = shared_roles[0] + others = sorted( + other_name + for other_arn, other_name in agents_by_role[role_arn] + if other_arn != agent.arn + ) + through = ( + "" + if role_arn == agent.role_arn + else f" through deployed version {sorted(version for version, version_role in agent.version_role_arns.items() if version_role == role_arn)[0]}" + ) + report.status = "FAIL" + report.status_extended = f"Bedrock Agent {name} shares execution role {role_arn}{through} with {', '.join(others)} in region {agent.region}, so each agent inherits the union of their permissions and CloudTrail cannot attribute an action to one of them." + elif incomplete: + report.status = "MANUAL" + # Each entry already reads as its own subject, because they are + # not all roles: an entry is an agent whose role is unknown, the + # deployed versions of an agent, or the agents of a Region that + # could not be listed. + report.status_extended = f"Bedrock Agent {name} execution role is used by no other agent whose role could be read in region {agent.region}, but {', '.join(incomplete)} could not be read; verify manually that none of them shares it." + else: + report.status = "PASS" + report.status_extended = f"Bedrock Agent {name} has a dedicated execution role in region {agent.region}." + findings.append(report) + + return findings + + def _roles_held_by(self, agent) -> set: + """Collect every execution role an agent can run under. + + Args: + agent: The Bedrock Agent to inspect. + + Returns: + The working draft's role plus the role of each deployed version an + alias routes to, skipping any that could not be read. + """ + roles = set() + if agent.detail_retrieved and agent.role_arn: + roles.add(agent.role_arn) + roles.update( + role_arn for role_arn in agent.version_role_arns.values() if role_arn + ) + return roles diff --git a/prowler/providers/aws/services/bedrock/bedrock_custom_model_encrypted_with_cmk/__init__.py b/prowler/providers/aws/services/bedrock/bedrock_custom_model_encrypted_with_cmk/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/prowler/providers/aws/services/bedrock/bedrock_custom_model_encrypted_with_cmk/bedrock_custom_model_encrypted_with_cmk.metadata.json b/prowler/providers/aws/services/bedrock/bedrock_custom_model_encrypted_with_cmk/bedrock_custom_model_encrypted_with_cmk.metadata.json new file mode 100644 index 0000000000..06f11e42a2 --- /dev/null +++ b/prowler/providers/aws/services/bedrock/bedrock_custom_model_encrypted_with_cmk/bedrock_custom_model_encrypted_with_cmk.metadata.json @@ -0,0 +1,43 @@ +{ + "Provider": "aws", + "CheckID": "bedrock_custom_model_encrypted_with_cmk", + "CheckTitle": "Bedrock custom model is encrypted with a customer-managed KMS key", + "CheckType": [ + "Software and Configuration Checks/AWS Security Best Practices", + "Software and Configuration Checks/AWS Security Best Practices/Data Encryption" + ], + "ServiceName": "bedrock", + "SubServiceName": "", + "ResourceIdTemplate": "", + "Severity": "critical", + "ResourceType": "Other", + "ResourceGroup": "ai_ml", + "Description": "**Bedrock custom models** produced by model customization can be encrypted at rest with a customer-managed KMS key rather than resting under an AWS-owned key the organization cannot audit, rotate, or revoke.", + "Risk": "A **fine-tuned model** is derived from the training data used to build it and can leak that data through inference.\n\nUnder an **AWS-owned key**, the organization has no key policy to restrict who may decrypt the artifacts, no customer-controlled rotation, no **CloudTrail** record of key usage, and no way to revoke access to the model by disabling a key.", + "RelatedUrl": "", + "AdditionalURLs": [ + "https://docs.aws.amazon.com/bedrock/latest/userguide/encryption-custom-job.html", + "https://docs.aws.amazon.com/bedrock/latest/APIReference/API_GetCustomModel.html" + ], + "Remediation": { + "Code": { + "CLI": "", + "NativeIaC": "", + "Other": "1. Open the AWS Console and go to Amazon Bedrock\n2. Select **Custom models**, then create a new customization job\n3. Under encryption, choose a customer-managed KMS key instead of the default\n4. Grant the customization service role kms:Decrypt, kms:GenerateDataKey and kms:DescribeKey on that key\n5. Re-run customization, because the key of an existing custom model cannot be changed in place", + "Terraform": "```hcl\nresource \"aws_bedrock_custom_model\" \"example_resource\" {\n custom_model_name = \"example-custom-model\"\n job_name = \"example-customization-job\"\n base_model_identifier = \"arn:aws:bedrock:us-east-1::foundation-model/amazon.titan-text-express-v1\"\n role_arn = \"arn:aws:iam::123456789012:role/example-bedrock-customization-role\"\n custom_model_kms_key_id = \"arn:aws:kms:us-east-1:123456789012:key/example-key-id\" # Critical: without this the model is encrypted with an AWS owned key\n\n hyperparameters = {\n epochCount = \"1\"\n }\n\n output_data_config {\n s3_uri = \"s3://example-bucket/output/\"\n }\n\n training_data_config {\n s3_uri = \"s3://example-bucket/training/\"\n }\n}\n```" + }, + "Recommendation": { + "Text": "Choose a customer-managed KMS key when creating each model customization job so the resulting custom model rests under a key with an auditable policy and rotation. The key cannot be changed after the model is created, so re-run customization for existing models.", + "Url": "https://hub.prowler.com/check/bedrock_custom_model_encrypted_with_cmk" + } + }, + "Categories": [ + "gen-ai", + "encryption" + ], + "DependsOn": [], + "RelatedTo": [ + "bedrock_prompt_encrypted_with_cmk" + ], + "Notes": "Reports one finding per custom model, reading modelKmsKeyArn from GetCustomModel because ListCustomModels summaries do not carry it. A model whose detail could not be retrieved returns MANUAL rather than PASS, because an absent key ARN means the answer is unknown, not that no key is set." +} diff --git a/prowler/providers/aws/services/bedrock/bedrock_custom_model_encrypted_with_cmk/bedrock_custom_model_encrypted_with_cmk.py b/prowler/providers/aws/services/bedrock/bedrock_custom_model_encrypted_with_cmk/bedrock_custom_model_encrypted_with_cmk.py new file mode 100644 index 0000000000..1ac9c7e8d8 --- /dev/null +++ b/prowler/providers/aws/services/bedrock/bedrock_custom_model_encrypted_with_cmk/bedrock_custom_model_encrypted_with_cmk.py @@ -0,0 +1,54 @@ +from prowler.lib.check.models import Check, Check_Report_AWS +from prowler.providers.aws.services.bedrock.bedrock_client import bedrock_client + + +class bedrock_custom_model_encrypted_with_cmk(Check): + """Ensure Bedrock custom models are encrypted with a customer-managed KMS key. + + - PASS: GetCustomModel returns a `modelKmsKeyArn`, so the model artifacts + are encrypted with a key the account controls. + - FAIL: No `modelKmsKeyArn` is set, so the model is encrypted with an + AWS-owned key that the organization cannot audit, rotate, or revoke. + - MANUAL: GetCustomModel failed, so the key could not be retrieved and an + absent value cannot be read as "no key"; or ListCustomModels failed for a + region, so that region's custom models are unknown rather than absent. + + Only models this account owns are audited (`ListCustomModels` is called with + `isOwned=True`): the KMS key of a model shared in through Resource Access + Manager belongs to the owning account and cannot be changed here. + """ + + def execute(self) -> list[Check_Report_AWS]: + """Execute the check logic. + + Returns: + A list of reports containing the result of the check. + """ + findings = [] + + for region, error in sorted(bedrock_client.custom_models_scan_errors.items()): + report = Check_Report_AWS( + metadata=self.metadata(), resource={"region": region} + ) + report.region = region + report.resource_id = "custom-model/unknown" + report.resource_arn = f"arn:{bedrock_client.audited_partition}:bedrock:{region}:{bedrock_client.audited_account}:custom-model/unknown" + report.status = "MANUAL" + report.status_extended = f"Bedrock custom models could not be listed in region {region} ({error}); verify manually that every custom model uses a customer-managed KMS key." + findings.append(report) + + for model in bedrock_client.custom_models.values(): + report = Check_Report_AWS(metadata=self.metadata(), resource=model) + + if not model.detail_retrieved: + report.status = "MANUAL" + report.status_extended = f"Bedrock custom model {model.name} encryption configuration could not be retrieved in region {model.region}; verify manually that it uses a customer-managed KMS key." + elif model.kms_key_arn: + report.status = "PASS" + report.status_extended = f"Bedrock custom model {model.name} is encrypted with a customer-managed KMS key in region {model.region}." + else: + report.status = "FAIL" + report.status_extended = f"Bedrock custom model {model.name} is not encrypted with a customer-managed KMS key in region {model.region}, so the fine-tuned weights rest under an AWS-owned key the organization cannot audit or revoke." + findings.append(report) + + return findings diff --git a/prowler/providers/aws/services/bedrock/bedrock_guardrail_contextual_grounding_filter_enabled/__init__.py b/prowler/providers/aws/services/bedrock/bedrock_guardrail_contextual_grounding_filter_enabled/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/prowler/providers/aws/services/bedrock/bedrock_guardrail_contextual_grounding_filter_enabled/bedrock_guardrail_contextual_grounding_filter_enabled.metadata.json b/prowler/providers/aws/services/bedrock/bedrock_guardrail_contextual_grounding_filter_enabled/bedrock_guardrail_contextual_grounding_filter_enabled.metadata.json new file mode 100644 index 0000000000..2fd1b3cb54 --- /dev/null +++ b/prowler/providers/aws/services/bedrock/bedrock_guardrail_contextual_grounding_filter_enabled/bedrock_guardrail_contextual_grounding_filter_enabled.metadata.json @@ -0,0 +1,41 @@ +{ + "Provider": "aws", + "CheckID": "bedrock_guardrail_contextual_grounding_filter_enabled", + "CheckTitle": "Bedrock guardrail blocks ungrounded and irrelevant model responses", + "CheckType": [ + "Software and Configuration Checks/AWS Security Best Practices" + ], + "ServiceName": "bedrock", + "SubServiceName": "", + "ResourceIdTemplate": "", + "Severity": "high", + "ResourceType": "Other", + "ResourceGroup": "ai_ml", + "Description": "**Bedrock guardrails** can attach a `contextualGroundingPolicy` whose filters score each response for `GROUNDING` (supported by the retrieved source) and `RELEVANCE` (answers the question asked). Both filter types must be enabled, set to `action: BLOCK`, and carry a `threshold` above 0.", + "Risk": "Without both filters the model can return confident, unsupported answers that a **RAG application** presents as sourced fact.\n\nThree settings make a configured filter **inert**: `enabled: false` runs no evaluation, so a `BLOCK` action and tuned threshold sit unused; `action: NONE` scores the response and reports it in the trace while still delivering it; a `threshold` of 0 can never be tripped.", + "RelatedUrl": "", + "AdditionalURLs": [ + "https://docs.aws.amazon.com/bedrock/latest/userguide/guardrails-contextual-grounding-check.html", + "https://docs.aws.amazon.com/bedrock/latest/APIReference/API_GuardrailContextualGroundingFilterConfig.html" + ], + "Remediation": { + "Code": { + "CLI": "aws bedrock update-guardrail --guardrail-identifier example-guardrail-id --name example-guardrail --blocked-input-messaging \"Sorry, I cannot answer that.\" --blocked-outputs-messaging \"Sorry, I cannot answer that.\" --contextual-grounding-policy-config '{\"filtersConfig\":[{\"type\":\"GROUNDING\",\"threshold\":0.75,\"action\":\"BLOCK\",\"enabled\":true},{\"type\":\"RELEVANCE\",\"threshold\":0.75,\"action\":\"BLOCK\",\"enabled\":true}]}'", + "NativeIaC": "```yaml\nResources:\n ExampleResource:\n Type: AWS::Bedrock::Guardrail\n Properties:\n Name: example_resource\n BlockedInputMessaging: Sorry, I cannot answer that.\n BlockedOutputsMessaging: Sorry, I cannot answer that.\n ContextualGroundingPolicyConfig:\n FiltersConfig:\n - Type: GROUNDING\n Threshold: 0.75\n Action: BLOCK # Critical: NONE only reports, it does not block\n Enabled: true # Critical: false runs no evaluation at all\n - Type: RELEVANCE\n Threshold: 0.75\n Action: BLOCK\n Enabled: true\n```", + "Other": "1. Open the AWS Console and go to Amazon Bedrock\n2. Select **Guardrails**, then the guardrail\n3. Edit the contextual grounding check\n4. Enable both the grounding and the relevance filter\n5. Set each threshold above 0 to a value tuned for your corpus\n6. Set each action to Block rather than None\n7. Save and create a new guardrail version", + "Terraform": "```hcl\nresource \"awscc_bedrock_guardrail\" \"example_resource\" {\n name = \"example-guardrail\"\n blocked_input_messaging = \"Sorry, I cannot answer that.\"\n blocked_outputs_messaging = \"Sorry, I cannot answer that.\"\n\n contextual_grounding_policy_config = {\n filters_config = [\n {\n type = \"GROUNDING\"\n threshold = 0.75 # Critical: 0 can never be tripped\n action = \"BLOCK\" # Critical: NONE scores without blocking\n enabled = true # Critical: when false the evaluation never runs\n },\n {\n type = \"RELEVANCE\"\n threshold = 0.75\n action = \"BLOCK\"\n enabled = true\n },\n ]\n }\n}\n```" + }, + "Recommendation": { + "Text": "Configure both a GROUNDING and a RELEVANCE contextual grounding filter on every guardrail used by a retrieval-augmented application, set each action to BLOCK, and tune each threshold above 0 against a labelled sample rather than leaving it at the minimum.", + "Url": "https://hub.prowler.com/check/bedrock_guardrail_contextual_grounding_filter_enabled" + } + }, + "Categories": [ + "gen-ai" + ], + "DependsOn": [], + "RelatedTo": [ + "bedrock_guardrail_prompt_attack_filter_enabled" + ], + "Notes": "Reports one finding per guardrail, reading contextualGroundingPolicy.filters from GetGuardrail. A guardrail whose detail could not be retrieved returns MANUAL rather than PASS, because an absent policy means the answer is unknown, not that the control is satisfied. enabled and action are both optional members with no documented default, so a filter that is otherwise compliant but omits either returns MANUAL rather than being read as blocking or not; an explicit enabled: false or a non-BLOCK action is a definite finding and FAILs." +} diff --git a/prowler/providers/aws/services/bedrock/bedrock_guardrail_contextual_grounding_filter_enabled/bedrock_guardrail_contextual_grounding_filter_enabled.py b/prowler/providers/aws/services/bedrock/bedrock_guardrail_contextual_grounding_filter_enabled/bedrock_guardrail_contextual_grounding_filter_enabled.py new file mode 100644 index 0000000000..d94255c835 --- /dev/null +++ b/prowler/providers/aws/services/bedrock/bedrock_guardrail_contextual_grounding_filter_enabled/bedrock_guardrail_contextual_grounding_filter_enabled.py @@ -0,0 +1,136 @@ +from prowler.lib.check.models import Check, Check_Report_AWS +from prowler.providers.aws.services.bedrock.bedrock_client import bedrock_client + +REQUIRED_FILTER_TYPES = frozenset({"GROUNDING", "RELEVANCE"}) +BLOCKING_ACTION = "BLOCK" + + +class bedrock_guardrail_contextual_grounding_filter_enabled(Check): + """Ensure Bedrock guardrails block ungrounded and irrelevant model responses. + + - PASS: The guardrail configures both a GROUNDING and a RELEVANCE + contextual grounding filter, each enabled, with action BLOCK and a + threshold above 0. + - FAIL: No contextual grounding policy is configured; one of the two + required filter types is missing; a present filter carries + enabled: false, so its evaluation never runs; a present filter uses + action NONE, which scores and reports without blocking; or a present + filter has a threshold of 0, which nothing can ever trip. + - MANUAL: GetGuardrail failed, so the policy could not be retrieved and + compliance cannot be asserted from an absent field; or a filter is + otherwise compliant but omits enabled or action, leaving it unknown + whether the evaluation runs or whether it blocks; or ListGuardrails failed + for a Region, so that Region's guardrails are unknown rather than absent. + + enabled and action are both optional members of + GuardrailContextualGroundingFilter (only type and threshold are required) + and AWS documents no default for either, so an omitted value is unknown + rather than false or NONE. Reading an omitted enabled as false would FAIL + filters that are in fact evaluating; reading an omitted action as NONE + would assert a misconfiguration the response never stated, and would print + the literal None into status_extended where it reads as an AWS enum value. + Both are therefore reported MANUAL, matching how this check already treats + a guardrail whose detail could not be retrieved. An explicit enabled: false + or an explicit non-BLOCK action is a definite finding and still FAILs, as + does a zero threshold, so an unknown never masks a real one. + """ + + def execute(self) -> list[Check_Report_AWS]: + """Execute the check logic. + + Returns: + A list of reports containing the result of the check. + """ + findings = [] + + for region, error in sorted(bedrock_client.guardrails_scan_errors.items()): + report = Check_Report_AWS( + metadata=self.metadata(), resource={"region": region} + ) + report.region = region + report.resource_id = "guardrail/unknown" + report.resource_arn = f"arn:{bedrock_client.audited_partition}:bedrock:{region}:{bedrock_client.audited_account}:guardrail/unknown" + report.status = "MANUAL" + report.status_extended = f"Bedrock guardrails could not be listed in region {region} ({error}); verify manually that each one blocks ungrounded and irrelevant responses." + findings.append(report) + + for guardrail in bedrock_client.guardrails.values(): + report = Check_Report_AWS(metadata=self.metadata(), resource=guardrail) + + if not guardrail.detail_retrieved: + # GetGuardrail failed (permissions, throttling, transient + # error). An absent policy is not evidence of a missing one. + report.status = "MANUAL" + report.status_extended = f"Bedrock Guardrail {guardrail.name} contextual grounding policy could not be retrieved in region {guardrail.region}; verify manually that GROUNDING and RELEVANCE filters block ungrounded responses." + findings.append(report) + continue + + filters_by_type = { + filter.type: filter + for filter in guardrail.contextual_grounding_filters + if filter.type + } + + if not filters_by_type: + report.status = "FAIL" + report.status_extended = f"Bedrock Guardrail {guardrail.name} has no contextual grounding policy configured in region {guardrail.region}, so ungrounded and irrelevant responses are never detected." + findings.append(report) + continue + + missing_types = sorted(REQUIRED_FILTER_TYPES - set(filters_by_type)) + if missing_types: + report.status = "FAIL" + # Both required types can be missing at once, so the nouns follow + # the count rather than assuming a single filter. + noun = "filter" if len(missing_types) == 1 else "filters" + classes = "that class" if len(missing_types) == 1 else "those classes" + report.status_extended = f"Bedrock Guardrail {guardrail.name} contextual grounding policy is missing the {', '.join(missing_types)} {noun} in region {guardrail.region}, leaving {classes} of ungrounded response unchecked." + findings.append(report) + continue + + reasons = [] + unknown_types = [] + for filter_type in sorted(REQUIRED_FILTER_TYPES): + filter = filters_by_type[filter_type] + action = filter.action + if filter.enabled is False: + reasons.append( + f"the {filter_type} filter is disabled, so its evaluation never runs regardless of its action or threshold" + ) + elif action is not None and action != BLOCKING_ACTION: + reasons.append( + f"the {filter_type} filter uses action {action} instead of BLOCK, so it scores and reports without blocking" + ) + elif not (filter.threshold or 0) > 0: + reasons.append( + f"the {filter_type} filter has a threshold of {filter.threshold}, which no response can ever trip" + ) + elif action is None or filter.enabled is None: + # Name only the attributes actually omitted: one of the two + # may be present, and claiming both are missing describes a + # response the guardrail did not return. + missing = [ + attribute + for attribute, value in ( + ("enabled", filter.enabled), + ("action", action), + ) + if value is None + ] + unknown_types.append( + f"{filter_type} filter omits {' and '.join(missing)}" + ) + + if reasons: + report.status = "FAIL" + report.status_extended = f"Bedrock Guardrail {guardrail.name} does not block ungrounded responses in region {guardrail.region}: {'; '.join(reasons)}." + elif unknown_types: + report.status = "MANUAL" + subject = "it blocks" if len(unknown_types) == 1 else "they block" + report.status_extended = f"Bedrock Guardrail {guardrail.name} has both required contextual grounding filters with a non-zero threshold in region {guardrail.region}, but the {', '.join(unknown_types)}, so whether {subject} is unknown; verify manually that the evaluation runs and blocks." + else: + report.status = "PASS" + report.status_extended = f"Bedrock Guardrail {guardrail.name} blocks ungrounded and irrelevant responses with GROUNDING and RELEVANCE filters in region {guardrail.region}." + findings.append(report) + + return findings diff --git a/prowler/providers/aws/services/bedrock/bedrock_knowledge_base_encrypted_with_cmk/__init__.py b/prowler/providers/aws/services/bedrock/bedrock_knowledge_base_encrypted_with_cmk/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/prowler/providers/aws/services/bedrock/bedrock_knowledge_base_encrypted_with_cmk/bedrock_knowledge_base_encrypted_with_cmk.metadata.json b/prowler/providers/aws/services/bedrock/bedrock_knowledge_base_encrypted_with_cmk/bedrock_knowledge_base_encrypted_with_cmk.metadata.json new file mode 100644 index 0000000000..a6ea8bfe5f --- /dev/null +++ b/prowler/providers/aws/services/bedrock/bedrock_knowledge_base_encrypted_with_cmk/bedrock_knowledge_base_encrypted_with_cmk.metadata.json @@ -0,0 +1,43 @@ +{ + "Provider": "aws", + "CheckID": "bedrock_knowledge_base_encrypted_with_cmk", + "CheckTitle": "Bedrock knowledge base data source is encrypted with a customer-managed KMS key", + "CheckType": [ + "Software and Configuration Checks/AWS Security Best Practices", + "Software and Configuration Checks/AWS Security Best Practices/Data Encryption" + ], + "ServiceName": "bedrock", + "SubServiceName": "", + "ResourceIdTemplate": "", + "Severity": "high", + "ResourceType": "Other", + "ResourceGroup": "ai_ml", + "Description": "Each data source of a **Bedrock knowledge base** can set `serverSideEncryptionConfiguration.kmsKeyArn`, which encrypts the transient storage used while documents are chunked and embedded, rather than relying on an AWS-owned key.", + "Risk": "**Knowledge base data sources** carry the private corpus a **RAG application** answers from, often the most sensitive content in the workload.\n\nUnder an **AWS-owned key**, there is no key policy restricting who may decrypt the ingestion storage, no account-controlled rotation, no **CloudTrail** record of key usage, and no way to revoke access by disabling a key.", + "RelatedUrl": "", + "AdditionalURLs": [ + "https://docs.aws.amazon.com/bedrock/latest/userguide/encryption-kb.html", + "https://docs.aws.amazon.com/bedrock/latest/APIReference/API_agent_GetDataSource.html" + ], + "Remediation": { + "Code": { + "CLI": "aws bedrock-agent update-data-source --knowledge-base-id example-kb-id --data-source-id example-ds-id --name example-data-source --data-source-configuration '{\"type\":\"S3\",\"s3Configuration\":{\"bucketArn\":\"arn:aws:s3:::example-bucket\"}}' --server-side-encryption-configuration '{\"kmsKeyArn\":\"arn:aws:kms:us-east-1:123456789012:key/example-key-id\"}'", + "NativeIaC": "```yaml\nResources:\n ExampleResource:\n Type: AWS::Bedrock::DataSource\n Properties:\n Name: example_resource\n KnowledgeBaseId: ABCDE12345\n DataSourceConfiguration:\n Type: S3\n S3Configuration:\n BucketArn: arn:aws:s3:::example-bucket\n ServerSideEncryptionConfiguration:\n KmsKeyArn: arn:aws:kms:us-east-1:123456789012:key/1234abcd-12ab-34cd-56ef-1234567890ab # Critical: omitting this uses an AWS-owned key\n```", + "Other": "1. Open the AWS Console and go to Amazon Bedrock\n2. Select **Knowledge bases**, then the knowledge base\n3. Select the data source and choose Edit\n4. Under transient data encryption, choose a customer-managed KMS key\n5. Grant the knowledge base service role kms:Decrypt and kms:GenerateDataKey on that key\n6. Save and re-sync the data source", + "Terraform": "```hcl\nresource \"aws_bedrockagent_data_source\" \"example_resource\" {\n name = \"example_resource\"\n knowledge_base_id = \"ABCDE12345\"\n data_source_configuration {\n type = \"S3\"\n s3_configuration {\n bucket_arn = \"arn:aws:s3:::example-bucket\"\n }\n }\n server_side_encryption_configuration {\n kms_key_arn = \"arn:aws:kms:us-east-1:123456789012:key/1234abcd-12ab-34cd-56ef-1234567890ab\" # Critical: omitting this uses an AWS-owned key\n }\n}\n```" + }, + "Recommendation": { + "Text": "Set a customer-managed KMS key on every knowledge base data source so the transient ingestion storage rests under a key with an auditable policy, and grant the knowledge base service role only kms:Decrypt and kms:GenerateDataKey on it.", + "Url": "https://hub.prowler.com/check/bedrock_knowledge_base_encrypted_with_cmk" + } + }, + "Categories": [ + "gen-ai", + "encryption" + ], + "DependsOn": [], + "RelatedTo": [ + "bedrock_custom_model_encrypted_with_cmk" + ], + "Notes": "Reports one finding per DATA SOURCE, not per knowledge base, because the key is configured on the data source; the knowledge base name is carried into the message. GetDataSource nests its payload under a top-level dataSource object. A data source whose detail could not be retrieved returns MANUAL rather than PASS, because an absent key ARN means the answer is unknown, not that no key is set." +} diff --git a/prowler/providers/aws/services/bedrock/bedrock_knowledge_base_encrypted_with_cmk/bedrock_knowledge_base_encrypted_with_cmk.py b/prowler/providers/aws/services/bedrock/bedrock_knowledge_base_encrypted_with_cmk/bedrock_knowledge_base_encrypted_with_cmk.py new file mode 100644 index 0000000000..b8598a3352 --- /dev/null +++ b/prowler/providers/aws/services/bedrock/bedrock_knowledge_base_encrypted_with_cmk/bedrock_knowledge_base_encrypted_with_cmk.py @@ -0,0 +1,80 @@ +from prowler.lib.check.models import Check, Check_Report_AWS +from prowler.providers.aws.services.bedrock.bedrock_agent_client import ( + bedrock_agent_client, +) + + +class bedrock_knowledge_base_encrypted_with_cmk(Check): + """Ensure Bedrock knowledge base data sources are encrypted with a CMK. + + One finding is reported per data source, because the key is configured on + the data source rather than on the knowledge base. + + - PASS: The data source sets `serverSideEncryptionConfiguration.kmsKeyArn`, + so the transient storage used during ingestion is encrypted with a key + the account controls. + - FAIL: No `kmsKeyArn` is set, so the ingested documents rest under an + AWS-owned key the organization cannot audit, rotate, or revoke. + - MANUAL: GetDataSource failed, so the key could not be retrieved and an + absent value cannot be read as "no key"; or ListDataSources failed for the + knowledge base, so its data sources are unknown; or ListKnowledgeBases + failed for a region, so the region's knowledge bases are unknown. + + A knowledge base whose data sources could not be listed is reported against + the knowledge base itself. Reporting nothing would drop it from the output + entirely, which reads as "no data sources to flag" and is indistinguishable + from a clean result. + """ + + def execute(self) -> list[Check_Report_AWS]: + """Execute the check logic. + + Returns: + A list of reports containing the result of the check. + """ + findings = [] + + for region, error in sorted( + bedrock_agent_client.knowledge_bases_scan_errors.items() + ): + report = Check_Report_AWS( + metadata=self.metadata(), resource={"region": region} + ) + report.region = region + report.resource_id = "knowledge-base/unknown" + report.resource_arn = f"arn:{bedrock_agent_client.audited_partition}:bedrock:{region}:{bedrock_agent_client.audited_account}:knowledge-base/unknown" + report.status = "MANUAL" + report.status_extended = f"Bedrock knowledge bases could not be listed in region {region} ({error}); verify manually that every knowledge base data source uses a customer-managed KMS key." + findings.append(report) + + for knowledge_base in bedrock_agent_client.knowledge_bases.values(): + if knowledge_base.data_sources_listed: + continue + report = Check_Report_AWS(metadata=self.metadata(), resource=knowledge_base) + report.status = "MANUAL" + reason = ( + f" ({knowledge_base.data_sources_error})" + if knowledge_base.data_sources_error + else "" + ) + report.status_extended = f"Bedrock knowledge base {knowledge_base.name} data sources could not be listed in region {knowledge_base.region}{reason}; verify manually that each one uses a customer-managed KMS key." + findings.append(report) + + for data_source in bedrock_agent_client.data_sources.values(): + report = Check_Report_AWS(metadata=self.metadata(), resource=data_source) + knowledge_base = ( + data_source.knowledge_base_name or data_source.knowledge_base_id + ) + + if not data_source.detail_retrieved: + report.status = "MANUAL" + report.status_extended = f"Bedrock knowledge base {knowledge_base} data source {data_source.name} encryption configuration could not be retrieved in region {data_source.region}; verify manually that it uses a customer-managed KMS key." + elif data_source.kms_key_arn: + report.status = "PASS" + report.status_extended = f"Bedrock knowledge base {knowledge_base} data source {data_source.name} is encrypted with a customer-managed KMS key in region {data_source.region}." + else: + report.status = "FAIL" + report.status_extended = f"Bedrock knowledge base {knowledge_base} data source {data_source.name} is not encrypted with a customer-managed KMS key in region {data_source.region}, so ingested documents rest under an AWS-owned key the organization cannot audit or revoke." + findings.append(report) + + return findings diff --git a/prowler/providers/aws/services/bedrock/bedrock_service.py b/prowler/providers/aws/services/bedrock/bedrock_service.py index aead63a67f..8e7ea7c843 100644 --- a/prowler/providers/aws/services/bedrock/bedrock_service.py +++ b/prowler/providers/aws/services/bedrock/bedrock_service.py @@ -16,10 +16,14 @@ class Bedrock(AWSService): self.guardrails = {} self.guardrails_scanned_regions = set() self.guardrails_scan_errors = {} + self.custom_models = {} + self.custom_models_scan_errors = {} self.__threading_call__(self._get_model_invocation_logging_configuration) self.__threading_call__(self._list_guardrails) self.__threading_call__(self._get_guardrail, self.guardrails.values()) self.__threading_call__(self._list_tags_for_resource, self.guardrails.values()) + self.__threading_call__(self._list_custom_models) + self.__threading_call__(self._get_custom_model, self.custom_models.values()) def _get_model_invocation_logging_arn_template(self, region): return ( @@ -56,6 +60,7 @@ class Bedrock(AWSService): ) def _list_guardrails(self, regional_client): + """List the guardrails in a region.""" logger.info("Bedrock - Listing Guardrails...") try: paginator = regional_client.get_paginator("list_guardrails") @@ -72,9 +77,11 @@ class Bedrock(AWSService): ) self.guardrails_scanned_regions.add(regional_client.region) except ClientError as error: - self.guardrails_scan_errors[regional_client.region] = error.response[ - "Error" - ].get("Code", error.__class__.__name__) + code = error.response["Error"].get("Code", error.__class__.__name__) + # ValidationException means Bedrock is unavailable in the region: + # a definite "no guardrails", so it must not become a MANUAL finding. + if code != "ValidationException": + self.guardrails_scan_errors[regional_client.region] = code logger.error( f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}" ) @@ -100,6 +107,19 @@ class Bedrock(AWSService): guardrail.prompt_attack_filter_strength = filter.get( "inputStrength", "NONE" ) + # Absent from the response when no grounding policy is configured. + guardrail.contextual_grounding_filters = [ + ContextualGroundingFilter( + type=filter.get("type"), + threshold=filter.get("threshold"), + action=filter.get("action"), + enabled=filter.get("enabled"), + ) + for filter in guardrail_info.get("contextualGroundingPolicy", {}).get( + "filters", [] + ) + ] + guardrail.detail_retrieved = True except Exception as error: logger.error( f"{guardrail.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}" @@ -118,6 +138,64 @@ class Bedrock(AWSService): f"{guardrail.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}" ) + def _list_custom_models(self, regional_client): + """List the customized models owned by the audited account. + + isOwned=True is required: without it the response also carries models + shared into this account through Resource Access Manager, whose KMS key + the audited account does not own and cannot set, so auditing them + produces a finding nobody here can remediate. + """ + logger.info("Bedrock - Listing Custom Models...") + try: + paginator = regional_client.get_paginator("list_custom_models") + for page in paginator.paginate(isOwned=True): + for model in page.get("modelSummaries", []): + model_arn = model.get("modelArn", "") + if model_arn and ( + not self.audit_resources + or is_resource_filtered(model_arn, self.audit_resources) + ): + self.custom_models[model_arn] = CustomModel( + name=model.get("modelName", ""), + arn=model_arn, + region=regional_client.region, + ) + except ClientError as error: + code = error.response["Error"].get("Code", error.__class__.__name__) + # ValidationException means Bedrock is unavailable in the region: + # a definite "no models", so it must not become a MANUAL finding. + if code != "ValidationException": + self.custom_models_scan_errors[regional_client.region] = code + logger.error( + f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}" + ) + except Exception as error: + self.custom_models_scan_errors[regional_client.region] = ( + error.__class__.__name__ + ) + logger.error( + f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}" + ) + + def _get_custom_model(self, model): + """Fetch the KMS key a custom model's weights are encrypted with. + + list_custom_models summaries do not carry modelKmsKeyArn, so a + per-model GetCustomModel call is required. + """ + logger.info("Bedrock - Getting Custom Model...") + try: + model_info = self.regional_clients[model.region].get_custom_model( + modelIdentifier=model.arn + ) + model.kms_key_arn = model_info.get("modelKmsKeyArn") + model.detail_retrieved = True + except Exception as error: + logger.error( + f"{model.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}" + ) + class LoggingConfiguration(BaseModel): enabled: bool = False @@ -125,6 +203,19 @@ class LoggingConfiguration(BaseModel): s3_bucket: Optional[str] = None +class ContextualGroundingFilter(BaseModel): + """One filter of a guardrail's contextualGroundingPolicy. + + type and threshold are required by the API; enabled and action are optional + and have no documented default, so an absent value means unknown. + """ + + type: Optional[str] = None + threshold: Optional[float] = None + action: Optional[str] = None + enabled: Optional[bool] = None + + class Guardrail(BaseModel): id: str name: str @@ -133,6 +224,20 @@ class Guardrail(BaseModel): tags: Optional[list] = [] sensitive_information_filter: bool = False prompt_attack_filter_strength: Optional[str] = None + contextual_grounding_filters: list[ContextualGroundingFilter] = [] + # False when GetGuardrail failed: absent policy is unknown, not unset. + detail_retrieved: bool = False + + +class CustomModel(BaseModel): + """Model representing a Bedrock customized (fine-tuned) model.""" + + name: str + arn: str + region: str + kms_key_arn: Optional[str] = None + # False when GetCustomModel failed: absent key is unknown, not unset. + detail_retrieved: bool = False class BedrockAgent(AWSService): @@ -143,13 +248,28 @@ class BedrockAgent(AWSService): # Call AWSService's __init__ super().__init__("bedrock-agent", provider) self.agents = {} + # Every agent in the account, including those --resource-arn excluded. A check whose verdict + # for one agent depends on another (role sharing) cannot read self.agents: the agent that + # proves the sharing may be the one the operator filtered out. Same shape as + # cloudwatch_service's all_log_groups/log_groups pair -- one object, two dicts. + self.all_agents = {} self.prompts = {} + self.knowledge_bases = {} + self.data_sources = {} + self.knowledge_bases_scan_errors = {} + self.agents_scan_errors = {} self.prompt_scanned_regions: set = set() self.__threading_call__(self._list_agents) - self.__threading_call__(self._get_agent, self.agents.values()) + # Detail collection runs over the COMPLETE inventory: an out-of-scope agent's role still + # determines whether an in-scope agent's role is shared. + self.__threading_call__(self._get_agent, self.all_agents.values()) + self.__threading_call__(self._get_agent_version_roles, self.all_agents.values()) self.__threading_call__(self._list_prompts) self.__threading_call__(self._get_prompt, self.prompts.values()) self.__threading_call__(self._list_tags_for_resource, self.agents.values()) + self.__threading_call__(self._list_knowledge_bases) + self.__threading_call__(self._list_data_sources, self.knowledge_bases.values()) + self.__threading_call__(self._get_data_source, self.data_sources.values()) def _list_agents(self, regional_client): logger.info("Bedrock Agent - Listing Agents...") @@ -157,20 +277,32 @@ class BedrockAgent(AWSService): paginator = regional_client.get_paginator("list_agents") for page in paginator.paginate(): for agent in page.get("agentSummaries", []): - agent_arn = f"arn:aws:bedrock:{regional_client.region}:{self.audited_account}:agent/{agent['agentId']}" + agent_arn = f"arn:{self.audited_partition}:bedrock:{regional_client.region}:{self.audited_account}:agent/{agent['agentId']}" + agent_object = Agent( + id=agent["agentId"], + name=agent["agentName"], + arn=agent_arn, + guardrail_id=agent.get("guardrailConfiguration", {}).get( + "guardrailIdentifier" + ), + region=regional_client.region, + ) + self.all_agents[agent_arn] = agent_object if not self.audit_resources or ( is_resource_filtered(agent_arn, self.audit_resources) ): - self.agents[agent_arn] = Agent( - id=agent["agentId"], - name=agent["agentName"], - arn=agent_arn, - guardrail_id=agent.get("guardrailConfiguration", {}).get( - "guardrailIdentifier" - ), - region=regional_client.region, - ) + self.agents[agent_arn] = agent_object + except ClientError as error: + code = error.response["Error"].get("Code", error.__class__.__name__) + # ValidationException means Bedrock Agent is unavailable in the + # region: a definite "no agents", so it must not become MANUAL. + if code != "ValidationException": + self.agents_scan_errors[regional_client.region] = code + logger.error( + f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}" + ) except Exception as error: + self.agents_scan_errors[regional_client.region] = error.__class__.__name__ logger.error( f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}" ) @@ -186,11 +318,86 @@ class BedrockAgent(AWSService): try: agent_info = self.regional_clients[agent.region].get_agent(agentId=agent.id) agent.role_arn = agent_info.get("agent", {}).get("agentResourceRoleArn") + agent.detail_retrieved = True except Exception as error: logger.error( f"{agent.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}" ) + # PREPARED is the only status AWS documents as ready to invoke. CREATING and + # UPDATING are in-flight, so their routed versions are unknown rather than + # definitely active or inactive. + TRANSITIONAL_ALIAS_STATUSES = {"CREATING", "UPDATING"} + + @staticmethod + def _is_alias_active(alias: dict) -> bool: + """Can this alias actually invoke the version it routes to? + + Args: + alias: One agentAliasSummaries entry from ListAgentAliases. + + Returns: + True only when the alias is prepared and does not reject invocations. + """ + if alias.get("aliasInvocationState") == "REJECT_INVOCATIONS": + return False + return alias.get("agentAliasStatus") == "PREPARED" + + def _get_agent_version_roles(self, agent): + """Fetch the execution role of every agent version an ACTIVE alias routes to. + + GetAgent returns only the working draft. An agent version is an + immutable snapshot that keeps the role it was cut with, and an alias + routes invocations at a specific version, so a deployed version can + still hold a role the draft no longer has. Only versions an active alias + routes to are fetched: a version nothing can invoke is not live + exposure, so reporting on it would be a false FAIL. + """ + logger.info("Bedrock Agent - Getting Agent Version Roles...") + try: + client = self.regional_clients[agent.region] + paginator = client.get_paginator("list_agent_aliases") + routed_versions = set() + inventory_complete = True + for page in paginator.paginate(agentId=agent.id): + for alias in page.get("agentAliasSummaries", []): + if ( + alias.get("agentAliasStatus") + in self.TRANSITIONAL_ALIAS_STATUSES + ): + inventory_complete = False + continue + if not self._is_alias_active(alias): + continue + for route in alias.get("routingConfiguration", []): + version = route.get("agentVersion") + # agentVersion is an optional member of the routing + # configuration, and DRAFT routes at the working draft + # whose role GetAgent already captured. + if version and version != "DRAFT": + routed_versions.add(version) + + for version in sorted(routed_versions): + version_info = client.get_agent_version( + agentId=agent.id, agentVersion=version + ) + agent.version_role_arns[version] = version_info.get( + "agentVersion", {} + ).get("agentResourceRoleArn") + agent.versions_listed = inventory_complete + except ClientError as error: + agent.versions_error = error.response["Error"].get( + "Code", error.__class__.__name__ + ) + logger.error( + f"{agent.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}" + ) + except Exception as error: + agent.versions_error = error.__class__.__name__ + logger.error( + f"{agent.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}" + ) + def _list_prompts(self, regional_client): """List all prompts in a region.""" logger.info("Bedrock Agent - Listing Prompts...") @@ -229,6 +436,111 @@ class BedrockAgent(AWSService): f"{prompt.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}" ) + def _list_knowledge_bases(self, regional_client): + """List the knowledge bases in a region.""" + logger.info("Bedrock Agent - Listing Knowledge Bases...") + try: + paginator = regional_client.get_paginator("list_knowledge_bases") + for page in paginator.paginate(): + for knowledge_base in page.get("knowledgeBaseSummaries", []): + knowledge_base_id = knowledge_base.get("knowledgeBaseId", "") + if not knowledge_base_id: + continue + knowledge_base_arn = f"arn:{self.audited_partition}:bedrock:{regional_client.region}:{self.audited_account}:knowledge-base/{knowledge_base_id}" + if not self.audit_resources or is_resource_filtered( + knowledge_base_arn, self.audit_resources + ): + self.knowledge_bases[knowledge_base_arn] = KnowledgeBase( + id=knowledge_base_id, + name=knowledge_base.get("name", ""), + arn=knowledge_base_arn, + region=regional_client.region, + ) + except ClientError as error: + code = error.response["Error"].get("Code", error.__class__.__name__) + # ValidationException means Bedrock Agent is unavailable in the + # region: a definite "none", so it must not become a MANUAL finding. + if code != "ValidationException": + self.knowledge_bases_scan_errors[regional_client.region] = code + logger.error( + f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}" + ) + except Exception as error: + self.knowledge_bases_scan_errors[regional_client.region] = ( + error.__class__.__name__ + ) + logger.error( + f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}" + ) + + def _list_data_sources(self, knowledge_base): + """List the data sources attached to one knowledge base. + + A failure is recorded on the knowledge base itself: findings are per data + source, so an unlisted knowledge base would otherwise vanish from the report. + """ + logger.info("Bedrock Agent - Listing Data Sources...") + try: + paginator = self.regional_clients[knowledge_base.region].get_paginator( + "list_data_sources" + ) + for page in paginator.paginate(knowledgeBaseId=knowledge_base.id): + for data_source in page.get("dataSourceSummaries", []): + data_source_id = data_source.get("dataSourceId", "") + if not data_source_id: + continue + data_source_arn = ( + f"{knowledge_base.arn}/data-source/{data_source_id}" + ) + # No filter here: the parent knowledge base was already + # filtered on its own ARN, and this ARN is synthetic because + # AWS exposes none for a data source, so it could never match + # a user-supplied --resource-arn and would silently drop every + # data source of an in-scope knowledge base. + self.data_sources[data_source_arn] = KnowledgeBaseDataSource( + id=data_source_id, + name=data_source.get("name", ""), + arn=data_source_arn, + region=knowledge_base.region, + knowledge_base_id=knowledge_base.id, + knowledge_base_name=knowledge_base.name, + ) + knowledge_base.data_sources_listed = True + except ClientError as error: + knowledge_base.data_sources_error = error.response["Error"].get( + "Code", error.__class__.__name__ + ) + logger.error( + f"{knowledge_base.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}" + ) + except Exception as error: + knowledge_base.data_sources_error = error.__class__.__name__ + logger.error( + f"{knowledge_base.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}" + ) + + def _get_data_source(self, data_source): + """Fetch the KMS key a data source's transient storage is encrypted with.""" + logger.info("Bedrock Agent - Getting Data Source...") + try: + # GetDataSource nests everything under a top-level dataSource object. + data_source_info = ( + self.regional_clients[data_source.region] + .get_data_source( + knowledgeBaseId=data_source.knowledge_base_id, + dataSourceId=data_source.id, + ) + .get("dataSource", {}) + ) + data_source.kms_key_arn = data_source_info.get( + "serverSideEncryptionConfiguration", {} + ).get("kmsKeyArn") + data_source.detail_retrieved = True + except Exception as error: + logger.error( + f"{data_source.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}" + ) + def _list_tags_for_resource(self, resource): """List tags for a Bedrock Agent resource.""" logger.info("Bedrock Agent - Listing Tags for Resource...") @@ -256,6 +568,17 @@ class Agent(BaseModel): role_arn: Optional[str] = None region: str tags: Optional[list] = [] + # False when GetAgent failed: absent role is unknown, not unset. + detail_retrieved: bool = False + # Execution role of each numbered version an alias routes to, keyed by + # version. A version is an immutable snapshot, so it keeps the role it was + # cut with even after the working draft's role changes. + version_role_arns: dict = {} + # True once the alias and version inventory was read in full, so an agent + # with no deployed versions genuinely has none rather than that they could + # not be listed. + versions_listed: bool = False + versions_error: Optional[str] = None class Prompt(BaseModel): @@ -266,3 +589,30 @@ class Prompt(BaseModel): arn: str region: str customer_encryption_key_arn: Optional[str] = None + + +class KnowledgeBase(BaseModel): + """Model representing a Bedrock Agent knowledge base.""" + + id: str + name: str + arn: str + region: str + # False when ListDataSources failed: empty set is unknown, not none. + data_sources_listed: bool = False + # The error code from a failed ListDataSources, for the finding message. + data_sources_error: Optional[str] = None + + +class KnowledgeBaseDataSource(BaseModel): + """Model representing a data source attached to a Bedrock knowledge base.""" + + id: str + name: str + arn: str + region: str + knowledge_base_id: str + knowledge_base_name: Optional[str] = None + kms_key_arn: Optional[str] = None + # False when GetDataSource failed: absent key is unknown, not unset. + detail_retrieved: bool = False diff --git a/tests/providers/aws/services/bedrock/bedrock_agent_role_not_shared_across_agents/bedrock_agent_role_not_shared_across_agents_test.py b/tests/providers/aws/services/bedrock/bedrock_agent_role_not_shared_across_agents/bedrock_agent_role_not_shared_across_agents_test.py new file mode 100644 index 0000000000..60cc40079c --- /dev/null +++ b/tests/providers/aws/services/bedrock/bedrock_agent_role_not_shared_across_agents/bedrock_agent_role_not_shared_across_agents_test.py @@ -0,0 +1,633 @@ +from unittest import mock + +import botocore +from botocore.exceptions import ClientError +from moto import mock_aws + +from tests.providers.aws.utils import ( + AWS_ACCOUNT_NUMBER, + AWS_REGION_US_EAST_1, + AWS_REGION_US_WEST_2, + set_mocked_aws_provider, +) + +make_api_call = botocore.client.BaseClient._make_api_call + +AGENT_A_ID = "test-agent-a" +AGENT_A_NAME = "agent-alpha" +AGENT_A_ARN = ( + f"arn:aws:bedrock:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:agent/{AGENT_A_ID}" +) +AGENT_B_ID = "test-agent-b" +AGENT_B_NAME = "agent-bravo" +AGENT_B_ARN = ( + f"arn:aws:bedrock:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:agent/{AGENT_B_ID}" +) +AGENT_C_ID = "test-agent-c" +AGENT_C_NAME = "agent-charlie" + +SHARED_ROLE_ARN = f"arn:aws:iam::{AWS_ACCOUNT_NUMBER}:role/SharedAgentRole" +ROLE_A_ARN = f"arn:aws:iam::{AWS_ACCOUNT_NUMBER}:role/AgentAlphaRole" +ROLE_B_ARN = f"arn:aws:iam::{AWS_ACCOUNT_NUMBER}:role/AgentBravoRole" + +# Operations the BedrockAgent constructor calls that these tests do not exercise. +_UNUSED_OPERATIONS = ( + "ListPrompts", + "GetPrompt", + "ListTagsForResource", + "ListKnowledgeBases", + "ListDataSources", + "GetDataSource", +) + +# These scenarios describe agents with no alias, so no version is deployed and +# the draft role is the only one in play. ListAgentAliases is stubbed empty +# rather than left to moto, which does not implement it: an unstubbed call would +# leave the version inventory unread and correctly downgrade every PASS to +# MANUAL, masking what these tests are actually asserting. +_NO_ALIASES = {"agentAliasSummaries": []} + + +def _agent_mock(agents, fail_get_for=()): + """Build a _make_api_call replacement returning the given agents. + + Args: + agents: list of (agentId, agentName, roleArn) tuples. + fail_get_for: agent ids whose GetAgent call must raise. + """ + + def _mock(self, operation_name, kwarg): + if operation_name in _UNUSED_OPERATIONS: + return {} + if operation_name == "ListAgentAliases": + return _NO_ALIASES + if operation_name == "ListAgents": + return { + "agentSummaries": [ + { + "agentId": agent_id, + "agentName": agent_name, + "agentStatus": "PREPARED", + } + for agent_id, agent_name, _ in agents + ] + } + if operation_name == "GetAgent": + agent_id = kwarg["agentId"] + if agent_id in fail_get_for: + raise ClientError( + {"Error": {"Code": "AccessDeniedException", "Message": "denied"}}, + operation_name, + ) + # GetAgent nests its payload under a top-level agent key. + for candidate_id, agent_name, role_arn in agents: + if candidate_id == agent_id: + agent = { + "agentId": candidate_id, + "agentName": agent_name, + "agentStatus": "PREPARED", + } + if role_arn is not None: + agent["agentResourceRoleArn"] = role_arn + return {"agent": agent} + return {"agent": {}} + return make_api_call(self, operation_name, kwarg) + + return _mock + + +_mock_dedicated_roles = _agent_mock( + [ + (AGENT_A_ID, AGENT_A_NAME, ROLE_A_ARN), + (AGENT_B_ID, AGENT_B_NAME, ROLE_B_ARN), + ] +) +_mock_shared_role = _agent_mock( + [ + (AGENT_A_ID, AGENT_A_NAME, SHARED_ROLE_ARN), + (AGENT_B_ID, AGENT_B_NAME, SHARED_ROLE_ARN), + ] +) +_mock_shared_role_three_agents = _agent_mock( + [ + (AGENT_A_ID, AGENT_A_NAME, SHARED_ROLE_ARN), + (AGENT_B_ID, AGENT_B_NAME, SHARED_ROLE_ARN), + (AGENT_C_ID, AGENT_C_NAME, SHARED_ROLE_ARN), + ] +) +_mock_single_agent = _agent_mock([(AGENT_A_ID, AGENT_A_NAME, ROLE_A_ARN)]) +_mock_role_missing = _agent_mock([(AGENT_A_ID, AGENT_A_NAME, None)]) +_mock_get_agent_fails = _agent_mock( + [(AGENT_A_ID, AGENT_A_NAME, ROLE_A_ARN)], fail_get_for=(AGENT_A_ID,) +) +# One agent's role is unreadable, so it must not count toward the other's share. +_mock_one_unreadable_one_readable = _agent_mock( + [ + (AGENT_A_ID, AGENT_A_NAME, SHARED_ROLE_ARN), + (AGENT_B_ID, AGENT_B_NAME, SHARED_ROLE_ARN), + ], + fail_get_for=(AGENT_B_ID,), +) + + +def _mock_empty(self, operation_name, kwarg): + """No agents at all.""" + if operation_name in _UNUSED_OPERATIONS: + return {} + if operation_name == "ListAgentAliases": + return _NO_ALIASES + if operation_name == "ListAgents": + return {"agentSummaries": []} + return make_api_call(self, operation_name, kwarg) + + +def _mock_unsupported_region(self, operation_name, kwarg): + """The API is not available in the audited region.""" + if operation_name in _UNUSED_OPERATIONS: + return {} + if operation_name == "ListAgentAliases": + return _NO_ALIASES + if operation_name == "ListAgents": + raise ClientError( + { + "Error": { + "Code": "ValidationException", + "Message": "Bedrock Agent is not supported in this region.", + } + }, + operation_name, + ) + return make_api_call(self, operation_name, kwarg) + + +def _inventory_mock(agents): + """Build a stub for one exhaustive-matrix inventory. + + Args: + agents: list of (name, roleArn, retrieved) tuples. retrieved=False makes + GetAgent raise for that agent. + """ + rows = [(name, name, role) for name, role, _ in agents] + fail = tuple(name for name, _, retrieved in agents if not retrieved) + return _agent_mock(rows, fail_get_for=fail) + + +def _mock_list_agents_denied(self, operation_name, kwarg): + """ListAgents is denied, so the region's agents are unknown.""" + if operation_name in _UNUSED_OPERATIONS: + return {} + if operation_name == "ListAgentAliases": + return _NO_ALIASES + if operation_name == "ListAgents": + raise ClientError( + {"Error": {"Code": "AccessDeniedException", "Message": "denied"}}, + operation_name, + ) + return make_api_call(self, operation_name, kwarg) + + +def _mock_dedicated_role_with_partial_inventory(self, operation_name, kwarg): + """us-east-1 lists one agent; us-west-2's ListAgents is denied. + + The listed agent's role is used by no other KNOWN agent, but an unlisted + Region could hold one sharing it, so PASS must not be asserted. + """ + if operation_name in _UNUSED_OPERATIONS: + return {} + if operation_name == "ListAgentAliases": + return _NO_ALIASES + region = self.meta.region_name + if operation_name == "ListAgents": + if region == AWS_REGION_US_EAST_1: + return { + "agentSummaries": [ + { + "agentId": AGENT_A_ID, + "agentName": AGENT_A_NAME, + "agentStatus": "PREPARED", + } + ] + } + raise ClientError( + {"Error": {"Code": "AccessDeniedException", "Message": "denied"}}, + operation_name, + ) + if operation_name == "GetAgent": + return { + "agent": { + "agentId": AGENT_A_ID, + "agentName": AGENT_A_NAME, + "agentStatus": "PREPARED", + "agentResourceRoleArn": ROLE_A_ARN, + } + } + return make_api_call(self, operation_name, kwarg) + + +class Test_bedrock_agent_role_not_shared_across_agents: + """Unit tests for the bedrock_agent_role_not_shared_across_agents check.""" + + def _run(self): + """Import the service + check under the active mocks and execute.""" + from prowler.providers.aws.services.bedrock.bedrock_service import BedrockAgent + + 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, + ), + mock.patch( + "prowler.providers.aws.services.bedrock.bedrock_agent_role_not_shared_across_agents.bedrock_agent_role_not_shared_across_agents.bedrock_agent_client", + new=BedrockAgent(aws_provider), + ), + ): + from prowler.providers.aws.services.bedrock.bedrock_agent_role_not_shared_across_agents.bedrock_agent_role_not_shared_across_agents import ( + bedrock_agent_role_not_shared_across_agents, + ) + + return bedrock_agent_role_not_shared_across_agents().execute() + + def _run_multi_region(self): + """Same as _run but with two Regions in scope.""" + from prowler.providers.aws.services.bedrock.bedrock_service import BedrockAgent + + aws_provider = set_mocked_aws_provider( + [AWS_REGION_US_EAST_1, AWS_REGION_US_WEST_2] + ) + with ( + mock.patch( + "prowler.providers.common.provider.Provider.get_global_provider", + return_value=aws_provider, + ), + mock.patch( + "prowler.providers.aws.services.bedrock.bedrock_agent_role_not_shared_across_agents.bedrock_agent_role_not_shared_across_agents.bedrock_agent_client", + new=BedrockAgent(aws_provider), + ), + ): + from prowler.providers.aws.services.bedrock.bedrock_agent_role_not_shared_across_agents.bedrock_agent_role_not_shared_across_agents import ( + bedrock_agent_role_not_shared_across_agents, + ) + + return bedrock_agent_role_not_shared_across_agents().execute() + + @mock.patch("botocore.client.BaseClient._make_api_call", new=_mock_empty) + @mock_aws + def test_no_resources(self): + """No resources means no findings, not a spurious FAIL.""" + assert self._run() == [] + + @mock.patch( + "botocore.client.BaseClient._make_api_call", new=_mock_unsupported_region + ) + @mock_aws + def test_region_not_supported(self): + """A ValidationException from the region must not raise; it yields no findings.""" + assert self._run() == [] + + @mock.patch("botocore.client.BaseClient._make_api_call", new=_mock_single_agent) + @mock_aws + def test_single_agent_passes(self): + """The only agent in the account cannot be sharing its role.""" + result = self._run() + assert len(result) == 1 + assert result[0].status == "PASS" + assert result[0].resource_id == AGENT_A_ID + assert result[0].resource_arn == AGENT_A_ARN + assert result[0].region == AWS_REGION_US_EAST_1 + assert ( + result[0].status_extended + == f"Bedrock Agent {AGENT_A_NAME} has a dedicated execution role in region {AWS_REGION_US_EAST_1}." + ) + + @mock.patch("botocore.client.BaseClient._make_api_call", new=_mock_dedicated_roles) + @mock_aws + def test_distinct_roles_pass(self): + """Two agents with distinct roles are both compliant.""" + result = self._run() + assert len(result) == 2 + assert {report.status for report in result} == {"PASS"} + + @mock.patch("botocore.client.BaseClient._make_api_call", new=_mock_shared_role) + @mock_aws + def test_shared_role_fails_for_both_agents(self): + """A role used by two agents fails for each of them, naming the other.""" + result = self._run() + assert len(result) == 2 + assert {report.status for report in result} == {"FAIL"} + by_id = {report.resource_id: report for report in result} + assert SHARED_ROLE_ARN in by_id[AGENT_A_ID].status_extended + assert AGENT_B_NAME in by_id[AGENT_A_ID].status_extended + assert AGENT_A_NAME in by_id[AGENT_B_ID].status_extended + assert "cannot attribute an action" in by_id[AGENT_A_ID].status_extended + + @mock.patch( + "botocore.client.BaseClient._make_api_call", new=_mock_shared_role_three_agents + ) + @mock_aws + def test_shared_role_names_others_sorted(self): + """The other sharing agents are listed in sorted order for determinism.""" + result = self._run() + assert len(result) == 3 + assert {report.status for report in result} == {"FAIL"} + by_id = {report.resource_id: report for report in result} + assert ( + f"with {AGENT_B_NAME}, {AGENT_C_NAME} in region" + in by_id[AGENT_A_ID].status_extended + ) + assert ( + f"with {AGENT_A_NAME}, {AGENT_C_NAME} in region" + in by_id[AGENT_B_ID].status_extended + ) + + @mock.patch("botocore.client.BaseClient._make_api_call", new=_mock_role_missing) + @mock_aws + def test_role_absent_is_manual_not_pass(self): + """An agent whose GetAgent returned no role must not be reported as compliant.""" + result = self._run() + assert len(result) == 1 + assert result[0].status == "MANUAL" + assert "could not be retrieved" in result[0].status_extended + + @mock.patch("botocore.client.BaseClient._make_api_call", new=_mock_get_agent_fails) + @mock_aws + def test_get_agent_failure_is_manual_not_pass(self): + """A failed GetAgent must not be reported as compliant.""" + result = self._run() + assert len(result) == 1 + assert result[0].status == "MANUAL" + assert "could not be retrieved" in result[0].status_extended + + @mock.patch( + "botocore.client.BaseClient._make_api_call", + new=_mock_one_unreadable_one_readable, + ) + @mock_aws + def test_unreadable_agent_does_not_inflate_share_count(self): + """An excluded agent must neither manufacture a FAIL nor allow a PASS. + + The unreadable agent is kept out of the share index, so the readable agent + is not reported as sharing. But its role is unknown and could be the same + one, so the readable agent cannot be asserted dedicated either: both are + MANUAL, and neither is FAIL. + """ + result = self._run() + assert len(result) == 2 + by_id = {report.resource_id: report for report in result} + assert by_id[AGENT_B_ID].status == "MANUAL" + assert by_id[AGENT_A_ID].status == "MANUAL" + assert "could not be read" in by_id[AGENT_A_ID].status_extended + # The entry names what could not be read, so the sentence stays + # grammatical for a role, a version set, or a Region alike. + assert ( + f"the execution role of {AGENT_B_NAME}" in by_id[AGENT_A_ID].status_extended + ) + assert {report.status for report in result} == {"MANUAL"} + + def test_every_inventory_shape_resolves_correctly(self): + """Exhaust the decision space instead of sampling it. + + Bedrock Agents cannot be created in every account (the service refuses new + agents for accounts without prior usage), so this check's behaviour is + pinned by enumerating every inventory of up to three agents over the cross + product of {role A, role B, no role} x {readable, unreadable}, and + asserting the verdict for each agent. A role counts as shared only when two + or more READABLE agents hold it. + + Each case is driven through the real BedrockAgent service so a renamed + service attribute breaks the test rather than passing silently. + """ + from itertools import product + + states = [ + (ROLE_A_ARN, True), + (ROLE_B_ARN, True), + (None, True), + (ROLE_A_ARN, False), + ] + + checked = 0 + for size in (1, 2, 3): + for combo in product(states, repeat=size): + agents = [ + (f"agent-{index}", role, retrieved) + for index, (role, retrieved) in enumerate(combo) + ] + + readable_per_role = {} + for _, role, retrieved in agents: + if retrieved and role: + readable_per_role[role] = readable_per_role.get(role, 0) + 1 + # Any agent whose own role could not be read leaves the picture + # incomplete, so no other agent can be asserted dedicated. A role + # already seen twice is shared regardless. + any_unresolved = any( + not retrieved or not role for _, role, retrieved in agents + ) + expected = sorted( + ( + "MANUAL" + if not retrieved or not role + else ( + "FAIL" + if readable_per_role[role] >= 2 + else ("MANUAL" if any_unresolved else "PASS") + ) + ) + for _, role, retrieved in agents + ) + + with mock.patch( + "botocore.client.BaseClient._make_api_call", + new=_inventory_mock(agents), + ): + with mock_aws(): + result = self._run() + + assert len(result) == size, combo + assert sorted(report.status for report in result) == expected, combo + assert all( + report.status_extended.endswith(".") for report in result + ), combo + checked += 1 + + # 4 + 16 + 64 inventories. + assert checked == 84 + + @mock.patch( + "botocore.client.BaseClient._make_api_call", new=_mock_list_agents_denied + ) + @mock_aws + def test_list_agents_denied_is_manual_not_silence(self): + """A denied ListAgents must report MANUAL for the region, not vanish.""" + result = self._run() + assert len(result) == 1 + assert result[0].status == "MANUAL" + assert result[0].region == AWS_REGION_US_EAST_1 + assert result[0].resource_id == "agent/unknown" + assert "could not be listed" in result[0].status_extended + assert "AccessDeniedException" in result[0].status_extended + assert result[0].status_extended.endswith(".") + + @mock.patch( + "botocore.client.BaseClient._make_api_call", + new=_mock_dedicated_role_with_partial_inventory, + ) + @mock_aws + def test_dedicated_role_is_manual_when_inventory_incomplete(self): + """A seemingly dedicated role cannot be asserted from a partial inventory. + + One Region lists an agent whose role no other known agent uses, while + another Region's ListAgents fails. An unlisted Region could hold an agent + sharing that role, so the verdict is MANUAL rather than PASS. + """ + result = self._run_multi_region() + by_status = {} + for report in result: + by_status.setdefault(report.status, []).append(report) + assert "PASS" not in by_status, [r.status_extended for r in result] + assert len(by_status["MANUAL"]) == 2 + agent_report = [ + r for r in by_status["MANUAL"] if r.resource_id != "agent/unknown" + ] + assert len(agent_report) == 1 + assert "agents in region" in agent_report[0].status_extended + assert all(r.status_extended.endswith(".") for r in result) + + @mock.patch( + "botocore.client.BaseClient._make_api_call", + new=_agent_mock( + [ + (AGENT_A_ID, AGENT_A_NAME, ROLE_A_ARN), + (AGENT_B_ID, AGENT_B_NAME, ROLE_B_ARN), + ], + fail_get_for=(AGENT_B_ID,), + ), + ) + @mock_aws + def test_unresolved_role_blocks_pass_for_a_distinct_role(self): + """One agent's unreadable role prevents asserting another's dedication. + + Agent A holds a role no other *readable* agent uses, so the old logic + returned PASS. Agent B's role could not be retrieved and may be the same + one, so PASS would be an assertion the data does not support. + """ + result = self._run() + assert len(result) == 2 + by_id = {report.resource_id: report for report in result} + assert by_id[AGENT_B_ID].status == "MANUAL" + assert by_id[AGENT_A_ID].status == "MANUAL" + assert AGENT_B_NAME in by_id[AGENT_A_ID].status_extended + assert "PASS" not in {report.status for report in result} + + +class Test_scoped_scan_still_sees_the_sharing: + """A --resource-arn scoped scan must not turn a shared role into a PASS. + + Collectors apply is_resource_filtered at COLLECTION time, so + bedrock_agent_client.agents holds only the agents the operator selected. + Whether a role is shared is a property of every agent that holds it, so an + index built from the filtered set cannot see the agent that proves the + violation -- and an operator filter is not a scan error, so nothing marks the + inventory incomplete either. The check therefore aggregates over all_agents + and filters only when emitting findings. + """ + + def _run_scoped(self, audit_resources): + """Execute the check with a scan scoped to the given resource ARNs.""" + from prowler.providers.aws.services.bedrock.bedrock_service import BedrockAgent + + aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1]) + aws_provider._audit_resources = audit_resources + with mock.patch( + "prowler.providers.common.provider.Provider.get_global_provider", + return_value=aws_provider, + ): + service = BedrockAgent(aws_provider) + with mock.patch( + "prowler.providers.aws.services.bedrock." + "bedrock_agent_role_not_shared_across_agents." + "bedrock_agent_role_not_shared_across_agents.bedrock_agent_client", + new=service, + ): + from prowler.providers.aws.services.bedrock.bedrock_agent_role_not_shared_across_agents.bedrock_agent_role_not_shared_across_agents import ( + bedrock_agent_role_not_shared_across_agents, + ) + + return service, bedrock_agent_role_not_shared_across_agents().execute() + + @mock_aws + @mock.patch( + "botocore.client.BaseClient._make_api_call", + new=_agent_mock( + [ + (AGENT_A_ID, AGENT_A_NAME, SHARED_ROLE_ARN), + (AGENT_B_ID, AGENT_B_NAME, SHARED_ROLE_ARN), + ] + ), + ) + def test_selecting_one_of_two_sharing_agents_still_fails(self): + """The unselected agent is what proves the role is shared. + + Only agent A is in scope, so exactly one finding is emitted -- but agent + B, filtered out of the report, still holds the same role, so agent A's + role is not dedicated. Reporting PASS here was the reproducible false + PASS a scoped scan produced. + """ + service, result = self._run_scoped([AGENT_A_ARN]) + + # The report set is narrowed; the role index is not. + assert list(service.agents) == [AGENT_A_ARN] + assert sorted(service.all_agents) == sorted([AGENT_A_ARN, AGENT_B_ARN]) + + assert len(result) == 1 + assert result[0].resource_id == AGENT_A_ID + assert result[0].resource_arn == AGENT_A_ARN + assert result[0].region == AWS_REGION_US_EAST_1 + assert result[0].status == "FAIL" + assert SHARED_ROLE_ARN in result[0].status_extended + # The out-of-scope agent is still named, because it is the evidence. + assert AGENT_B_NAME in result[0].status_extended + + @mock_aws + @mock.patch( + "botocore.client.BaseClient._make_api_call", + new=_agent_mock( + [ + (AGENT_A_ID, AGENT_A_NAME, ROLE_A_ARN), + (AGENT_B_ID, AGENT_B_NAME, ROLE_B_ARN), + ] + ), + ) + def test_scoped_scan_on_a_genuinely_dedicated_role_passes(self): + """Completeness must not manufacture a FAIL either. + + Aggregating over the whole account is only correct if a genuinely + dedicated role still passes when the scan is scoped. + """ + service, result = self._run_scoped([AGENT_A_ARN]) + + assert list(service.agents) == [AGENT_A_ARN] + assert len(service.all_agents) == 2 + assert len(result) == 1 + assert result[0].resource_id == AGENT_A_ID + assert result[0].status == "PASS" + assert "has a dedicated execution role" in result[0].status_extended + + @mock_aws + @mock.patch( + "botocore.client.BaseClient._make_api_call", + new=_agent_mock( + [ + (AGENT_A_ID, AGENT_A_NAME, SHARED_ROLE_ARN), + (AGENT_B_ID, AGENT_B_NAME, SHARED_ROLE_ARN), + ] + ), + ) + def test_unscoped_scan_is_unchanged(self): + """With no filter, every agent is both aggregated and reported.""" + service, result = self._run_scoped(None) + + assert sorted(service.agents) == sorted(service.all_agents) + assert len(result) == 2 + assert {report.status for report in result} == {"FAIL"} diff --git a/tests/providers/aws/services/bedrock/bedrock_agent_role_not_shared_across_agents/bedrock_agent_role_not_shared_across_agents_versions_test.py b/tests/providers/aws/services/bedrock/bedrock_agent_role_not_shared_across_agents/bedrock_agent_role_not_shared_across_agents_versions_test.py new file mode 100644 index 0000000000..37035b95b7 --- /dev/null +++ b/tests/providers/aws/services/bedrock/bedrock_agent_role_not_shared_across_agents/bedrock_agent_role_not_shared_across_agents_versions_test.py @@ -0,0 +1,556 @@ +"""Tests that a role shared through a deployed agent version is still reported. + +GetAgent returns only the working draft. An agent version is an immutable +snapshot that keeps the role it was cut with, and an alias routes invocations at +a specific version, so two agents whose drafts hold distinct roles can still be +invoking one shared role in production. Judging the draft alone reports that as +compliant. +""" + +from unittest import mock + +import botocore +import pytest +from botocore.exceptions import ClientError +from moto import mock_aws + +from tests.providers.aws.utils import ( + AWS_ACCOUNT_NUMBER, + AWS_REGION_US_EAST_1, + set_mocked_aws_provider, +) + +make_api_call = botocore.client.BaseClient._make_api_call + +AGENT_A_ID = "test-agent-a" +AGENT_A_NAME = "agent-alpha" +AGENT_B_ID = "test-agent-b" +AGENT_B_NAME = "agent-bravo" + +DRAFT_A_ARN = f"arn:aws:iam::{AWS_ACCOUNT_NUMBER}:role/AgentAlphaDraftRole" +DRAFT_B_ARN = f"arn:aws:iam::{AWS_ACCOUNT_NUMBER}:role/AgentBravoDraftRole" +SHARED_VERSION_ARN = f"arn:aws:iam::{AWS_ACCOUNT_NUMBER}:role/SharedVersionRole" +DEDICATED_VERSION_ARN = f"arn:aws:iam::{AWS_ACCOUNT_NUMBER}:role/AlphaVersionRole" + +# Collectors the BedrockAgent constructor runs that these tests do not exercise. +_UNUSED_OPERATIONS = ( + "ListPrompts", + "GetPrompt", + "ListTagsForResource", + "ListKnowledgeBases", + "ListDataSources", + "GetDataSource", +) + +DRAFT_ROLES = {AGENT_A_ID: DRAFT_A_ARN, AGENT_B_ID: DRAFT_B_ARN} + + +def _mock( + version_roles, + routed_version="3", + fail_list_aliases=(), + fail_get_version=(), + route_draft=False, + no_aliases=(), + alias_status="PREPARED", + alias_invocation_state=None, + aliases=None, + draft_roles=DRAFT_ROLES, +): + """Build a _make_api_call replacement for a two-agent account. + + Args: + version_roles: agent id -> role ARN its deployed version was cut with. + routed_version: the version each alias routes to. + fail_list_aliases: agent ids whose ListAgentAliases must raise. + fail_get_version: agent ids whose GetAgentVersion must raise. + route_draft: route the alias at DRAFT instead of a numbered version. + no_aliases: agent ids that have no alias at all. + alias_status: agentAliasStatus each alias reports. + alias_invocation_state: aliasInvocationState each alias reports; omitted + from the response entirely when None, which is what the API does for + an alias never set to reject. + aliases: agentAliasSummaries to return instead of building one alias. + draft_roles: agent id -> role ARN returned by GetAgent. + """ + + def _call(self, operation_name, kwarg): + if operation_name in _UNUSED_OPERATIONS: + return {} + if operation_name == "ListAgents": + return { + "agentSummaries": [ + { + "agentId": AGENT_A_ID, + "agentName": AGENT_A_NAME, + "agentStatus": "PREPARED", + }, + { + "agentId": AGENT_B_ID, + "agentName": AGENT_B_NAME, + "agentStatus": "PREPARED", + }, + ] + } + if operation_name == "GetAgent": + agent_id = kwarg["agentId"] + return { + "agent": { + "agentId": agent_id, + "agentName": agent_id, + "agentStatus": "PREPARED", + "agentResourceRoleArn": draft_roles[agent_id], + } + } + if operation_name == "ListAgentAliases": + agent_id = kwarg["agentId"] + if agent_id in fail_list_aliases: + raise ClientError( + {"Error": {"Code": "AccessDeniedException", "Message": "denied"}}, + operation_name, + ) + if aliases is not None: + return {"agentAliasSummaries": aliases} + if agent_id in no_aliases: + return {"agentAliasSummaries": []} + alias = { + "agentAliasId": "alias-1", + "agentAliasName": "production", + "agentAliasStatus": alias_status, + "routingConfiguration": [ + {"agentVersion": "DRAFT" if route_draft else routed_version} + ], + } + if alias_invocation_state is not None: + alias["aliasInvocationState"] = alias_invocation_state + return {"agentAliasSummaries": [alias]} + if operation_name == "GetAgentVersion": + agent_id = kwarg["agentId"] + if agent_id in fail_get_version: + raise ClientError( + {"Error": {"Code": "AccessDeniedException", "Message": "denied"}}, + operation_name, + ) + return { + "agentVersion": { + "agentId": agent_id, + "agentName": agent_id, + "version": kwarg["agentVersion"], + "agentStatus": "PREPARED", + "agentResourceRoleArn": version_roles[agent_id], + } + } + return make_api_call(self, operation_name, kwarg) + + return _call + + +def _run(stub): + """Import the service and check under the stub and execute.""" + from prowler.providers.aws.services.bedrock.bedrock_service import BedrockAgent + + check_name = "bedrock_agent_role_not_shared_across_agents" + aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1]) + with ( + mock.patch("botocore.client.BaseClient._make_api_call", new=stub), + mock.patch( + "prowler.providers.common.provider.Provider.get_global_provider", + return_value=aws_provider, + ), + ): + service = BedrockAgent(aws_provider) + with mock.patch( + f"prowler.providers.aws.services.bedrock.{check_name}.{check_name}.bedrock_agent_client", + new=service, + ): + module = __import__( + f"prowler.providers.aws.services.bedrock.{check_name}.{check_name}", + fromlist=[check_name], + ) + return service, getattr(module, check_name)().execute() + + +class Test_agent_version_roles: + """Tests for sharing judged across the draft and every routed version.""" + + @mock_aws + def test_shared_version_role_fails_despite_distinct_drafts(self): + """Distinct drafts do not make two agents dedicated. + + Both agents' aliases route at a version cut with one shared role, so + both are invoking it in production. Judging the draft alone would call + this compliant. + """ + service, results = _run( + _mock({AGENT_A_ID: SHARED_VERSION_ARN, AGENT_B_ID: SHARED_VERSION_ARN}) + ) + + for agent in service.agents.values(): + assert agent.versions_listed is True + assert agent.version_role_arns == {"3": SHARED_VERSION_ARN} + + assert len(results) == 2 + assert {report.status for report in results} == {"FAIL"} + for report in results: + assert SHARED_VERSION_ARN in report.status_extended + # The message names the version, so the reader knows the draft is + # not where the sharing lives. + assert "through deployed version 3" in report.status_extended + assert report.status_extended.endswith(".") + + @mock_aws + def test_distinct_version_roles_pass(self): + """Distinct roles on both the draft and the deployed version comply.""" + _, results = _run( + _mock( + { + AGENT_A_ID: DEDICATED_VERSION_ARN, + AGENT_B_ID: f"arn:aws:iam::{AWS_ACCOUNT_NUMBER}:role/BravoVersionRole", + } + ) + ) + + assert len(results) == 2 + assert {report.status for report in results} == {"PASS"} + + @mock_aws + def test_draft_sharing_still_names_no_version(self): + """Sharing on the draft is reported without a version attribution.""" + stub = _mock( + { + AGENT_A_ID: DEDICATED_VERSION_ARN, + AGENT_B_ID: f"arn:aws:iam::{AWS_ACCOUNT_NUMBER}:role/BravoVersionRole", + } + ) + + def _shared_draft(self, operation_name, kwarg): + if operation_name == "GetAgent": + return { + "agent": { + "agentId": kwarg["agentId"], + "agentName": kwarg["agentId"], + "agentStatus": "PREPARED", + "agentResourceRoleArn": DRAFT_A_ARN, + } + } + return stub(self, operation_name, kwarg) + + _, results = _run(_shared_draft) + + assert len(results) == 2 + assert {report.status for report in results} == {"FAIL"} + for report in results: + assert DRAFT_A_ARN in report.status_extended + assert "through deployed version" not in report.status_extended + + @mock_aws + def test_alias_routing_at_draft_needs_no_version_call(self): + """DRAFT resolves to the role GetAgent already returned.""" + service, results = _run( + _mock( + {AGENT_A_ID: SHARED_VERSION_ARN, AGENT_B_ID: SHARED_VERSION_ARN}, + route_draft=True, + ) + ) + + for agent in service.agents.values(): + assert agent.versions_listed is True + assert agent.version_role_arns == {} + + assert len(results) == 2 + assert {report.status for report in results} == {"PASS"} + + @mock_aws + def test_agent_without_an_alias_deploys_no_version(self): + """No alias means no version is reachable, so the draft is the verdict.""" + service, results = _run( + _mock( + {AGENT_A_ID: SHARED_VERSION_ARN, AGENT_B_ID: SHARED_VERSION_ARN}, + no_aliases=(AGENT_A_ID, AGENT_B_ID), + ) + ) + + for agent in service.agents.values(): + assert agent.versions_listed is True + assert agent.version_role_arns == {} + + assert {report.status for report in results} == {"PASS"} + + @mock_aws + def test_unlistable_aliases_block_pass(self): + """An unread version inventory may hold the same role, so PASS is unsafe.""" + service, results = _run( + _mock( + {AGENT_A_ID: DEDICATED_VERSION_ARN, AGENT_B_ID: DEDICATED_VERSION_ARN}, + fail_list_aliases=(AGENT_B_ID,), + ) + ) + + agent_b = next(a for a in service.agents.values() if a.id == AGENT_B_ID) + assert agent_b.versions_listed is False + assert agent_b.versions_error == "AccessDeniedException" + + assert "PASS" not in {report.status for report in results} + assert {report.status for report in results} == {"MANUAL"} + assert any( + "deployed versions of" in report.status_extended for report in results + ) + + @mock_aws + def test_unreadable_version_blocks_pass(self): + """A failed GetAgentVersion leaves that version's role unknown.""" + service, results = _run( + _mock( + {AGENT_A_ID: DEDICATED_VERSION_ARN, AGENT_B_ID: DEDICATED_VERSION_ARN}, + fail_get_version=(AGENT_A_ID,), + ) + ) + + agent_a = next(a for a in service.agents.values() if a.id == AGENT_A_ID) + assert agent_a.versions_listed is False + + assert "PASS" not in {report.status for report in results} + + @mock_aws + def test_shared_version_role_outranks_an_incomplete_inventory(self): + """A role seen on two agents is shared whatever else is missing. + + Both agents route at two versions and the second version is unreadable, + so the inventory is incomplete and the read half already shows the role + on both. A partial answer must not downgrade a definite finding, so FAIL + stands rather than MANUAL. + """ + + def _two_versions_one_unreadable(self, operation_name, kwarg): + if operation_name == "ListAgentAliases": + return { + "agentAliasSummaries": [ + { + "agentAliasId": "alias-1", + "agentAliasName": "production", + "agentAliasStatus": "PREPARED", + "routingConfiguration": [ + {"agentVersion": "2"}, + {"agentVersion": "3"}, + ], + } + ] + } + if operation_name == "GetAgentVersion" and kwarg["agentVersion"] == "3": + raise ClientError( + {"Error": {"Code": "AccessDeniedException", "Message": "denied"}}, + operation_name, + ) + return _mock( + {AGENT_A_ID: SHARED_VERSION_ARN, AGENT_B_ID: SHARED_VERSION_ARN} + )(self, operation_name, kwarg) + + service, results = _run(_two_versions_one_unreadable) + + for agent in service.agents.values(): + assert agent.versions_listed is False, "the inventory must be incomplete" + assert agent.version_role_arns == {"2": SHARED_VERSION_ARN} + + assert {report.status for report in results} == {"FAIL"} + for report in results: + assert "through deployed version 2" in report.status_extended + + +class Test_alias_must_be_invocable: + """A version only an unreachable alias routes to is not live exposure. + + Widening the audit from the working draft to every routed version closes a + false PASS and opens the symmetric false FAIL: an alias that cannot invoke + the version it points at contributes no exposure, so the role on that + version must not count as shared. Both state fields come from + ListAgentAliases -- `aliasInvocationState` is ACCEPT_INVOCATIONS | + REJECT_INVOCATIONS, `agentAliasStatus` is CREATING | PREPARED | FAILED | + UPDATING | DELETING | DISSOCIATED. + """ + + @mock_aws + def test_reject_invocations_alias_does_not_share_its_version_role(self): + """An alias set to REJECT_INVOCATIONS cannot invoke the routed version. + + Both agents' aliases route at a version cut with one shared role, so + judging routing alone reports FAIL -- but neither alias will accept an + invocation, so nothing is running under that role. + """ + service, results = _run( + _mock( + {AGENT_A_ID: SHARED_VERSION_ARN, AGENT_B_ID: SHARED_VERSION_ARN}, + alias_invocation_state="REJECT_INVOCATIONS", + ) + ) + + for agent in service.all_agents.values(): + assert agent.versions_listed is True, "the aliases WERE listed" + assert agent.version_role_arns == {}, "a rejecting alias routes nothing" + + assert len(results) == 2 + assert {report.status for report in results} == {"PASS"} + for report in results: + assert "has a dedicated execution role" in report.status_extended + + @pytest.mark.parametrize("alias_status", ["FAILED", "DELETING", "DISSOCIATED"]) + @mock_aws + def test_terminal_alias_status_does_not_share_its_version_role(self, alias_status): + """A failed, deleting or dissociated alias routes no live invocation.""" + service, results = _run( + _mock( + {AGENT_A_ID: SHARED_VERSION_ARN, AGENT_B_ID: SHARED_VERSION_ARN}, + alias_status=alias_status, + ) + ) + + for agent in service.all_agents.values(): + assert agent.version_role_arns == {} + + assert len(results) == 2 + assert {report.status for report in results} == {"PASS"} + + @pytest.mark.parametrize("alias_status", ["CREATING", "UPDATING"]) + @mock_aws + def test_in_flight_alias_status_makes_version_inventory_incomplete( + self, alias_status + ): + """CREATING and UPDATING do not prove that their version is invocable.""" + service, results = _run( + _mock( + {AGENT_A_ID: SHARED_VERSION_ARN, AGENT_B_ID: SHARED_VERSION_ARN}, + alias_status=alias_status, + ) + ) + + for agent in service.all_agents.values(): + assert agent.versions_listed is False + assert agent.version_role_arns == {} + + assert len(results) == 2 + assert {report.status for report in results} == {"MANUAL"} + for report in results: + assert "deployed versions of" in report.status_extended + + @pytest.mark.parametrize("alias_status", ["CREATING", "UPDATING"]) + @mock_aws + def test_shared_draft_role_fails_despite_in_flight_alias(self, alias_status): + """Definite draft sharing outranks an incomplete alias inventory.""" + service, results = _run( + _mock( + {AGENT_A_ID: DEDICATED_VERSION_ARN, AGENT_B_ID: DEDICATED_VERSION_ARN}, + alias_status=alias_status, + draft_roles={AGENT_A_ID: DRAFT_A_ARN, AGENT_B_ID: DRAFT_A_ARN}, + ) + ) + + assert all( + agent.versions_listed is False for agent in service.all_agents.values() + ) + assert {report.status for report in results} == {"FAIL"} + assert all(DRAFT_A_ARN in report.status_extended for report in results) + + @mock_aws + def test_prepared_accepting_alias_shares_its_version_role(self): + """A PREPARED alias accepting invocations is active.""" + service, results = _run( + _mock( + {AGENT_A_ID: SHARED_VERSION_ARN, AGENT_B_ID: SHARED_VERSION_ARN}, + alias_status="PREPARED", + alias_invocation_state="ACCEPT_INVOCATIONS", + ) + ) + + for agent in service.all_agents.values(): + assert agent.versions_listed is True + assert agent.version_role_arns == {"3": SHARED_VERSION_ARN} + + assert {report.status for report in results} == {"FAIL"} + + @mock_aws + def test_prepared_alias_sharing_fails_despite_in_flight_alias(self): + """Definite prepared-version sharing outranks incomplete inventory.""" + service, results = _run( + _mock( + {AGENT_A_ID: SHARED_VERSION_ARN, AGENT_B_ID: SHARED_VERSION_ARN}, + aliases=[ + { + "agentAliasId": "alias-prepared", + "agentAliasName": "production", + "agentAliasStatus": "PREPARED", + "routingConfiguration": [{"agentVersion": "3"}], + }, + { + "agentAliasId": "alias-updating", + "agentAliasName": "next", + "agentAliasStatus": "UPDATING", + "routingConfiguration": [{"agentVersion": "4"}], + }, + ], + ) + ) + + for agent in service.all_agents.values(): + assert agent.versions_listed is False + assert agent.version_role_arns == {"3": SHARED_VERSION_ARN} + + assert {report.status for report in results} == {"FAIL"} + assert all("through deployed version 3" in r.status_extended for r in results) + + @mock_aws + def test_absent_invocation_state_still_shares_its_version_role(self): + """aliasInvocationState is optional; absent means never set to reject. + + Reading absence as "not accepting" would report every alias that was + never explicitly enabled as dead, and a genuinely shared role as PASS. + """ + service, results = _run( + _mock( + {AGENT_A_ID: SHARED_VERSION_ARN, AGENT_B_ID: SHARED_VERSION_ARN}, + alias_invocation_state=None, + ) + ) + + for agent in service.all_agents.values(): + assert agent.version_role_arns == {"3": SHARED_VERSION_ARN} + + assert {report.status for report in results} == {"FAIL"} + + @mock_aws + def test_rejecting_alias_does_not_mask_a_shared_draft_role(self): + """The predicate must gate ROUTING only, never the draft. + + The draft role comes from GetAgent, not from an alias, so an unreachable + alias has no bearing on it. Without this, a predicate applied one level + too high would silence a genuinely shared draft role. + """ + service, results = _run( + _mock( + {AGENT_A_ID: DEDICATED_VERSION_ARN, AGENT_B_ID: DEDICATED_VERSION_ARN}, + alias_invocation_state="REJECT_INVOCATIONS", + ) + ) + # No routed version survives the predicate... + for agent in service.all_agents.values(): + assert agent.version_role_arns == {} + # ...so make both DRAFTS share one role and re-run the verdict. + for agent in service.all_agents.values(): + agent.role_arn = DRAFT_A_ARN + + check_name = "bedrock_agent_role_not_shared_across_agents" + with mock.patch( + f"prowler.providers.aws.services.bedrock.{check_name}.{check_name}" + ".bedrock_agent_client", + new=service, + ): + module = __import__( + f"prowler.providers.aws.services.bedrock.{check_name}.{check_name}", + fromlist=[check_name], + ) + results = getattr(module, check_name)().execute() + + assert len(results) == 2 + assert {report.status for report in results} == {"FAIL"} + for report in results: + assert DRAFT_A_ARN in report.status_extended + # The sharing is on the draft, so no version is named. + assert "through deployed version" not in report.status_extended diff --git a/tests/providers/aws/services/bedrock/bedrock_custom_model_encrypted_with_cmk/bedrock_custom_model_encrypted_with_cmk_test.py b/tests/providers/aws/services/bedrock/bedrock_custom_model_encrypted_with_cmk/bedrock_custom_model_encrypted_with_cmk_test.py new file mode 100644 index 0000000000..ddace1cd14 --- /dev/null +++ b/tests/providers/aws/services/bedrock/bedrock_custom_model_encrypted_with_cmk/bedrock_custom_model_encrypted_with_cmk_test.py @@ -0,0 +1,287 @@ +from unittest import mock + +import botocore +from botocore.exceptions import ClientError +from moto import mock_aws + +from tests.providers.aws.utils import ( + AWS_ACCOUNT_NUMBER, + AWS_REGION_US_EAST_1, + set_mocked_aws_provider, +) + +make_api_call = botocore.client.BaseClient._make_api_call + +MODEL_NAME = "test-custom-model" +MODEL_ARN = f"arn:aws:bedrock:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:custom-model/example.base-v1/{MODEL_NAME}" +KMS_KEY_ARN = f"arn:aws:kms:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:key/test-key-id" +FOREIGN_ACCOUNT = "999999999999" +FOREIGN_MODEL_ARN = f"arn:aws:bedrock:{AWS_REGION_US_EAST_1}:{FOREIGN_ACCOUNT}:custom-model/example.base-v1/foreign-model" + +# Operations the Bedrock constructor calls that these tests do not exercise. +_UNUSED_OPERATIONS = ( + "GetModelInvocationLoggingConfiguration", + "ListGuardrails", + "GetGuardrail", + "ListTagsForResource", +) + + +def _custom_model_mock(kms_key_arn=None, fail_get=False): + """Build a _make_api_call replacement returning one custom model.""" + + def _mock(self, operation_name, kwarg): + if operation_name in _UNUSED_OPERATIONS: + return {} + if operation_name == "ListCustomModels": + return { + "modelSummaries": [ + {"modelArn": MODEL_ARN, "modelName": MODEL_NAME}, + ] + } + if operation_name == "GetCustomModel": + if fail_get: + raise ClientError( + {"Error": {"Code": "AccessDeniedException", "Message": "denied"}}, + operation_name, + ) + response = {"modelArn": MODEL_ARN, "modelName": MODEL_NAME} + if kms_key_arn is not None: + response["modelKmsKeyArn"] = kms_key_arn + return response + return make_api_call(self, operation_name, kwarg) + + return _mock + + +_mock_with_cmk = _custom_model_mock(KMS_KEY_ARN) +_mock_without_cmk = _custom_model_mock(None) +_mock_empty_cmk = _custom_model_mock("") +_mock_unreadable = _custom_model_mock(fail_get=True) + + +def _mock_empty(self, operation_name, kwarg): + """No custom models at all.""" + if operation_name in _UNUSED_OPERATIONS: + return {} + if operation_name == "ListCustomModels": + return {"modelSummaries": []} + return make_api_call(self, operation_name, kwarg) + + +def _mock_unsupported_region(self, operation_name, kwarg): + """The API is not available in the audited region.""" + if operation_name in _UNUSED_OPERATIONS: + return {} + if operation_name == "ListCustomModels": + raise ClientError( + { + "Error": { + "Code": "ValidationException", + "Message": "Bedrock is not supported in this region.", + } + }, + operation_name, + ) + return make_api_call(self, operation_name, kwarg) + + +def _mock_list_denied(self, operation_name, kwarg): + """ListCustomModels is denied, so the region's models are unknown.""" + if operation_name in _UNUSED_OPERATIONS: + return {} + if operation_name == "ListCustomModels": + raise ClientError( + {"Error": {"Code": "AccessDeniedException", "Message": "denied"}}, + operation_name, + ) + return make_api_call(self, operation_name, kwarg) + + +def _mock_shared_in_model(self, operation_name, kwarg): + """Assert isOwned=True is sent, and return a model only when it is. + + A caller that omits isOwned would also receive models shared into this + account through RAM, whose KMS key this account cannot set. + """ + if operation_name in _UNUSED_OPERATIONS: + return {} + if operation_name == "ListCustomModels": + if kwarg.get("isOwned") is not True: + # What the API would return without the filter: a foreign-owned + # model this account cannot remediate. + return { + "modelSummaries": [ + { + "modelArn": FOREIGN_MODEL_ARN, + "modelName": "foreign-model", + "ownerAccountId": FOREIGN_ACCOUNT, + } + ] + } + return {"modelSummaries": []} + return make_api_call(self, operation_name, kwarg) + + +class Test_bedrock_custom_model_encrypted_with_cmk: + """Unit tests for the bedrock_custom_model_encrypted_with_cmk check.""" + + def _run(self): + """Import the service + check under the active mocks and execute.""" + from prowler.providers.aws.services.bedrock.bedrock_service import Bedrock + + 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, + ), + mock.patch( + "prowler.providers.aws.services.bedrock.bedrock_custom_model_encrypted_with_cmk.bedrock_custom_model_encrypted_with_cmk.bedrock_client", + new=Bedrock(aws_provider), + ), + ): + from prowler.providers.aws.services.bedrock.bedrock_custom_model_encrypted_with_cmk.bedrock_custom_model_encrypted_with_cmk import ( + bedrock_custom_model_encrypted_with_cmk, + ) + + return bedrock_custom_model_encrypted_with_cmk().execute() + + @mock.patch("botocore.client.BaseClient._make_api_call", new=_mock_empty) + @mock_aws + def test_no_resources(self): + """No resources means no findings, not a spurious FAIL.""" + assert self._run() == [] + + @mock.patch( + "botocore.client.BaseClient._make_api_call", new=_mock_unsupported_region + ) + @mock_aws + def test_region_not_supported(self): + """A ValidationException from the region must not raise; it yields no findings.""" + assert self._run() == [] + + @mock.patch("botocore.client.BaseClient._make_api_call", new=_mock_with_cmk) + @mock_aws + def test_cmk_present_passes(self): + """A model with modelKmsKeyArn set is compliant.""" + result = self._run() + assert len(result) == 1 + assert result[0].status == "PASS" + assert result[0].resource_id == MODEL_NAME + assert result[0].resource_arn == MODEL_ARN + assert result[0].region == AWS_REGION_US_EAST_1 + assert ( + result[0].status_extended + == f"Bedrock custom model {MODEL_NAME} is encrypted with a customer-managed KMS key in region {AWS_REGION_US_EAST_1}." + ) + + @mock.patch("botocore.client.BaseClient._make_api_call", new=_mock_without_cmk) + @mock_aws + def test_no_cmk_fails(self): + """An absent modelKmsKeyArn means an AWS-owned key is in use.""" + result = self._run() + assert len(result) == 1 + assert result[0].status == "FAIL" + assert "is not encrypted with a customer-managed KMS key" in ( + result[0].status_extended + ) + + @mock.patch("botocore.client.BaseClient._make_api_call", new=_mock_empty_cmk) + @mock_aws + def test_empty_cmk_fails(self): + """An empty modelKmsKeyArn string is not a key.""" + result = self._run() + assert len(result) == 1 + assert result[0].status == "FAIL" + assert "is not encrypted with a customer-managed KMS key" in ( + result[0].status_extended + ) + + @mock.patch("botocore.client.BaseClient._make_api_call", new=_mock_unreadable) + @mock_aws + def test_detail_unreadable_is_manual_not_pass(self): + """A failed GetCustomModel must not be reported as compliant.""" + result = self._run() + assert len(result) == 1 + assert result[0].status == "MANUAL" + assert "could not be retrieved" in result[0].status_extended + + @mock.patch("botocore.client.BaseClient._make_api_call", new=_mock_list_denied) + @mock_aws + def test_list_denied_is_manual_not_silence(self): + """A denied ListCustomModels must report MANUAL, not vanish. + + Without a region-level report the region is indistinguishable from one + that genuinely holds no custom models. + """ + result = self._run() + assert len(result) == 1 + assert result[0].status == "MANUAL" + assert result[0].region == AWS_REGION_US_EAST_1 + assert result[0].resource_id == "custom-model/unknown" + assert ( + result[0].resource_arn + == f"arn:aws:bedrock:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:custom-model/unknown" + ) + assert "could not be listed" in result[0].status_extended + assert "AccessDeniedException" in result[0].status_extended + assert result[0].status_extended.endswith(".") + + @mock.patch("botocore.client.BaseClient._make_api_call", new=_mock_shared_in_model) + @mock_aws + def test_shared_in_models_are_not_audited(self): + """ListCustomModels must be called with isOwned=True. + + The mock returns a foreign-owned model only when the filter is absent, so + a regression that drops isOwned produces a FAIL naming another account's + model — a finding this account cannot remediate. + """ + result = self._run() + assert result == [] + + def test_every_model_state_resolves_correctly(self): + """Exhaust the decision space instead of sampling it. + + A Bedrock custom model cannot be created without a model customization + job, so this check's behaviour is pinned by enumerating every combination + of retrieval outcome, key value (absent, empty, set) and regional listing + outcome, and asserting the per-model verdict plus the presence of the + region-level report. + + Each case is driven through the real Bedrock service so a renamed service + attribute breaks the test rather than passing silently. + """ + from itertools import product + + for key, fail_get, list_denied in product( + [None, "", KMS_KEY_ARN], [False, True], [False, True] + ): + case = (key, fail_get, list_denied) + stub = ( + _mock_list_denied + if list_denied + else _custom_model_mock(key, fail_get=fail_get) + ) + with mock.patch("botocore.client.BaseClient._make_api_call", new=stub): + with mock_aws(): + result = self._run() + + per_model = [ + r for r in result if "custom-model/unknown" not in r.resource_arn + ] + region_level = [ + r for r in result if "custom-model/unknown" in r.resource_arn + ] + + if list_denied: + # No model reaches the inventory; only the region-level report. + assert per_model == [], case + assert len(region_level) == 1, case + assert region_level[0].status == "MANUAL", case + else: + assert len(per_model) == 1, case + expected = "MANUAL" if fail_get else ("PASS" if key else "FAIL") + assert per_model[0].status == expected, case + assert region_level == [], case + assert all(r.status_extended.endswith(".") for r in result), case diff --git a/tests/providers/aws/services/bedrock/bedrock_guardrail_contextual_grounding_filter_enabled/bedrock_guardrail_contextual_grounding_filter_enabled_test.py b/tests/providers/aws/services/bedrock/bedrock_guardrail_contextual_grounding_filter_enabled/bedrock_guardrail_contextual_grounding_filter_enabled_test.py new file mode 100644 index 0000000000..ae19e43e00 --- /dev/null +++ b/tests/providers/aws/services/bedrock/bedrock_guardrail_contextual_grounding_filter_enabled/bedrock_guardrail_contextual_grounding_filter_enabled_test.py @@ -0,0 +1,466 @@ +from unittest import mock + +import botocore +from botocore.exceptions import ClientError +from moto import mock_aws + +from tests.providers.aws.utils import ( + AWS_ACCOUNT_NUMBER, + AWS_REGION_US_EAST_1, + set_mocked_aws_provider, +) + +make_api_call = botocore.client.BaseClient._make_api_call + +GUARDRAIL_ID = "test-guardrail-id" +GUARDRAIL_NAME = "test-guardrail" +GUARDRAIL_ARN = f"arn:aws:bedrock:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:guardrail/{GUARDRAIL_ID}" + +# Operations the Bedrock constructor calls that these tests do not exercise. +_UNUSED_OPERATIONS = ( + "GetModelInvocationLoggingConfiguration", + "ListTagsForResource", + "ListCustomModels", +) + + +def _guardrail_mock(grounding_policy=None, fail_get=False): + """Build a _make_api_call replacement returning one guardrail.""" + + def _mock(self, operation_name, kwarg): + if operation_name in _UNUSED_OPERATIONS: + return {} + if operation_name == "ListGuardrails": + return { + "guardrails": [ + { + "id": GUARDRAIL_ID, + "name": GUARDRAIL_NAME, + "arn": GUARDRAIL_ARN, + "status": "READY", + } + ] + } + if operation_name == "GetGuardrail": + if fail_get: + raise ClientError( + {"Error": {"Code": "AccessDeniedException", "Message": "denied"}}, + operation_name, + ) + response = { + "guardrailId": GUARDRAIL_ID, + "guardrailArn": GUARDRAIL_ARN, + "name": GUARDRAIL_NAME, + "status": "READY", + } + if grounding_policy is not None: + response["contextualGroundingPolicy"] = grounding_policy + return response + return make_api_call(self, operation_name, kwarg) + + return _mock + + +def _filter(filter_type, threshold=0.75, action="BLOCK", enabled=True): + """Build one contextual grounding filter; a None value omits that key. + + Both action and enabled are optional members of the API shape, so omitting + either has to be expressible here to test the unknown paths. + """ + filter = {"type": filter_type, "threshold": threshold} + if action is not None: + filter["action"] = action + if enabled is not None: + filter["enabled"] = enabled + return filter + + +_mock_action_absent = _guardrail_mock( + { + "filters": [ + _filter("GROUNDING", action=None), + _filter("RELEVANCE", action=None), + ] + } +) +_mock_action_absent_one_filter = _guardrail_mock( + {"filters": [_filter("GROUNDING", action=None), _filter("RELEVANCE")]} +) +_mock_action_absent_with_zero_threshold = _guardrail_mock( + { + "filters": [ + _filter("GROUNDING", action=None, threshold=0.0), + _filter("RELEVANCE"), + ] + } +) + + +_mock_both_blocking = _guardrail_mock( + {"filters": [_filter("GROUNDING"), _filter("RELEVANCE")]} +) +_mock_no_policy = _guardrail_mock(None) +_mock_empty_filters = _guardrail_mock({"filters": []}) +_mock_missing_relevance = _guardrail_mock({"filters": [_filter("GROUNDING")]}) +_mock_missing_grounding = _guardrail_mock({"filters": [_filter("RELEVANCE")]}) +# A policy carrying only an unrecognised filter type: the policy exists, so it is +# not the "no policy" case, yet both required types are absent at once. +_mock_missing_both = _guardrail_mock({"filters": [_filter("UNKNOWN_TYPE")]}) +_mock_action_none = _guardrail_mock( + {"filters": [_filter("GROUNDING", action="NONE"), _filter("RELEVANCE")]} +) +_mock_zero_threshold = _guardrail_mock( + {"filters": [_filter("GROUNDING"), _filter("RELEVANCE", threshold=0.0)]} +) +_mock_disabled = _guardrail_mock( + {"filters": [_filter("GROUNDING", enabled=False), _filter("RELEVANCE")]} +) +_mock_both_disabled = _guardrail_mock( + { + "filters": [ + _filter("GROUNDING", enabled=False), + _filter("RELEVANCE", enabled=False), + ] + } +) +_mock_disabled_and_action_none = _guardrail_mock( + { + "filters": [ + _filter("GROUNDING", enabled=False, action="NONE"), + _filter("RELEVANCE"), + ] + } +) +_mock_enabled_absent = _guardrail_mock( + { + "filters": [ + _filter("GROUNDING", enabled=None), + _filter("RELEVANCE", enabled=None), + ] + } +) +_mock_enabled_absent_one_filter = _guardrail_mock( + {"filters": [_filter("GROUNDING", enabled=None), _filter("RELEVANCE")]} +) +_mock_enabled_absent_with_action_none = _guardrail_mock( + { + "filters": [ + _filter("GROUNDING", enabled=None, action="NONE"), + _filter("RELEVANCE", enabled=None), + ] + } +) +_mock_unreadable = _guardrail_mock(fail_get=True) + + +def _mock_empty(self, operation_name, kwarg): + """No guardrails at all.""" + if operation_name in _UNUSED_OPERATIONS: + return {} + if operation_name == "ListGuardrails": + return {"guardrails": []} + return make_api_call(self, operation_name, kwarg) + + +def _mock_list_guardrails_denied(self, operation_name, kwarg): + """ListGuardrails is denied, so the region's guardrails are unknown.""" + if operation_name in _UNUSED_OPERATIONS: + return {} + if operation_name == "ListGuardrails": + raise ClientError( + {"Error": {"Code": "AccessDeniedException", "Message": "denied"}}, + operation_name, + ) + return make_api_call(self, operation_name, kwarg) + + +def _mock_unsupported_region(self, operation_name, kwarg): + """The API is not available in the audited region.""" + if operation_name in _UNUSED_OPERATIONS: + return {} + if operation_name == "ListGuardrails": + raise ClientError( + { + "Error": { + "Code": "ValidationException", + "Message": "Bedrock is not supported in this region.", + } + }, + operation_name, + ) + return make_api_call(self, operation_name, kwarg) + + +class Test_bedrock_guardrail_contextual_grounding_filter_enabled: + """Unit tests for the bedrock_guardrail_contextual_grounding_filter_enabled check.""" + + def _run(self): + """Import the service + check under the active mocks and execute.""" + from prowler.providers.aws.services.bedrock.bedrock_service import Bedrock + + 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, + ), + mock.patch( + "prowler.providers.aws.services.bedrock.bedrock_guardrail_contextual_grounding_filter_enabled.bedrock_guardrail_contextual_grounding_filter_enabled.bedrock_client", + new=Bedrock(aws_provider), + ), + ): + from prowler.providers.aws.services.bedrock.bedrock_guardrail_contextual_grounding_filter_enabled.bedrock_guardrail_contextual_grounding_filter_enabled import ( + bedrock_guardrail_contextual_grounding_filter_enabled, + ) + + return bedrock_guardrail_contextual_grounding_filter_enabled().execute() + + @mock.patch("botocore.client.BaseClient._make_api_call", new=_mock_empty) + @mock_aws + def test_no_resources(self): + """No resources means no findings, not a spurious FAIL.""" + assert self._run() == [] + + @mock.patch( + "botocore.client.BaseClient._make_api_call", new=_mock_unsupported_region + ) + @mock_aws + def test_region_not_supported(self): + """A ValidationException from the region must not raise; it yields no findings.""" + assert self._run() == [] + + @mock.patch("botocore.client.BaseClient._make_api_call", new=_mock_both_blocking) + @mock_aws + def test_both_filters_blocking_passes(self): + """Both filter types enabled and blocking above a zero threshold is compliant.""" + result = self._run() + assert len(result) == 1 + assert result[0].status == "PASS" + assert result[0].resource_id == GUARDRAIL_ID + assert result[0].resource_arn == GUARDRAIL_ARN + assert result[0].region == AWS_REGION_US_EAST_1 + assert "blocks ungrounded and irrelevant responses" in result[0].status_extended + + @mock.patch("botocore.client.BaseClient._make_api_call", new=_mock_no_policy) + @mock_aws + def test_no_grounding_policy_fails(self): + """No contextual grounding policy at all means nothing is ever detected.""" + result = self._run() + assert len(result) == 1 + assert result[0].status == "FAIL" + assert "no contextual grounding policy" in result[0].status_extended + + @mock.patch("botocore.client.BaseClient._make_api_call", new=_mock_empty_filters) + @mock_aws + def test_empty_filter_list_fails(self): + """A policy present but carrying no filters is equivalent to no policy.""" + result = self._run() + assert len(result) == 1 + assert result[0].status == "FAIL" + assert "no contextual grounding policy" in result[0].status_extended + + @mock.patch( + "botocore.client.BaseClient._make_api_call", new=_mock_missing_relevance + ) + @mock_aws + def test_missing_relevance_filter_fails(self): + """A GROUNDING filter alone leaves irrelevant answers unchecked.""" + result = self._run() + assert len(result) == 1 + assert result[0].status == "FAIL" + assert "missing the RELEVANCE filter" in result[0].status_extended + + @mock.patch( + "botocore.client.BaseClient._make_api_call", new=_mock_missing_grounding + ) + @mock_aws + def test_missing_grounding_filter_fails(self): + """A RELEVANCE filter alone leaves unsupported answers unchecked.""" + result = self._run() + assert len(result) == 1 + assert result[0].status == "FAIL" + assert "missing the GROUNDING filter" in result[0].status_extended + + @mock.patch("botocore.client.BaseClient._make_api_call", new=_mock_missing_both) + @mock_aws + def test_missing_both_filters_fails_with_plural_wording(self): + """Both required types can be absent at once, so the nouns must agree. + + The message lists the missing types, so hard-coding "filter" and "that + class" would read "missing the GROUNDING, RELEVANCE filter ... leaving + that class of ungrounded response unchecked". + """ + result = self._run() + assert len(result) == 1 + assert result[0].status == "FAIL" + assert "missing the GROUNDING, RELEVANCE filters" in result[0].status_extended + assert "leaving those classes of ungrounded response" in ( + result[0].status_extended + ) + assert result[0].status_extended.endswith(".") + + @mock.patch("botocore.client.BaseClient._make_api_call", new=_mock_action_none) + @mock_aws + def test_action_none_fails(self): + """Action NONE scores and reports without blocking, so it must FAIL.""" + result = self._run() + assert len(result) == 1 + assert result[0].status == "FAIL" + assert "GROUNDING filter uses action NONE" in result[0].status_extended + assert "without blocking" in result[0].status_extended + + @mock.patch("botocore.client.BaseClient._make_api_call", new=_mock_zero_threshold) + @mock_aws + def test_zero_threshold_fails(self): + """A threshold of 0 can never be tripped, so the filter blocks nothing.""" + result = self._run() + assert len(result) == 1 + assert result[0].status == "FAIL" + assert "RELEVANCE filter has a threshold of 0" in result[0].status_extended + + @mock.patch("botocore.client.BaseClient._make_api_call", new=_mock_disabled) + @mock_aws + def test_disabled_filter_fails(self): + """enabled: false runs no evaluation, so BLOCK and a real threshold are inert.""" + result = self._run() + assert len(result) == 1 + assert result[0].status == "FAIL" + assert "GROUNDING filter is disabled" in result[0].status_extended + assert "its evaluation never runs" in result[0].status_extended + + @mock.patch("botocore.client.BaseClient._make_api_call", new=_mock_both_disabled) + @mock_aws + def test_both_filters_disabled_fails(self): + """Both filters disabled reports both, not just the first.""" + result = self._run() + assert len(result) == 1 + assert result[0].status == "FAIL" + assert "GROUNDING filter is disabled" in result[0].status_extended + assert "RELEVANCE filter is disabled" in result[0].status_extended + + @mock.patch( + "botocore.client.BaseClient._make_api_call", new=_mock_disabled_and_action_none + ) + @mock_aws + def test_disabled_reported_ahead_of_action(self): + """Disabled is the operative defect: the action is never reached.""" + result = self._run() + assert len(result) == 1 + assert result[0].status == "FAIL" + assert "GROUNDING filter is disabled" in result[0].status_extended + assert "action NONE" not in result[0].status_extended + + @mock.patch("botocore.client.BaseClient._make_api_call", new=_mock_enabled_absent) + @mock_aws + def test_absent_enabled_is_manual_not_pass(self): + """enabled is optional with no documented default, so absent is unknown.""" + result = self._run() + assert len(result) == 1 + assert result[0].status == "MANUAL" + assert ( + "GROUNDING filter omits enabled, RELEVANCE filter omits enabled" + in result[0].status_extended + ) + # Two unknown filters take the plural subject. + assert "so whether they block is unknown" in result[0].status_extended + + @mock.patch( + "botocore.client.BaseClient._make_api_call", new=_mock_enabled_absent_one_filter + ) + @mock_aws + def test_absent_enabled_on_one_filter_is_manual(self): + """One filter omitting enabled is enough to make the answer unknown.""" + result = self._run() + assert len(result) == 1 + assert result[0].status == "MANUAL" + assert "GROUNDING filter omits enabled" in result[0].status_extended + # action was present, so it must not be reported as missing. + assert "omits action" not in result[0].status_extended + assert "RELEVANCE" not in result[0].status_extended + assert "so whether it blocks is unknown" in result[0].status_extended + + @mock.patch("botocore.client.BaseClient._make_api_call", new=_mock_action_absent) + @mock_aws + def test_absent_action_is_manual_not_fail(self): + """action is optional with no documented default, so omitting it is unknown. + + Treating an absent action as NONE would assert a misconfiguration the + response never stated, and would print the literal None as if it were an + AWS enum value. It is reported the same way as an absent enabled. + """ + result = self._run() + assert len(result) == 1 + assert result[0].status == "MANUAL" + assert "None" not in result[0].status_extended + + @mock.patch( + "botocore.client.BaseClient._make_api_call", new=_mock_action_absent_one_filter + ) + @mock_aws + def test_absent_action_on_one_filter_is_manual(self): + """One filter omitting action is enough to make the answer unknown.""" + result = self._run() + assert len(result) == 1 + assert result[0].status == "MANUAL" + assert "GROUNDING filter omits action" in result[0].status_extended + # enabled was present, so it must not be reported as missing. + assert "omits enabled" not in result[0].status_extended + assert "RELEVANCE" not in result[0].status_extended + assert "so whether it blocks is unknown" in result[0].status_extended + + @mock.patch( + "botocore.client.BaseClient._make_api_call", + new=_mock_action_absent_with_zero_threshold, + ) + @mock_aws + def test_absent_action_does_not_mask_a_definite_defect(self): + """An unknown action must not downgrade a real threshold defect to MANUAL.""" + result = self._run() + assert len(result) == 1 + assert result[0].status == "FAIL" + assert "threshold" in result[0].status_extended + + @mock.patch( + "botocore.client.BaseClient._make_api_call", + new=_mock_enabled_absent_with_action_none, + ) + @mock_aws + def test_absent_enabled_does_not_mask_a_definite_defect(self): + """An unknown enabled must not downgrade a real action defect to MANUAL.""" + result = self._run() + assert len(result) == 1 + assert result[0].status == "FAIL" + assert "GROUNDING filter uses action NONE" in result[0].status_extended + + @mock.patch("botocore.client.BaseClient._make_api_call", new=_mock_unreadable) + @mock_aws + def test_detail_unreadable_is_manual_not_pass(self): + """A failed GetGuardrail must not be reported as compliant.""" + result = self._run() + assert len(result) == 1 + assert result[0].status == "MANUAL" + assert "could not be retrieved" in result[0].status_extended + + @mock.patch( + "botocore.client.BaseClient._make_api_call", new=_mock_list_guardrails_denied + ) + @mock_aws + def test_list_guardrails_denied_is_manual_not_silence(self): + """A denied ListGuardrails must report MANUAL for the region, not vanish. + + Without this the Region is indistinguishable from one holding no + guardrails, which is the same silent-inventory gap the sibling checks + report against custom-model/unknown and knowledge-base/unknown. + """ + result = self._run() + assert len(result) == 1 + assert result[0].status == "MANUAL" + assert result[0].region == AWS_REGION_US_EAST_1 + assert result[0].resource_id == "guardrail/unknown" + assert ( + result[0].resource_arn + == f"arn:aws:bedrock:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:guardrail/unknown" + ) + assert "could not be listed" in result[0].status_extended + assert "AccessDeniedException" in result[0].status_extended + assert result[0].status_extended.endswith(".") diff --git a/tests/providers/aws/services/bedrock/bedrock_knowledge_base_encrypted_with_cmk/bedrock_knowledge_base_encrypted_with_cmk_test.py b/tests/providers/aws/services/bedrock/bedrock_knowledge_base_encrypted_with_cmk/bedrock_knowledge_base_encrypted_with_cmk_test.py new file mode 100644 index 0000000000..f15916959b --- /dev/null +++ b/tests/providers/aws/services/bedrock/bedrock_knowledge_base_encrypted_with_cmk/bedrock_knowledge_base_encrypted_with_cmk_test.py @@ -0,0 +1,338 @@ +from unittest import mock + +import botocore +from botocore.exceptions import ClientError +from moto import mock_aws + +from tests.providers.aws.utils import ( + AWS_ACCOUNT_NUMBER, + AWS_REGION_US_EAST_1, + set_mocked_aws_provider, +) + +make_api_call = botocore.client.BaseClient._make_api_call + +KB_ID = "test-kb-id" +KB_NAME = "test-knowledge-base" +KB_ARN = f"arn:aws:bedrock:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:knowledge-base/{KB_ID}" +DS_ID = "test-ds-id" +DS_NAME = "test-data-source" +DS_ARN = f"{KB_ARN}/data-source/{DS_ID}" +KMS_KEY_ARN = f"arn:aws:kms:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:key/test-key-id" + +# Operations the BedrockAgent constructor calls that these tests do not exercise. +_UNUSED_OPERATIONS = ( + "ListAgents", + "GetAgent", + "ListPrompts", + "GetPrompt", + "ListTagsForResource", +) + + +def _knowledge_base_mock(kms_key_arn=None, fail_get=False): + """Build a _make_api_call replacement returning one KB with one data source.""" + + def _mock(self, operation_name, kwarg): + if operation_name in _UNUSED_OPERATIONS: + return {} + if operation_name == "ListKnowledgeBases": + return { + "knowledgeBaseSummaries": [ + { + "knowledgeBaseId": KB_ID, + "name": KB_NAME, + "status": "ACTIVE", + } + ] + } + if operation_name == "ListDataSources": + return { + "dataSourceSummaries": [ + { + "knowledgeBaseId": KB_ID, + "dataSourceId": DS_ID, + "name": DS_NAME, + "status": "AVAILABLE", + } + ] + } + if operation_name == "GetDataSource": + if fail_get: + raise ClientError( + {"Error": {"Code": "AccessDeniedException", "Message": "denied"}}, + operation_name, + ) + # GetDataSource nests its payload under a top-level dataSource key. + data_source = { + "knowledgeBaseId": KB_ID, + "dataSourceId": DS_ID, + "name": DS_NAME, + "status": "AVAILABLE", + } + if kms_key_arn is not None: + data_source["serverSideEncryptionConfiguration"] = { + "kmsKeyArn": kms_key_arn + } + return {"dataSource": data_source} + return make_api_call(self, operation_name, kwarg) + + return _mock + + +_mock_with_cmk = _knowledge_base_mock(KMS_KEY_ARN) +_mock_without_cmk = _knowledge_base_mock(None) +_mock_empty_cmk = _knowledge_base_mock("") +_mock_unreadable = _knowledge_base_mock(fail_get=True) + + +def _mock_empty(self, operation_name, kwarg): + """No knowledge bases at all.""" + if operation_name in _UNUSED_OPERATIONS: + return {} + if operation_name == "ListKnowledgeBases": + return {"knowledgeBaseSummaries": []} + return make_api_call(self, operation_name, kwarg) + + +def _mock_no_data_sources(self, operation_name, kwarg): + """A knowledge base with no data sources produces no findings.""" + if operation_name in _UNUSED_OPERATIONS: + return {} + if operation_name == "ListKnowledgeBases": + return { + "knowledgeBaseSummaries": [ + {"knowledgeBaseId": KB_ID, "name": KB_NAME, "status": "ACTIVE"} + ] + } + if operation_name == "ListDataSources": + return {"dataSourceSummaries": []} + return make_api_call(self, operation_name, kwarg) + + +def _mock_unsupported_region(self, operation_name, kwarg): + """The API is not available in the audited region.""" + if operation_name in _UNUSED_OPERATIONS: + return {} + if operation_name == "ListKnowledgeBases": + raise ClientError( + { + "Error": { + "Code": "ValidationException", + "Message": "Bedrock Agent is not supported in this region.", + } + }, + operation_name, + ) + return make_api_call(self, operation_name, kwarg) + + +def _mock_list_kb_denied(self, operation_name, kwarg): + """ListKnowledgeBases is denied, so the region's knowledge bases are unknown.""" + if operation_name in _UNUSED_OPERATIONS: + return {} + if operation_name == "ListKnowledgeBases": + raise ClientError( + {"Error": {"Code": "AccessDeniedException", "Message": "denied"}}, + operation_name, + ) + return make_api_call(self, operation_name, kwarg) + + +def _mock_list_ds_denied(self, operation_name, kwarg): + """The knowledge base is visible but its data sources cannot be listed.""" + if operation_name in _UNUSED_OPERATIONS: + return {} + if operation_name == "ListKnowledgeBases": + return { + "knowledgeBaseSummaries": [ + {"knowledgeBaseId": KB_ID, "name": KB_NAME, "status": "ACTIVE"} + ] + } + if operation_name == "ListDataSources": + raise ClientError( + {"Error": {"Code": "AccessDeniedException", "Message": "denied"}}, + operation_name, + ) + return make_api_call(self, operation_name, kwarg) + + +class Test_bedrock_knowledge_base_encrypted_with_cmk: + """Unit tests for the bedrock_knowledge_base_encrypted_with_cmk check.""" + + def _run(self): + """Import the service + check under the active mocks and execute.""" + from prowler.providers.aws.services.bedrock.bedrock_service import BedrockAgent + + 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, + ), + mock.patch( + "prowler.providers.aws.services.bedrock.bedrock_knowledge_base_encrypted_with_cmk.bedrock_knowledge_base_encrypted_with_cmk.bedrock_agent_client", + new=BedrockAgent(aws_provider), + ), + ): + from prowler.providers.aws.services.bedrock.bedrock_knowledge_base_encrypted_with_cmk.bedrock_knowledge_base_encrypted_with_cmk import ( + bedrock_knowledge_base_encrypted_with_cmk, + ) + + return bedrock_knowledge_base_encrypted_with_cmk().execute() + + @mock.patch("botocore.client.BaseClient._make_api_call", new=_mock_empty) + @mock_aws + def test_no_resources(self): + """No resources means no findings, not a spurious FAIL.""" + assert self._run() == [] + + @mock.patch("botocore.client.BaseClient._make_api_call", new=_mock_no_data_sources) + @mock_aws + def test_knowledge_base_without_data_sources(self): + """Findings are per data source, so a KB with none produces nothing.""" + assert self._run() == [] + + @mock.patch( + "botocore.client.BaseClient._make_api_call", new=_mock_unsupported_region + ) + @mock_aws + def test_region_not_supported(self): + """A ValidationException from the region must not raise; it yields no findings.""" + assert self._run() == [] + + @mock.patch("botocore.client.BaseClient._make_api_call", new=_mock_with_cmk) + @mock_aws + def test_cmk_present_passes(self): + """A data source with kmsKeyArn set is compliant.""" + result = self._run() + assert len(result) == 1 + assert result[0].status == "PASS" + assert result[0].resource_id == DS_ID + assert result[0].resource_arn == DS_ARN + assert result[0].region == AWS_REGION_US_EAST_1 + assert ( + result[0].status_extended + == f"Bedrock knowledge base {KB_NAME} data source {DS_NAME} is encrypted with a customer-managed KMS key in region {AWS_REGION_US_EAST_1}." + ) + + @mock.patch("botocore.client.BaseClient._make_api_call", new=_mock_without_cmk) + @mock_aws + def test_no_cmk_fails(self): + """An absent kmsKeyArn means an AWS-owned key is in use.""" + result = self._run() + assert len(result) == 1 + assert result[0].status == "FAIL" + assert KB_NAME in result[0].status_extended + assert "is not encrypted with a customer-managed KMS key" in ( + result[0].status_extended + ) + + @mock.patch("botocore.client.BaseClient._make_api_call", new=_mock_empty_cmk) + @mock_aws + def test_empty_cmk_fails(self): + """An empty kmsKeyArn string is not a key.""" + result = self._run() + assert len(result) == 1 + assert result[0].status == "FAIL" + assert "is not encrypted with a customer-managed KMS key" in ( + result[0].status_extended + ) + + @mock.patch("botocore.client.BaseClient._make_api_call", new=_mock_unreadable) + @mock_aws + def test_detail_unreadable_is_manual_not_pass(self): + """A failed GetDataSource must not be reported as compliant.""" + result = self._run() + assert len(result) == 1 + assert result[0].status == "MANUAL" + assert "could not be retrieved" in result[0].status_extended + + @mock.patch("botocore.client.BaseClient._make_api_call", new=_mock_list_kb_denied) + @mock_aws + def test_list_knowledge_bases_denied_is_manual_not_silence(self): + """A denied ListKnowledgeBases must report MANUAL for the region.""" + result = self._run() + assert len(result) == 1 + assert result[0].status == "MANUAL" + assert result[0].region == AWS_REGION_US_EAST_1 + assert result[0].resource_id == "knowledge-base/unknown" + assert ( + result[0].resource_arn + == f"arn:aws:bedrock:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:knowledge-base/unknown" + ) + assert "could not be listed" in result[0].status_extended + assert "AccessDeniedException" in result[0].status_extended + assert result[0].status_extended.endswith(".") + + @mock.patch("botocore.client.BaseClient._make_api_call", new=_mock_list_ds_denied) + @mock_aws + def test_list_data_sources_denied_is_manual_not_silence(self): + """A knowledge base whose data sources cannot be listed must still report. + + Reporting nothing would drop the knowledge base from the output, which is + indistinguishable from one that genuinely has no data sources. + """ + result = self._run() + assert len(result) == 1 + assert result[0].status == "MANUAL" + assert result[0].resource_arn == KB_ARN + assert KB_NAME in result[0].status_extended + assert "data sources could not be listed" in result[0].status_extended + # The message names why, like the region-level and detail-level ones do. + assert "AccessDeniedException" in result[0].status_extended + assert result[0].status_extended.endswith(".") + + @mock.patch("botocore.client.BaseClient._make_api_call", new=_mock_no_data_sources) + @mock_aws + def test_listed_but_empty_marks_the_knowledge_base_as_listed(self): + """A successful but empty ListDataSources is "none", not "unknown". + + Distinct from test_knowledge_base_without_data_sources, which only asserts + the empty result: this asserts the service state that produces it, so the + over-correction of reporting MANUAL for a genuinely empty knowledge base + cannot regress silently. + """ + from prowler.providers.aws.services.bedrock.bedrock_service import BedrockAgent + + 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, + ): + service = BedrockAgent(aws_provider) + + assert service.knowledge_bases, "the knowledge base must be discovered" + assert all( + knowledge_base.data_sources_listed + for knowledge_base in service.knowledge_bases.values() + ) + assert service.data_sources == {} + assert service.knowledge_bases_scan_errors == {} + assert self._run() == [] + + @mock.patch("botocore.client.BaseClient._make_api_call", new=_mock_without_cmk) + @mock_aws + def test_scoping_by_knowledge_base_arn_keeps_its_data_sources(self): + """A scan scoped to the knowledge base ARN must still see its data sources. + + AWS exposes no ARN for a Bedrock data source, so the one built here is + synthetic and can never equal a user-supplied --resource-arn. Filtering on + it would keep the knowledge base, silently drop every data source, and + leave the check reporting nothing for an in-scope knowledge base. + """ + from prowler.providers.aws.services.bedrock.bedrock_service import BedrockAgent + + aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1]) + aws_provider._audit_resources = [KB_ARN] + with mock.patch( + "prowler.providers.common.provider.Provider.get_global_provider", + return_value=aws_provider, + ): + service = BedrockAgent(aws_provider) + + assert service.knowledge_bases, "the scoped knowledge base must be kept" + assert service.data_sources, "its data sources must not be filtered out" + assert all( + ds.knowledge_base_id == KB_ID for ds in service.data_sources.values() + ) diff --git a/tests/providers/aws/services/bedrock/bedrock_service_test.py b/tests/providers/aws/services/bedrock/bedrock_service_test.py index f8a1fc8996..276dc44a4f 100644 --- a/tests/providers/aws/services/bedrock/bedrock_service_test.py +++ b/tests/providers/aws/services/bedrock/bedrock_service_test.py @@ -2,6 +2,7 @@ from unittest import mock from unittest.mock import MagicMock import botocore +import pytest from boto3 import client from moto import mock_aws @@ -283,11 +284,20 @@ class TestBedrockPagination: class TestBedrockAgentPagination: """Test suite for Bedrock Agent pagination logic.""" - def test_list_agents_pagination(self): - """Test that list_agents iterates through all pages.""" - # Mock the audit_info + @pytest.mark.parametrize("partition", ["aws", "aws-us-gov", "aws-cn"]) + def test_list_agents_pagination(self, partition): + """Test that list_agents iterates through all pages, in every partition. + + The ARN is built from the audited partition, so GovCloud and China must + produce aws-us-gov/aws-cn ARNs. A hardcoded `arn:aws:` here yielded an ARN + that does not exist in those partitions, and exact --resource-arn matching + against it could never succeed. + """ + # Mock the audit_info. AWSService reads the partition off provider.identity, + # so setting audited_partition alone leaves a MagicMock in the ARN. audit_info = MagicMock() - audit_info.audited_partition = "aws" + audit_info.identity.partition = partition + audit_info.audited_partition = partition audit_info.audited_account = "123456789012" audit_info.audit_resources = None @@ -322,6 +332,7 @@ class TestBedrockAgentPagination: bedrock_agent_service = BedrockAgent(audit_info) bedrock_agent_service.regional_clients = {"us-east-1": regional_client} bedrock_agent_service.agents = {} # Clear init side effects + bedrock_agent_service.all_agents = {} bedrock_agent_service.audited_account = "123456789012" # Run method @@ -329,14 +340,16 @@ class TestBedrockAgentPagination: # Assertions assert len(bedrock_agent_service.agents) == 2 - assert ( - "arn:aws:bedrock:us-east-1:123456789012:agent/agent-1" - in bedrock_agent_service.agents - ) - assert ( - "arn:aws:bedrock:us-east-1:123456789012:agent/agent-2" - in bedrock_agent_service.agents - ) + for agent_id in ("agent-1", "agent-2"): + expected_arn = ( + f"arn:{partition}:bedrock:us-east-1:123456789012:agent/{agent_id}" + ) + assert expected_arn in bedrock_agent_service.agents + # With no --resource-arn, the complete inventory and the reported set + # hold the very same objects. + assert bedrock_agent_service.all_agents[expected_arn] is ( + bedrock_agent_service.agents[expected_arn] + ) # Verify paginator was used regional_client.get_paginator.assert_called_once_with("list_agents")