Alibaba Cloud SDK：管理訪問憑據

前提條件

• 使用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

目前只有Cloud Call CenterCCC這款產品支援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泄露處理方案