fix(aws): delegated administrator lookup and reporting (#12342)

Co-authored-by: Pedro Martín <pedromarting3@gmail.com>

prowler/changelog.d/delegated-admin-unknown-manual.changed.md

@@ -0,0 +1 @@
+The `securityhub_delegated_admin_enabled_all_regions`, `guardduty_delegated_admin_enabled_all_regions` and `config_delegated_admin_and_org_aggregator_all_regions` checks now report MANUAL instead of FAIL when the delegated administrator status cannot be read and no independent misconfiguration is detected, which happens on member accounts that are not registered as delegated administrators because the API is restricted to the organization management account and to delegated administrator accounts

prowler/changelog.d/guardduty-delegated-admin-lookup.fixed.md

@@ -0,0 +1 @@
+`guardduty_delegated_admin_enabled_all_regions` no longer reports `no delegated administrator configured` when the lookup was denied or failed, which asserted absence where there was only lack of visibility

prowler/changelog.d/securityhub-admin-lookup-per-region.fixed.md

@@ -0,0 +1 @@
+A failed `ListOrganizationAdminAccounts` lookup in one region no longer marks the Security Hub delegated administrator status as undetermined in every other region

prowler/changelog.d/securityhub-delegated-admin-fields.fixed.md

@@ -0,0 +1 @@
+`securityhub_delegated_admin_enabled_all_regions` no longer reports FAIL with `delegated administrator status could not be determined` on accounts that do have a Security Hub delegated administrator; `ListOrganizationAdminAccounts` responses are now parsed with the `AccountId` and `Status` fields the API actually returns

prowler/providers/aws/services/config/config_delegated_admin_and_org_aggregator_all_regions/config_delegated_admin_and_org_aggregator_all_regions.py

@@ -71,12 +71,7 @@ class config_delegated_admin_and_org_aggregator_all_regions(Check):
                 covers_all = aggregator.all_aws_regions
 
                 issues = []
-                if delegated_admin_unknown:
-                    issues.append(
-                        "delegated administrator status for config.amazonaws.com "
-                        "could not be determined"
-                    )
-                elif not has_delegated_admin:
+                if not delegated_admin_unknown and not has_delegated_admin:
                     issues.append(
                         "no delegated administrator registered for config.amazonaws.com"
                     )
@@ -95,6 +90,25 @@ class config_delegated_admin_and_org_aggregator_all_regions(Check):
                         f"AWS Config aggregator {aggregator.name} in region "
                         f"{region} has issues: {', '.join(issues)}."
                     )
+                    if delegated_admin_unknown:
+                        report.status_extended = (
+                            f"{report.status_extended[:-1]}; the delegated "
+                            f"administrator status for config.amazonaws.com could "
+                            f"not be determined."
+                        )
+                elif delegated_admin_unknown:
+                    # Not being able to read the delegated administrator is a lack
+                    # of visibility, not a misconfiguration: the Organizations API
+                    # is only available to the management or delegated
+                    # administrator account.
+                    report.status = "MANUAL"
+                    report.status_extended = (
+                        f"AWS Config aggregator {aggregator.name} in region {region} "
+                        f"is an organization aggregator covering all AWS regions, but "
+                        f"the delegated administrator status for config.amazonaws.com "
+                        f"could not be determined; run this check from the "
+                        f"organization management or delegated administrator account."
+                    )
                 else:
                     report.status = "PASS"
                     report.status_extended = (

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

@@ -32,6 +32,15 @@ class guardduty_delegated_admin_enabled_all_regions(Check):
             # Check if this region has a delegated admin
             has_delegated_admin = detector.region in regions_with_admin
 
+            # The lookup is tracked per region so that a failure in one region does
+            # not mask the result of the others. A failure is only relevant when no
+            # delegated admin was found: if one was, the status is known.
+            admin_lookup_failed = (
+                not has_delegated_admin
+                and detector.region
+                in guardduty_client.organization_admin_lookup_failed_regions
+            )
+
             # Check if detector is enabled
             detector_enabled = detector.enabled_in_account and detector.status
 
@@ -43,7 +52,7 @@ class guardduty_delegated_admin_enabled_all_regions(Check):
 
             # Determine overall status
             issues = []
-            if not has_delegated_admin:
+            if not admin_lookup_failed and not has_delegated_admin:
                 issues.append("no delegated administrator configured")
             if not detector_enabled:
                 issues.append("detector not enabled")
@@ -57,6 +66,22 @@ class guardduty_delegated_admin_enabled_all_regions(Check):
                     f"GuardDuty in region {detector.region} has issues: "
                     f"{', '.join(issues)}."
                 )
