Object Storage Service:OSS Python SDK V1

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

# Sample code for querying endpoint information using the OSS Python SDK

import oss2
from oss2.credentials import EnvironmentVariableCredentialsProvider

def main():

    print("=== Querying endpoint information for all supported regions ===\n")
    
    # Obtain access credentials from environment variables. You must set the OSS_ACCESS_KEY_ID and OSS_ACCESS_KEY_SECRET environment variables.
    auth = oss2.ProviderAuthV4(EnvironmentVariableCredentialsProvider())
    
    # Specify the endpoint for the region where your bucket is located.
    endpoint = "<endpoint>"
    
    # Specify the region that corresponds to the endpoint. This parameter is required when you use Signature V4.
    region = "<region-id>"
    
    # Initialize an OSS Service instance.
    service = oss2.Service(auth, endpoint, region=region)
    
    try:
        # Query endpoint information for all supported regions.
        result = service.describe_regions()
        
        # Iterate through the region information.
        for r in result.regions:
            print(f"Region: {r.region}")
            print(f"  Public (IPv4) Endpoint: {r.internet_endpoint}")
            print(f"  Internal (Classic Network or VPC) Endpoint: {r.internal_endpoint}")
            print(f"  Transfer Acceleration Endpoint: {r.accelerate_endpoint}")
            print("-" * 80)
        
        # Print statistics.
        print(f"\nFound endpoint information for {len(result.regions)} regions.")
            
    except oss2.exceptions.OssError as e:
        print(f"OSS error: {e}")
        raise
    except Exception as e:
        print(f"An error occurred: {e}")
        raise

if __name__ == "__main__":
    # Entry point of the program.
    try:
        main()
    except Exception as e:
        print(f"An error occurred during execution: {e}")
        raise

result = service.describe_regions("oss-<region-id>")

AK de usuário RAM

Ideal para ambientes seguros e estáveis que precisam de acesso de longo prazo ao OSS sem rotação frequente de credenciais. Inicialize o provedor de credenciais com a AccessKey (AK) de um usuário RAM. Isso requer manutenção manual da AK, o que introduz riscos de segurança.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

# OSS Python SDK V1: Sample code for client initialization

import oss2
from oss2.credentials import EnvironmentVariableCredentialsProvider

def main():

    # Specify the AccessKey ID and AccessKey Secret of a RAM user.
    access_key_id = 'yourAccessKeyID'
    access_key_secret = 'yourAccessKeySecret'
    # Configure access credentials using the Access Key (AK) of the RAM user.
    auth = oss2.AuthV4(access_key_id, access_key_secret)
    # Set the OSS service region.
    region = "<region-id>"
    # Set the endpoint.
    endpoint = "<endpoint>"
    # Set the bucket name.
    bucket_name = "example-bucket-hz"

    # Initialize an OSS Bucket instance.
    bucket = oss2.Bucket(auth, endpoint, bucket_name, region=region)
    
    # List objects in the bucket.
    print("=== List objects in the bucket ===")
    try:
        # Use the list_objects method to list objects.
        result = bucket.list_objects()
        
        # Iterate through the object list and print only the object keys.
        for obj in result.object_list:
            print(obj.key)
        
        # Print statistics.
        print(f"\nTotal objects listed: {len(result.object_list)}")
        if result.is_truncated:
            print("Note: The result is truncated. More objects are not listed.")
            
    except oss2.exceptions.OssError as e:
        print(f"OSS error: {e}")
        raise
    except Exception as e:
        print(f"An error occurred: {e}")
        raise

if __name__ == "__main__":
    # Entry point of the program.
    try:
        main()
    except Exception as e:
        print(f"An error occurred during execution: {e}")
        raise

Credenciais temporárias STS

Indicado para aplicativos que precisam de acesso temporário ao OSS. Inicialize o provedor de credenciais com credenciais temporárias (AccessKey ID, AccessKey Secret e token de segurança) do STS. Você deve atualizar o token STS manualmente antes que ele expire.

# -*- coding: utf-8 -*-
import oss2

# Specify the temporary AccessKey ID and AccessKey Secret obtained from STS, not the AccessKey ID and AccessKey Secret of your Alibaba Cloud account.
# Note that the Access Key ID obtained from STS starts with "STS."
sts_access_key_id = 'STS.****************'
sts_access_key_secret = 'yourAccessKeySecret'
# Specify the security token obtained from STS.
security_token = 'yourSecurityToken'

# yourEndpoint specifies the endpoint for the region where the bucket is located.
endpoint = '<endpoint>'

# yourBucketName specifies the bucket name.
bucket_name = 'yourBucketName'

