管理访问凭据

前提条件

• 使用Credentials工具要求Python版本 >= 3.7，详见阿里云SDK支持策略。

• 使用V2.0代系的阿里云SDK，请参见V2.0 SDK 和 V1.0 SDK。

安装Credentials工具

推荐使用pip安装。

pip install alibabacloud_credentials

alibabacloud_credentials库的最新版本信息请参见alibabacloud-credentials · PyPI或Github，建议使用最新版本，以确保对所有功能的支持。

Credentials工具支持的凭据类型及其配置参数

Credentials工具支持的凭据类型及其配置参数均在alibabacloud_credentials.models模块的Config类中定义。凭据类型由参数type指定，不同的凭据类型所需的配置参数各不相同。下表将详细介绍type的取值范围及其支持的配置参数。其中，支持表示必填参数，-表示可选参数，不支持表示不支持参数。

方式一：默认凭据链

在使用Credentials工具时，如果不传入任何配置参数，Credentials工具将从默认凭据链中获取凭据。在采用此方式时，必须确保在应用程序运行环境中已配置默认凭据链支持的任意一种凭据获取方式。

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

# 不指定任何配置参数，将从默认凭据链中获取凭据。
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()

# 若使用云产品 V2.0 SDK 时，使用alibabacloud_tea_openapi.models模块的Config类传递credential
config = open_api_models.Config(
    credential=credentialsClient,
    endpoint='<endpoint>'
)
# 省略使用 config 初始化云产品 client 的代码...

方式二：AK

Credentials工具将使用您提供的AccessKey作为访问凭据。

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',
    # 必填参数，此处以从环境变量中获取AccessKey ID为例
    access_key_id=os.environ.get('ALIBABA_CLOUD_ACCESS_KEY_ID'),
    # 必填参数，此处以从环境变量中获取AccessKey Secret为例
    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()

# 若使用云产品 V2.0 SDK 时，使用alibabacloud_tea_openapi.models模块的Config类传递credential
config = open_api_models.Config(
    credential=credentialsClient,
    endpoint='<endpoint>'
)
# 省略使用 config 初始化云产品 client 的代码...

方式三：STS Token

Credentials工具将使用您提供的静态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',
    # 必填参数，此处以从环境变量中获取AccessKey ID为例
    access_key_id=os.environ.get('ALIBABA_CLOUD_ACCESS_KEY_ID'),
    # 必填参数，此处以从环境变量中获取AccessKey Secret为例
    access_key_secret=os.environ.get('ALIBABA_CLOUD_ACCESS_KEY_SECRET'),
    # 必填参数，此处以从环境变量中获取临时SecurityToken为例
    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()

# 若使用云产品 V2.0 SDK 时，使用alibabacloud_tea_openapi.models模块的Config类传递credential
config = open_api_models.Config(
    credential=credentialsClient,
    endpoint='<endpoint>'
)
# 省略使用 config 初始化云产品 client 的代码...

方式四：AK及RamRoleArn

Credentials工具将使用您提供的AK及RAM角色ARN调用AssumeRole获取STS Token，并将该STS Token作为访问凭据。通过此方式获取的凭据支持自动刷新，详情请参见Session类型凭据自动刷新机制。

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',
    # 必填参数，此处以从环境变量中获取AccessKey ID为例
    access_key_id=os.environ.get('ALIBABA_CLOUD_ACCESS_KEY_ID'),
    # 必填参数，此处以从环境变量中获取AccessKey Secret为例
    access_key_secret=os.environ.get('ALIBABA_CLOUD_ACCESS_KEY_SECRET'),
    # 必填参数，要扮演的RAM角色ARN，示例值：acs:ram::123456789012****:role/adminrole，支持通过环境变量ALIBABA_CLOUD_ROLE_ARN设置。
    role_arn='<role_arn>',
    # 可选参数，角色会话名称，支持通过环境变量ALIBABA_CLOUD_ROLE_SESSION_NAME设置。
    role_session_name='<role_session_name>',
    # 可选参数，设置更小的权限策略，非必填。示例值：{"Statement": [{"Action": ["*"],"Effect": "Allow","Resource": ["*"]}],"Version":"1"}
    policy='<policy>',
    # 可选参数，角色外部ID，主要功能是防止混淆代理人问题。
    external_id='<external_id>',
    # 可选参数，会话过期时间，默认3600秒。
    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()

# 若使用云产品 V2.0 SDK 时，使用alibabacloud_tea_openapi.models模块的Config类传递credential
config = open_api_models.Config(
    credential=credentialsClient,
    endpoint='<endpoint>'
)
# 省略使用 config 初始化云产品 client 的代码...

方式五：ECS实例RAM角色

若您的应用程序运行于ECS或ECI实例，并且该实例已被授予RAM角色，Credentials工具支持通过实例元数据获取RAM角色的STS Token作为访问凭据。通过此方式获取的凭据支持自动刷新，详情请参见Session类型凭据自动刷新机制。

Credentials工具将默认采用加固模式（IMDSv2）访问ECS实例元数据。若在加固模式下发生异常，您可以通过参数disable_imds_v1或环境变量 ALIBABA_CLOUD_IMDSV1_DISABLED 来控制异常处理逻辑：

• 当值为False（默认）时，系统将尝试切换到普通模式继续获取凭据；

• 当值为True时，表示仅允许使用加固模式获取凭据，若加固模式访问失败则会抛出异常。

服务端是否支持IMDSv2，取决于您在服务器的配置。

