From cb125d5ceef470d1c31969dcbb9416537ca9e074 Mon Sep 17 00:00:00 2001 From: pedrooot Date: Tue, 9 Dec 2025 11:23:53 +0100 Subject: [PATCH] feat(alibabacloud): add needed static credentials changes --- .../alibabacloud/alibabacloud_provider.py | 66 +- .../alibabacloud_provider_test.py | 671 ++++++++++++++++++ 2 files changed, 715 insertions(+), 22 deletions(-) create mode 100644 tests/providers/alibabacloud/alibabacloud_provider_test.py diff --git a/prowler/providers/alibabacloud/alibabacloud_provider.py b/prowler/providers/alibabacloud/alibabacloud_provider.py index 23ce01c18d..82e48e2f14 100644 --- a/prowler/providers/alibabacloud/alibabacloud_provider.py +++ b/prowler/providers/alibabacloud/alibabacloud_provider.py @@ -75,6 +75,9 @@ class AlibabacloudProvider(Provider): mutelist_path: str = None, mutelist_content: dict = None, fixer_config: dict = {}, + access_key_id: str = None, + access_key_secret: str = None, + security_token: str = None, ): """ Initialize the AlibabaCloudProvider. @@ -91,6 +94,9 @@ class AlibabacloudProvider(Provider): mutelist_path: Path to the mutelist file mutelist_content: Content of the mutelist file fixer_config: Fixer configuration dictionary + access_key_id: Alibaba Cloud Access Key ID + access_key_secret: Alibaba Cloud Access Key Secret + security_token: STS Security Token (for temporary credentials) Raises: AlibabaCloudSetUpSessionError: If an error occurs during the setup process. @@ -107,6 +113,7 @@ class AlibabacloudProvider(Provider): - alibabacloud = AlibabacloudProvider(regions=["cn-hangzhou", "cn-shanghai"]) # Specific regions - alibabacloud = AlibabacloudProvider(role_arn="acs:ram::...:role/ProwlerRole") - alibabacloud = AlibabacloudProvider(ecs_ram_role="ECS-Prowler-Role") + - alibabacloud = AlibabacloudProvider(access_key_id="LTAI...", access_key_secret="...") """ logger.info("Initializing Alibaba Cloud Provider ...") @@ -118,6 +125,9 @@ class AlibabacloudProvider(Provider): ecs_ram_role=ecs_ram_role, oidc_role_arn=oidc_role_arn, credentials_uri=credentials_uri, + access_key_id=access_key_id, + access_key_secret=access_key_secret, + security_token=security_token, ) logger.info("Alibaba Cloud session configured successfully") @@ -234,6 +244,9 @@ class AlibabacloudProvider(Provider): ecs_ram_role: str = None, oidc_role_arn: str = None, credentials_uri: str = None, + access_key_id: str = None, + access_key_secret: str = None, + security_token: str = None, ) -> AlibabaCloudSession: """ Set up the Alibaba Cloud session. @@ -244,6 +257,9 @@ class AlibabacloudProvider(Provider): ecs_ram_role: Name of the RAM role attached to an ECS instance oidc_role_arn: ARN of the RAM role for OIDC authentication credentials_uri: URI to retrieve credentials from an external service + access_key_id: Alibaba Cloud Access Key ID + access_key_secret: Alibaba Cloud Access Key Secret + security_token: STS Security Token (for temporary credentials) Returns: AlibabaCloudSession object @@ -275,25 +291,22 @@ class AlibabacloudProvider(Provider): if not ecs_ram_role and "ALIBABA_CLOUD_ECS_METADATA" in os.environ: ecs_ram_role = os.environ["ALIBABA_CLOUD_ECS_METADATA"] - # Check for access key credentials from environment variables only + # Check for access key credentials from parameters first, then fall back to environment variables # Support both ALIBABA_CLOUD_* and ALIYUN_* prefixes for compatibility - # Note: We intentionally do NOT support credentials via CLI arguments for security reasons - access_key_id = None - access_key_secret = None - security_token = None + if not access_key_id: + if "ALIBABA_CLOUD_ACCESS_KEY_ID" in os.environ: + access_key_id = os.environ["ALIBABA_CLOUD_ACCESS_KEY_ID"] + elif "ALIYUN_ACCESS_KEY_ID" in os.environ: + access_key_id = os.environ["ALIYUN_ACCESS_KEY_ID"] - if "ALIBABA_CLOUD_ACCESS_KEY_ID" in os.environ: - access_key_id = os.environ["ALIBABA_CLOUD_ACCESS_KEY_ID"] - elif "ALIYUN_ACCESS_KEY_ID" in os.environ: - access_key_id = os.environ["ALIYUN_ACCESS_KEY_ID"] - - if "ALIBABA_CLOUD_ACCESS_KEY_SECRET" in os.environ: - access_key_secret = os.environ["ALIBABA_CLOUD_ACCESS_KEY_SECRET"] - elif "ALIYUN_ACCESS_KEY_SECRET" in os.environ: - access_key_secret = os.environ["ALIYUN_ACCESS_KEY_SECRET"] + if not access_key_secret: + if "ALIBABA_CLOUD_ACCESS_KEY_SECRET" in os.environ: + access_key_secret = os.environ["ALIBABA_CLOUD_ACCESS_KEY_SECRET"] + elif "ALIYUN_ACCESS_KEY_SECRET" in os.environ: + access_key_secret = os.environ["ALIYUN_ACCESS_KEY_SECRET"] # Check for STS security token (for temporary credentials) - if "ALIBABA_CLOUD_SECURITY_TOKEN" in os.environ: + if not security_token and "ALIBABA_CLOUD_SECURITY_TOKEN" in os.environ: security_token = os.environ["ALIBABA_CLOUD_SECURITY_TOKEN"] # Check for RAM role assumption from CLI arguments or environment @@ -695,6 +708,9 @@ class AlibabacloudProvider(Provider): @staticmethod def test_connection( + access_key_id: str = None, + access_key_secret: str = None, + security_token: str = None, role_arn: str = None, role_session_name: str = None, ecs_ram_role: str = None, @@ -707,6 +723,9 @@ class AlibabacloudProvider(Provider): Test the connection to Alibaba Cloud with the provided credentials. Args: + access_key_id: Alibaba Cloud Access Key ID (for static credentials) + access_key_secret: Alibaba Cloud Access Key Secret (for static credentials) + security_token: STS Security Token (for temporary credentials) role_arn: ARN of the RAM role to assume role_session_name: Session name when assuming the RAM role ecs_ram_role: Name of the RAM role attached to an ECS instance @@ -734,17 +753,24 @@ class AlibabacloudProvider(Provider): raise_on_exception=False ) Connection(is_connected=True, Error=None) + >>> AlibabacloudProvider.test_connection( + access_key_id="LTAI...", + access_key_secret="...", + raise_on_exception=False + ) + Connection(is_connected=True, Error=None) """ try: - session = None - - # Setup session + # Setup session - pass credentials directly instead of using env vars session = AlibabacloudProvider.setup_session( role_arn=role_arn, role_session_name=role_session_name, ecs_ram_role=ecs_ram_role, oidc_role_arn=oidc_role_arn, credentials_uri=credentials_uri, + access_key_id=access_key_id, + access_key_secret=access_key_secret, + security_token=security_token, ) # Validate credentials @@ -755,10 +781,6 @@ class AlibabacloudProvider(Provider): # Validate provider_id if provided if provider_id and caller_identity.account_id != provider_id: - from prowler.providers.alibabacloud.exceptions.exceptions import ( - AlibabaCloudInvalidCredentialsError, - ) - raise AlibabaCloudInvalidCredentialsError( file=pathlib.Path(__file__).name, message=f"Provider ID mismatch: expected '{provider_id}', got '{caller_identity.account_id}'", diff --git a/tests/providers/alibabacloud/alibabacloud_provider_test.py b/tests/providers/alibabacloud/alibabacloud_provider_test.py new file mode 100644 index 0000000000..8fd23acdf4 --- /dev/null +++ b/tests/providers/alibabacloud/alibabacloud_provider_test.py @@ -0,0 +1,671 @@ +import os +from unittest.mock import MagicMock, patch + +import pytest + +from prowler.providers.alibabacloud.alibabacloud_provider import AlibabacloudProvider +from prowler.providers.alibabacloud.exceptions.exceptions import ( + AlibabaCloudInvalidCredentialsError, + AlibabaCloudSetUpSessionError, +) +from prowler.providers.alibabacloud.models import AlibabaCloudCallerIdentity +from prowler.providers.common.models import Connection + + +class TestAlibabacloudProviderTestConnection: + """Tests for the AlibabacloudProvider.test_connection method.""" + + def test_test_connection_with_static_credentials_success(self): + """Test successful connection with static access key credentials.""" + mock_session = MagicMock() + mock_caller_identity = AlibabaCloudCallerIdentity( + account_id="1234567890", + principal_id="123456", + arn="acs:ram::1234567890:user/test-user", + identity_type="RamUser", + ) + + with ( + patch.object( + AlibabacloudProvider, + "setup_session", + return_value=mock_session, + ) as mock_setup_session, + patch.object( + AlibabacloudProvider, + "validate_credentials", + return_value=mock_caller_identity, + ) as mock_validate_credentials, + ): + result = AlibabacloudProvider.test_connection( + access_key_id="LTAI1234567890", + access_key_secret="test-secret-key", + raise_on_exception=False, + ) + + assert isinstance(result, Connection) + assert result.is_connected is True + assert result.error is None + mock_setup_session.assert_called_once() + mock_validate_credentials.assert_called_once() + + def test_test_connection_with_sts_token_success(self): + """Test successful connection with STS temporary credentials.""" + mock_session = MagicMock() + mock_caller_identity = AlibabaCloudCallerIdentity( + account_id="1234567890", + principal_id="123456", + arn="acs:ram::1234567890:user/test-user", + identity_type="RamUser", + ) + + with ( + patch.object( + AlibabacloudProvider, + "setup_session", + return_value=mock_session, + ), + patch.object( + AlibabacloudProvider, + "validate_credentials", + return_value=mock_caller_identity, + ), + ): + result = AlibabacloudProvider.test_connection( + access_key_id="STS.LTAI1234567890", + access_key_secret="test-secret-key", + security_token="test-security-token", + raise_on_exception=False, + ) + + assert isinstance(result, Connection) + assert result.is_connected is True + assert result.error is None + + def test_test_connection_with_role_arn_success(self): + """Test successful connection with RAM role assumption.""" + mock_session = MagicMock() + mock_caller_identity = AlibabaCloudCallerIdentity( + account_id="1234567890", + principal_id="123456", + arn="acs:ram::1234567890:role/ProwlerRole", + identity_type="AssumedRoleUser", + ) + + with ( + patch.object( + AlibabacloudProvider, + "setup_session", + return_value=mock_session, + ) as mock_setup_session, + patch.object( + AlibabacloudProvider, + "validate_credentials", + return_value=mock_caller_identity, + ), + ): + result = AlibabacloudProvider.test_connection( + role_arn="acs:ram::1234567890:role/ProwlerRole", + role_session_name="prowler-session", + raise_on_exception=False, + ) + + assert isinstance(result, Connection) + assert result.is_connected is True + assert result.error is None + mock_setup_session.assert_called_once_with( + role_arn="acs:ram::1234567890:role/ProwlerRole", + role_session_name="prowler-session", + ecs_ram_role=None, + oidc_role_arn=None, + credentials_uri=None, + access_key_id=None, + access_key_secret=None, + security_token=None, + ) + + def test_test_connection_with_provider_id_validation_success(self): + """Test successful connection with provider_id validation.""" + mock_session = MagicMock() + mock_caller_identity = AlibabaCloudCallerIdentity( + account_id="1234567890", + principal_id="123456", + arn="acs:ram::1234567890:user/test-user", + identity_type="RamUser", + ) + + with ( + patch.object( + AlibabacloudProvider, + "setup_session", + return_value=mock_session, + ), + patch.object( + AlibabacloudProvider, + "validate_credentials", + return_value=mock_caller_identity, + ), + ): + result = AlibabacloudProvider.test_connection( + access_key_id="LTAI1234567890", + access_key_secret="test-secret-key", + provider_id="1234567890", + raise_on_exception=False, + ) + + assert isinstance(result, Connection) + assert result.is_connected is True + assert result.error is None + + def test_test_connection_with_provider_id_mismatch_raises_exception(self): + """Test connection with provider_id mismatch raises exception.""" + mock_session = MagicMock() + mock_caller_identity = AlibabaCloudCallerIdentity( + account_id="1234567890", + principal_id="123456", + arn="acs:ram::1234567890:user/test-user", + identity_type="RamUser", + ) + + with ( + patch.object( + AlibabacloudProvider, + "setup_session", + return_value=mock_session, + ), + patch.object( + AlibabacloudProvider, + "validate_credentials", + return_value=mock_caller_identity, + ), + ): + with pytest.raises(AlibabaCloudInvalidCredentialsError) as exception: + AlibabacloudProvider.test_connection( + access_key_id="LTAI1234567890", + access_key_secret="test-secret-key", + provider_id="different-account-id", + raise_on_exception=True, + ) + + assert "Provider ID mismatch" in str(exception.value) + assert "expected 'different-account-id'" in str(exception.value) + assert "got '1234567890'" in str(exception.value) + + def test_test_connection_with_provider_id_mismatch_no_raise(self): + """Test connection with provider_id mismatch returns error without raising.""" + mock_session = MagicMock() + mock_caller_identity = AlibabaCloudCallerIdentity( + account_id="1234567890", + principal_id="123456", + arn="acs:ram::1234567890:user/test-user", + identity_type="RamUser", + ) + + with ( + patch.object( + AlibabacloudProvider, + "setup_session", + return_value=mock_session, + ), + patch.object( + AlibabacloudProvider, + "validate_credentials", + return_value=mock_caller_identity, + ), + ): + result = AlibabacloudProvider.test_connection( + access_key_id="LTAI1234567890", + access_key_secret="test-secret-key", + provider_id="different-account-id", + raise_on_exception=False, + ) + + assert isinstance(result, Connection) + assert result.is_connected is False + assert result.error is not None + assert isinstance(result.error, AlibabaCloudInvalidCredentialsError) + + def test_test_connection_setup_session_error_raises_exception(self): + """Test connection when setup_session raises an exception.""" + with patch.object( + AlibabacloudProvider, + "setup_session", + side_effect=AlibabaCloudSetUpSessionError( + file="test_file", + original_exception=Exception("Simulated setup error"), + ), + ): + with pytest.raises(AlibabaCloudSetUpSessionError) as exception: + AlibabacloudProvider.test_connection( + access_key_id="LTAI1234567890", + access_key_secret="test-secret-key", + raise_on_exception=True, + ) + + assert exception.type == AlibabaCloudSetUpSessionError + + def test_test_connection_setup_session_error_no_raise(self): + """Test connection when setup_session raises an exception without raising.""" + setup_error = AlibabaCloudSetUpSessionError( + file="test_file", + original_exception=Exception("Simulated setup error"), + ) + + with patch.object( + AlibabacloudProvider, + "setup_session", + side_effect=setup_error, + ): + result = AlibabacloudProvider.test_connection( + access_key_id="LTAI1234567890", + access_key_secret="test-secret-key", + raise_on_exception=False, + ) + + assert isinstance(result, Connection) + assert result.is_connected is False + assert result.error is setup_error + + def test_test_connection_invalid_credentials_raises_exception(self): + """Test connection when validate_credentials raises an exception.""" + mock_session = MagicMock() + + with ( + patch.object( + AlibabacloudProvider, + "setup_session", + return_value=mock_session, + ), + patch.object( + AlibabacloudProvider, + "validate_credentials", + side_effect=AlibabaCloudInvalidCredentialsError( + file="test_file", + original_exception=Exception("Invalid credentials"), + ), + ), + ): + with pytest.raises(AlibabaCloudInvalidCredentialsError) as exception: + AlibabacloudProvider.test_connection( + access_key_id="LTAI-invalid", + access_key_secret="invalid-secret", + raise_on_exception=True, + ) + + assert exception.type == AlibabaCloudInvalidCredentialsError + + def test_test_connection_invalid_credentials_no_raise(self): + """Test connection when validate_credentials raises an exception without raising.""" + mock_session = MagicMock() + auth_error = AlibabaCloudInvalidCredentialsError( + file="test_file", + original_exception=Exception("Invalid credentials"), + ) + + with ( + patch.object( + AlibabacloudProvider, + "setup_session", + return_value=mock_session, + ), + patch.object( + AlibabacloudProvider, + "validate_credentials", + side_effect=auth_error, + ), + ): + result = AlibabacloudProvider.test_connection( + access_key_id="LTAI-invalid", + access_key_secret="invalid-secret", + raise_on_exception=False, + ) + + assert isinstance(result, Connection) + assert result.is_connected is False + assert result.error is auth_error + + def test_test_connection_generic_exception_raises(self): + """Test connection when a generic exception occurs.""" + mock_session = MagicMock() + + with ( + patch.object( + AlibabacloudProvider, + "setup_session", + return_value=mock_session, + ), + patch.object( + AlibabacloudProvider, + "validate_credentials", + side_effect=Exception("Unexpected error"), + ), + ): + with pytest.raises(Exception) as exception: + AlibabacloudProvider.test_connection( + access_key_id="LTAI1234567890", + access_key_secret="test-secret-key", + raise_on_exception=True, + ) + + assert str(exception.value) == "Unexpected error" + + def test_test_connection_generic_exception_no_raise(self): + """Test connection when a generic exception occurs without raising.""" + mock_session = MagicMock() + generic_error = Exception("Unexpected error") + + with ( + patch.object( + AlibabacloudProvider, + "setup_session", + return_value=mock_session, + ), + patch.object( + AlibabacloudProvider, + "validate_credentials", + side_effect=generic_error, + ), + ): + result = AlibabacloudProvider.test_connection( + access_key_id="LTAI1234567890", + access_key_secret="test-secret-key", + raise_on_exception=False, + ) + + assert isinstance(result, Connection) + assert result.is_connected is False + assert result.error is generic_error + + def test_test_connection_passes_credentials_to_setup_session(self): + """Test that credentials are passed directly to setup_session.""" + mock_session = MagicMock() + mock_caller_identity = AlibabaCloudCallerIdentity( + account_id="1234567890", + principal_id="123456", + arn="acs:ram::1234567890:user/test-user", + identity_type="RamUser", + ) + + with ( + patch.object( + AlibabacloudProvider, + "setup_session", + return_value=mock_session, + ) as mock_setup_session, + patch.object( + AlibabacloudProvider, + "validate_credentials", + return_value=mock_caller_identity, + ), + ): + result = AlibabacloudProvider.test_connection( + access_key_id="LTAI1234567890", + access_key_secret="test-secret-key", + security_token="test-token", + raise_on_exception=False, + ) + + assert result.is_connected is True + + # Verify credentials are passed directly to setup_session + mock_setup_session.assert_called_once_with( + role_arn=None, + role_session_name=None, + ecs_ram_role=None, + oidc_role_arn=None, + credentials_uri=None, + access_key_id="LTAI1234567890", + access_key_secret="test-secret-key", + security_token="test-token", + ) + + def test_test_connection_does_not_set_environment_variables(self): + """Test that test_connection does not set environment variables.""" + mock_session = MagicMock() + mock_caller_identity = AlibabaCloudCallerIdentity( + account_id="1234567890", + principal_id="123456", + arn="acs:ram::1234567890:user/test-user", + identity_type="RamUser", + ) + + # Ensure env vars don't exist before the test + for var in [ + "ALIBABA_CLOUD_ACCESS_KEY_ID", + "ALIBABA_CLOUD_ACCESS_KEY_SECRET", + "ALIBABA_CLOUD_SECURITY_TOKEN", + ]: + if var in os.environ: + del os.environ[var] + + with ( + patch.object( + AlibabacloudProvider, + "setup_session", + return_value=mock_session, + ), + patch.object( + AlibabacloudProvider, + "validate_credentials", + return_value=mock_caller_identity, + ), + ): + result = AlibabacloudProvider.test_connection( + access_key_id="LTAI1234567890", + access_key_secret="test-secret-key", + security_token="test-token", + raise_on_exception=False, + ) + + assert result.is_connected is True + + # Verify environment variables are not set + assert "ALIBABA_CLOUD_ACCESS_KEY_ID" not in os.environ + assert "ALIBABA_CLOUD_ACCESS_KEY_SECRET" not in os.environ + assert "ALIBABA_CLOUD_SECURITY_TOKEN" not in os.environ + + def test_test_connection_with_ecs_ram_role(self): + """Test successful connection with ECS RAM role.""" + mock_session = MagicMock() + mock_caller_identity = AlibabaCloudCallerIdentity( + account_id="1234567890", + principal_id="123456", + arn="acs:ram::1234567890:role/ECS-Prowler-Role", + identity_type="AssumedRoleUser", + ) + + with ( + patch.object( + AlibabacloudProvider, + "setup_session", + return_value=mock_session, + ) as mock_setup_session, + patch.object( + AlibabacloudProvider, + "validate_credentials", + return_value=mock_caller_identity, + ), + ): + result = AlibabacloudProvider.test_connection( + ecs_ram_role="ECS-Prowler-Role", + raise_on_exception=False, + ) + + assert isinstance(result, Connection) + assert result.is_connected is True + assert result.error is None + mock_setup_session.assert_called_once_with( + role_arn=None, + role_session_name=None, + ecs_ram_role="ECS-Prowler-Role", + oidc_role_arn=None, + credentials_uri=None, + access_key_id=None, + access_key_secret=None, + security_token=None, + ) + + def test_test_connection_with_oidc_role_arn(self): + """Test successful connection with OIDC role ARN.""" + mock_session = MagicMock() + mock_caller_identity = AlibabaCloudCallerIdentity( + account_id="1234567890", + principal_id="123456", + arn="acs:ram::1234567890:role/OIDCRole", + identity_type="AssumedRoleUser", + ) + + with ( + patch.object( + AlibabacloudProvider, + "setup_session", + return_value=mock_session, + ) as mock_setup_session, + patch.object( + AlibabacloudProvider, + "validate_credentials", + return_value=mock_caller_identity, + ), + ): + result = AlibabacloudProvider.test_connection( + oidc_role_arn="acs:ram::1234567890:role/OIDCRole", + raise_on_exception=False, + ) + + assert isinstance(result, Connection) + assert result.is_connected is True + assert result.error is None + mock_setup_session.assert_called_once_with( + role_arn=None, + role_session_name=None, + ecs_ram_role=None, + oidc_role_arn="acs:ram::1234567890:role/OIDCRole", + credentials_uri=None, + access_key_id=None, + access_key_secret=None, + security_token=None, + ) + + def test_test_connection_with_credentials_uri(self): + """Test successful connection with credentials URI.""" + mock_session = MagicMock() + mock_caller_identity = AlibabaCloudCallerIdentity( + account_id="1234567890", + principal_id="123456", + arn="acs:ram::1234567890:user/test-user", + identity_type="RamUser", + ) + + with ( + patch.object( + AlibabacloudProvider, + "setup_session", + return_value=mock_session, + ) as mock_setup_session, + patch.object( + AlibabacloudProvider, + "validate_credentials", + return_value=mock_caller_identity, + ), + ): + result = AlibabacloudProvider.test_connection( + credentials_uri="http://localhost:8080/credentials", + raise_on_exception=False, + ) + + assert isinstance(result, Connection) + assert result.is_connected is True + assert result.error is None + mock_setup_session.assert_called_once_with( + role_arn=None, + role_session_name=None, + ecs_ram_role=None, + oidc_role_arn=None, + credentials_uri="http://localhost:8080/credentials", + access_key_id=None, + access_key_secret=None, + security_token=None, + ) + + def test_test_connection_without_any_credentials(self): + """Test connection without any credentials uses default credential chain.""" + mock_session = MagicMock() + mock_caller_identity = AlibabaCloudCallerIdentity( + account_id="1234567890", + principal_id="123456", + arn="acs:ram::1234567890:user/test-user", + identity_type="RamUser", + ) + + with ( + patch.object( + AlibabacloudProvider, + "setup_session", + return_value=mock_session, + ) as mock_setup_session, + patch.object( + AlibabacloudProvider, + "validate_credentials", + return_value=mock_caller_identity, + ), + ): + result = AlibabacloudProvider.test_connection( + raise_on_exception=False, + ) + + assert isinstance(result, Connection) + assert result.is_connected is True + assert result.error is None + # Should call setup_session with all None values + mock_setup_session.assert_called_once_with( + role_arn=None, + role_session_name=None, + ecs_ram_role=None, + oidc_role_arn=None, + credentials_uri=None, + access_key_id=None, + access_key_secret=None, + security_token=None, + ) + + def test_test_connection_preserves_existing_env_vars(self): + """Test that existing environment variables are not affected by test_connection.""" + # Set up existing env vars + original_key = "original-key-id" + os.environ["ALIBABA_CLOUD_ACCESS_KEY_ID"] = original_key + + mock_session = MagicMock() + mock_caller_identity = AlibabaCloudCallerIdentity( + account_id="1234567890", + principal_id="123456", + arn="acs:ram::1234567890:user/test-user", + identity_type="RamUser", + ) + + try: + with ( + patch.object( + AlibabacloudProvider, + "setup_session", + return_value=mock_session, + ), + patch.object( + AlibabacloudProvider, + "validate_credentials", + return_value=mock_caller_identity, + ), + ): + result = AlibabacloudProvider.test_connection( + access_key_id="LTAI1234567890", + access_key_secret="test-secret-key", + raise_on_exception=False, + ) + + assert result.is_connected is True + # Verify test_connection does not modify existing env vars + # (credentials are passed directly to setup_session, not via env vars) + assert os.environ.get("ALIBABA_CLOUD_ACCESS_KEY_ID") == original_key + finally: + # Clean up + if "ALIBABA_CLOUD_ACCESS_KEY_ID" in os.environ: + del os.environ["ALIBABA_CLOUD_ACCESS_KEY_ID"]