# Specify the region where the bucket is located.
region = '<region-id>'

# Initialize the StsAuth instance. Note that setting auth_version to "v4" enables OSS V4 signatures.
auth = oss2.StsAuth(sts_access_key_id,
                    sts_access_key_secret,
                    security_token,
                    auth_version="v4")

# Initialize the bucket using the StsAuth instance. Note that the region parameter is required when using V4 signatures;
# otherwise, an error occurs.
bucket = oss2.Bucket(auth, endpoint, bucket_name, region=region)

# Use the bucket object for subsequent operations...

ARN de função RAM

Recomendado para acesso entre contas. Inicialize o provedor de credenciais com um ARN de função RAM. A ferramenta de credenciais atualiza automaticamente o token STS chamando AssumeRole antes que ele expire. Você pode definir o parâmetro policy para restringir ainda mais as permissões.

1. Adicione a dependência de credenciais.

   pip install alibabacloud_credentials

2. Configure a Access Key (AK) e o ARN da função RAM como credenciais de acesso.

   Antes de executar o código de exemplo, substitua espaços reservados como <region-id> pela região e endpoint reais. Por exemplo, substitua <region-id> por ap-southeast-1 .

   # -*- coding: utf-8 -*-
   import oss2
   from alibabacloud_credentials.client import Client
   from alibabacloud_credentials.models import Config
   from oss2 import CredentialsProvider
   from oss2.credentials import Credentials
   import os
   
   class CredentialProviderWrapper(CredentialsProvider):
       def __init__(self, client):
           self.client = client
   
       def get_credentials(self):
           credential = self.client.get_credential()
           access_key_id = credential.access_key_id
           access_key_secret = credential.access_key_secret
           security_token = credential.security_token
           return Credentials(access_key_id, access_key_secret, security_token)
   
   config = Config(
       # Obtain the Access Key (AccessKey ID and AccessKey Secret) of the RAM user from environment variables.
       access_key_id=os.getenv('ALIBABA_CLOUD_ACCESS_KEY_ID'),
       access_key_secret=os.getenv('ALIBABA_CLOUD_ACCESS_KEY_SECRET'),
       type='ram_role_arn',
       # The ARN of the RAM role to assume. Example: acs:ram::123456789012****:role/adminrole. You can set the RoleArn using the ALIBABA_CLOUD_ROLE_ARN environment variable.
       role_arn='<RoleArn>',
       # The role session name. You can set the RoleSessionName using the ALIBABA_CLOUD_ROLE_SESSION_NAME environment variable.
       role_session_name='<RoleSessionName>',
       # Optional. A more restrictive permission policy. Example: {"Statement": [{"Action": ["*"],"Effect": "Allow","Resource": ["*"]}],"Version":"1"}
       policy='<Policy>',
       # Optional. The validity period of the role session.
       role_session_expiration=3600
   )
   
   cred = Client(config)
   
   credentials_provider = CredentialProviderWrapper(cred)
   
   # Note that ProviderAuthV4 indicates the use of V4 signatures.
   auth = oss2.ProviderAuthV4(credentials_provider)
   
   # yourEndpoint specifies the endpoint for the region where the bucket is located.
   endpoint = '<endpoint>'
   
   # yourBucketName specifies the bucket name.
   bucket_name = 'yourBucketName'
   
   # Specify the region where the bucket is located.
   region = '<region-id>'
   
   # Initialize the bucket by using the Auth instance. Note that the region parameter is required when using V4 signatures;
   # otherwise, an error occurs.
   bucket = oss2.Bucket(auth, endpoint, bucket_name, region=region)
   
   # Use the bucket object for subsequent operations...

Função RAM de instância ECS

Projetado para aplicativos em instâncias ECS, instâncias ECI ou nós de trabalho do Container Service for Kubernetes (ACK). Uma função RAM de instância ECS permite a atualização automática do token STS em instâncias ECS, instâncias ECI ou nós de trabalho do Container Service for Kubernetes (ACK) sem fornecer uma AK ou token STS. Para obter uma função, consulte Criar uma função RAM. Para anexá-la a uma instância, consulte Função RAM de instância.

1. Adicione a dependência de credenciais.

   pip install alibabacloud_credentials

