Alibaba Cloud SDK:Manage access credentials

Prerequisites

• Python 3.7 or later. For more information, see Alibaba Cloud SDK support policy.

• An Alibaba Cloud SDK for Python V2.0. For more information, see SDKs V2.0 and V1.0.

Install the Credentials tool

We recommend that you use pip to install the tool.

pip install alibabacloud_credentials

For information about the latest version of the alibabacloud_credentials library, see alibabacloud-credentials · PyPI or Github. We recommend that you use the latest version to ensure support for all features.

Parameters of the Credentials tools

The credential types supported by the Credentials tool and their configuration parameters are defined in the Config class of the alibabacloud_credentials.models module. The credential type is specified by the type parameter. Different credential types require different configuration parameters. The following table describes the valid values for type and the configuration parameters supported by each type. In the table, Supported indicates a required parameter, - indicates an optional parameter, and Not supported indicates that the parameter is not supported.

Method 1: Default credential provider chain

When you use the Credentials tool without passing any configuration parameters, the tool retrieves credentials from the default credential provider chain. If you use this method, you must ensure that at least one credential retrieval method supported by the default credential provider chain is configured in your application's runtime environment.

from alibabacloud_credentials.client import Client as CredentialClient
from alibabacloud_tea_openapi import models as open_api_models

# Do not specify any configuration parameters. Credentials will be retrieved from the default credential provider chain.
credentialsClient = CredentialClient()

credential = credentialsClient.get_credential()
access_key_id = credential.get_access_key_id()
access_key_secret = credential.get_access_key_secret()
security_token = credential.get_security_token()

# If you use a V2.0 SDK for an Alibaba Cloud product, use the Config class from the alibabacloud_tea_openapi.models module to pass the credential.
config = open_api_models.Config(
    credential=credentialsClient,
    endpoint='<endpoint>'
)
# ... Code to initialize the product client using the config object is omitted.

Method 2: AccessKey pair

This method lets you create an AccessKey pair to initialize a Credentials client. For more information, see Create an AccessKey pair.

import os

from alibabacloud_credentials.client import Client as CredentialClient
from alibabacloud_credentials.models import Config as CredentialConfig
from alibabacloud_tea_openapi import models as open_api_models

credentialsConfig = CredentialConfig(
    type='access_key',
    # Required. This example shows how to obtain the AccessKey ID from an environment variable.
    access_key_id=os.environ.get('ALIBABA_CLOUD_ACCESS_KEY_ID'),
    # Required. This example shows how to obtain the AccessKey secret from an environment variable.
    access_key_secret=os.environ.get('ALIBABA_CLOUD_ACCESS_KEY_SECRET')
)
credentialsClient = CredentialClient(credentialsConfig)

credential = credentialsClient.get_credential()
access_key_id = credential.get_access_key_id()
access_key_secret = credential.get_access_key_secret()
security_token = credential.get_security_token()

# If you use a V2.0 SDK for an Alibaba Cloud product, use the Config class from the alibabacloud_tea_openapi.models module to pass the credential.
config = open_api_models.Config(
    credential=credentialsClient,
    endpoint='<endpoint>'
)
# ... Code to initialize the product client using the config object is omitted.

Method 3: STS token

This method lets you use a static STS token to initialize a Credentials client. For more information about how to obtain an STS token, see What is STS? The following example shows how to initialize a Credentials client using an STS token. The example does not show how to obtain an STS token.

import os

from alibabacloud_credentials.client import Client as CredentialClient
from alibabacloud_credentials.models import Config as CredentialConfig
from alibabacloud_tea_openapi import models as open_api_models

credentialsConfig = CredentialConfig(
    type='sts',
    # Required. This example shows how to obtain the AccessKey ID from an environment variable.
    access_key_id=os.environ.get('ALIBABA_CLOUD_ACCESS_KEY_ID'),
    # Required. This example shows how to obtain the AccessKey secret from an environment variable.
    access_key_secret=os.environ.get('ALIBABA_CLOUD_ACCESS_KEY_SECRET'),
    # Required. This example shows how to obtain the temporary security token from an environment variable.
    security_token=os.environ.get('ALIBABA_CLOUD_SECURITY_TOKEN')
)
credentialsClient = CredentialClient(credentialsConfig)