另外，您可以通过配置环境变量ALIBABA_CLOUD_ECS_METADATA_DISABLED=true来关闭实例元数据的凭据访问。

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',
    # 可选参数，该ECS角色的角色名称，不填会自动获取，但是建议加上以减少请求次数，支持通过环境变量ALIBABA_CLOUD_ECS_METADATA设置。
    role_name='<role_name>',
    # 可选参数，默认值：False。True：表示强制使用加固模式。False：系统将首先尝试在加固模式下获取凭据。如果失败，则会切换到普通模式（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()

# 若使用云产品 V2.0 SDK 时，使用alibabacloud_tea_openapi.models模块的Config类传递credential
config = open_api_models.Config(
    credential=credentialsClient,
    endpoint='<endpoint>'
)
# 省略使用 config 初始化云产品 client 的代码...

方式六：OIDCRoleArn

若您使用OIDC认证协议，并已创建OIDC身份提供商的RAM角色，您可以通过Credentials工具传入OIDC身份提供商ARN、OIDC Token及RAM角色ARN，系统将自动调用STS服务的AssumeRoleWithOIDC接口获取RAM角色的STS Token，并将该STS Token作为访问凭据。通过此方式获取的凭据支持自动刷新，详情请参见Session类型凭据自动刷新机制。例如当您的应用程序运行于已启用RRSA功能的ACK集群中时，Credentials工具支持通过读取Pod环境变量中的OIDC配置信息，调用AssumeRoleWithOIDC接口获取服务角色的STS Token，从而利用该STS Token访问阿里云相关服务。

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',
    # 必填参数，RAM角色ARN，支持通过环境变量ALIBABA_CLOUD_ROLE_ARN设置。
    role_arn='<role_arn>',
    # 必填参数，OIDC身份提供商ARN，支持通过环境变量ALIBABA_CLOUD_OIDC_PROVIDER_ARN设置。
    oidc_provider_arn='<oidc_provider_arn>',
    # 必填参数，OIDC Token文件路径，支持通过环境变量ALIBABA_CLOUD_OIDC_TOKEN_FILE设置。
    oidc_token_file_path='<oidc_token_file_path>',
    # 可选参数，角色会话名称，支持通过环境变量ALIBABA_CLOUD_ROLE_SESSION_NAME设置。
    role_session_name='<role_session_name>',
    # 可选参数，设置更小的权限策略，非必填。示例值：{"Statement": [{"Action": ["*"],"Effect": "Allow","Resource": ["*"]}],"Version":"1"}
    policy='<policy>',
    # 可选参数，设置session过期时间，默认3600秒。
    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()

# 若使用云产品 V2.0 SDK 时，使用alibabacloud_tea_openapi.models模块的Config类传递credential
config = open_api_models.Config(
    credential=credentialsClient,
    endpoint='<endpoint>'
)
# 省略使用 config 初始化云产品 client 的代码...

方式七：URI凭据

通过封装STS服务并对外暴露服务的URI，外部服务能够通过该URI获取STS Token，从而有效降低AK等敏感信息的暴露风险。Credentials工具支持通过访问您所提供的URI来获取STS Token，并将该STS Token作为访问凭据。通过此方式获取的凭据支持自动刷新，详情请参见Session类型凭据自动刷新机制。

该URI必须满足如下条件：

• 支持GET请求。

• 响应状态码为200。

• 响应体为如下的结构：

  {
    "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',
    # 必填参数，外部获取凭据的URI，格式为http://local_or_remote_uri/，支持通过环境变量ALIBABA_CLOUD_CREDENTIALS_URI设置。
    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()

# 若使用云产品 V2.0 SDK 时，使用alibabacloud_tea_openapi.models模块的Config类传递credential
config = open_api_models.Config(
    credential=credentialsClient,
    endpoint='<endpoint>'
)
# 省略使用 config 初始化云产品 client 的代码...

方式八：Bearer Token

目前只有云呼叫中心CCC这款产品支持Bearer Token。

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',
    # 必填参数，填入您的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()

# 若使用CCC V2.0 SDK 时，使用alibabacloud_tea_openapi.models模块的Config类传递credential
config = open_api_models.Config(
    credential=credentialsClient,
    endpoint='<endpoint>'
)
# 省略使用 config 初始化云产品 client 的代码...

默认凭据链

默认凭据链是一种兜底策略，按照内置的凭据查找顺序逐层向下查找，直至获取凭据信息。若所有情况均未成功获取到凭据信息，则会抛出CredentialException。以下为查找顺序：

{
  "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
    }
  ]
}

Session类型凭据自动刷新机制

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


#
# 单例装饰器（线程安全）
#

def singleton(cls):
    """实现线程安全的单例模式"""
    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
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


#
# 任务函数
#

def execute_task():
    try:
        # 获取凭据和ECS客户端
        credential = Credential().client.get_credential()
        ecs_client = EcsClient(Credential().client).client

        # 打印凭据信息
        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)

        # 调用API查询Region列表
        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

    # 定时任务延迟定义
    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()

相关文档

• RAM相关的基本概念，请参见基本概念。

• 如何创建AccessKey，请参见创建AccessKey。

• 如何通过程序方式创建RAM用户、AccessKey、RAM角色、权限策略及进行授权等操作，请参见RAM SDK概览。

• 如何通过程序方式进行角色扮演，请参见STS SDK概览。

• 关于RAM及STS相关的API信息，请参见API参考。

• 使用访问凭据访问阿里云OpenAPI最佳实践

• AccessKey泄露处理方案