+                if admin_lookup_failed:
+                    report.status_extended = (
+                        f"{report.status_extended[:-1]}; the delegated administrator "
+                        f"status could not be determined."
+                    )
+            elif admin_lookup_failed:
+                # Not being able to read the delegated administrator is a lack of
+                # visibility, not a misconfiguration: the API is only available to
+                # the management or delegated administrator account.
+                report.status = "MANUAL"
+                report.status_extended = (
+                    f"GuardDuty delegated administrator status in region "
+                    f"{detector.region} could not be determined; run this check "
+                    f"from the organization management or delegated administrator "
+                    f"account."
+                )
             else:
                 report.status = "PASS"
                 report.status_extended = (

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

@@ -14,6 +14,7 @@ class GuardDuty(AWSService):
         super().__init__(__class__.__name__, provider)
         self.detectors = []
         self.organization_admin_accounts = []
+        self.organization_admin_lookup_failed_regions: set = set()
         self.__threading_call__(self._list_detectors)
         self.__threading_call__(self._get_detector, self.detectors)
         self._list_findings()
@@ -227,6 +228,9 @@ class GuardDuty(AWSService):
 
         This API is only available to the organization management account or
         a delegated administrator account.
+
+        Args:
+            regional_client: Regional client object.
         """
         logger.info("GuardDuty - listing organization admin accounts...")
         try:
@@ -235,12 +239,30 @@ class GuardDuty(AWSService):
             )
             for page in paginator.paginate():
                 for admin in page.get("AdminAccounts", []):
+                    # GuardDuty returns AdminAccountId/AdminStatus, unlike Security
+                    # Hub's AccountId/Status for the same operation name.
+                    account_id = admin.get("AdminAccountId")
+                    status = admin.get("AdminStatus")
+                    if not account_id or not status:
+                        # An entry we cannot interpret means the delegated admin
+                        # status for this region is unknown, not absent.
+                        if (
+                            regional_client.region
+                            not in self.organization_admin_lookup_failed_regions
+                        ):
+                            logger.warning(
+                                f"{regional_client.region} -- Unexpected admin account entry with keys {sorted(admin)}"
+                            )
+                        self.organization_admin_lookup_failed_regions.add(
+                            regional_client.region
+                        )
+                        continue
                     admin_account = OrganizationAdminAccount(
-                        admin_account_id=admin.get("AdminAccountId"),
-                        admin_status=admin.get("AdminStatus"),
+                        admin_account_id=account_id,
+                        admin_status=status,
                         region=regional_client.region,
                     )
-                    # Avoid duplicates across regions for the same admin account
+                    # Avoid duplicates across pages for the same admin account
                     if not any(
                         existing.admin_account_id == admin_account.admin_account_id
                         and existing.region == admin_account.region
@@ -248,6 +270,7 @@ class GuardDuty(AWSService):
                     ):
                         self.organization_admin_accounts.append(admin_account)
         except ClientError as error:
+            self.organization_admin_lookup_failed_regions.add(regional_client.region)
             if error.response["Error"]["Code"] in (
                 "AccessDeniedException",
                 "BadRequestException",
@@ -260,6 +283,7 @@ class GuardDuty(AWSService):
                     f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
                 )
         except Exception as error:
+            self.organization_admin_lookup_failed_regions.add(regional_client.region)
             logger.error(
                 f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
             )

prowler/providers/aws/services/securityhub/securityhub_delegated_admin_enabled_all_regions/securityhub_delegated_admin_enabled_all_regions.py

@@ -27,7 +27,6 @@ class securityhub_delegated_admin_enabled_all_regions(Check):
             for admin in securityhub_client.organization_admin_accounts
             if admin.admin_status == "ENABLED"
         }
-        admin_lookup_failed = securityhub_client.organization_admin_lookup_failed
 
         for securityhub in securityhub_client.securityhubs:
             report = Check_Report_AWS(metadata=self.metadata(), resource=securityhub)
@@ -35,6 +34,15 @@ class securityhub_delegated_admin_enabled_all_regions(Check):
             # Check if this region has a delegated admin
             has_delegated_admin = securityhub.region in regions_with_admin
 
+            # The lookup is tracked per region so that a failure in one region does
+            # not mask the result of the others. A partial failure is only relevant
+            # when no delegated admin was found: if one was, the status is known.
+            admin_lookup_failed = (
+                not has_delegated_admin
+                and securityhub.region
+                in securityhub_client.organization_admin_lookup_failed_regions
+            )
+
             # Check if hub is active
             hub_active = securityhub.status == "ACTIVE"
 
@@ -43,9 +51,7 @@ class securityhub_delegated_admin_enabled_all_regions(Check):
 
             # Determine overall status
             issues = []
-            if admin_lookup_failed:
-                issues.append("delegated administrator status could not be determined")
-            elif not has_delegated_admin:
+            if not admin_lookup_failed and not has_delegated_admin:
                 issues.append("no delegated administrator configured")
             if not hub_active:
                 issues.append("Security Hub not enabled")
@@ -64,6 +70,22 @@ class securityhub_delegated_admin_enabled_all_regions(Check):
                     f"Security Hub in region {securityhub.region} has issues: "
                     f"{', '.join(issues)}."
                 )
+                if admin_lookup_failed:
+                    report.status_extended = (
+                        f"{report.status_extended[:-1]}; the delegated administrator "
+                        f"status could not be determined."
+                    )
+            elif admin_lookup_failed:
+                # Not being able to read the delegated administrator is a lack of
+                # visibility, not a misconfiguration: the API is only available to
+                # the management or delegated administrator account.
+                report.status = "MANUAL"
+                report.status_extended = (
+                    f"Security Hub delegated administrator status in region "
+                    f"{securityhub.region} could not be determined; run this check "
+                    f"from the organization management or delegated administrator "
+                    f"account."
+                )
             else:
                 report.status = "PASS"
                 report.status_extended = (

prowler/providers/aws/services/securityhub/securityhub_service.py

@@ -14,7 +14,7 @@ class SecurityHub(AWSService):
         super().__init__(__class__.__name__, provider)
         self.securityhubs = []
         self.organization_admin_accounts = []
-        self.organization_admin_lookup_failed: bool = False
+        self.organization_admin_lookup_failed_regions: set = set()
         self.__threading_call__(self._describe_hub)
         self.__threading_call__(self._list_tags, self.securityhubs)
         self.__threading_call__(self._list_organization_admin_accounts)
@@ -115,6 +115,9 @@ class SecurityHub(AWSService):
 
         This API is only available to the organization management account or
         a delegated administrator account.
+
+        Args:
+            regional_client: Regional client object.
         """
         logger.info("SecurityHub - listing organization admin accounts...")
         try:
@@ -123,12 +126,30 @@ class SecurityHub(AWSService):
             )
             for page in paginator.paginate():
                 for admin in page.get("AdminAccounts", []):
+                    # Security Hub returns AccountId/Status, unlike GuardDuty's
+                    # AdminAccountId/AdminStatus for the same operation name.
+                    account_id = admin.get("AccountId")
+                    status = admin.get("Status")
+                    if not account_id or not status:
+                        # An entry we cannot interpret means the delegated admin
+                        # status for this region is unknown, not absent.
+                        if (
+                            regional_client.region
+                            not in self.organization_admin_lookup_failed_regions
+                        ):
+                            logger.warning(
+                                f"{regional_client.region} -- Unexpected admin account entry with keys {sorted(admin)}"
+                            )
+                        self.organization_admin_lookup_failed_regions.add(
+                            regional_client.region
+                        )
+                        continue
                     admin_account = OrganizationAdminAccount(
-                        admin_account_id=admin.get("AdminAccountId"),
-                        admin_status=admin.get("AdminStatus"),
+                        admin_account_id=account_id,
+                        admin_status=status,
                         region=regional_client.region,
                     )
-                    # Avoid duplicates across regions for the same admin account
+                    # Avoid duplicates across pages for the same admin account
                     if not any(
                         existing.admin_account_id == admin_account.admin_account_id
                         and existing.region == admin_account.region
@@ -136,7 +157,7 @@ class SecurityHub(AWSService):
                     ):
                         self.organization_admin_accounts.append(admin_account)
         except ClientError as error:
-            self.organization_admin_lookup_failed = True
+            self.organization_admin_lookup_failed_regions.add(regional_client.region)
             if error.response["Error"]["Code"] in (
                 "AccessDeniedException",
                 "InvalidAccessException",
@@ -150,7 +171,7 @@ class SecurityHub(AWSService):
                     f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
                 )
         except Exception as error:
-            self.organization_admin_lookup_failed = True
+            self.organization_admin_lookup_failed_regions.add(regional_client.region)
             logger.error(
                 f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
             )

tests/providers/aws/services/config/config_delegated_admin_and_org_aggregator_all_regions/config_delegated_admin_and_org_aggregator_all_regions_test.py

@@ -430,10 +430,15 @@ class Test_config_delegated_admin_and_org_aggregator_all_regions:
 
                 assert eu_west_1_result is not None
                 # The check still runs; aggregator coverage is satisfied but the
-                # delegated-admin status is unknown, so it must FAIL.
-                assert eu_west_1_result.status == "FAIL"
+                # delegated-admin status is unknown, which is a lack of visibility
+                # rather than a misconfiguration.
+                assert eu_west_1_result.status == "MANUAL"
                 assert (
-                    "delegated administrator status for config.amazonaws.com could not be determined"
+                    "delegated administrator status for config.amazonaws.com could "
+                    "not be determined" in eu_west_1_result.status_extended
+                )
+                assert (
+                    "organization management or delegated administrator account"
                     in eu_west_1_result.status_extended
                 )
 

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

@@ -7,56 +7,114 @@ from moto import mock_aws
 from tests.providers.aws.utils import (
     AWS_ACCOUNT_NUMBER,
     AWS_REGION_EU_WEST_1,
+    mocked_api_response,
     set_mocked_aws_provider,
 )
 
 orig = botocore.client.BaseClient._make_api_call
 
 
+def mocked_response(operation_name, response):
+    """Validate a mocked GuardDuty response against the real API model."""
+    return mocked_api_response("guardduty", operation_name, response)
+
+
 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",
-                }
-            ]
-        }
+        return mocked_response(
+            operation_name,
+            {
+                "AdminAccounts": [
+                    {"AdminAccountId": "123456789012", "AdminStatus": "ENABLED"}
+                ]
+            },
+        )
     if operation_name == "DescribeOrganizationConfiguration":