credential = credentialsClient.get_credential()
access_key_id = credential.get_access_key_id()
access_key_secret = credential.get_access_key_secret()
security_token = credential.get_security_token()

# If you use a V2.0 SDK for an Alibaba Cloud product, use the Config class from the alibabacloud_tea_openapi.models module to pass the credential.
config = open_api_models.Config(
    credential=credentialsClient,
    endpoint='<endpoint>'
)
# ... Code to initialize the product client using the config object is omitted.

Method 4: AccessKey pair and RamRoleArn

The Credentials tool uses the AccessKey and RAM role ARN that you provide to call AssumeRole, obtain an STS token, and use the token as an access credential. The credentials obtained in this way support auto-refresh. For more information, see Auto-refresh mechanism for session credentials.

import os

from alibabacloud_credentials.client import Client as CredentialClient
from alibabacloud_credentials.models import Config as CredentialConfig
from alibabacloud_tea_openapi import models as open_api_models

credentialsConfig = CredentialConfig(
    type='ram_role_arn',
    # Required. This example shows how to obtain the AccessKey ID from an environment variable.
    access_key_id=os.environ.get('ALIBABA_CLOUD_ACCESS_KEY_ID'),
    # Required. This example shows how to obtain the AccessKey secret from an environment variable.
    access_key_secret=os.environ.get('ALIBABA_CLOUD_ACCESS_KEY_SECRET'),
    # Required. The ARN of the RAM role to assume. Example: acs:ram::123456789012****:role/adminrole. You can set this using the ALIBABA_CLOUD_ROLE_ARN environment variable.
    role_arn='<role_arn>',
    # Optional. The name of the role session. You can set this using the ALIBABA_CLOUD_ROLE_SESSION_NAME environment variable.
    role_session_name='<role_session_name>',
    # Optional. A smaller permission policy. Example: {"Statement": [{"Action": ["*"],"Effect": "Allow","Resource": ["*"]}],"Version":"1"}
    policy='<policy>',
    # Optional. The external ID of the role, which is used to prevent the confused deputy problem.
    external_id='<external_id>',
    # Optional. The session expiration time. The default value is 3600 seconds.
    role_session_expiration=3600
)

credentialsClient = CredentialClient(credentialsConfig)

credential = credentialsClient.get_credential()
access_key_id = credential.get_access_key_id()
access_key_secret = credential.get_access_key_secret()
security_token = credential.get_security_token()

# If you use a V2.0 SDK for an Alibaba Cloud product, use the Config class from the alibabacloud_tea_openapi.models module to pass the credential.
config = open_api_models.Config(
    credential=credentialsClient,
    endpoint='<endpoint>'
)
# ... Code to initialize the product client using the config object is omitted.

Method 5: ECS instance RAM role

If your application runs on an ECS or ECI instance that is assigned a RAM role, the Credentials tool can retrieve the STS token of the RAM role from the instance metadata. This STS token is then used as the access credential. Credentials retrieved in this way support auto-refresh. For more information, see Auto-refresh mechanism for session credentials.

The Credentials tool uses the security-hardened mode (IMDSv2) by default to access ECS instance metadata. If an exception occurs in security-hardened mode, you can use the disable_imds_v1 parameter or the ALIBABA_CLOUD_IMDSV1_DISABLED environment variable to control the exception handling logic:

• If the value is False (the default), the system attempts to switch to NAT mode and continue to retrieve credentials.

• If the value is True, credentials can be retrieved only in security-hardened mode. If access in security-hardened mode fails, an exception is thrown.

Whether the server supports IMDSv2 depends on your server configuration.

You can also set the ALIBABA_CLOUD_ECS_METADATA_DISABLED=true environment variable to disable credential access from instance metadata.

from alibabacloud_credentials.client import Client as CredentialClient
from alibabacloud_credentials.models import Config as CredentialConfig
from alibabacloud_tea_openapi import models as open_api_models

