feat(guardduty): add org-wide delegated admin check across all regions (#9867)

Co-authored-by: Daniel Barranquero <danielbo2001@gmail.com>

prowler/CHANGELOG.md

@@ -9,6 +9,7 @@ All notable changes to the **Prowler SDK** are documented in this file.
 - `entra_conditional_access_policy_approved_client_app_required_for_mobile` check for m365 provider [(#10216)](https://github.com/prowler-cloud/prowler/pull/10216)
 - `entra_conditional_access_policy_compliant_device_hybrid_joined_device_mfa_required` check for M365 provider [(#10197)](https://github.com/prowler-cloud/prowler/pull/10197)
 - Add `trusted_ips` configurable option to `opensearch_service_domains_not_publicly_accessible` check to reduce false positives on IP-restricted policies [(#8631)](https://github.com/prowler-cloud/prowler/pull/8631)
+- `guardduty_delegated_admin_enabled_all_regions` check for AWS provider [(#9867)](https://github.com/prowler-cloud/prowler/pull/9867)
 
 ### 🔄 Changed
 

prowler/providers/aws/services/guardduty/guardduty_delegated_admin_enabled_all_regions/guardduty_delegated_admin_enabled_all_regions.metadata.json

@@ -0,0 +1,44 @@
+{
+  "Provider": "aws",
+  "CheckID": "guardduty_delegated_admin_enabled_all_regions",
+  "CheckTitle": "GuardDuty has delegated admin configured and is enabled in all regions with organization auto-enable",
+  "CheckType": [
+    "Software and Configuration Checks/AWS Security Best Practices",
+    "Software and Configuration Checks/Industry and Regulatory Standards/AWS Foundational Security Best Practices"
+  ],
+  "ServiceName": "guardduty",
+  "SubServiceName": "",
+  "ResourceIdTemplate": "",
+  "Severity": "high",
+  "ResourceType": "AwsGuardDutyDetector",
+  "ResourceGroup": "security",
+  "Description": "**Amazon GuardDuty** has a delegated administrator configured at the organization level, detectors are enabled in all opted-in regions, and organization auto-enable is active for new member accounts.",
+  "Risk": "Without org-wide **Amazon GuardDuty** configuration, gaps can occur where **GuardDuty** may be enabled in some regions but not others. Delegated admin may not be set consistently, and new accounts may not be automatically enrolled. This fragments **threat visibility**, delays **incident response**, and allows adversaries to exploit unmonitored regions or accounts for **lateral movement**, **persistence**, and **data exfiltration**.",
+  "RelatedUrl": "",
+  "AdditionalURLs": [
+    "https://docs.aws.amazon.com/guardduty/latest/ug/guardduty_organizations.html",
+    "https://docs.aws.amazon.com/guardduty/latest/ug/guardduty_multi-account.html",
+    "https://docs.aws.amazon.com/organizations/latest/userguide/services-that-can-integrate-guardduty.html"
+  ],
+  "Remediation": {
+    "Code": {
+      "CLI": "aws guardduty enable-organization-admin-account --admin-account-id <ADMIN_ACCOUNT_ID> && aws guardduty update-organization-configuration --detector-id <DETECTOR_ID> --auto-enable-organization-members ALL",
+      "NativeIaC": "",
+      "Other": "1. Sign in to the AWS Organizations management account\n2. Open the AWS Organizations console\n3. Navigate to Services > Amazon GuardDuty\n4. Click Register delegated administrator and enter the security account ID\n5. Switch to the delegated admin account\n6. In GuardDuty console, go to Settings > Accounts\n7. Enable auto-enable for all organization members\n8. Repeat detector enablement for all opted-in regions",
+      "Terraform": ""
+    },
+    "Recommendation": {
+      "Text": "Configure a **delegated administrator** for GuardDuty via AWS Organizations. Enable GuardDuty detectors in **all opted-in regions** and configure **auto-enable** to automatically enroll new member accounts. This ensures consistent threat detection coverage across the entire organization.",
+      "Url": "https://hub.prowler.com/check/guardduty_delegated_admin_enabled_all_regions"
+    }
+  },
+  "Categories": [
+    "forensics-ready"
+  ],
+  "DependsOn": [],
+  "RelatedTo": [
+    "guardduty_is_enabled",
+    "guardduty_centrally_managed"
+  ],
+  "Notes": "This check requires execution from the organization management account or delegated administrator account to access organization-level APIs."
+}

prowler/providers/aws/services/guardduty/guardduty_delegated_admin_enabled_all_regions/guardduty_delegated_admin_enabled_all_regions.py

@@ -0,0 +1,76 @@
+from prowler.lib.check.models import Check, Check_Report_AWS
+from prowler.providers.aws.services.guardduty.guardduty_client import guardduty_client
+
+
+class guardduty_delegated_admin_enabled_all_regions(Check):
+    """Ensure GuardDuty has a delegated admin and is enabled in all regions.
+
+    This check verifies that:
+    1. A delegated administrator account is configured for GuardDuty
+    2. GuardDuty detectors are enabled in each region
+    3. Organization auto-enable is configured for new member accounts
+    """
+
+    def execute(self) -> list[Check_Report_AWS]:
+        """Execute the check logic.
+
+        Returns:
+            A list of reports containing the result of the check for each region.
+        """
+        findings = []
+
+        # Build a set of regions that have an organization admin account configured
+        regions_with_admin = {
+            admin.region
+            for admin in guardduty_client.organization_admin_accounts
+            if admin.admin_status == "ENABLED"
+        }
+
+        for detector in guardduty_client.detectors:
+            report = Check_Report_AWS(metadata=self.metadata(), resource=detector)
+
+            # Check if this region has a delegated admin
+            has_delegated_admin = detector.region in regions_with_admin
+
+            # Check if detector is enabled
+            detector_enabled = detector.enabled_in_account and detector.status
+
+            # Check if auto-enable is configured for organization members
+            auto_enable_configured = detector.organization_auto_enable_members in (
+                "NEW",
+                "ALL",
+            )
+
+            # Determine overall status
+            issues = []
+            if not has_delegated_admin:
+                issues.append("no delegated administrator configured")
+            if not detector_enabled:
+                issues.append("detector not enabled")
+            if not auto_enable_configured and detector.organization_config_available:
+                # Only report auto-enable issue if org config data is available
+                issues.append("organization auto-enable not configured")
+
+            if issues:
+                report.status = "FAIL"
+                report.status_extended = (
+                    f"GuardDuty in region {detector.region} has issues: "
+                    f"{', '.join(issues)}."
+                )
+            else:
+                report.status = "PASS"
+                report.status_extended = (
+                    f"GuardDuty in region {detector.region} has delegated admin "
+                    f"configured with detector enabled and organization auto-enable active."
+                )
+
+            # Support muting non-default regions if configured
+            if report.status == "FAIL" and (
+                guardduty_client.audit_config.get("mute_non_default_regions", False)
+                and detector.region != guardduty_client.region
+            ):
+                report.muted = True
+
+            findings.append(report)
+
+        return findings

prowler/providers/aws/services/guardduty/guardduty_service.py

@@ -1,5 +1,6 @@
 from typing import Optional
 
+from botocore.exceptions import ClientError
 from pydantic.v1 import BaseModel
 
 from prowler.lib.logger import logger
@@ -12,12 +13,17 @@ class GuardDuty(AWSService):
         # Call AWSService's __init__
         super().__init__(__class__.__name__, provider)
         self.detectors = []
+        self.organization_admin_accounts = []
         self.__threading_call__(self._list_detectors)
         self.__threading_call__(self._get_detector, self.detectors)
         self._list_findings()
         self._list_members()
         self._get_administrator_account()
         self._list_tags_for_resource()
+        self.__threading_call__(self._list_organization_admin_accounts)
+        self.__threading_call__(
+            self._describe_organization_configuration, self.detectors
+        )
 
     def _list_detectors(self, regional_client):
         logger.info("GuardDuty - listing detectors...")
@@ -216,13 +222,97 @@ class GuardDuty(AWSService):
                 f"{error.__class__.__name__}:{error.__traceback__.tb_lineno} -- {error}"
             )
 
+    def _list_organization_admin_accounts(self, regional_client):
+        """List GuardDuty delegated administrator accounts for the organization.
+
+        This API is only available to the organization management account or
+        a delegated administrator account.
+        """
+        logger.info("GuardDuty - listing organization admin accounts...")
+        try:
+            paginator = regional_client.get_paginator(
+                "list_organization_admin_accounts"
+            )
+            for page in paginator.paginate():
+                for admin in page.get("AdminAccounts", []):
+                    admin_account = OrganizationAdminAccount(
+                        admin_account_id=admin.get("AdminAccountId"),
+                        admin_status=admin.get("AdminStatus"),
+                        region=regional_client.region,
+                    )
+                    # Avoid duplicates across regions for the same admin account
+                    if not any(
+                        existing.admin_account_id == admin_account.admin_account_id
+                        and existing.region == admin_account.region
+                        for existing in self.organization_admin_accounts
+                    ):
+                        self.organization_admin_accounts.append(admin_account)
+        except ClientError as error:
+            if error.response["Error"]["Code"] in (
+                "AccessDeniedException",
+                "BadRequestException",
+            ):
+                logger.warning(
+                    f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
+                )
+            else:
+                logger.error(
+                    f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
+                )
+        except Exception as error:
+            logger.error(
+                f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
+            )
+
+    def _describe_organization_configuration(self, detector):
+        """Describe the organization configuration for a GuardDuty detector.
+
+        This provides information about auto-enable settings for the organization.
+        """
+        logger.info("GuardDuty - describing organization configuration...")
+        try:
+            if detector.id and detector.enabled_in_account:
+                regional_client = self.regional_clients[detector.region]
+                org_config = regional_client.describe_organization_configuration(
+                    DetectorId=detector.id
+                )
+                detector.organization_auto_enable_members = org_config.get(
+                    "AutoEnableOrganizationMembers", "NONE"
+                )
+                detector.organization_config_available = True
+        except ClientError as error:
+            if error.response["Error"]["Code"] in (
+                "AccessDeniedException",
+                "BadRequestException",
+            ):
+                # Expected when not running from management or delegated admin account
+                logger.warning(
+                    f"{detector.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
+                )
+            else:
+                logger.error(
+                    f"{detector.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
+                )
+        except Exception as error:
+            logger.error(
+                f"{detector.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
+            )
+
+
+class OrganizationAdminAccount(BaseModel):
+    """Represents a GuardDuty delegated administrator account."""
+
+    admin_account_id: str
+    admin_status: str  # ENABLED or DISABLE_IN_PROGRESS
+    region: str
+
 
 class Detector(BaseModel):
     id: str
     arn: str
     region: str
     enabled_in_account: bool
-    status: bool = None
+    status: Optional[bool] = None
     findings: list = []
     member_accounts: list = []
     administrator_account: str = None
@@ -233,3 +323,6 @@ class Detector(BaseModel):
     eks_runtime_monitoring: bool = False
     lambda_protection: bool = False
     ec2_malware_protection: bool = False
+    # Organization configuration fields
+    organization_auto_enable_members: str = "NONE"  # NEW, ALL, or NONE
+    organization_config_available: bool = False

tests/providers/aws/services/guardduty/guardduty_delegated_admin_enabled_all_regions/guardduty_delegated_admin_enabled_all_regions_test.py

@@ -0,0 +1,233 @@
+from unittest.mock import patch
+
+import botocore
+from boto3 import client
+from moto import mock_aws
+
+from tests.providers.aws.utils import (
+    AWS_ACCOUNT_NUMBER,
+    AWS_REGION_EU_WEST_1,
+    set_mocked_aws_provider,
+)
+
+orig = botocore.client.BaseClient._make_api_call
+
+
+def mock_make_api_call_org_admin_and_config(self, operation_name, api_params):
+    """Mock organization admin accounts and configuration APIs."""
+    if operation_name == "ListOrganizationAdminAccounts":
+        return {
+            "AdminAccounts": [
+                {
+                    "AdminAccountId": "123456789012",
+                    "AdminStatus": "ENABLED",
+                }
+            ]
+        }
+    if operation_name == "DescribeOrganizationConfiguration":
+        return {
+            "AutoEnableOrganizationMembers": "ALL",
+        }
+    return orig(self, operation_name, api_params)
+
+
+def mock_make_api_call_org_admin_no_auto_enable(self, operation_name, api_params):
+    """Mock organization admin configured but auto-enable disabled."""
+    if operation_name == "ListOrganizationAdminAccounts":
+        return {
+            "AdminAccounts": [
+                {
+                    "AdminAccountId": "123456789012",
+                    "AdminStatus": "ENABLED",
+                }
+            ]
+        }
+    if operation_name == "DescribeOrganizationConfiguration":
+        return {
+            "AutoEnableOrganizationMembers": "NONE",
+        }
+    return orig(self, operation_name, api_params)
+
+
+def mock_make_api_call_no_org_admin(self, operation_name, api_params):
+    """Mock no organization admin configured."""
+    if operation_name == "ListOrganizationAdminAccounts":
+        return {"AdminAccounts": []}
+    if operation_name == "DescribeOrganizationConfiguration":
+        return {
+            "AutoEnableOrganizationMembers": "NONE",
+        }
+    return orig(self, operation_name, api_params)
+
+
+class Test_guardduty_delegated_admin_enabled_all_regions:
+    @mock_aws
+    def test_no_detectors(self):
+        """Test when no GuardDuty detectors exist."""
+        aws_provider = set_mocked_aws_provider()
+
+        from prowler.providers.aws.services.guardduty.guardduty_service import GuardDuty
+
+        with (
+            patch(
+                "prowler.providers.common.provider.Provider.get_global_provider",
+                return_value=aws_provider,
+            ),
+            patch(
+                "prowler.providers.aws.services.guardduty.guardduty_delegated_admin_enabled_all_regions.guardduty_delegated_admin_enabled_all_regions.guardduty_client",
+                new=GuardDuty(aws_provider),
+            ),
+        ):
+            from prowler.providers.aws.services.guardduty.guardduty_delegated_admin_enabled_all_regions.guardduty_delegated_admin_enabled_all_regions import (
+                guardduty_delegated_admin_enabled_all_regions,
+            )
+
+            check = guardduty_delegated_admin_enabled_all_regions()
+            result = check.execute()
+
+            # Should have findings for each region (with unknown detectors)
+            assert len(result) > 0
+            # All should fail since no detectors are enabled
+            for finding in result:
+                assert finding.status == "FAIL"
+                assert "detector not enabled" in finding.status_extended
+
+    @patch(
+        "botocore.client.BaseClient._make_api_call",
+        new=mock_make_api_call_no_org_admin,
+    )
+    @mock_aws
+    def test_detector_enabled_no_delegated_admin(self):
+        """Test when detector is enabled but no delegated admin is configured."""
+        guardduty_client_boto = client("guardduty", region_name=AWS_REGION_EU_WEST_1)
+        detector_id = guardduty_client_boto.create_detector(Enable=True)["DetectorId"]
+
+        aws_provider = set_mocked_aws_provider()
+
+        from prowler.providers.aws.services.guardduty.guardduty_service import GuardDuty
+
+        with (
+            patch(
+                "prowler.providers.common.provider.Provider.get_global_provider",
+                return_value=aws_provider,
+            ),
+            patch(
+                "prowler.providers.aws.services.guardduty.guardduty_delegated_admin_enabled_all_regions.guardduty_delegated_admin_enabled_all_regions.guardduty_client",
+                new=GuardDuty(aws_provider),
+            ),
+        ):
+            from prowler.providers.aws.services.guardduty.guardduty_delegated_admin_enabled_all_regions.guardduty_delegated_admin_enabled_all_regions import (
+                guardduty_delegated_admin_enabled_all_regions,
+            )
+
+            check = guardduty_delegated_admin_enabled_all_regions()
+            result = check.execute()
+
+            # Find the result for our region
+            eu_west_1_result = None
+            for finding in result:
+                if finding.region == AWS_REGION_EU_WEST_1:
+                    eu_west_1_result = finding
+                    break
+
+            assert eu_west_1_result is not None
+            assert eu_west_1_result.status == "FAIL"
+            assert (
+                "no delegated administrator configured"
+                in eu_west_1_result.status_extended
+            )
+            assert eu_west_1_result.resource_id == detector_id
+
+    @patch(
+        "botocore.client.BaseClient._make_api_call",
+        new=mock_make_api_call_org_admin_no_auto_enable,
+    )
+    @mock_aws
+    def test_detector_enabled_with_admin_no_auto_enable(self):
+        """Test when detector is enabled with delegated admin but auto-enable is off."""
+        guardduty_client_boto = client("guardduty", region_name=AWS_REGION_EU_WEST_1)
+        detector_id = guardduty_client_boto.create_detector(Enable=True)["DetectorId"]
+
+        aws_provider = set_mocked_aws_provider()
+
+        from prowler.providers.aws.services.guardduty.guardduty_service import GuardDuty
+
+        with (
+            patch(
+                "prowler.providers.common.provider.Provider.get_global_provider",
+                return_value=aws_provider,
+            ),
+            patch(
+                "prowler.providers.aws.services.guardduty.guardduty_delegated_admin_enabled_all_regions.guardduty_delegated_admin_enabled_all_regions.guardduty_client",
+                new=GuardDuty(aws_provider),
+            ),
+        ):
+            from prowler.providers.aws.services.guardduty.guardduty_delegated_admin_enabled_all_regions.guardduty_delegated_admin_enabled_all_regions import (
+                guardduty_delegated_admin_enabled_all_regions,
+            )
+
+            check = guardduty_delegated_admin_enabled_all_regions()
+            result = check.execute()
+
+            # Find the result for our region
+            eu_west_1_result = None
+            for finding in result:
+                if finding.region == AWS_REGION_EU_WEST_1:
+                    eu_west_1_result = finding
+                    break
+
+            assert eu_west_1_result is not None
+            assert eu_west_1_result.status == "FAIL"
+            assert (
+                "organization auto-enable not configured"
+                in eu_west_1_result.status_extended
+            )
+            assert eu_west_1_result.resource_id == detector_id
+
+    @patch(
+        "botocore.client.BaseClient._make_api_call",
+        new=mock_make_api_call_org_admin_and_config,
+    )
+    @mock_aws
+    def test_detector_enabled_with_admin_and_auto_enable(self):
+        """Test when detector is enabled with delegated admin and auto-enable is on (PASS)."""
+        guardduty_client_boto = client("guardduty", region_name=AWS_REGION_EU_WEST_1)
+        detector_id = guardduty_client_boto.create_detector(Enable=True)["DetectorId"]
+
+        aws_provider = set_mocked_aws_provider()
+
+        from prowler.providers.aws.services.guardduty.guardduty_service import GuardDuty
+
+        with (
+            patch(
+                "prowler.providers.common.provider.Provider.get_global_provider",
+                return_value=aws_provider,
+            ),
+            patch(
+                "prowler.providers.aws.services.guardduty.guardduty_delegated_admin_enabled_all_regions.guardduty_delegated_admin_enabled_all_regions.guardduty_client",
+                new=GuardDuty(aws_provider),
+            ),
+        ):
+            from prowler.providers.aws.services.guardduty.guardduty_delegated_admin_enabled_all_regions.guardduty_delegated_admin_enabled_all_regions import (
+                guardduty_delegated_admin_enabled_all_regions,
+            )
+
+            check = guardduty_delegated_admin_enabled_all_regions()
+            result = check.execute()
+
+            # Find the result for our region
+            eu_west_1_result = None
+            for finding in result:
+                if finding.region == AWS_REGION_EU_WEST_1:
+                    eu_west_1_result = finding
+                    break
+
+            assert eu_west_1_result is not None
+            assert eu_west_1_result.status == "PASS"
+            assert "delegated admin configured" in eu_west_1_result.status_extended
+            assert "auto-enable active" in eu_west_1_result.status_extended
+            assert eu_west_1_result.resource_id == detector_id
+            assert (
+                eu_west_1_result.resource_arn
+                == f"arn:aws:guardduty:{AWS_REGION_EU_WEST_1}:{AWS_ACCOUNT_NUMBER}:detector/{detector_id}"
+            )