fix(alibabacloud): retry STS connection failures (#12353)

tests/providers/alibabacloud/alibabacloud_provider_test.py

@@ -1,10 +1,16 @@
 import os
-from unittest.mock import MagicMock, patch
+from unittest.mock import MagicMock, call, patch
 
 import pytest
+from alibabacloud_tea_openapi.exceptions import ClientException
+from darabonba.core import DaraCore
+from darabonba.exceptions import RetryError, UnretryableException
+from Tea.exceptions import UnretryableException as TeaUnretryableException
+from Tea.response import TeaResponse
 
 from prowler.providers.alibabacloud.alibabacloud_provider import AlibabacloudProvider
 from prowler.providers.alibabacloud.exceptions.exceptions import (
+    AlibabaCloudConnectionError,
     AlibabaCloudInvalidCredentialsError,
     AlibabaCloudSetUpSessionError,
 )
@@ -12,9 +18,243 @@ from prowler.providers.alibabacloud.models import AlibabaCloudCallerIdentity
 from prowler.providers.common.models import Connection
 
 
+def test_validate_credentials_non_authentication_api_error_is_not_invalid_credentials():
+    """Test non-authentication STS API errors are not relabeled as credentials."""
+    session = MagicMock()
+    session.get_credentials.return_value = MagicMock(
+        access_key_id="LTAI1234567890",
+        access_key_secret="test-secret-key",
+        security_token=None,
+    )
+    api_error = ClientException(
+        code="Forbidden",
+        message="The caller is not authorized",
+        status_code=403,
+    )
+    wrapped_api_error = TeaUnretryableException(request=None, ex=api_error)
+
+    with (
+        patch.object(DaraCore, "do_action", side_effect=wrapped_api_error),
+        patch.object(DaraCore, "sleep") as sleep,
+    ):
+        with pytest.raises(UnretryableException) as exception:
+            AlibabacloudProvider.validate_credentials(session)
+
+    assert not isinstance(exception.value, AlibabaCloudInvalidCredentialsError)
+    assert exception.value.inner_exception is wrapped_api_error
+    assert exception.value.inner_exception.inner_exception is api_error
+    sleep.assert_not_called()
+
+
+def test_validate_credentials_retries_transport_failure_then_succeeds():
+    """Test STS caller identity retries a transient transport failure."""
+    session = MagicMock()
+    session.get_credentials.return_value = MagicMock(
+        access_key_id="LTAI1234567890",
+        access_key_secret="test-secret-key",
+        security_token=None,
+    )
+    response = TeaResponse()
+    response.status_code = 200
+    response.headers = {"content-type": "application/json"}
+    response.body = (
+        b'{"AccountId":"1234567890","PrincipalId":"123456",'
+        b'"Arn":"acs:ram::1234567890:user/test-user",'
+        b'"IdentityType":"RamUser"}'
+    )
+
+    with (
+        patch.object(
+            DaraCore,
+            "do_action",
+            side_effect=[RetryError("connection reset"), response],
+        ) as do_action,
+        patch.object(DaraCore, "sleep") as sleep,
+    ):
+        caller_identity = AlibabacloudProvider.validate_credentials(session)
+
+    assert caller_identity.account_id == "1234567890"
+    assert do_action.call_count == 2
+    sleep.assert_called_once_with(1000)
+
+
+def test_validate_credentials_connection_failure_is_not_invalid_credentials():
+    """Test exhausted STS transport retries raise a connection-specific error."""
+    session = MagicMock()
+    session.get_credentials.return_value = MagicMock(
+        access_key_id="LTAI1234567890",
+        access_key_secret="test-secret-key",
+        security_token=None,
+    )
+    retry_errors = []
+    for _ in range(3):
+        connection_reset = ConnectionResetError(104, "Connection reset by peer")
+        retry_error = RetryError(str(connection_reset))
+        retry_error.__cause__ = connection_reset
+        retry_errors.append(retry_error)
+
+    with (
+        patch.object(DaraCore, "do_action", side_effect=retry_errors) as do_action,
+        patch.object(DaraCore, "sleep") as sleep,
+    ):
+        with pytest.raises(AlibabaCloudConnectionError) as exception:
+            AlibabacloudProvider.validate_credentials(session)
+
+    assert not isinstance(exception.value, AlibabaCloudInvalidCredentialsError)
+    assert do_action.call_count == 3
+    assert sleep.call_args_list == [call(1000), call(1000)]
+    assert isinstance(exception.value.original_exception, UnretryableException)
+    assert exception.value.original_exception.inner_exception is retry_errors[-1]
+    assert exception.value.__cause__ is exception.value.original_exception
+
+
+def test_validate_credentials_genuine_invalid_credentials():
+    """Test an explicit STS authentication failure remains a credentials error."""
+    session = MagicMock()
+    session.get_credentials.return_value = MagicMock(
+        access_key_id="LTAI-invalid",
+        access_key_secret="invalid-secret",
+        security_token=None,
+    )
+    authentication_error = ClientException(
+        code="InvalidAccessKeyId.NotFound",
+        message="The AccessKey ID does not exist",
+        status_code=400,
+    )
+    wrapped_authentication_error = TeaUnretryableException(
+        request=None, ex=authentication_error
+    )
+
+    with (
+        patch.object(DaraCore, "do_action", side_effect=wrapped_authentication_error),
+        patch.object(DaraCore, "sleep") as sleep,
+    ):
+        with pytest.raises(AlibabaCloudInvalidCredentialsError) as exception:
+            AlibabacloudProvider.validate_credentials(session)
+
+    assert isinstance(exception.value.original_exception, UnretryableException)
+    assert (
+        exception.value.original_exception.inner_exception
+        is wrapped_authentication_error
+    )
+    assert (
+        exception.value.original_exception.inner_exception.inner_exception
+        is authentication_error
+    )
+    assert exception.value.__cause__ is exception.value.original_exception
+    sleep.assert_not_called()
+
+
+def test_validate_credentials_authentication_error_wins_over_transport_error():
+    """Test a definitive nested authentication error takes precedence over transport."""
+    session = MagicMock()
+    session.get_credentials.return_value = MagicMock(
+        access_key_id="LTAI-invalid",
+        access_key_secret="invalid-secret",
+        security_token=None,
+    )
+    authentication_error = ClientException(
+        code="InvalidAccessKeyId.NotFound",
+        message="The AccessKey ID does not exist",
+        status_code=400,
+    )
+    retry_errors = []
+    for _ in range(3):
+        retry_error = RetryError("connection reset")
+        retry_error.__cause__ = authentication_error
+        retry_errors.append(retry_error)
+
+    with (
+        patch.object(DaraCore, "do_action", side_effect=retry_errors),
+        patch.object(DaraCore, "sleep") as sleep,
+    ):
+        with pytest.raises(AlibabaCloudInvalidCredentialsError) as exception:
+            AlibabacloudProvider.validate_credentials(session)
+
+    assert isinstance(exception.value.original_exception, UnretryableException)
+    assert exception.value.original_exception.inner_exception is retry_errors[-1]
+    assert exception.value.__cause__ is exception.value.original_exception
+    assert sleep.call_args_list == [call(1000), call(1000)]
+
+
 class TestAlibabacloudProviderTestConnection:
     """Tests for the AlibabacloudProvider.test_connection method."""
 
+    def test_test_connection_connection_error_no_raise(self):
+        """Test connection failures are returned with their dedicated type."""
+        mock_session = MagicMock()
+        connection_error = AlibabaCloudConnectionError(
+            file="test_file",
+            original_exception=RetryError("connection reset"),
+        )
+
+        with (
+            patch.object(
+                AlibabacloudProvider,
+                "setup_session",
+                return_value=mock_session,
+            ),
+            patch.object(
+                AlibabacloudProvider,
+                "validate_credentials",
+                side_effect=connection_error,
+            ),
+            patch(
+                "prowler.providers.alibabacloud.alibabacloud_provider.logger.error"
+            ) as logger_error,
+            patch(
+                "prowler.providers.alibabacloud.alibabacloud_provider.logger.critical"
+            ) as logger_critical,
+        ):
+            result = AlibabacloudProvider.test_connection(
+                access_key_id="LTAI1234567890",
+                access_key_secret="test-secret-key",
+                raise_on_exception=False,
+            )
+
+        assert result.is_connected is False
+        assert result.error is connection_error
+        assert result.error.code == 10008
+        logger_error.assert_called_once()
+        logger_critical.assert_not_called()
+
+    def test_test_connection_connection_error_raises(self):
+        """Test connection failures retain raise-on-exception behavior."""
+        mock_session = MagicMock()
+        connection_error = AlibabaCloudConnectionError(
+            file="test_file",
+            original_exception=RetryError("connection reset"),
+        )
+
+        with (
+            patch.object(
+                AlibabacloudProvider,
+                "setup_session",
+                return_value=mock_session,
+            ),
+            patch.object(
+                AlibabacloudProvider,
+                "validate_credentials",
+                side_effect=connection_error,
+            ),
+            patch(
+                "prowler.providers.alibabacloud.alibabacloud_provider.logger.error"
+            ) as logger_error,
+            patch(
+                "prowler.providers.alibabacloud.alibabacloud_provider.logger.critical"
+            ) as logger_critical,
+        ):
+            with pytest.raises(AlibabaCloudConnectionError) as exception:
+                AlibabacloudProvider.test_connection(
+                    access_key_id="LTAI1234567890",
+                    access_key_secret="test-secret-key",
+                    raise_on_exception=True,
+                )
+
+        assert exception.value is connection_error
+        logger_error.assert_called_once()
+        logger_critical.assert_not_called()
+
     def test_test_connection_with_static_credentials_success(self):
         """Test successful connection with static access key credentials."""
         mock_session = MagicMock()

tests/providers/alibabacloud/conftest.py

@@ -5,6 +5,7 @@ Mocks Alibaba Cloud SDK modules to avoid import issues when the real
 dependencies are not installed in the test environment.
 """
 
+import importlib
 import sys
 from unittest.mock import MagicMock
 
@@ -38,4 +39,7 @@ MOCKED_MODULES = [
 ]
 
 for module_name in MOCKED_MODULES:
-    sys.modules.setdefault(module_name, MagicMock())
+    try:
+        importlib.import_module(module_name)
+    except ModuleNotFoundError:
+        sys.modules.setdefault(module_name, MagicMock())