credentialsConfig = CredentialConfig(
    type='ecs_ram_role',
    # Optional. The name of the ECS role. If you do not specify this parameter, the name is automatically retrieved. We recommend that you specify this parameter to reduce the number of requests. You can set this using the ALIBABA_CLOUD_ECS_METADATA environment variable.
    role_name='<role_name>',
    # Optional. The default value is False. If set to True, the security-hardened mode is forcibly used. If set to False, the system first attempts to retrieve the credential in security-hardened mode. If the attempt fails, the system switches to the normal mode (IMDSv1).
    disable_imds_v1=True
)

credentialsClient = CredentialClient(credentialsConfig)

credential = credentialsClient.get_credential()
access_key_id = credential.get_access_key_id()
access_key_secret = credential.get_access_key_secret()
security_token = credential.get_security_token()

# If you use a V2.0 SDK for an Alibaba Cloud product, use the Config class from the alibabacloud_tea_openapi.models module to pass the credential.
config = open_api_models.Config(
    credential=credentialsClient,
    endpoint='<endpoint>'
)
# ... Code to initialize the product client using the config object is omitted.

Method 6: OIDCRoleArn

If you use the OIDC authentication protocol and have already created a RAM role for an OIDC IdP, you can provide the OIDC IdP ARN, OIDC token, and RAM role ARN to the Credentials tool. The system automatically calls the AssumeRoleWithOIDC API of STS to obtain an STS token for the RAM role and uses this STS token as an access credential. The credentials obtained using this method support auto-refresh. For more information, see Auto-refresh mechanism for session credentials. For example, if your application runs in an ACK cluster that has the RRSA feature enabled, the Credentials tool can read the OIDC configuration information from the pod's environment variables, call the AssumeRoleWithOIDC API to obtain an STS token for the service role, and then use the STS token to access Alibaba Cloud services.

from alibabacloud_credentials.client import Client as CredentialClient
from alibabacloud_credentials.models import Config as CredentialConfig
from alibabacloud_tea_openapi import models as open_api_models

credentialsConfig = CredentialConfig(
    type='oidc_role_arn',
    # Required. The ARN of the RAM role. You can set this using the ALIBABA_CLOUD_ROLE_ARN environment variable.
    role_arn='<role_arn>',
    # Required. The ARN of the OIDC IdP. You can set this using the ALIBABA_CLOUD_OIDC_PROVIDER_ARN environment variable.
    oidc_provider_arn='<oidc_provider_arn>',
    # Required. The path of the OIDC token file. You can set this using the ALIBABA_CLOUD_OIDC_TOKEN_FILE environment variable.
    oidc_token_file_path='<oidc_token_file_path>',
    # Optional. The name of the role session. You can set this using the ALIBABA_CLOUD_ROLE_SESSION_NAME environment variable.
    role_session_name='<role_session_name>',
    # Optional. A smaller permission policy. Example: {"Statement": [{"Action": ["*"],"Effect": "Allow","Resource": ["*"]}],"Version":"1"}
    policy='<policy>',
    # Optional. The session expiration time. The default value is 3600 seconds.
    role_session_expiration=3600
)
credentialsClient = CredentialClient(credentialsConfig)

credential = credentialsClient.get_credential()
access_key_id = credential.get_access_key_id()
access_key_secret = credential.get_access_key_secret()
security_token = credential.get_security_token()

# If you use a V2.0 SDK for an Alibaba Cloud product, use the Config class from the alibabacloud_tea_openapi.models module to pass the credential.
config = open_api_models.Config(
    credential=credentialsClient,
    endpoint='<endpoint>'
)
# ... Code to initialize the product client using the config object is omitted.

Method 7: URI credential

You can encapsulate the STS service and expose its URI to allow external services to retrieve an STS token from the URI. This reduces the risk of exposing sensitive information such as AccessKeys. The Credentials tool can access the URI you provide to retrieve an STS token. This STS token is then used as the access credential. Credentials retrieved in this way support auto-refresh. For more information, see Auto-refresh mechanism for session credentials.

The URI must meet the following conditions:

• Supports GET requests.

• Returns a 200 status code.

• The response body has the following structure:

  {
    "Code": "Success",
    "AccessKeySecret": "yourAccessKeySecret",
    "AccessKeyId": "STS.****************",
    "Expiration": "2021-09-26T03:46:38Z",
    "SecurityToken": "yourSecurityToken"
  }

