ak-kms怎么构造

[ycq0125]

from aliyun_sdk_secretsmanager_oss_plugin.proxy_bucket import ProxyBucket
from itertools import islice

secret_name = "/test-"
endpoint = "oss-cn-shanghai-internal.aliyuncs.com"
bucket_name = "xxxx"
bucket = ProxyBucket(secret_name=secret_name,
                     endpoint=endpoint, bucket_name=bucket_name)
objects = bucket.list_objects()

for b in islice(objects.object_list, 10):
    print(b.key)
bucket.shutdown()

@alibaba-oss @huiguangjun @zhuxiaolong37 想知道这个包有上面方法对应的等价实现吗？

[huiguangjun]

SDK 提供了 CredentialsProvider 机制，可以很方便与 第三方凭证实现结合在一起。 具体做法如下：

1）参考 kms 凭据客户端 文档，初始化 client，并获取 凭证信息

2）通过 自定义凭证者 使用 KMS 凭证

示例如下：

构建KMS凭据客户端
from alibaba_cloud_secretsmanager_client.secret_manager_cache_client_builder import SecretManagerCacheClientBuilder

secret_cache_client = SecretManagerCacheClientBuilder.new_client()

自定义OSS SDK 凭证者
import alibabacloud_oss_v2 as oss

def get_credentials_wrapper():
返回临时凭证
使用构建好的客户端获取凭据信息
secret_info = secret_cache_client.get_secret_info("secretName")

提取 secret_info里的 ak，sk 和 token 信息，转成 sdk 凭证 
return oss.credentials.Credentials(access_key_id=ak, access_key_secret=sk , security_token=token )

credentials_provider = oss.credentials.CredentialsProviderFunc(func=get_credentials_wrapper)

cfg = oss.config.load_default()
cfg.credentials_provider = credentials_provider
cfg.region = 'cn-hangzhou'

client = oss.Client(cfg)

[ycq0125]

SDK 提供了 CredentialsProvider 机制，可以很方便与 第三方凭证实现结合在一起。 具体做法如下：

1）参考 kms 凭据客户端 文档，初始化 client，并获取 凭证信息

2）通过 自定义凭证者 使用 KMS 凭证

示例如下：

构建KMS凭据客户端 from alibaba_cloud_secretsmanager_client.secret_manager_cache_client_builder import SecretManagerCacheClientBuilder

secret_cache_client = SecretManagerCacheClientBuilder.new_client()

自定义OSS SDK 凭证者 import alibabacloud_oss_v2 as oss

def get_credentials_wrapper(): 返回临时凭证 使用构建好的客户端获取凭据信息 secret_info = secret_cache_client.get_secret_info("secretName")

提取 secret_info里的 ak，sk 和 token 信息，转成 sdk 凭证 
return oss.credentials.Credentials(access_key_id=ak, access_key_secret=sk , security_token=token )

credentials_provider = oss.credentials.CredentialsProviderFunc(func=get_credentials_wrapper)

cfg = oss.config.load_default() cfg.credentials_provider = credentials_provider cfg.region = 'cn-hangzhou'

client = oss.Client(cfg)

@huiguangjun @zhuxiaolong37 @alibaba-oss 目前我使用的是python3.11版本的python，但是aliyun-secret-manager-client不支持这种高版本，还有其他的方式吗？

[ycq0125]

SDK 提供了 CredentialsProvider 机制，可以很方便与 第三方凭证实现结合在一起。 具体做法如下：
1）参考 kms 凭据客户端 文档，初始化 client，并获取 凭证信息
2）通过 自定义凭证者 使用 KMS 凭证
示例如下：
构建KMS凭据客户端 from alibaba_cloud_secretsmanager_client.secret_manager_cache_client_builder import SecretManagerCacheClientBuilder
secret_cache_client = SecretManagerCacheClientBuilder.new_client()
自定义OSS SDK 凭证者 import alibabacloud_oss_v2 as oss
def get_credentials_wrapper(): 返回临时凭证 使用构建好的客户端获取凭据信息 secret_info = secret_cache_client.get_secret_info("secretName")

提取 secret_info里的 ak，sk 和 token 信息，转成 sdk 凭证 
return oss.credentials.Credentials(access_key_id=ak, access_key_secret=sk , security_token=token )

credentials_provider = oss.credentials.CredentialsProviderFunc(func=get_credentials_wrapper)
cfg = oss.config.load_default() cfg.credentials_provider = credentials_provider cfg.region = 'cn-hangzhou'
client = oss.Client(cfg)

@huiguangjun @zhuxiaolong37 @alibaba-oss 目前我使用的是python3.11版本的python，但是aliyun-secret-manager-client不支持这种高版本，还有其他的方式吗？

from functools import cached_property
from typing import Optional
from conf.db_config import OSSConfig
import alibabacloud_oss_v2 as oss
from alibaba_cloud_secretsmanager_client.secret_manager_cache_client_builder import SecretManagerCacheClientBuilder
import json
__all__ = ["OssDB", "oss_db"]
def apply_python3_compatibility_patch():
    """修复alibaba_cloud_secretsmanager_client在Python 3中的兼容性问题"""
    import alibaba_cloud_secretsmanager_client.utils.config_utils as config_utils
    def patched_get_properties(self):
        config_dict = {}
        try:
            with open(self.file_name, 'r', encoding='utf-8') as pro_file:
                for line in pro_file:
                    line = line.strip()
                    if line and not line.startswith('#'):
                        if '=' in line:
                            key, value = line.split('=', 1)
                            config_dict[key.strip()] = value.strip()
            return config_dict
        except Exception as e:
            raise e
    config_utils.Properties.get_properties = patched_get_properties
# 应用补丁
apply_python3_compatibility_patch()
class OssDB:
    def __init__(self, config: OSSConfig):
        self.__client: Optional[oss.Client] = None
        self.__config: OSSConfig = config
        self.__secret_cache_client = SecretManagerCacheClientBuilder.new_client()
    def get_credentials_wrapper(self):
        secret_info = self.__secret_cache_client.get_secret_info(self.__config.user_path)
        secret_value = json.loads(secret_info.secret_value)
        access_key_id = secret_value.get('AccessKeyId')
        access_key_secret = secret_value.get('AccessKeySecret')
        return oss.credentials.Credentials(
            access_key_id=access_key_id,
            access_key_secret=access_key_secret,
        )
    @cached_property
    def client(self):
        if self.__client:
            return self.__client
        credentials_provider = oss.credentials.CredentialsProviderFunc(func=self.get_credentials_wrapper)
        cfg = oss.config.load_default()
        cfg.credentials_provider = credentials_provider
        cfg.region = self.__config.region
        cfg.endpoint = self.__config.endpoint
        cfg.retry_max_attempts = self.__config.retry_max_attempts
        self.__client = oss.Client(config=cfg)
        return self.__client
oss_db = OssDB(OSSConfig())

python3.11需要加补丁