-        return {
-            "AutoEnableOrganizationMembers": "ALL",
-        }
+        return mocked_response(
+            operation_name,
+            {
+                "AutoEnableOrganizationMembers": "ALL",
+                "MemberAccountLimitReached": False,
+            },
+        )
     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",
-                }
-            ]
-        }
+        return mocked_response(
+            operation_name,
+            {
+                "AdminAccounts": [
+                    {"AdminAccountId": "123456789012", "AdminStatus": "ENABLED"}
+                ]
+            },
+        )
     if operation_name == "DescribeOrganizationConfiguration":
-        return {
-            "AutoEnableOrganizationMembers": "NONE",
-        }
+        return mocked_response(
+            operation_name,
+            {
+                "AutoEnableOrganizationMembers": "NONE",
+                "MemberAccountLimitReached": False,
+            },
+        )
     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": []}
+        return mocked_response(operation_name, {"AdminAccounts": []})
     if operation_name == "DescribeOrganizationConfiguration":
-        return {
-            "AutoEnableOrganizationMembers": "NONE",
-        }
+        return mocked_response(
+            operation_name,
+            {
+                "AutoEnableOrganizationMembers": "NONE",
+                "MemberAccountLimitReached": False,
+            },
+        )
+    return orig(self, operation_name, api_params)
+
+
+def mock_make_api_call_admin_lookup_access_denied(self, operation_name, api_params):
+    """ListOrganizationAdminAccounts is denied — lookup-failed path."""
+    if operation_name == "ListOrganizationAdminAccounts":
+        raise botocore.exceptions.ClientError(
+            {
+                "Error": {
+                    "Code": "AccessDeniedException",
+                    "Message": "User is not authorized to perform: guardduty:ListOrganizationAdminAccounts",
+                }
+            },
+            operation_name,
+        )
+    if operation_name == "DescribeOrganizationConfiguration":
+        return mocked_response(
+            operation_name,
+            {
+                "AutoEnableOrganizationMembers": "ALL",
+                "MemberAccountLimitReached": False,
+            },
+        )
+    return orig(self, operation_name, api_params)
+
+
+def mock_make_api_call_admin_account_missing_fields(self, operation_name, api_params):
+    """AdminAccounts entry without the documented fields."""
+    if operation_name == "ListOrganizationAdminAccounts":
+        # Deliberately not validated against the API model: this simulates the
+        # response drifting away from what botocore currently describes.
+        return {"AdminAccounts": [{"SomethingElse": "unexpected"}]}
+    if operation_name == "DescribeOrganizationConfiguration":
+        return mocked_response(
+            operation_name,
+            {
+                "AutoEnableOrganizationMembers": "ALL",
+                "MemberAccountLimitReached": False,
+            },
+        )
     return orig(self, operation_name, api_params)
 
 
@@ -231,3 +289,67 @@ class Test_guardduty_delegated_admin_enabled_all_regions:
                 eu_west_1_result.resource_arn
                 == f"arn:aws:guardduty:{AWS_REGION_EU_WEST_1}:{AWS_ACCOUNT_NUMBER}:detector/{detector_id}"
             )
+
+    @patch(
+        "botocore.client.BaseClient._make_api_call",
+        new=mock_make_api_call_admin_lookup_access_denied,
+    )
+    @mock_aws
+    def test_admin_lookup_access_denied(self):
+        """A denied lookup is unknown, not absent: MANUAL instead of FAIL."""
+        guardduty_client_boto = client("guardduty", region_name=AWS_REGION_EU_WEST_1)
+        guardduty_client_boto.create_detector(Enable=True)
+
+        aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
+
+        from prowler.providers.aws.services.guardduty.guardduty_service import GuardDuty
+
+        service = GuardDuty(aws_provider)
+        assert AWS_REGION_EU_WEST_1 in service.organization_admin_lookup_failed_regions
+
+        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=service,
+            ),
+        ):
+            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,
+            )
+
+            result = guardduty_delegated_admin_enabled_all_regions().execute()
+
+            assert result and result[0].status == "MANUAL"
+            assert "could not be determined" in result[0].status_extended
+            assert (
+                "management or delegated administrator account"
+                in result[0].status_extended
+            )
+            assert (
+                "no delegated administrator configured" not in result[0].status_extended
+            )
+
+    @patch(
+        "botocore.client.BaseClient._make_api_call",
+        new=mock_make_api_call_admin_account_missing_fields,
+    )
+    @mock_aws
+    def test_admin_account_missing_fields(self):
+        """An unparseable admin entry marks the region as unknown instead of raising."""
+        guardduty_client_boto = client("guardduty", region_name=AWS_REGION_EU_WEST_1)
+        guardduty_client_boto.create_detector(Enable=True)
+
+        aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
+
+        from prowler.providers.aws.services.guardduty.guardduty_service import GuardDuty
+
+        service = GuardDuty(aws_provider)
+
+        assert service.organization_admin_accounts == []
+        assert service.organization_admin_lookup_failed_regions == {
+            AWS_REGION_EU_WEST_1
+        }

tests/providers/aws/services/securityhub/securityhub_delegated_admin_enabled_all_regions/securityhub_delegated_admin_enabled_all_regions_test.py

@@ -6,6 +6,8 @@ from moto import mock_aws
 from tests.providers.aws.utils import (
     AWS_ACCOUNT_NUMBER,
     AWS_REGION_EU_WEST_1,
+    AWS_REGION_EU_WEST_2,
+    mocked_api_response,
     set_mocked_aws_provider,
 )
 
@@ -14,6 +16,11 @@ orig = botocore.client.BaseClient._make_api_call
 HUB_ARN = f"arn:aws:securityhub:{AWS_REGION_EU_WEST_1}:{AWS_ACCOUNT_NUMBER}:hub/default"
 
 
+def mocked_response(operation_name, response):
+    """Validate a mocked Security Hub response against the real API model."""
+    return mocked_api_response("securityhub", operation_name, response)
+
+
 def _active_hub_responses(operation_name):
     """Return a moto-friendly response for hub-describing API calls.
 
@@ -21,17 +28,20 @@ def _active_hub_responses(operation_name):
     can fall back to the default behavior).
     """
     if operation_name == "DescribeHub":
-        return {
-            "HubArn": HUB_ARN,
-            "SubscribedAt": "2024-01-01T00:00:00.000Z",
-            "AutoEnableControls": True,
-        }
+        return mocked_response(
+            operation_name,
+            {
+                "HubArn": HUB_ARN,
+                "SubscribedAt": "2024-01-01T00:00:00.000Z",
+                "AutoEnableControls": True,
+            },
+        )
     if operation_name == "GetEnabledStandards":
-        return {"StandardsSubscriptions": []}
+        return mocked_response(operation_name, {"StandardsSubscriptions": []})
     if operation_name == "ListEnabledProductsForImport":
-        return {"ProductSubscriptions": []}
+        return mocked_response(operation_name, {"ProductSubscriptions": []})
     if operation_name == "ListTagsForResource":
-        return {"Tags": {}}
+        return mocked_response(operation_name, {"Tags": {}})
     return None
 
 
@@ -41,19 +51,14 @@ def mock_make_api_call_org_admin_and_config(self, operation_name, api_params):
     if hub_resp is not None:
         return hub_resp
     if operation_name == "ListOrganizationAdminAccounts":
-        return {
-            "AdminAccounts": [
-                {
-                    "AdminAccountId": "123456789012",
-                    "AdminStatus": "ENABLED",
-                }
-            ]
-        }
+        return mocked_response(
+            operation_name,
+            {"AdminAccounts": [{"AccountId": "123456789012", "Status": "ENABLED"}]},
+        )
     if operation_name == "DescribeOrganizationConfiguration":
