Alibaba Cloud SDK:Gerencie credenciais de acesso

pip install alibabacloud_credentials

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.

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.

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.

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.

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.

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.

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.

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.

1. Uso de variáveis de ambiente

2. Função RAM de IdP OIDC

3. Uso de arquivos de configuração

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

4. Função RAM de instância ECS

5. Credenciais baseadas em URI

Tipos de credenciais de sessão, como ram_role_arn, ecs_ram_role, oidc_role_arn e credentials_uri, oferecem suporte à atualização automática por meio de um mecanismo integrado ao provedor de credenciais. Quando um cliente de credenciais obtém uma credencial pela primeira vez, o provedor a armazena em cache. Nas operações subsequentes, a mesma instância do cliente recupera automaticamente a credencial desse cache. Se a credencial em cache tiver expirado, a instância do cliente busca uma nova e atualiza o cache adequadamente.

O exemplo a seguir usa o padrão singleton para criar um cliente de credenciais. Ele demonstra o mecanismo de atualização obtendo uma credencial em intervalos de tempo diferentes e chamando uma operação OpenAPI para verificar se a credencial está utilizável.

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()

Análise baseada na saída do log:

• Na primeira chamada, o cache está vazio. O sistema obtém uma credencial com base na sua configuração e a armazena no cache.

• A segunda chamada usa a mesma credencial da primeira, indicando que ela foi recuperada do cache.

• Na terceira chamada, a credencial em cache expirou. Seu tempo de expiração (RoleSessionExpiration) é de 3.600 segundos, mas esta chamada ocorre 4.200 segundos após a primeira. Consequentemente, o mecanismo de atualização automática do SDK busca uma nova credencial e atualiza o cache.

• A quarta chamada usa a mesma credencial da terceira, confirmando que o cache foi atualizado.