2. Configure a função RAM da instância ECS como credencial de acesso.

   Antes de executar o código de exemplo, substitua espaços reservados como <region-id> pela região e endpoint reais. Por exemplo, substitua <region-id> por ap-southeast-1 .

   # -*- coding: utf-8 -*-
   import oss2
   from alibabacloud_credentials.client import Client
   from alibabacloud_credentials.models import Config
   from oss2 import CredentialsProvider
   from oss2.credentials import Credentials
   
   class CredentialProviderWrapper(CredentialsProvider):
       def __init__(self, client):
           self.client = client
   
       def get_credentials(self):
           credential = self.client.get_credential()
           access_key_id = credential.access_key_id
           access_key_secret = credential.access_key_secret
           security_token = credential.security_token
           return Credentials(access_key_id, access_key_secret, security_token)
   
   config = Config(
       type='ecs_ram_role',      # Credential type. Fixed value: ecs_ram_role.
       role_name='<RoleName>'    # The name of the RAM role attached to the ECS instance. Optional. If not set, it is retrieved automatically. We strongly recommend setting this parameter to reduce requests.
   )
   
   cred = Client(config)
   
   credentials_provider = CredentialProviderWrapper(cred)
   
   # Note that ProviderAuthV4 indicates the use of V4 signatures.
   auth = oss2.ProviderAuthV4(credentials_provider)
   
   # yourEndpoint specifies the endpoint for the region where the bucket is located.
   endpoint = '<endpoint>'
   
   # yourBucketName specifies the bucket name.
   bucket_name = 'yourBucketName'
   
   # Specify the region where the bucket is located.
   region = '<region-id>'
   
   # Initialize the bucket by using the Auth instance. Note that the region parameter is required when using V4 signatures;
   # otherwise, an error occurs.
   bucket = oss2.Bucket(auth, endpoint, bucket_name, region=region)
   
   # Use the bucket object for subsequent operations...

ARN de função OIDC

Quando aplicativos não confiáveis são executados em nós de trabalho do ACK, use RRSA para impedir que eles obtenham o token STS da função RAM da instância do nó. O cluster monta um arquivo de token OIDC de conta de serviço por pod e injeta a configuração nas variáveis de ambiente. A ferramenta de credenciais troca o token OIDC por um token STS via AssumeRoleWithOIDC, sem exigir uma AK ou token STS. Configure permissões RAM para uma ServiceAccount para implementar isolamento de permissões de pod usando RRSA.

1. Adicione a dependência de credenciais.

   pip install alibabacloud_credentials

2. Configure a função OIDC como credencial de acesso.

   Antes de executar o código de exemplo, substitua espaços reservados como <region-id> pela região e endpoint reais. Por exemplo, substitua <region-id> por ap-southeast-1 .

   # -*- coding: utf-8 -*-
   import oss2
   from alibabacloud_credentials.client import Client
   from alibabacloud_credentials.models import Config
   from oss2 import CredentialsProvider
   from oss2.credentials import Credentials
   import os
   
   class CredentialProviderWrapper(CredentialsProvider):
       def __init__(self, client):
           self.client = client
   
       def get_credentials(self):
           credential = self.client.get_credential()
           access_key_id = credential.access_key_id
           access_key_secret = credential.access_key_secret
           security_token = credential.security_token
           return Credentials(access_key_id, access_key_secret, security_token)
   
   config = Config(
       # Credential type. Fixed value: oidc_role_arn.
       type='oidc_role_arn',
       # The ARN of the RAM role. You can set the RoleArn using the ALIBABA_CLOUD_ROLE_ARN environment variable.
       role_arn=os.environ.get('<RoleArn>'),
       # The ARN of the OIDC provider. You can set the OidcProviderArn using the ALIBABA_CLOUD_OIDC_PROVIDER_ARN environment variable.
       oidc_provider_arn=os.environ.get('<OidcProviderArn>'),
       # The OIDC token file path. You can set the OidcTokenFilePath using the ALIBABA_CLOUD_OIDC_TOKEN_FILE environment variable.
       oidc_token_file_path=os.environ.get('<OidcTokenFilePath>'),
       # The role session name. You can set the RoleSessionName using the ALIBABA_CLOUD_ROLE_SESSION_NAME environment variable.
       role_session_name='<RoleSessionName>',
       # Optional. A more restrictive permission policy. Example: {"Statement": [{"Action": ["*"],"Effect": "Allow","Resource": ["*"]}],"Version":"1"}
       policy='<Policy>',
       # The validity period of the session.
       role_session_expiration=3600
   )
   
   cred = Client(config)
   
   credentials_provider = CredentialProviderWrapper(cred)
   
   # Note that ProviderAuthV4 indicates the use of V4 signatures.
   auth = oss2.ProviderAuthV4(credentials_provider)
   
   # yourEndpoint specifies the endpoint for the region where the bucket is located.
   endpoint = '<endpoint>'
   
   # yourBucketName specifies the bucket name.
   bucket_name = 'yourBucketName'
   
   # Specify the region where the bucket is located.
   region = '<region-id>'
   
   # Initialize the bucket by using the Auth instance. Note that the region parameter is required when using V4 signatures;
   # otherwise, an error occurs.
   bucket = oss2.Bucket(auth, endpoint, bucket_name, region=region)
   
   # Use the bucket object for subsequent operations...

