toolbox/aliyun_kms/aliyun_kms.py

#!/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()