fix(aws): query Organizations in the audited partition (#12988)

prowler/changelog.d/aws-organizations-partition-region.fixed.md

@@ -0,0 +1 @@
+AWS Organizations metadata is fetched from the partition of the audited account, using the region that answered the STS bootstrap, instead of the commercial `us-east-1` endpoint a session without a region resolves to, which timed out in GovCloud

prowler/providers/aws/aws_provider.py

@@ -390,7 +390,7 @@ class AwsProvider(Provider):
             )
 
         self._organizations_metadata = self.get_organizations_info(
-            aws_organizations_session, self._identity.account
+            aws_organizations_session, self._identity.account, sts_region
         )
 
         # Fallback to the original (pre-assume) session when no explicit
@@ -412,7 +412,7 @@ class AwsProvider(Provider):
                 "Retrying AWS Organizations metadata retrieval with the original session"
             )
             self._organizations_metadata = self.get_organizations_info(
-                self._session.original_session, self._identity.account
+                self._session.original_session, self._identity.account, sts_region
             )
         ########
 
@@ -508,7 +508,10 @@ class AwsProvider(Provider):
 
     # TODO: This can be moved to another class since it doesn't need self
     def get_organizations_info(
-        self, organizations_session: Session, aws_account_id: str
+        self,
+        organizations_session: Session,
+        aws_account_id: str,
+        region: str | None = None,
     ) -> AWSOrganizationsInfo:
         """
         get_organizations_info returns a AWSOrganizationsInfo object if the account to be audited is a delegated administrator for AWS Organizations or if the AWS Organizations Role ARN (--organizations-role) is passed.
@@ -516,6 +519,7 @@ class AwsProvider(Provider):
         Args:
         - organizations_session: needs to be a Session object with permissions to do organizations:DescribeAccount and organizations:ListTagsForResource.
         - aws_account_id: is the AWS Account ID from which we want to get the AWS Organizations account metadata
+        - region: a region of the audited partition, so the client reaches that partition's Organizations endpoint
 
         Returns:
         - AWSOrganizationsInfo object with the AWS Organizations metadata for the account to be audited.
@@ -529,6 +533,7 @@ class AwsProvider(Provider):
                 get_organizations_metadata(
                     aws_account_id=aws_account_id,
                     session=organizations_session,
+                    region=region,
                 )
             )
 

prowler/providers/aws/lib/organizations/organizations.py

@@ -42,9 +42,11 @@ def _get_ou_metadata(organizations_client, account_id):
 def get_organizations_metadata(
     aws_account_id: str,
     session: session.Session,
+    region: str | None = None,
 ) -> tuple[dict, dict, dict]:
     try:
-        organizations_client = session.client("organizations")
+        # Without a region botocore resolves the commercial endpoint, whatever the partition
+        organizations_client = session.client("organizations", region_name=region)
 
         organizations_metadata = organizations_client.describe_account(
             AccountId=aws_account_id

tests/providers/aws/aws_provider_test.py

@@ -490,10 +490,10 @@ class TestAWSProvider:
         captured_sessions = []
         original_get_organizations_info = AwsProvider.get_organizations_info
 
-        def capture(self, organizations_session, aws_account_id):
+        def capture(self, organizations_session, aws_account_id, region=None):
             captured_sessions.append(organizations_session)
             return original_get_organizations_info(
-                self, organizations_session, aws_account_id
+                self, organizations_session, aws_account_id, region
             )
 
         with patch.object(AwsProvider, "get_organizations_info", capture):
@@ -513,10 +513,10 @@ class TestAWSProvider:
         captured_sessions = []
         original_get_organizations_info = AwsProvider.get_organizations_info
 
-        def capture(self, organizations_session, aws_account_id):
+        def capture(self, organizations_session, aws_account_id, region=None):
             captured_sessions.append(organizations_session)
             return original_get_organizations_info(
-                self, organizations_session, aws_account_id
+                self, organizations_session, aws_account_id, region
             )
 
         with patch.object(AwsProvider, "get_organizations_info", capture):
@@ -2132,6 +2132,40 @@ aws:
 
         assert sts_regions == [AWS_REGION_GOV_CLOUD_US_WEST_1]
 
+    @mock_aws
+    def test_aws_provider_queries_organizations_where_validation_got_an_answer(
+        self, monkeypatch
+    ):
+        monkeypatch.setenv("PROWLER_AWS_PARTITION", AWS_GOV_CLOUD_PARTITION)
+        # No region in the session: Organizations would resolve to us-east-1
+        monkeypatch.delenv("AWS_DEFAULT_REGION", raising=False)
+        monkeypatch.delenv("AWS_REGION", raising=False)
+        answered = AWSCallerIdentity(
+            user_id="test-user-id",
+            account=AWS_ACCOUNT_NUMBER,
+            arn=ARN(AWS_GOV_CLOUD_ACCOUNT_ARN),
+            region=AWS_REGION_GOV_CLOUD_US_WEST_1,
+        )
+        organizations_regions = []
+
+        def get_organizations_metadata(aws_account_id, session, region=None):
+            organizations_regions.append(region)
+            return {}, {}, {}
+
+        with (
+            patch(
+                "prowler.providers.aws.aws_provider.AwsProvider.validate_credentials",
+                return_value=answered,
+            ),
+            patch(
+                "prowler.providers.aws.aws_provider.get_organizations_metadata",
+                side_effect=get_organizations_metadata,
+            ),
+        ):
+            AwsProvider()
+
+        assert organizations_regions == [AWS_REGION_GOV_CLOUD_US_WEST_1]
+
     @mock_aws
     def test_test_connection_with_env_credentials(self, monkeypatch):
         # Create a mock IAM user

tests/providers/aws/lib/organizations/organizations_test.py

@@ -238,6 +238,17 @@ class Test_AWS_Organizations:
         assert default_config is not None
         assert BOTO3_USER_AGENT_EXTRA in default_config.user_agent_extra
 
+    def test_get_organizations_metadata_uses_the_given_region(self):
+        # A session with no region resolves Organizations to the commercial
+        # global endpoint, unreachable from GovCloud
+        wrapper = MagicMock(wraps=boto3.Session())
+
+        get_organizations_metadata("123456789012", wrapper, region="us-gov-west-1")
+
+        wrapper.client.assert_called_once_with(
+            "organizations", region_name="us-gov-west-1"
+        )
+
     def test_parse_organizations_metadata_with_empty_ou_metadata(self):
         tags = {"Tags": []}
         metadata = {