Contexto do Function Compute

Mais indicado para funções executadas no FC. O FC assume uma função de serviço para obter um token STS (válido por 36 horas) e o passa através do parâmetro Credentials do contexto. Como o tempo máximo de execução da função é de 24 horas, o token não expira durante a execução, portanto, nenhuma atualização é necessária. Para conceder permissões ao FC para acessar o OSS, consulte Usar uma função de função para conceder permissões ao Function Compute para acessar outros serviços de nuvem.

Inicialize o provedor de credenciais com credenciais do contexto do FC.

Antes de executar o código de exemplo, substitua espaços reservados como <region-id> pela região e endpoint reais. Por exemplo, substitua <region-id> por ap-southeast-1 .

# -*- coding: utf-8 -*-
import oss2
from oss2 import CredentialsProvider
from oss2.credentials import Credentials

def handler(event, context):

    class CredentialProviderWrapper(CredentialsProvider):
        def get_credentials(self):
            creds = context.credentials
            return Credentials(creds.access_key_id, creds.access_key_secret, creds.security_token)

    credentials_provider = CredentialProviderWrapper()
    # Note that ProviderAuthV4 indicates the use of V4 signatures.
    auth = oss2.ProviderAuthV4(credentials_provider)

    # yourEndpoint specifies the endpoint for the region where the bucket is located.
    endpoint = '<endpoint>'

    # yourBucketName specifies the bucket name.
    bucket_name = 'yourBucketName'

    # Specify the region where the bucket is located.
    region = '<region-id>'

    # Initialize the bucket by using the Auth instance. Note that the region parameter is required when using V4 signatures;
    # otherwise, an error occurs.
    bucket = oss2.Bucket(auth, endpoint, bucket_name, region=region)

    # Use the bucket object for subsequent operations...

    return 'success'

URI de credenciais

Utilize este método para aplicativos que obtêm credenciais de um sistema externo. A ferramenta de credenciais recupera um token STS do URI fornecido e lida com a atualização automática.

1. Para garantir que a ferramenta de credenciais possa analisar e usar corretamente o token STS, a resposta do URI deve aderir ao seguinte protocolo:

   • Código de status da resposta: 200

   • Estrutura do corpo da resposta:

     {
         "Code": "Success",
         "AccessKeySecret": "AccessKeySecret",
         "AccessKeyId": "AccessKeyId",
         "Expiration": "2021-09-26T03:46:38Z",
         "SecurityToken": "SecurityToken"
     }

2. Adicione a dependência de credenciais.

   pip install alibabacloud_credentials

3. Configure o URI de credenciais como credencial de acesso.

   Antes de executar o código de exemplo, substitua espaços reservados como <region-id> pela região e endpoint reais. Por exemplo, substitua <region-id> por ap-southeast-1 .

   # -*- coding: utf-8 -*-
   import oss2
   from alibabacloud_credentials.client import Client
   from alibabacloud_credentials.models import Config
   from oss2 import CredentialsProvider
   from oss2.credentials import Credentials
   
   class CredentialProviderWrapper(CredentialsProvider):
       def __init__(self, client):
           self.client = client
   
       def get_credentials(self):
           credential = self.client.get_credential()
           access_key_id = credential.access_key_id
           access_key_secret = credential.access_key_secret
           security_token = credential.security_token
           return Credentials(access_key_id, access_key_secret, security_token)
   
   config = Config(
       type='credentials_uri',
       # The URI for the credentials, in the format http://local_or_remote_uri/. You can set the CredentialsUri using the ALIBABA_CLOUD_CREDENTIALS_URI environment variable.
       credentials_uri='<CredentialsUri>',
   )
   
   cred = Client(config)
   
   credentials_provider = CredentialProviderWrapper(cred)
   
   # Note that ProviderAuthV4 indicates the use of V4 signatures.
   auth = oss2.ProviderAuthV4(credentials_provider)
   
   # yourEndpoint specifies the endpoint for the region where the bucket is located.
   endpoint = '<endpoint>'
   
   # yourBucketName specifies the bucket name.
   bucket_name = 'yourBucketName'
   
   # Specify the region where the bucket is located.
   region = '<region-id>'
   
   # Initialize the bucket by using the Auth instance. Note that the region parameter is required when using V4 signatures;
   # otherwise, an error occurs.
   bucket = oss2.Bucket(auth, endpoint, bucket_name, region=region)
   
   # Use the bucket object for subsequent operations...

