mirror of
https://github.com/prowler-cloud/prowler.git
synced 2026-10-04 02:04:06 +00:00
feat(bedrock): add model artifact and guardrail grounding security checks for the AWS provider (#12459)
Co-authored-by: Hugo P.Brito <hugopbrit@gmail.com>
This commit is contained in:
Jonathan Nguyen
co-authored by
Hugo P.Brito
Hugo P.Brito
parent
823efc5ab1
commit
f39c92b8f8
+633
@@ -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"}
|
||||
+556
@@ -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
|
||||
+287
@@ -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
|
||||
+466
@@ -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(".")
|
||||
+338
@@ -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()
|
||||
)
|
||||
@@ -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")
|
||||
|
||||
Reference in New Issue
Block a user