#!/usr/bin/python3 # -*- coding: utf-8 -*- import argparse import json import os from alibabacloud_kms20160120.client import Client as Kms20160120Client from alibabacloud_credentials.client import Client as CredentialClient from alibabacloud_kms20160120.models import GetSecretValueResponse from alibabacloud_tea_openapi import models as open_api_models from alibabacloud_kms20160120 import models as kms_20160120_models from alibabacloud_tea_util import models as util_models from alibabacloud_tea_util.client import Client as UtilClient from alibabacloud_credentials.models import Config class AliyunKMS(object): """ https://help.aliyun.com/zh/sdk/developer-reference/v2-manage-python-access-credentials """ def __init__(self, access_key_id: str = None, access_key_secret: str = None, endpoint: str = "kms.ap-southeast-1.aliyuncs.com", ): self.access_key_id = access_key_id self.access_key_secret = access_key_secret self.endpoint = endpoint self.client = self.get_client() def get_client(self): credential = CredentialClient( config=Config( type="access_key", access_key_id=self.access_key_id, access_key_secret=self.access_key_secret, ) ) config = open_api_models.Config( credential=credential ) # Endpoint 请参考 https://api.aliyun.com/product/Kms config.endpoint = self.endpoint client = Kms20160120Client(config) return client def create_secret(self, secret_name: str, secret_data: str, version_id: str): """ https://next.api.aliyun.com/api/Kms/2016-01-20/CreateSecret """ create_secret_request = kms_20160120_models.CreateSecretRequest( secret_name=secret_name, secret_data=secret_data, version_id=version_id, ) runtime = util_models.RuntimeOptions() result = self.client.create_secret_with_options(create_secret_request, runtime) return result def get_secret_value(self, secret_name: str, version_id: str = None): """ https://next.api.aliyun.com/api/Kms/2016-01-20/GetSecretValue """ get_secret_value_request = kms_20160120_models.GetSecretValueRequest( secret_name=secret_name, version_id=version_id, ) runtime = util_models.RuntimeOptions() response: GetSecretValueResponse = self.client.get_secret_value_with_options(get_secret_value_request, runtime) js = response.to_map() return js async def async_get_secret_value(self, secret_name: str, version_id: str = None): """ https://next.api.aliyun.com/api/Kms/2016-01-20/GetSecretValue """ get_secret_value_request = kms_20160120_models.GetSecretValueRequest( secret_name=secret_name, version_id=version_id, ) runtime = util_models.RuntimeOptions() response: GetSecretValueResponse = await self.client.get_secret_value_with_options_async(get_secret_value_request, runtime) js = response.to_map() return js def main(): from settings import environment access_key_id = environment.get(key="ALIBABA_CLOUD_ACCESS_KEY_ID", dtype=str) access_key_secret = environment.get(key="ALIBABA_CLOUD_ACCESS_KEY_SECRET", dtype=str) print(f"access_key_id: {access_key_id}") print(f"access_key_secret: {access_key_secret}") # os.environ["ALIBABA_CLOUD_ACCESS_KEY_ID"] = access_key_id # os.environ["ALIBABA_CLOUD_ACCESS_KEY_SECRET"] = access_key_secret manager = AliyunKMS( access_key_id=access_key_id, access_key_secret=access_key_secret, ) # result = manager.get_secret_value( # secret_name="azure-east-asia-asr-dev", # version_id="v1", # ) # print(result) result = manager.get_secret_value( secret_name="aliyun-nxai123-oss-dev", # version_id="d5b82ac1ee63d748b25bf7be6c75695e", ) print(result) return if __name__ == "__main__": main()