feat(guardduty): assess unified Runtime Monitoring and AI Protection (#12564)

This commit is contained in:
Jonathan Nguyen
2026-09-01 13:18:57 +02:00
committed by GitHub
parent fb7064401b
commit b6a8af3c54
15 changed files with 884 additions and 7 deletions
@@ -0,0 +1,215 @@
from unittest import mock
import botocore
from boto3 import client
from moto import mock_aws
from prowler.providers.aws.services.guardduty.guardduty_service import GuardDuty
from tests.providers.aws.utils import (
AWS_ACCOUNT_NUMBER,
AWS_REGION_US_EAST_1,
set_mocked_aws_provider,
)
CHECK_CLIENT_PATH = "prowler.providers.aws.services.guardduty.guardduty_ai_protection_enabled.guardduty_ai_protection_enabled.guardduty_client"
make_api_call = botocore.client.BaseClient._make_api_call
def mock_get_detector_raises(self, operation_name, kwarg):
"""Deny GetDetector only, leaving every other GuardDuty operation intact."""
if operation_name == "GetDetector":
raise botocore.exceptions.ClientError(
{"Error": {"Code": "AccessDeniedException", "Message": "denied"}},
operation_name,
)
return make_api_call(self, operation_name, kwarg)
def _run_check(aws_provider):
"""Run the check against a GuardDuty service built from the mocked provider."""
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=aws_provider,
),
mock.patch(CHECK_CLIENT_PATH, new=GuardDuty(aws_provider)),
):
from prowler.providers.aws.services.guardduty.guardduty_ai_protection_enabled.guardduty_ai_protection_enabled import (
guardduty_ai_protection_enabled,
)
return guardduty_ai_protection_enabled().execute()
class Test_guardduty_ai_protection_enabled:
@mock_aws
def test_no_detectors(self):
"""A Region with no detector has no resource to judge; guardduty_is_enabled owns it."""
client("guardduty", region_name=AWS_REGION_US_EAST_1)
result = _run_check(set_mocked_aws_provider([AWS_REGION_US_EAST_1]))
assert len(result) == 0
@mock_aws
def test_ai_protection_enabled(self):
"""An enabled feature PASSes and carries the detector's own resource fields."""
guardduty_client = client("guardduty", region_name=AWS_REGION_US_EAST_1)
response = guardduty_client.create_detector(
Enable=True,
Features=[{"Name": "AI_PROTECTION", "Status": "ENABLED"}],
)
result = _run_check(set_mocked_aws_provider([AWS_REGION_US_EAST_1]))
assert len(result) == 1
assert result[0].status == "PASS"
assert (
result[0].status_extended
== f"GuardDuty detector {response['DetectorId']} has AI Protection enabled."
)
assert result[0].resource_id == response["DetectorId"]
assert result[0].region == AWS_REGION_US_EAST_1
assert (
result[0].resource_arn
== f"arn:aws:guardduty:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:detector/{response['DetectorId']}"
)
assert result[0].resource_tags == []
@mock_aws
def test_ai_protection_disabled(self):
"""A reported-disabled feature FAILs and names the detector."""
guardduty_client = client("guardduty", region_name=AWS_REGION_US_EAST_1)
response = guardduty_client.create_detector(
Enable=True,
Features=[{"Name": "AI_PROTECTION", "Status": "DISABLED"}],
)
result = _run_check(set_mocked_aws_provider([AWS_REGION_US_EAST_1]))
assert len(result) == 1
assert result[0].status == "FAIL"
assert (
result[0].status_extended
== f"GuardDuty detector {response['DetectorId']} does not have AI Protection enabled."
)
assert result[0].resource_id == response["DetectorId"]
assert result[0].region == AWS_REGION_US_EAST_1
@mock_aws
def test_ai_protection_feature_absent(self):
"""A feature GuardDuty does not report is not a feature GuardDuty turned off.
The Region or the GuardDuty version may not offer AI Protection at all.
"""
guardduty_client = client("guardduty", region_name=AWS_REGION_US_EAST_1)
response = guardduty_client.create_detector(
Enable=True,
Features=[
{"Name": "S3_DATA_EVENTS", "Status": "ENABLED"},
{"Name": "LAMBDA_NETWORK_LOGS", "Status": "ENABLED"},
],
)
result = _run_check(set_mocked_aws_provider([AWS_REGION_US_EAST_1]))
assert len(result) == 1
assert result[0].status == "MANUAL"
assert (
result[0].status_extended
== f"GuardDuty detector {response['DetectorId']} does not report the AI Protection feature, so verify manually whether AI Protection is available in region {AWS_REGION_US_EAST_1}."
)
assert result[0].resource_id == response["DetectorId"]
assert result[0].region == AWS_REGION_US_EAST_1
@mock_aws
def test_no_features_reported(self):
"""An empty features array reads the same as an absent AI_PROTECTION entry."""
guardduty_client = client("guardduty", region_name=AWS_REGION_US_EAST_1)
response = guardduty_client.create_detector(Enable=True)
result = _run_check(set_mocked_aws_provider([AWS_REGION_US_EAST_1]))
assert len(result) == 1
assert result[0].status == "MANUAL"
assert (
result[0].status_extended
== f"GuardDuty detector {response['DetectorId']} does not report the AI Protection feature, so verify manually whether AI Protection is available in region {AWS_REGION_US_EAST_1}."
)
@mock_aws
def test_detector_not_enabled(self):
"""A suspended detector is MANUAL: its feature state is unknown, not absent."""
guardduty_client = client("guardduty", region_name=AWS_REGION_US_EAST_1)
response = guardduty_client.create_detector(
Enable=False,
Features=[{"Name": "AI_PROTECTION", "Status": "ENABLED"}],
)
result = _run_check(set_mocked_aws_provider([AWS_REGION_US_EAST_1]))
assert len(result) == 1
assert result[0].status == "MANUAL"
assert (
result[0].status_extended
== f"GuardDuty detector {response['DetectorId']} is not enabled or could not be read, so AI Protection coverage could not be determined."
)
@mock_aws
def test_get_detector_unreadable(self):
"""A denied GetDetector is unknown, not absent: MANUAL instead of FAIL."""
guardduty_client = client("guardduty", region_name=AWS_REGION_US_EAST_1)
response = guardduty_client.create_detector(
Enable=True,
Features=[{"Name": "AI_PROTECTION", "Status": "DISABLED"}],
)
aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
with mock.patch(
"botocore.client.BaseClient._make_api_call", new=mock_get_detector_raises
):
result = _run_check(aws_provider)
assert len(result) == 1
assert result[0].status == "MANUAL"
assert (
result[0].status_extended
== f"GuardDuty detector {response['DetectorId']} is not enabled or could not be read, so AI Protection coverage could not be determined."
)
@mock_aws
def test_real_get_detector_payload_ai_protection_disabled(self):
"""The real GetDetector payload carries AI_PROTECTION alongside AI_ANALYST.
AI_PROTECTION is absent from the pinned DetectorFeatureResult enum, so this
asserts the name still reaches the check alongside the two runtime feature names.
"""
guardduty_client = client("guardduty", region_name=AWS_REGION_US_EAST_1)
response = guardduty_client.create_detector(
Enable=True,
Features=[
{"Name": "CLOUD_TRAIL", "Status": "ENABLED"},
{"Name": "DNS_LOGS", "Status": "ENABLED"},
{"Name": "FLOW_LOGS", "Status": "ENABLED"},
{"Name": "S3_DATA_EVENTS", "Status": "ENABLED"},
{"Name": "EKS_AUDIT_LOGS", "Status": "ENABLED"},
{"Name": "EBS_MALWARE_PROTECTION", "Status": "ENABLED"},
{"Name": "RDS_LOGIN_EVENTS", "Status": "ENABLED"},
{"Name": "AI_PROTECTION", "Status": "DISABLED"},
{"Name": "AI_ANALYST", "Status": "ENABLED"},
{"Name": "EKS_RUNTIME_MONITORING", "Status": "DISABLED"},
{"Name": "LAMBDA_NETWORK_LOGS", "Status": "ENABLED"},
{"Name": "RUNTIME_MONITORING", "Status": "DISABLED"},
],
)
result = _run_check(set_mocked_aws_provider([AWS_REGION_US_EAST_1]))
assert len(result) == 1
assert result[0].status == "FAIL"
assert (
result[0].status_extended
== f"GuardDuty detector {response['DetectorId']} does not have AI Protection enabled."
)
@@ -0,0 +1,270 @@
from unittest import mock
import botocore
from boto3 import client
from moto import mock_aws
from prowler.providers.aws.services.guardduty.guardduty_service import GuardDuty
from tests.providers.aws.utils import (
AWS_ACCOUNT_NUMBER,
AWS_REGION_US_EAST_1,
set_mocked_aws_provider,
)
CHECK_CLIENT_PATH = "prowler.providers.aws.services.guardduty.guardduty_runtime_monitoring_enabled.guardduty_runtime_monitoring_enabled.guardduty_client"
make_api_call = botocore.client.BaseClient._make_api_call
def mock_get_detector_raises(self, operation_name, kwarg):
"""Deny GetDetector only, leaving every other GuardDuty operation intact."""
if operation_name == "GetDetector":
raise botocore.exceptions.ClientError(
{"Error": {"Code": "AccessDeniedException", "Message": "denied"}},
operation_name,
)
return make_api_call(self, operation_name, kwarg)
def _run_check(aws_provider):
"""Run the check against a GuardDuty service built from the mocked provider."""
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=aws_provider,
),
mock.patch(CHECK_CLIENT_PATH, new=GuardDuty(aws_provider)),
):
from prowler.providers.aws.services.guardduty.guardduty_runtime_monitoring_enabled.guardduty_runtime_monitoring_enabled import (
guardduty_runtime_monitoring_enabled,
)
return guardduty_runtime_monitoring_enabled().execute()
class Test_guardduty_runtime_monitoring_enabled:
@mock_aws
def test_no_detectors(self):
"""A Region with no detector has no resource to judge; guardduty_is_enabled owns it."""
client("guardduty", region_name=AWS_REGION_US_EAST_1)
result = _run_check(set_mocked_aws_provider([AWS_REGION_US_EAST_1]))
assert len(result) == 0
@mock_aws
def test_detector_disabled(self):
"""A suspended detector is MANUAL, never dropped.
The detector exists, so omitting it would leave the Region unreported, which
reads as compliant -- and it is reported here with Runtime Monitoring ENABLED,
the case where the omission was hardest to notice.
"""
guardduty_client = client("guardduty", region_name=AWS_REGION_US_EAST_1)
response = guardduty_client.create_detector(
Enable=False,
Features=[{"Name": "RUNTIME_MONITORING", "Status": "ENABLED"}],
)
result = _run_check(set_mocked_aws_provider([AWS_REGION_US_EAST_1]))
assert len(result) == 1
assert result[0].status == "MANUAL"
assert (
result[0].status_extended
== f"GuardDuty detector {response['DetectorId']} is not enabled or could not be read, so Runtime Monitoring coverage could not be determined."
)
assert result[0].resource_id == response["DetectorId"]
assert result[0].region == AWS_REGION_US_EAST_1
@mock_aws
def test_get_detector_unreadable(self):
"""A denied GetDetector is unknown, not absent: MANUAL instead of FAIL.
Detector.status cannot distinguish this from a suspended detector, which is why
both carry the same MANUAL wording rather than a definite verdict.
"""
guardduty_client = client("guardduty", region_name=AWS_REGION_US_EAST_1)
response = guardduty_client.create_detector(
Enable=True,
Features=[{"Name": "RUNTIME_MONITORING", "Status": "DISABLED"}],
)
aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
with mock.patch(
"botocore.client.BaseClient._make_api_call", new=mock_get_detector_raises
):
result = _run_check(aws_provider)
assert len(result) == 1
assert result[0].status == "MANUAL"
assert (
result[0].status_extended
== f"GuardDuty detector {response['DetectorId']} is not enabled or could not be read, so Runtime Monitoring coverage could not be determined."
)
@mock_aws
def test_runtime_monitoring_enabled(self):
"""An enabled unified feature PASSes and carries the detector's own resource fields."""
guardduty_client = client("guardduty", region_name=AWS_REGION_US_EAST_1)
response = guardduty_client.create_detector(
Enable=True,
Features=[{"Name": "RUNTIME_MONITORING", "Status": "ENABLED"}],
)
result = _run_check(set_mocked_aws_provider([AWS_REGION_US_EAST_1]))
assert len(result) == 1
assert result[0].status == "PASS"
assert (
result[0].status_extended
== f"GuardDuty detector {response['DetectorId']} has Runtime Monitoring enabled."
)
assert result[0].resource_id == response["DetectorId"]
assert result[0].region == AWS_REGION_US_EAST_1
assert (
result[0].resource_arn
== f"arn:aws:guardduty:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:detector/{response['DetectorId']}"
)
assert result[0].resource_tags == []
@mock_aws
def test_runtime_monitoring_disabled(self):
"""A reported-disabled unified feature FAILs and names the detector."""
guardduty_client = client("guardduty", region_name=AWS_REGION_US_EAST_1)
response = guardduty_client.create_detector(
Enable=True,
Features=[{"Name": "RUNTIME_MONITORING", "Status": "DISABLED"}],
)
result = _run_check(set_mocked_aws_provider([AWS_REGION_US_EAST_1]))
assert len(result) == 1
assert result[0].status == "FAIL"
assert (
result[0].status_extended
== f"GuardDuty detector {response['DetectorId']} does not have Runtime Monitoring enabled."
)
assert result[0].resource_id == response["DetectorId"]
assert result[0].region == AWS_REGION_US_EAST_1
@mock_aws
def test_unreported_runtime_feature_is_manual_not_fail(self):
"""A feature the detector never reports is not the same as one it reports DISABLED.
GuardDuty returns a disabled feature with Status DISABLED rather than omitting
it -- which is exactly why AI_PROTECTION is recorded even when off -- so an
omitted RUNTIME_MONITORING means the Region does not offer the unified feature,
or the features array could not be read. Neither is a definite absence of
runtime coverage.
"""
guardduty_client = client("guardduty", region_name=AWS_REGION_US_EAST_1)
response = guardduty_client.create_detector(
Enable=True,
Features=[{"Name": "S3_DATA_EVENTS", "Status": "ENABLED"}],
)
result = _run_check(set_mocked_aws_provider([AWS_REGION_US_EAST_1]))
assert len(result) == 1
assert result[0].status == "MANUAL"
assert (
result[0].status_extended
== f"GuardDuty detector {response['DetectorId']} did not report the Runtime Monitoring feature, so runtime coverage could not be determined; verify manually."
)
@mock_aws
def test_runtime_monitoring_reported_disabled_still_fails(self):
"""The complement of the unreported case: reported and DISABLED stays a FAIL.
Making the unreported case MANUAL must not soften a definite absence of coverage.
"""
guardduty_client = client("guardduty", region_name=AWS_REGION_US_EAST_1)
response = guardduty_client.create_detector(
Enable=True,
Features=[{"Name": "RUNTIME_MONITORING", "Status": "DISABLED"}],
)
result = _run_check(set_mocked_aws_provider([AWS_REGION_US_EAST_1]))
assert len(result) == 1
assert result[0].status == "FAIL"
assert (
result[0].status_extended
== f"GuardDuty detector {response['DetectorId']} does not have Runtime Monitoring enabled."
)
@mock_aws
def test_legacy_eks_runtime_monitoring_only_fails(self):
"""The legacy EKS-only feature covers Amazon EKS and nothing else.
It must not PASS a check about Amazon EC2 and Amazon ECS on Fargate coverage.
"""
guardduty_client = client("guardduty", region_name=AWS_REGION_US_EAST_1)
response = guardduty_client.create_detector(
Enable=True,
Features=[
{"Name": "EKS_RUNTIME_MONITORING", "Status": "ENABLED"},
{"Name": "RUNTIME_MONITORING", "Status": "DISABLED"},
],
)
result = _run_check(set_mocked_aws_provider([AWS_REGION_US_EAST_1]))
assert len(result) == 1
assert result[0].status == "FAIL"
assert (
result[0].status_extended
== f"GuardDuty detector {response['DetectorId']} only has the legacy EKS Runtime Monitoring enabled, leaving Amazon EC2 instances and Amazon ECS on Fargate tasks without runtime coverage."
)
@mock_aws
def test_legacy_eks_only_without_any_unified_entry_fails(self):
"""The real legacy shape: the unified feature is absent, not reported DISABLED.
The two features are mutually exclusive at the API, so a detector on the legacy
one has no RUNTIME_MONITORING entry at all -- which is the same absence that makes
an unknown detector MANUAL. The test above supplies a DISABLED unified entry as
well, so it never reaches that ambiguity. Here the legacy verdict has to win on
ordering alone: EKS coverage is stated, so EC2 and Fargate are definitively
uncovered rather than undetermined.
"""
guardduty_client = client("guardduty", region_name=AWS_REGION_US_EAST_1)
response = guardduty_client.create_detector(
Enable=True,
Features=[{"Name": "EKS_RUNTIME_MONITORING", "Status": "ENABLED"}],
)
result = _run_check(set_mocked_aws_provider([AWS_REGION_US_EAST_1]))
assert len(result) == 1
assert result[0].status == "FAIL"
assert (
result[0].status_extended
== f"GuardDuty detector {response['DetectorId']} only has the legacy EKS Runtime Monitoring enabled, leaving Amazon EC2 instances and Amazon ECS on Fargate tasks without runtime coverage."
)
@mock_aws
def test_unified_enabled_with_legacy_disabled(self):
"""GetDetector returns an entry for both feature names on the same detector.
A DISABLED legacy feature must not mask the ENABLED unified one.
"""
guardduty_client = client("guardduty", region_name=AWS_REGION_US_EAST_1)
response = guardduty_client.create_detector(
Enable=True,
Features=[
{"Name": "EKS_RUNTIME_MONITORING", "Status": "DISABLED"},
{"Name": "RUNTIME_MONITORING", "Status": "ENABLED"},
],
)
result = _run_check(set_mocked_aws_provider([AWS_REGION_US_EAST_1]))
assert len(result) == 1
assert result[0].status == "PASS"
assert (
result[0].status_extended
== f"GuardDuty detector {response['DetectorId']} has Runtime Monitoring enabled."
)
@@ -2,6 +2,7 @@ from datetime import datetime
from unittest.mock import patch
import botocore
import pytest
from boto3 import client
from moto import mock_aws
@@ -143,6 +144,142 @@ class Test_GuardDuty_Service:
assert guardduty.detectors[0].region == AWS_REGION_EU_WEST_1
assert guardduty.detectors[0].tags == [{"test": "test"}]
@mock_aws
@pytest.mark.parametrize(
"feature_name", ["EKS_RUNTIME_MONITORING", "RUNTIME_MONITORING"]
)
def test_get_detector_eks_runtime_monitoring(self, feature_name):
"""Both feature names set eks_runtime_monitoring.
Unified Runtime Monitoring supersedes EKS Runtime Monitoring and covers EKS.
"""
guardduty_client = client("guardduty", region_name=AWS_REGION_EU_WEST_1)
guardduty_client.create_detector(
Enable=True,
Features=[{"Name": feature_name, "Status": "ENABLED"}],
)
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
guardduty = GuardDuty(aws_provider)
assert len(guardduty.detectors) == 1
assert guardduty.detectors[0].eks_runtime_monitoring
@mock_aws
@pytest.mark.parametrize(
"feature_name", ["EKS_RUNTIME_MONITORING", "RUNTIME_MONITORING"]
)
def test_get_detector_eks_runtime_monitoring_disabled(self, feature_name):
"""Neither feature name sets eks_runtime_monitoring while it is DISABLED."""
guardduty_client = client("guardduty", region_name=AWS_REGION_EU_WEST_1)
guardduty_client.create_detector(
Enable=True,
Features=[{"Name": feature_name, "Status": "DISABLED"}],
)
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
guardduty = GuardDuty(aws_provider)
assert len(guardduty.detectors) == 1
assert not guardduty.detectors[0].eks_runtime_monitoring
@mock_aws
def test_get_detector_unified_runtime_monitoring_with_disabled_eks_feature(self):
"""GetDetector returns an entry for both feature names.
The DISABLED legacy feature must not mask the ENABLED unified one regardless of
the order they arrive in.
"""
guardduty_client = client("guardduty", region_name=AWS_REGION_EU_WEST_1)
guardduty_client.create_detector(
Enable=True,
Features=[
{"Name": "EKS_RUNTIME_MONITORING", "Status": "DISABLED"},
{"Name": "RUNTIME_MONITORING", "Status": "ENABLED"},
],
)
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
guardduty = GuardDuty(aws_provider)
assert len(guardduty.detectors) == 1
assert guardduty.detectors[0].eks_runtime_monitoring
@mock_aws
def test_get_detector_runtime_monitoring_is_unified_only(self):
"""The legacy EKS feature must not set runtime_monitoring.
Only the unified feature covers Amazon EC2 and Amazon ECS on Fargate.
"""
guardduty_client = client("guardduty", region_name=AWS_REGION_EU_WEST_1)
guardduty_client.create_detector(
Enable=True,
Features=[{"Name": "EKS_RUNTIME_MONITORING", "Status": "ENABLED"}],
)
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
guardduty = GuardDuty(aws_provider)
assert len(guardduty.detectors) == 1
assert guardduty.detectors[0].eks_runtime_monitoring
assert not guardduty.detectors[0].runtime_monitoring
@mock_aws
@pytest.mark.parametrize("status", ["ENABLED", "DISABLED"])
def test_get_detector_runtime_monitoring(self, status):
"""runtime_monitoring tracks the reported status of the unified feature."""
guardduty_client = client("guardduty", region_name=AWS_REGION_EU_WEST_1)
guardduty_client.create_detector(
Enable=True,
Features=[{"Name": "RUNTIME_MONITORING", "Status": status}],
)
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
guardduty = GuardDuty(aws_provider)
assert len(guardduty.detectors) == 1
assert guardduty.detectors[0].runtime_monitoring == (status == "ENABLED")
@mock_aws
@pytest.mark.parametrize(
"status, expected", [("ENABLED", True), ("DISABLED", False)]
)
def test_get_detector_ai_protection(self, status, expected):
"""ai_protection is recorded as a bool for both reported statuses."""
guardduty_client = client("guardduty", region_name=AWS_REGION_EU_WEST_1)
guardduty_client.create_detector(
Enable=True,
Features=[{"Name": "AI_PROTECTION", "Status": status}],
)
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
guardduty = GuardDuty(aws_provider)
assert len(guardduty.detectors) == 1
assert guardduty.detectors[0].ai_protection is expected
@mock_aws
def test_get_detector_ai_protection_absent_stays_none(self):
"""An AI_PROTECTION entry GuardDuty never returned must stay None.
That is what lets a check tell a Region without AI Protection apart from a
Region that offers the feature and disabled it.
"""
guardduty_client = client("guardduty", region_name=AWS_REGION_EU_WEST_1)
guardduty_client.create_detector(
Enable=True,
Features=[
{"Name": "S3_DATA_EVENTS", "Status": "ENABLED"},
{"Name": "AI_ANALYST", "Status": "ENABLED"},
],
)
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
guardduty = GuardDuty(aws_provider)
assert len(guardduty.detectors) == 1
assert guardduty.detectors[0].ai_protection is None
@mock_aws
# Test GuardDuty session
def test_list_findings(self):