-        return {
-            "AutoEnable": True,
-            "AutoEnableStandards": "DEFAULT",
-        }
+        return mocked_response(
+            operation_name, {"AutoEnable": True, "AutoEnableStandards": "DEFAULT"}
+        )
     return orig(self, operation_name, api_params)
 
 
@@ -63,19 +68,14 @@ def mock_make_api_call_org_admin_no_auto_enable(self, operation_name, api_params
     if hub_resp is not None:
         return hub_resp
     if operation_name == "ListOrganizationAdminAccounts":
-        return {
-            "AdminAccounts": [
-                {
-                    "AdminAccountId": "123456789012",
-                    "AdminStatus": "ENABLED",
-                }
-            ]
-        }
+        return mocked_response(
+            operation_name,
+            {"AdminAccounts": [{"AccountId": "123456789012", "Status": "ENABLED"}]},
+        )
     if operation_name == "DescribeOrganizationConfiguration":
-        return {
-            "AutoEnable": False,
-            "AutoEnableStandards": "NONE",
-        }
+        return mocked_response(
+            operation_name, {"AutoEnable": False, "AutoEnableStandards": "NONE"}
+        )
     return orig(self, operation_name, api_params)
 
 
@@ -85,12 +85,11 @@ def mock_make_api_call_no_org_admin(self, operation_name, api_params):
     if hub_resp is not None:
         return hub_resp
     if operation_name == "ListOrganizationAdminAccounts":
-        return {"AdminAccounts": []}
+        return mocked_response(operation_name, {"AdminAccounts": []})
     if operation_name == "DescribeOrganizationConfiguration":
-        return {
-            "AutoEnable": False,
-            "AutoEnableStandards": "NONE",
-        }
+        return mocked_response(
+            operation_name, {"AutoEnable": False, "AutoEnableStandards": "NONE"}
+        )
     return orig(self, operation_name, api_params)
 
 
@@ -107,7 +106,7 @@ def mock_make_api_call_securityhub_not_subscribed(self, operation_name, api_para
             operation_name,
         )
     if operation_name == "ListOrganizationAdminAccounts":
-        return {"AdminAccounts": []}
+        return mocked_response(operation_name, {"AdminAccounts": []})
     return orig(self, operation_name, api_params)
 
 
@@ -127,7 +126,9 @@ def mock_make_api_call_admin_lookup_access_denied(self, operation_name, api_para
             operation_name,
         )
     if operation_name == "DescribeOrganizationConfiguration":
-        return {"AutoEnable": True, "AutoEnableStandards": "DEFAULT"}
+        return mocked_response(
+            operation_name, {"AutoEnable": True, "AutoEnableStandards": "DEFAULT"}
+        )
     return orig(self, operation_name, api_params)
 
 
@@ -139,7 +140,9 @@ def mock_make_api_call_admin_lookup_unexpected(self, operation_name, api_params)
     if operation_name == "ListOrganizationAdminAccounts":
         raise RuntimeError("simulated transient error")
     if operation_name == "DescribeOrganizationConfiguration":
-        return {"AutoEnable": True, "AutoEnableStandards": "DEFAULT"}
+        return mocked_response(
+            operation_name, {"AutoEnable": True, "AutoEnableStandards": "DEFAULT"}
+        )
     return orig(self, operation_name, api_params)
 
 
@@ -151,11 +154,10 @@ def mock_make_api_call_describe_org_config_other_client_error(
     if hub_resp is not None:
         return hub_resp
     if operation_name == "ListOrganizationAdminAccounts":
-        return {
-            "AdminAccounts": [
-                {"AdminAccountId": "123456789012", "AdminStatus": "ENABLED"}
-            ]
-        }
+        return mocked_response(
+            operation_name,
+            {"AdminAccounts": [{"AccountId": "123456789012", "Status": "ENABLED"}]},
+        )
     if operation_name == "DescribeOrganizationConfiguration":
         raise botocore.exceptions.ClientError(
             {"Error": {"Code": "InternalServerError", "Message": "boom"}},
@@ -170,28 +172,92 @@ def mock_make_api_call_describe_org_config_unexpected(self, operation_name, api_
     if hub_resp is not None:
         return hub_resp
     if operation_name == "ListOrganizationAdminAccounts":
-        return {
-            "AdminAccounts": [
-                {"AdminAccountId": "123456789012", "AdminStatus": "ENABLED"}
-            ]
-        }
+        return mocked_response(
+            operation_name,
+            {"AdminAccounts": [{"AccountId": "123456789012", "Status": "ENABLED"}]},
+        )
     if operation_name == "DescribeOrganizationConfiguration":
         raise RuntimeError("simulated transient error")
     return orig(self, operation_name, api_params)
 
 
+def mock_make_api_call_admin_account_missing_fields(self, operation_name, api_params):
+    """AdminAccounts entry without the expected keys — must not raise."""
+    hub_resp = _active_hub_responses(operation_name)
+    if hub_resp is not None:
+        return hub_resp
+    if operation_name == "ListOrganizationAdminAccounts":
+        # Deliberately not validated against the API model: this simulates the
+        # response drifting away from what botocore currently describes.
+        return {"AdminAccounts": [{"SomethingElse": "unexpected"}]}
+    if operation_name == "DescribeOrganizationConfiguration":
+        return mocked_response(
+            operation_name, {"AutoEnable": True, "AutoEnableStandards": "DEFAULT"}
+        )
+    return orig(self, operation_name, api_params)
+
+
+def mock_make_api_call_admin_account_partially_unparseable(
+    self, operation_name, api_params
+):
+    """A valid admin account alongside an entry that cannot be parsed."""
+    hub_resp = _active_hub_responses(operation_name)
+    if hub_resp is not None:
+        return hub_resp
+    if operation_name == "ListOrganizationAdminAccounts":
+        # Deliberately not validated against the API model: the second entry
+        # simulates the response drifting away from what botocore describes.
+        return {
+            "AdminAccounts": [
+                {"AccountId": "123456789012", "Status": "ENABLED"},
+                {"SomethingElse": "unexpected"},
+            ]
+        }
+    if operation_name == "DescribeOrganizationConfiguration":
+        return mocked_response(
+            operation_name, {"AutoEnable": True, "AutoEnableStandards": "DEFAULT"}
+        )
+    return orig(self, operation_name, api_params)
+
+
+def mock_make_api_call_admin_lookup_denied_in_one_region(
+    self, operation_name, api_params
+):
+    """ListOrganizationAdminAccounts fails only in eu-west-2."""
+    hub_resp = _active_hub_responses(operation_name)
+    if hub_resp is not None:
+        if operation_name == "DescribeHub":
+            return mocked_response(
+                operation_name,
+                {
+                    **hub_resp,
+                    "HubArn": f"arn:aws:securityhub:{self.meta.region_name}:{AWS_ACCOUNT_NUMBER}:hub/default",
+                },
+            )
+        return hub_resp
+    if operation_name == "ListOrganizationAdminAccounts":
+        if self.meta.region_name == AWS_REGION_EU_WEST_2:
+            raise botocore.exceptions.ClientError(
+                {
+                    "Error": {
+                        "Code": "AccessDeniedException",
+                        "Message": "User is not authorized to perform: securityhub:ListOrganizationAdminAccounts",
+                    }
+                },
+                operation_name,
+            )
+        return mocked_response(
+            operation_name,
+            {"AdminAccounts": [{"AccountId": "123456789012", "Status": "ENABLED"}]},
+        )
+    if operation_name == "DescribeOrganizationConfiguration":
+        return mocked_response(
+            operation_name, {"AutoEnable": True, "AutoEnableStandards": "DEFAULT"}
+        )
+    return orig(self, operation_name, api_params)
+
+
 class Test_securityhub_delegated_admin_enabled_all_regions:
-    def teardown_method(self):
-        """Evict cached securityhub modules so legacy mock.patch-based tests
-        in the same session see a fresh import path."""
-        import sys
-
-        for mod in (
-            "prowler.providers.aws.services.securityhub.securityhub_client",
-            "prowler.providers.aws.services.securityhub.securityhub_delegated_admin_enabled_all_regions.securityhub_delegated_admin_enabled_all_regions",
-        ):
-            sys.modules.pop(mod, None)
-
     @patch(
         "botocore.client.BaseClient._make_api_call",
         new=mock_make_api_call_securityhub_not_subscribed,
@@ -395,9 +461,10 @@ class Test_securityhub_delegated_admin_enabled_all_regions:
                     break
 
             assert eu_west_1_result is not None
-            assert eu_west_1_result.status == "FAIL"
+            assert eu_west_1_result.status == "MANUAL"
             assert (
-                "delegated administrator status could not be determined"
+                "could not be determined" in eu_west_1_result.status_extended
+                and "management or delegated administrator account"
                 in eu_west_1_result.status_extended
             )
             assert (
@@ -419,7 +486,7 @@ class Test_securityhub_delegated_admin_enabled_all_regions:
         )
 
         service = SecurityHub(aws_provider)
-        assert service.organization_admin_lookup_failed is True
+        assert AWS_REGION_EU_WEST_1 in service.organization_admin_lookup_failed_regions
 
         with (
             patch(
@@ -436,11 +503,8 @@ class Test_securityhub_delegated_admin_enabled_all_regions:
             )
 
             result = securityhub_delegated_admin_enabled_all_regions().execute()
-            assert result and result[0].status == "FAIL"
-            assert (
-                "delegated administrator status could not be determined"
-                in result[0].status_extended
-            )
+            assert result and result[0].status == "MANUAL"
+            assert "could not be determined" in result[0].status_extended
 
     @patch(
         "botocore.client.BaseClient._make_api_call",
@@ -510,3 +574,112 @@ class Test_securityhub_delegated_admin_enabled_all_regions:
 
             result = securityhub_delegated_admin_enabled_all_regions().execute()
             assert result and result[0].status == "PASS"
+
+    @patch(
+        "botocore.client.BaseClient._make_api_call",
+        new=mock_make_api_call_admin_account_missing_fields,
+    )
+    @mock_aws
+    def test_admin_account_missing_fields(self):
+        """An unparseable admin entry is reported as unknown, not as absent."""
+        aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
+
+        from prowler.providers.aws.services.securityhub.securityhub_service import (
+            SecurityHub,
+        )
+
+        service = SecurityHub(aws_provider)
+
+        with (
+            patch(
+                "prowler.providers.common.provider.Provider.get_global_provider",
+                return_value=aws_provider,
+            ),
+            patch(
+                "prowler.providers.aws.services.securityhub.securityhub_delegated_admin_enabled_all_regions.securityhub_delegated_admin_enabled_all_regions.securityhub_client",
+                new=service,
+            ),
+        ):
+            from prowler.providers.aws.services.securityhub.securityhub_delegated_admin_enabled_all_regions.securityhub_delegated_admin_enabled_all_regions import (
+                securityhub_delegated_admin_enabled_all_regions,
+            )
+
+            result = securityhub_delegated_admin_enabled_all_regions().execute()
+            assert result and result[0].status == "MANUAL"
+            assert "could not be determined" in result[0].status_extended
+
+    @patch(
+        "botocore.client.BaseClient._make_api_call",
+        new=mock_make_api_call_admin_lookup_denied_in_one_region,
+    )
+    @mock_aws
+    def test_admin_lookup_failure_is_isolated_per_region(self):
+        """A lookup failure in one region must not degrade the other regions."""
+        aws_provider = set_mocked_aws_provider(
+            [AWS_REGION_EU_WEST_1, AWS_REGION_EU_WEST_2]
+        )
+
+        from prowler.providers.aws.services.securityhub.securityhub_service import (
+            SecurityHub,
+        )
+
+        service = SecurityHub(aws_provider)
+
+        with (
+            patch(
+                "prowler.providers.common.provider.Provider.get_global_provider",
+                return_value=aws_provider,
+            ),
+            patch(
+                "prowler.providers.aws.services.securityhub.securityhub_delegated_admin_enabled_all_regions.securityhub_delegated_admin_enabled_all_regions.securityhub_client",
+                new=service,
+            ),
+        ):
+            from prowler.providers.aws.services.securityhub.securityhub_delegated_admin_enabled_all_regions.securityhub_delegated_admin_enabled_all_regions import (
+                securityhub_delegated_admin_enabled_all_regions,
+            )
+
+            results = {
+                finding.region: finding
+                for finding in securityhub_delegated_admin_enabled_all_regions().execute()
+            }
+
+            assert results[AWS_REGION_EU_WEST_1].status == "PASS"
+            assert results[AWS_REGION_EU_WEST_2].status == "MANUAL"
+            assert (
+                "could not be determined"
+                in results[AWS_REGION_EU_WEST_2].status_extended
+            )
+
+    @patch(
+        "botocore.client.BaseClient._make_api_call",
+        new=mock_make_api_call_admin_account_partially_unparseable,
+    )
+    @mock_aws
+    def test_admin_account_partially_unparseable(self):
+        """A known delegated admin is not downgraded to unknown by a partial failure."""
+        aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
+
+        from prowler.providers.aws.services.securityhub.securityhub_service import (
+            SecurityHub,
+        )
+
+        service = SecurityHub(aws_provider)
+
+        with (
+            patch(
+                "prowler.providers.common.provider.Provider.get_global_provider",
+                return_value=aws_provider,
+            ),
+            patch(
+                "prowler.providers.aws.services.securityhub.securityhub_delegated_admin_enabled_all_regions.securityhub_delegated_admin_enabled_all_regions.securityhub_client",
+                new=service,
+            ),
+        ):
+            from prowler.providers.aws.services.securityhub.securityhub_delegated_admin_enabled_all_regions.securityhub_delegated_admin_enabled_all_regions import (
+                securityhub_delegated_admin_enabled_all_regions,
+            )
+
+            result = securityhub_delegated_admin_enabled_all_regions().execute()
+            assert result and result[0].status == "PASS"
+            assert "could not be determined" not in result[0].status_extended

tests/providers/aws/services/securityhub/securityhub_enabled/securityhub_enabled_fixer_test.py

@@ -1,34 +1,80 @@
 from unittest import mock
 
-from moto import mock_aws
-
-from prowler.providers.aws.services.securityhub.securityhub_service import (
-    SecurityHubHub,
-)
 from tests.providers.aws.utils import AWS_REGION_EU_WEST_1
 
+# Patching the fixer client imports securityhub_client, which instantiates
+# SecurityHub against the global provider at module level. Stubbing the class
+# first keeps that import from reaching AWS when this file runs on its own.
+SERVICE_MODULE = "prowler.providers.aws.services.securityhub.securityhub_service"
+FIXER_MODULE = "prowler.providers.aws.services.securityhub.securityhub_enabled.securityhub_enabled_fixer"
 
-class test_securityhub_enabled_fixer:
-    @mock_aws
+
+def _mocked_securityhub_client(fixer_config: dict) -> tuple:
+    """Build a Security Hub client mock with a single regional client."""
+    regional_client = mock.MagicMock()
+    securityhub_client = mock.MagicMock()
+    securityhub_client.fixer_config = fixer_config
+    securityhub_client.regional_clients = {AWS_REGION_EU_WEST_1: regional_client}
+    return securityhub_client, regional_client
+
+
+class Test_securityhub_enabled_fixer:
     def test_securityhub_enabled_fixer(self):
-        securityhub_client = mock.MagicMock
-        securityhub_client.securityhubs = [
-            SecurityHubHub(
-                arn="arn:aws:securityhub:us-east-1:0123456789012:hub/default",
-                id="default",
-                status="ACTIVE",
-                standards="cis-aws-foundations-benchmark/v/1.2.0",
-                integrations="",
-                region="eu-west-1",
-            )
-        ]
-        with mock.patch(
-            "prowler.providers.aws.services.securityhub.securityhub_service.SecurityHub",
-            new=securityhub_client,
+        """Security Hub is enabled with the default standards from the fixer config."""
+        securityhub_client, regional_client = _mocked_securityhub_client(
+            {"securityhub_enabled": {"EnableDefaultStandards": True}}
+        )
+
+        with (
+            mock.patch(f"{SERVICE_MODULE}.SecurityHub", new=mock.MagicMock()),
+            mock.patch(f"{FIXER_MODULE}.securityhub_client", new=securityhub_client),
         ):
-            # Test Check
+            # Test Fixer
             from prowler.providers.aws.services.securityhub.securityhub_enabled.securityhub_enabled_fixer import (
                 fixer,
             )
 
             assert fixer(AWS_REGION_EU_WEST_1)
+
+        regional_client.enable_security_hub.assert_called_once_with(
+            EnableDefaultStandards=True
+        )
+
+    def test_securityhub_enabled_fixer_default_standards_disabled(self):
+        """EnableDefaultStandards must be taken from the fixer configuration."""
+        securityhub_client, regional_client = _mocked_securityhub_client(
+            {"securityhub_enabled": {"EnableDefaultStandards": False}}
+        )
+
+        with (
+            mock.patch(f"{SERVICE_MODULE}.SecurityHub", new=mock.MagicMock()),
+            mock.patch(f"{FIXER_MODULE}.securityhub_client", new=securityhub_client),
+        ):
+            # Test Fixer
+            from prowler.providers.aws.services.securityhub.securityhub_enabled.securityhub_enabled_fixer import (
+                fixer,
+            )
+
+            assert fixer(AWS_REGION_EU_WEST_1)
+
+        regional_client.enable_security_hub.assert_called_once_with(
+            EnableDefaultStandards=False
+        )
+
+    def test_securityhub_enabled_fixer_error(self):
+        """A failing EnableSecurityHub call must return False instead of raising."""
+        securityhub_client, regional_client = _mocked_securityhub_client({})
+        regional_client.enable_security_hub.side_effect = Exception(
+            "AccessDeniedException"
+        )
+
+        with (
+            mock.patch(f"{SERVICE_MODULE}.SecurityHub", new=mock.MagicMock()),
+            mock.patch(f"{FIXER_MODULE}.securityhub_client", new=securityhub_client),
+        ):
+            # Test Fixer
+            from prowler.providers.aws.services.securityhub.securityhub_enabled.securityhub_enabled_fixer import (
+                fixer,
+            )
+
+            assert not fixer(AWS_REGION_EU_WEST_1)

tests/providers/aws/services/securityhub/securityhub_enabled/securityhub_enabled_test.py

@@ -3,36 +3,52 @@ from unittest import mock
 from prowler.providers.aws.services.securityhub.securityhub_service import (
     SecurityHubHub,
 )
-from tests.providers.aws.utils import AWS_REGION_EU_WEST_1
+from tests.providers.aws.utils import AWS_ACCOUNT_NUMBER, AWS_REGION_EU_WEST_1
+
+# Patching the check client imports securityhub_client, which instantiates
+# SecurityHub against the global provider at module level. Stubbing the class
+# first keeps that import from reaching AWS when this file runs on its own.
+SERVICE_MODULE = "prowler.providers.aws.services.securityhub.securityhub_service"
+CHECK_MODULE = (
+    "prowler.providers.aws.services.securityhub.securityhub_enabled.securityhub_enabled"
+)
+
+HUB_ARN = f"arn:aws:securityhub:{AWS_REGION_EU_WEST_1}:{AWS_ACCOUNT_NUMBER}:hub/default"
+UNKNOWN_HUB_ARN = (
+    f"arn:aws:securityhub:{AWS_REGION_EU_WEST_1}:{AWS_ACCOUNT_NUMBER}:hub/unknown"
+)
+
+
+def _mocked_securityhub_client(securityhubs: list, audit_config: dict = None):
+    """Build a Security Hub client mock holding the given hubs."""
+    securityhub_client = mock.MagicMock()
+    securityhub_client.region = AWS_REGION_EU_WEST_1
+    securityhub_client.audited_partition = "aws"
+    securityhub_client.audited_account = AWS_ACCOUNT_NUMBER
+    securityhub_client.audit_config = audit_config if audit_config is not None else {}
+    securityhub_client.securityhubs = securityhubs
+    return securityhub_client
 
 
 class Test_securityhub_enabled:
     def test_securityhub_hub_inactive(self):
-        securityhub_client = mock.MagicMock
-        securityhub_client.region = AWS_REGION_EU_WEST_1
-        securityhub_client.get_unknown_arn = (
-            lambda x: f"arn:aws:securityhub:{x}:0123456789012:hub/unknown"
+        securityhub_client = _mocked_securityhub_client(
+            [
+                SecurityHubHub(
+                    arn=UNKNOWN_HUB_ARN,
+                    id="hub/unknown",
+                    status="NOT_AVAILABLE",
+                    standards="",
+                    integrations="",
+                    region=AWS_REGION_EU_WEST_1,
+                    tags=[{"test_key": "test_value"}],
+                )
+            ]
         )
-        securityhub_client.securityhubs = [
-            SecurityHubHub(
-                arn=f"arn:aws:securityhub:{AWS_REGION_EU_WEST_1}:0123456789012:hub/unknown",
-                id="hub/unknown",
-                status="NOT_AVAILABLE",
-                standards="",
-                integrations="",
-                region=AWS_REGION_EU_WEST_1,
-                tags=[{"test_key": "test_value"}],
-            )
-        ]
+
         with (
-            mock.patch(
-                "prowler.providers.aws.services.securityhub.securityhub_service.SecurityHub",
-                new=securityhub_client,
-            ),
-            mock.patch(
-                "prowler.providers.aws.services.securityhub.securityhub_service.SecurityHub.get_unknown_arn",
-                return_value="arn:aws:securityhub:eu-west-1:0123456789012:hub/unknown",
-            ),
+            mock.patch(f"{SERVICE_MODULE}.SecurityHub", new=mock.MagicMock()),
+            mock.patch(f"{CHECK_MODULE}.securityhub_client", new=securityhub_client),
         ):
             # Test Check
             from prowler.providers.aws.services.securityhub.securityhub_enabled.securityhub_enabled import (
@@ -42,32 +58,32 @@ class Test_securityhub_enabled:
             check = securityhub_enabled()
             result = check.execute()
 
+            assert len(result) == 1
             assert result[0].status == "FAIL"
             assert result[0].status_extended == "Security Hub is not enabled."
             assert result[0].resource_id == "hub/unknown"
-            assert (
-                result[0].resource_arn
-                == "arn:aws:securityhub:eu-west-1:0123456789012:hub/unknown"
-            )
+            assert result[0].resource_arn == UNKNOWN_HUB_ARN
             assert result[0].region == AWS_REGION_EU_WEST_1
             assert result[0].resource_tags == [{"test_key": "test_value"}]
 
     def test_securityhub_hub_active_with_standards(self):
-        securityhub_client = mock.MagicMock
-        securityhub_client.securityhubs = [
-            SecurityHubHub(
-                arn="arn:aws:securityhub:us-east-1:0123456789012:hub/default",
-                id="default",
-                status="ACTIVE",
-                standards="cis-aws-foundations-benchmark/v/1.2.0",
-                integrations="",
-                region=AWS_REGION_EU_WEST_1,
-                tags=[{"test_key": "test_value"}],
-            )
-        ]
-        with mock.patch(
-            "prowler.providers.aws.services.securityhub.securityhub_service.SecurityHub",
-            new=securityhub_client,
+        securityhub_client = _mocked_securityhub_client(
+            [
+                SecurityHubHub(
+                    arn=HUB_ARN,
+                    id="default",
+                    status="ACTIVE",
+                    standards="cis-aws-foundations-benchmark/v/1.2.0",
+                    integrations="",
+                    region=AWS_REGION_EU_WEST_1,
+                    tags=[{"test_key": "test_value"}],
+                )
+            ]
+        )
+
+        with (
+            mock.patch(f"{SERVICE_MODULE}.SecurityHub", new=mock.MagicMock()),
+            mock.patch(f"{CHECK_MODULE}.securityhub_client", new=securityhub_client),
         ):
             # Test Check
             from prowler.providers.aws.services.securityhub.securityhub_enabled.securityhub_enabled import (
@@ -77,35 +93,35 @@ class Test_securityhub_enabled:
             check = securityhub_enabled()
             result = check.execute()
 
+            assert len(result) == 1
             assert result[0].status == "PASS"
             assert (
                 result[0].status_extended
                 == "Security Hub is enabled with standards: cis-aws-foundations-benchmark/v/1.2.0."
             )
             assert result[0].resource_id == "default"
-            assert (
-                result[0].resource_arn
-                == "arn:aws:securityhub:us-east-1:0123456789012:hub/default"
-            )
+            assert result[0].resource_arn == HUB_ARN
             assert result[0].region == AWS_REGION_EU_WEST_1
             assert result[0].resource_tags == [{"test_key": "test_value"}]
 
     def test_securityhub_hub_active_with_integrations(self):
-        securityhub_client = mock.MagicMock
-        securityhub_client.securityhubs = [
-            SecurityHubHub(
-                arn="arn:aws:securityhub:us-east-1:0123456789012:hub/default",
-                id="default",
-                status="ACTIVE",
-                standards="",
-                integrations="prowler",
-                region=AWS_REGION_EU_WEST_1,
-                tags=[{"test_key": "test_value"}],
-            )
-        ]
-        with mock.patch(
-            "prowler.providers.aws.services.securityhub.securityhub_service.SecurityHub",
-            new=securityhub_client,
+        securityhub_client = _mocked_securityhub_client(
+            [
+                SecurityHubHub(
+                    arn=HUB_ARN,
+                    id="default",
+                    status="ACTIVE",
+                    standards="",
+                    integrations="prowler",
+                    region=AWS_REGION_EU_WEST_1,
+                    tags=[{"test_key": "test_value"}],
+                )
+            ]
+        )
+
+        with (
+            mock.patch(f"{SERVICE_MODULE}.SecurityHub", new=mock.MagicMock()),
+            mock.patch(f"{CHECK_MODULE}.securityhub_client", new=securityhub_client),
         ):
             # Test Check
             from prowler.providers.aws.services.securityhub.securityhub_enabled.securityhub_enabled import (
@@ -115,38 +131,35 @@ class Test_securityhub_enabled:
             check = securityhub_enabled()
             result = check.execute()
 
+            assert len(result) == 1
             assert result[0].status == "PASS"
             assert (
                 result[0].status_extended
                 == "Security Hub is enabled without standards but with integrations: prowler."
             )
             assert result[0].resource_id == "default"
-            assert (
-                result[0].resource_arn
-                == "arn:aws:securityhub:us-east-1:0123456789012:hub/default"
-            )
+            assert result[0].resource_arn == HUB_ARN
             assert result[0].region == AWS_REGION_EU_WEST_1
             assert result[0].resource_tags == [{"test_key": "test_value"}]
 
     def test_securityhub_hub_active_without_integrations_or_standards(self):
-        securityhub_client = mock.MagicMock
-        securityhub_client.region = AWS_REGION_EU_WEST_1
-        securityhub_client.audited_partition = "aws"
-        securityhub_client.audited_account = "0123456789012"
-        securityhub_client.securityhubs = [
-            SecurityHubHub(
-                arn="arn:aws:securityhub:us-east-1:0123456789012:hub/default",
-                id="default",
-                status="ACTIVE",
-                standards="",
-                integrations="",
-                region=AWS_REGION_EU_WEST_1,
-                tags=[{"test_key": "test_value"}],
-            )
-        ]
-        with mock.patch(
-            "prowler.providers.aws.services.securityhub.securityhub_service.SecurityHub",
-            new=securityhub_client,
+        securityhub_client = _mocked_securityhub_client(
+            [
+                SecurityHubHub(
+                    arn=HUB_ARN,
+                    id="default",
+                    status="ACTIVE",
+                    standards="",
+                    integrations="",
+                    region=AWS_REGION_EU_WEST_1,
+                    tags=[{"test_key": "test_value"}],
+                )
+            ]
+        )
+
+        with (
+            mock.patch(f"{SERVICE_MODULE}.SecurityHub", new=mock.MagicMock()),
+            mock.patch(f"{CHECK_MODULE}.securityhub_client", new=securityhub_client),
         ):
             # Test Check
             from prowler.providers.aws.services.securityhub.securityhub_enabled.securityhub_enabled import (
@@ -156,39 +169,37 @@ class Test_securityhub_enabled:
             check = securityhub_enabled()
             result = check.execute()
 
+            assert len(result) == 1
             assert result[0].status == "FAIL"
+            assert not result[0].muted
             assert (
                 result[0].status_extended
                 == "Security Hub is enabled but without any standard or integration."
             )
             assert result[0].resource_id == "default"
-            assert (
-                result[0].resource_arn
-                == "arn:aws:securityhub:us-east-1:0123456789012:hub/default"
-            )
+            assert result[0].resource_arn == HUB_ARN
             assert result[0].region == AWS_REGION_EU_WEST_1
             assert result[0].resource_tags == [{"test_key": "test_value"}]
 
     def test_securityhub_hub_active_without_integrations_or_standards_muted(self):
-        securityhub_client = mock.MagicMock
-        securityhub_client.audit_config = {"mute_non_default_regions": True}
-        securityhub_client.region = AWS_REGION_EU_WEST_1
-        securityhub_client.audited_partition = "aws"
-        securityhub_client.audited_account = "0123456789012"
-        securityhub_client.securityhubs = [
-            SecurityHubHub(
-                arn="arn:aws:securityhub:us-east-1:0123456789012:hub/default",
-                id="default",
-                status="ACTIVE",
-                standards="",
-                integrations="",
-                region="eu-south-2",
-                tags=[],
-            )
-        ]
-        with mock.patch(
-            "prowler.providers.aws.services.securityhub.securityhub_service.SecurityHub",
-            new=securityhub_client,
+        securityhub_client = _mocked_securityhub_client(
+            [
+                SecurityHubHub(
+                    arn=HUB_ARN,
+                    id="default",
+                    status="ACTIVE",
+                    standards="",
+                    integrations="",
+                    region="eu-south-2",
+                    tags=[],
+                )
+            ],
+            audit_config={"mute_non_default_regions": True},
+        )
+
+        with (
+            mock.patch(f"{SERVICE_MODULE}.SecurityHub", new=mock.MagicMock()),
+            mock.patch(f"{CHECK_MODULE}.securityhub_client", new=securityhub_client),
         ):
             # Test Check
             from prowler.providers.aws.services.securityhub.securityhub_enabled.securityhub_enabled import (
@@ -198,6 +209,7 @@ class Test_securityhub_enabled:
             check = securityhub_enabled()
             result = check.execute()
 
+            assert len(result) == 1
             assert result[0].status == "FAIL"
             assert result[0].muted
             assert (
@@ -205,9 +217,6 @@ class Test_securityhub_enabled:
                 == "Security Hub is enabled but without any standard or integration."
             )
             assert result[0].resource_id == "default"
-            assert (
-                result[0].resource_arn
-                == "arn:aws:securityhub:us-east-1:0123456789012:hub/default"
-            )
+            assert result[0].resource_arn == HUB_ARN
             assert result[0].region == "eu-south-2"
             assert result[0].resource_tags == []

tests/providers/aws/services/securityhub/securityhub_service_test.py

@@ -3,7 +3,12 @@ from unittest.mock import patch
 import botocore
 
 from prowler.providers.aws.services.securityhub.securityhub_service import SecurityHub
-from tests.providers.aws.utils import AWS_REGION_EU_WEST_1, set_mocked_aws_provider
+from tests.providers.aws.utils import (
+    AWS_ACCOUNT_NUMBER,
+    AWS_REGION_EU_WEST_1,
+    mocked_api_response,
+    set_mocked_aws_provider,
+)
 
 # Mocking Access Analyzer Calls
 make_api_call = botocore.client.BaseClient._make_api_call
@@ -41,10 +46,44 @@ def mock_make_api_call(self, operation_name, kwarg):
         return {
             "Tags": {"test_key": "test_value"},
         }
+    if operation_name == "ListOrganizationAdminAccounts":
+        # Security Hub returns AccountId/Status, unlike GuardDuty's
+        # AdminAccountId/AdminStatus for the same operation name.
+        return mocked_api_response(
+            "securityhub",
+            operation_name,
+            {"AdminAccounts": [{"AccountId": AWS_ACCOUNT_NUMBER, "Status": "ENABLED"}]},
+        )
 
     return make_api_call(self, operation_name, kwarg)
 
 
+def mock_make_api_call_admin_account_missing_fields(self, operation_name, kwarg):
+    """Return an admin account entry without the documented fields."""
+    if operation_name == "ListOrganizationAdminAccounts":
+        # Deliberately not validated against the API model: this simulates the
+        # response drifting away from what botocore currently describes.
+        return {"AdminAccounts": [{"SomethingElse": "unexpected"}]}
+
+    return mock_make_api_call(self, operation_name, kwarg)
+
+
+def mock_make_api_call_admin_account_access_denied(self, operation_name, kwarg):
+    """Deny ListOrganizationAdminAccounts, as AWS does outside the management account."""
+    if operation_name == "ListOrganizationAdminAccounts":
+        raise botocore.exceptions.ClientError(
+            {
+                "Error": {
+                    "Code": "AccessDeniedException",
+                    "Message": "User is not authorized to perform: securityhub:ListOrganizationAdminAccounts",
+                }
+            },
+            operation_name,
+        )
+
+    return mock_make_api_call(self, operation_name, kwarg)
+
+
 # Mock generate_regional_clients()
 def mock_generate_regional_clients(provider, service):
     regional_client = provider._session.current_session.client(
@@ -91,3 +130,46 @@ class Test_SecurityHub_Service:
         securityhub = SecurityHub(set_mocked_aws_provider([AWS_REGION_EU_WEST_1]))
         assert len(securityhub.securityhubs) == 1
         assert securityhub.securityhubs[0].tags == [{"test_key": "test_value"}]
+
+    def test_list_organization_admin_accounts(self):
+        """Security Hub returns AccountId/Status, not GuardDuty's AdminAccountId/AdminStatus."""
+        securityhub = SecurityHub(set_mocked_aws_provider([AWS_REGION_EU_WEST_1]))
+
+        assert securityhub.organization_admin_lookup_failed_regions == set()
+        assert len(securityhub.organization_admin_accounts) == 1
+        assert (
+            securityhub.organization_admin_accounts[0].admin_account_id
+            == AWS_ACCOUNT_NUMBER
+        )
+        assert securityhub.organization_admin_accounts[0].admin_status == "ENABLED"
+        assert securityhub.organization_admin_accounts[0].region == AWS_REGION_EU_WEST_1
+
+    def test_list_organization_admin_accounts_missing_fields(self):
+        """An unparseable entry marks the region as unknown instead of raising."""
+        aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
+
+        with patch(
+            "botocore.client.BaseClient._make_api_call",
+            new=mock_make_api_call_admin_account_missing_fields,
+        ):
+            securityhub = SecurityHub(aws_provider)
+
+        assert securityhub.organization_admin_accounts == []
+        assert securityhub.organization_admin_lookup_failed_regions == {
+            AWS_REGION_EU_WEST_1
+        }
+
+    def test_list_organization_admin_accounts_access_denied(self):
+        """A denied lookup only marks its own region as unknown."""
+        aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
+
+        with patch(
+            "botocore.client.BaseClient._make_api_call",
+            new=mock_make_api_call_admin_account_access_denied,
+        ):
+            securityhub = SecurityHub(aws_provider)
+
+        assert securityhub.organization_admin_accounts == []
+        assert securityhub.organization_admin_lookup_failed_regions == {
+            AWS_REGION_EU_WEST_1
+        }

tests/providers/aws/utils.py

@@ -1,7 +1,10 @@
 from argparse import Namespace
+from functools import lru_cache
 from json import dumps
 
+import botocore.session
 from boto3 import client, session
+from botocore.validate import ParamValidator
 from moto import mock_aws
 
 from prowler.config.config import (
@@ -235,3 +238,38 @@ def create_role(
         PolicyArn=policy["Arn"],
     )
     return administrator_role["Arn"]
+
+
+@lru_cache(maxsize=None)
+def _service_model(service_name: str):
+    return botocore.session.get_session().get_service_model(service_name)
+
+
+def mocked_api_response(service_name: str, operation_name: str, response: dict) -> dict:
+    """Validate a hand-written mocked response against the real AWS API model.
+
+    Responses returned from a `botocore.client.BaseClient._make_api_call` mock are
+    not validated by botocore, so a mock can return fields that the API never
+    sends and the test will still pass. Wrapping the response with this helper
+    turns that silent mismatch into a test failure.
+
+    Args:
+        service_name: Boto3 service name, e.g. `securityhub`.
+        operation_name: API operation name in PascalCase, e.g. `DescribeHub`.
+        response: The mocked response to validate and return.
+
+    Returns:
+        The response, unchanged.
+
+    Raises:
+        AssertionError: If the response does not match the operation output shape.
+    """
+    output_shape = (
+        _service_model(service_name).operation_model(operation_name).output_shape
+    )
+    report = ParamValidator().validate(response, output_shape)
+    assert not report.has_errors(), (
+        f"Mocked {service_name}:{operation_name} response does not match the API "
+        f"model: {report.generate_report()}"
+    )
+    return response