AK com rotação automática

Adequado para ambientes onde a AK corre risco de comprometimento. Inicialize o provedor de credenciais com uma ClientKey do KMS, que rotaciona automaticamente a AK do usuário RAM gerenciado. O KMS também suporta rotação imediata se uma AK for comprometida. Para obter uma ClientKey, consulte Criar um ponto de acesso de aplicativo.

1. Adicione a dependência do cliente Secrets Manager.

   pip install aliyun-secret-manager-client

2. Crie um arquivo de configuração chamado secretsmanager.properties.

   # The type of access credential. Fixed value: client_key
   credentials_type=client_key
   
   # The decryption password for the ClientKey. You can read it from an environment variable or a file. Set only one.
   client_key_password_from_env_variable=<your client key private key password environment variable name>
   client_key_password_from_file_path=<your client key private key password file path>
   
   # The path to the private key file of the ClientKey.
   client_key_private_key_path=<your client key private key file path>
   
   # The region of the associated KMS.
   cache_client_region_id=[{"regionId":"<regionId>"}]

3. Use o arquivo de configuração para fornecer informações de credenciais.

   Antes de executar o código de exemplo, substitua espaços reservados como <region-id> pela região e endpoint reais. Por exemplo, substitua <region-id> por ap-southeast-1 .

   # -*- coding: utf-8 -*-
   import json
   import oss2
   from oss2 import CredentialsProvider
   from oss2.credentials import Credentials
   from alibaba_cloud_secretsmanager_client.secret_manager_cache_client_builder import SecretManagerCacheClientBuilder
   
   class CredentialProviderWrapper(CredentialsProvider):
       def get_credentials(self):
           secret_cache_client = SecretManagerCacheClientBuilder.new_client()
           secret_info = secret_cache_client.get_secret_info("<secretName>")
           secret_value_json = json.loads(secret_info.secret_value)
           return Credentials(secret_value_json["AccessKeyId"], secret_value_json["AccessKeySecret"])
   
   credentials_provider = CredentialProviderWrapper()
   
   # Note that ProviderAuthV4 indicates the use of V4 signatures.
   auth = oss2.ProviderAuthV4(credentials_provider)
   
   # yourEndpoint specifies the endpoint for the region where the bucket is located.
   endpoint = '<endpoint>'
   
   # yourBucketName specifies the bucket name.
   bucket_name = 'yourBucketName'
   
   # Specify the region where the bucket is located.
   region = '<region-id>'
   
   # Initialize the bucket by using the Auth instance. Note that the region parameter is required when using V4 signatures;
   # otherwise, an error occurs.
   bucket = oss2.Bucket(auth, endpoint, bucket_name, region=region)
   
   # Use the bucket object for subsequent operations...

Provedor de credenciais personalizado

Se nenhum dos métodos de configuração de credenciais acima atender aos seus requisitos, você pode definir um provedor de credenciais personalizado implementando a interface CredentialProvider. Se sua implementação usar um token STS, você também deverá lidar com a atualização das credenciais.

Antes de executar o código de exemplo, substitua espaços reservados como <region-id> pela região e endpoint reais. Por exemplo, substitua <region-id> por ap-southeast-1 .

# -*- coding: utf-8 -*-
import oss2
from oss2 import CredentialsProvider
from oss2.credentials import Credentials

class CredentialProviderWrapper(CredentialsProvider):
    def get_credentials(self):
        # TODO
        # Implement your custom method for obtaining access credentials.

        # Return long-term credentials: access_key_id, access_key_secret
        return Credentials('<access_key_id>', '<access_key_secrect>')

        # Return temporary credentials: access_key_id, access_key_secret, token
        # For temporary credentials, you need to refresh them based on their expiration time.
        # return Credentials('<access_key_id>', '<access_key_secrect>', '<token>');

credentials_provider = CredentialProviderWrapper()

# Note that ProviderAuthV4 indicates the use of V4 signatures.
auth = oss2.ProviderAuthV4(credentials_provider)

# yourEndpoint specifies the endpoint for the region where the bucket is located.
endpoint = '<endpoint>'

# yourBucketName specifies the bucket name.
bucket_name = 'yourBucketName'

# Specify the region where the bucket is located.
region = '<region-id>'

# Initialize the bucket by using the Auth instance. Note that the region parameter is required when using V4 signatures;
# otherwise, an error occurs.
bucket = oss2.Bucket(auth, endpoint, bucket_name, region=region)

# Use the bucket object for subsequent operations...