from alibabacloud_credentials.client import Client as CredentialClient
from alibabacloud_credentials.models import Config as CredentialConfig
from alibabacloud_tea_openapi import models as open_api_models

credentialsConfig = CredentialConfig(
    type='credentials_uri',
    # Required. The external URI for retrieving credentials. The format is http://local_or_remote_uri/. You can set this using the ALIBABA_CLOUD_CREDENTIALS_URI environment variable.
    credentials_uri='<credentials_uri>',
)
credentialsClient = CredentialClient(credentialsConfig)

credential = credentialsClient.get_credential()
access_key_id = credential.get_access_key_id()
access_key_secret = credential.get_access_key_secret()
security_token = credential.get_security_token()

# If you use a V2.0 SDK for an Alibaba Cloud product, use the Config class from the alibabacloud_tea_openapi.models module to pass the credential.
config = open_api_models.Config(
    credential=credentialsClient,
    endpoint='<endpoint>'
)
# ... Code to initialize the product client using the config object is omitted.

Method 8: Bearer token

Only Cloud Call Center lets you use a bearer token to initialize an SDK client.

from alibabacloud_credentials.client import Client as CredentialClient
from alibabacloud_credentials.models import Config as CredentialConfig
from alibabacloud_tea_openapi import models as open_api_models

credentialsConfig = CredentialConfig(
    type='bearer',
    # Required. Enter your bearer token.
    bearer_token='<BearerToken>',
)
credentialsClient = CredentialClient(credentialsConfig)

credential = credentialsClient.get_credential()
access_key_id = credential.get_access_key_id()
access_key_secret = credential.get_access_key_secret()
security_token = credential.get_security_token()

# If you use the CCC V2.0 SDK, use the Config class from the alibabacloud_tea_openapi.models module to pass the credential.
config = open_api_models.Config(
    credential=credentialsClient,
    endpoint='<endpoint>'
)
# ... Code to initialize the product client using the config object is omitted.

Default credential provider chain

If you want to use different types of credentials in the development and production environments of your application, you generally need to obtain the environment information from the code and write code branches to obtain different credentials for the development and production environments. The default credential provider chain of Alibaba Cloud Credentials for Java lets you obtain credentials for different environments based on configurations independent of the application, without the need for code modification. If you use Client client = new Client() to initialize a Credentials client without specifying an initialization method, the Credentials tool obtains the credential information in the following order:

{
	"current": "<PROFILE_NAME>",
	"profiles": [
		{
			"name": "<PROFILE_NAME>",
			"mode": "AK",
			"access_key_id": "<ALIBABA_CLOUD_ACCESS_KEY_ID>",
			"access_key_secret": "<ALIBABA_CLOUD_ACCESS_KEY_SECRET>"
		},
		{
			"name": "<PROFILE_NAME1>",
			"mode": "StsToken",
			"access_key_id": "<ALIBABA_CLOUD_ACCESS_KEY_ID>",
			"access_key_secret": "<ALIBABA_CLOUD_ACCESS_KEY_SECRET>",
			"sts_token": "<SECURITY_TOKEN>"
		},
		{
			"name":"<PROFILE_NAME2>",
			"mode":"RamRoleArn",
			"access_key_id":"<ALIBABA_CLOUD_ACCESS_KEY_ID>",
			"access_key_secret":"<ALIBABA_CLOUD_ACCESS_KEY_SECRET>",
			"ram_role_arn":"<ROLE_ARN>",
			"ram_session_name":"<ROLE_SESSION_NAME>",
			"expired_seconds":3600
		},
		{
			"name":"<PROFILE_NAME3>",
			"mode":"EcsRamRole",
			"ram_role_name":"<RAM_ROLE_ARN>"
		},
		{
			"name":"<PROFILE_NAME4>",
			"mode":"OIDC",
			"oidc_provider_arn":"<OIDC_PROVIDER_ARN>",
			"oidc_token_file":"<OIDC_TOKEN_FILE>",
			"ram_role_arn":"<ROLE_ARN>",
			"ram_session_name":"<ROLE_SESSION_NAME>",
			"expired_seconds":3600
		},
		{
			"name":"<PROFILE_NAME5>",
			"mode":"ChainableRamRoleArn",
			"source_profile":"<PROFILE_NAME>",
			"ram_role_arn":"<ROLE_ARN>",
			"ram_session_name":"<ROLE_SESSION_NAME>",
			"expired_seconds":3600
		}
	]
}

Automatic update mechanism of session credentials

import os
import sys
import time
import threading
from functools import wraps
from threading import Lock

from alibabacloud_credentials.client import Client as CredentialClient
from alibabacloud_credentials.models import Config as CredentialConfig
from alibabacloud_ecs20140526.client import Client
from alibabacloud_ecs20140526.models import DescribeRegionsRequest
from alibabacloud_tea_openapi.models import Config as EcsConfig


#
# Thread-safe singleton decorator
#

def singleton(cls):
    """Implements a thread-safe singleton pattern"""
    instances = {}
    lock = Lock()

    @wraps(cls)
    def wrapper(*args, **kwargs):
        with lock:
            if cls not in instances:
                instances[cls] = cls(*args, **kwargs)
        return instances[cls]

    return wrapper


#
# Singleton service class
#

@singleton
class Credential:
    def __init__(self):
        self._client = self._init_client()

    @staticmethod
    def _init_client() -> CredentialClient:
        try:
            config = CredentialConfig(
                type="ram_role_arn",
                access_key_id=os.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID"),
                access_key_secret=os.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET"),
                role_arn=os.getenv("ALIBABA_CLOUD_ROLE_ARN"),
                role_session_name="RamRoleArnTest",
                role_session_expiration=3600,
            )
            return CredentialClient(config)
        except Exception as e:
            raise RuntimeError(f"Credential initialization failed: {e}") from e

    @property
    def client(self) -> CredentialClient:
        return self._client


@singleton
class EcsClient:
    def __init__(self, credential: CredentialClient):
        self._client = self._init_client(credential)

    @staticmethod
    def _init_client(credential: CredentialClient) -> Client:
        try:
            ecs_config = EcsConfig(credential=credential)
            ecs_config.endpoint = 'ecs.cn-hangzhou.aliyuncs.com'
            return Client(ecs_config)
        except Exception as e:
            raise RuntimeError(f"ECS client initialization failed: {e}") from e

    @property
    def client(self) -> Client:
        return self._client


#
# Task function
#

def execute_task():
    try:
        # Get the credential and the ECS client.
        credential = Credential().client.get_credential()
        ecs_client = EcsClient(Credential().client).client

        # Print the credential information.
        print("Time:", time.strftime("%Y-%m-%d %H:%M:%S"))
        print("AK ID:", credential.access_key_id)
        print("AK Secret:", credential.access_key_secret)
        print("STS Token:", credential.security_token)

        # Call the API operation to query the list of regions.
        request = DescribeRegionsRequest()
        response = ecs_client.describe_regions(request)
        print("Invoke result:", response.to_map()["statusCode"])

    except Exception as e:
        print(f"ECS client execution failed: {e}")
        raise


def schedule_task():
    def run_execute_task():
        try:
            execute_task()
        except Exception as e:
            print(f"Task execution failed: {e}")
        finally:
            nonlocal task_count
            task_count += 1
            if task_count == len(DELAYS):
                print("All tasks completed. Exiting program.")
                sys.exit(0)

    def schedule_execution(delay):
        task_timer = threading.Timer(delay, run_execute_task)
        task_timer.start()
        return task_timer

    # Define the delays for the scheduled tasks.
    DELAYS = [0, 600, 4200, 4300]
    task_count = 0

    timers = [schedule_execution(delay) for delay in DELAYS]

    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        print("Program interrupted by user.")
        for timer in timers:
            timer.cancel()


if __name__ == "__main__":
    schedule_task()

References

• For more information about RAM, see Terms.

• For more information about how to create an AccessKey pair, see Create an AccessKey pair.

• For more information about how to create a RAM user, an AccessKey pair, a RAM role, and a policy and grant permissions to a RAM user, see RAM SDK overview.

• For more information about how to assume a role using a program, see STS SDK overview.

• For more information about RAM and STS-related API operations, see API Reference.

• Best practices for using an access credential to call API operations

• Solutions to AccessKey pair leaks