def to_bytes(data):
def to_string(data):
def to_unicode(data):
def stringify(input):
def to_bytes(data):
def to_string(data):
def to_unicode(data):
def stringify(input): class LocalRsaProvider(BaseCryptoProvider):
    """使用本地RSA加密数据密钥。

        :param str dir: 本地RSA公钥私钥存储路径
        :param str key: 本地RSA公钥私钥名称前缀
        :param str passphrase: 本地RSA公钥私钥密码
        :param class cipher: 数据加密，默认aes256，用户可自行实现对称加密算法，需符合AESCipher注释规则
    """

    DEFAULT_PUB_KEY_SUFFIX = '.public_key.pem'
    DEFAULT_PRIV_KEY_SUFFIX = '.private_key.pem'

    def __init__(self, dir=None, key='', passphrase=None, cipher=utils.AESCTRCipher(),
                 pub_key_suffix=DEFAULT_PUB_KEY_SUFFIX, private_key_suffix=DEFAULT_PRIV_KEY_SUFFIX):

        super(LocalRsaProvider, self).__init__(cipher=cipher)

        self.wrap_alg = headers.RSA_NONE_OAEPWithSHA1AndMGF1Padding
        keys_dir = dir or os.path.join(os.path.expanduser('~'), _LOCAL_RSA_TMP_DIR)

        priv_key_path = os.path.join(keys_dir, key + private_key_suffix)
        pub_key_path = os.path.join(keys_dir, key + pub_key_suffix)
        try:
            if os.path.exists(priv_key_path) and os.path.exists(pub_key_path):
                with open(priv_key_path, 'rb') as f:
                    self.__decrypt_obj = PKCS1_OAEP.new(RSA.importKey(f.read(), passphrase=passphrase))

                with open(pub_key_path, 'rb') as f:
                    self.__encrypt_obj = PKCS1_OAEP.new(RSA.importKey(f.read(), passphrase=passphrase))

            else:
                logger.warn('The file path of private key or public key is not exist, will generate key pair')
                private_key = RSA.generate(2048)
                public_key = private_key.publickey()

                self.__encrypt_obj = PKCS1_OAEP.new(public_key)
                self.__decrypt_obj = PKCS1_OAEP.new(private_key)

                utils.makedir_p(keys_dir)
                with open(priv_key_path, 'wb') as f:
                    f.write(private_key.exportKey(passphrase=passphrase))

                with open(pub_key_path, 'wb') as f:
                    f.write(public_key.exportKey(passphrase=passphrase))
        except (ValueError, TypeError, IndexError) as e:
            raise ClientError(str(e))

    def get_key(self):
        return self.cipher.get_key()

    def decrypt_encrypted_key(self, encrypted_key):
        try:
            return self.__decrypt_data(encrypted_key)
        except (TypeError, ValueError) as e:
            raise ClientError(str(e))

    def decrypt_encrypted_iv(self, encrypted_iv):
        try:
            return self.__decrypt_data(encrypted_iv)
        except (TypeError, ValueError) as e:
            raise ClientError(str(e))

    def reset_encryption_materials(self, encryption_materials):
        raise ClientError("do not support reset_encryption_materials!")

    def create_content_material(self):
        plain_key = self.get_key()
        encrypted_key = self.__encrypt_data(plain_key)
        plain_iv = self.get_iv()
        encrypted_iv = self.__encrypt_data(plain_iv)
        cipher = copy.copy(self.cipher)
        wrap_alg = self.wrap_alg
        mat_desc = self.mat_desc

        cipher.initialize(plain_key, plain_iv)

        content_crypto_material = models.ContentCryptoMaterial(cipher, wrap_alg, encrypted_key, encrypted_iv,
                                                               mat_desc)
        return content_crypto_material

    def __encrypt_data(self, data):
        return self.__encrypt_obj.encrypt(data)

    def __decrypt_data(self, data):
        return self.__decrypt_obj.decrypt(data) class AliKMSProvider(BaseCryptoProvider):
    """使用aliyun kms服务加密数据密钥。kms的详细说明参见
        https://help.aliyun.com/product/28933.html?spm=a2c4g.11186623.3.1.jlYT4v
        此接口在py3.3下暂时不可用，详见
        https://github.com/aliyun/aliyun-openapi-python-sdk/issues/61

        :param str access_key_id: 可以访问kms密钥服务的access_key_id
        :param str access_key_secret: 可以访问kms密钥服务的access_key_secret
        :param str region: kms密钥服务地区
        :param str cmkey: 用户主密钥
        :param str sts_token: security token，如果使用的是临时AK需提供
        :param str passphrase: kms密钥服务密码
        :param class cipher: 数据加密，默认aes256，当前仅支持默认实现
    """

    def __init__(self, access_key_id, access_key_secret, region, cmk_id, sts_token=None, passphrase=None,
                 cipher=utils.AESCTRCipher(), mat_desc=None):

        super(AliKMSProvider, self).__init__(cipher=cipher, mat_desc=mat_desc)
        if not isinstance(cipher, utils.AESCTRCipher):
            raise ClientError('AliKMSProvider only support AES256 cipher now')
        self.wrap_alg = headers.KMS_ALI_WRAP_ALGORITHM
        self.custom_master_key_id = cmk_id
        self.sts_token = sts_token
        self.context = '{"x-passphrase":"' + passphrase + '"}' if passphrase else ''
        self.kms_client = client.AcsClient(access_key_id, access_key_secret, region)

    def get_key(self):
        plain_key, encrypted_key = self.__generate_data_key()
        return plain_key, encrypted_key

    def decrypt_encrypted_key(self, encrypted_key):
        return b64decode_from_string(self.__decrypt_data(encrypted_key))

    def decrypt_encrypted_iv(self, encrypted_iv, deprecated=False):
        if deprecated:
            return self.__decrypt_data(encrypted_iv)
        return b64decode_from_string(self.__decrypt_data(encrypted_iv))

    def reset_encryption_materials(self, encryption_materials):
        provider = copy.copy(self)
        provider.custom_master_key_id = encryption_materials.custom_master_key_id
        provider.context = '{"x-passphrase":"' + encryption_materials.passphrase + '"}' if encryption_materials.passphrase else ''
        provider.mat_desc = encryption_materials.desc
        return provider

    def create_content_material(self):
        plain_key, encrypted_key = self.get_key()
        plain_iv = self.get_iv()
        encrypted_iv = self.__encrypt_data(b64encode_as_string(plain_iv))
        cipher = copy.copy(self.cipher)
        wrap_alg = self.wrap_alg
        mat_desc = self.mat_desc

        cipher.initialize(plain_key, plain_iv)

        content_crypto_material = models.ContentCryptoMaterial(cipher, wrap_alg, encrypted_key, encrypted_iv,
                                                               mat_desc)
        return content_crypto_material

    def __generate_data_key(self):
        req = GenerateDataKeyRequest.GenerateDataKeyRequest()

        req.set_accept_format(format_type.JSON)
        req.set_method(method_type.POST)

        req.set_KeyId(self.custom_master_key_id)
        req.set_KeySpec('AES_256')
        req.set_NumberOfBytes(32)
        req.set_EncryptionContext(self.context)
        if self.sts_token:
            req.set_STSToken(self.sts_token)

        resp = self.__do(req)

        return b64decode_from_string(resp['Plaintext']), resp['CiphertextBlob']

    def __encrypt_data(self, data):
        req = EncryptRequest.EncryptRequest()

        req.set_accept_format(format_type.JSON)
        req.set_method(method_type.POST)
        req.set_KeyId(self.custom_master_key_id)
        req.set_Plaintext(data)
        req.set_EncryptionContext(self.context)
        if self.sts_token:
            req.set_STSToken(self.sts_token)

        resp = self.__do(req)

        return resp['CiphertextBlob']

    def __decrypt_data(self, data):
        req = DecryptRequest.DecryptRequest()

        req.set_accept_format(format_type.JSON)
        req.set_method(method_type.POST)
        req.set_CiphertextBlob(data)
        req.set_EncryptionContext(self.context)
        if self.sts_token:
            req.set_STSToken(self.sts_token)

        resp = self.__do(req)
        return resp['Plaintext']

    def __do(self, req):

        try:
            body = self.kms_client.do_action_with_exception(req)
            return json.loads(to_unicode(body))
        except ServerException as e:
            raise OpenApiServerError(e.http_status, e.request_id, e.message, e.error_code)
        except ClientException as e:
            raise ClientError(e.message)
        except (KeyError, ValueError, TypeError) as e:
            raise OpenApiFormatError('Json Error: ' + str(e)) class RsaProvider(BaseCryptoProvider):
    """使用本地RSA加密数据密钥。

        :param str dir: 本地RSA公钥私钥存储路径
        :param str key: 本地RSA公钥私钥名称前缀
        :param str passphrase: 本地RSA公钥私钥密码
        :param class cipher: 数据加密，默认aes256，用户可自行实现对称加密算法，需符合AESCipher注释规则
    """

    def __init__(self, key_pair, passphrase=None, cipher=utils.AESCTRCipher(), mat_desc=None):

        super(RsaProvider, self).__init__(cipher=cipher, mat_desc=mat_desc)
        self.wrap_alg = headers.RSA_NONE_PKCS1Padding_WRAP_ALGORITHM

        if key_pair and not isinstance(key_pair, dict):
            raise ClientError('Invalid type, the type of key_pair must be dict!')

        try:
            if 'public_key' in key_pair:
                self.__encrypt_obj = PKCS1_v1_5.new(RSA.importKey(key_pair['public_key'], passphrase=passphrase))

            if 'private_key' in key_pair:
                self.__decrypt_obj = PKCS1_v1_5.new(RSA.importKey(key_pair['private_key'], passphrase=passphrase))
        except (ValueError, TypeError) as e:
            raise ClientError(str(e))

    def get_key(self):
        return self.cipher.get_key()

    def decrypt_encrypted_key(self, encrypted_key):
        try:
            return self.__decrypt_data(encrypted_key)
        except (TypeError, ValueError) as e:
            raise ClientError(str(e))

    def decrypt_encrypted_iv(self, encrypted_iv):
        try:
            return self.__decrypt_data(encrypted_iv)
        except (TypeError, ValueError) as e:
            raise ClientError(str(e))

    def reset_encryption_materials(self, encryption_materials):
        return RsaProvider(encryption_materials.key_pair, encryption_materials.passphrase, self.cipher,
                           encryption_materials.desc)

    def create_content_material(self):
        plain_key = self.get_key()
        encrypted_key = self.__encrypt_data(plain_key)
        plain_iv = self.get_iv()
        encrypted_iv = self.__encrypt_data(plain_iv)
        cipher = copy.copy(self.cipher)
        wrap_alg = self.wrap_alg
        mat_desc = self.mat_desc

        cipher.initialize(plain_key, plain_iv)

        content_crypto_material = models.ContentCryptoMaterial(cipher, wrap_alg, encrypted_key, encrypted_iv,
                                                               mat_desc)
        return content_crypto_material

    def __encrypt_data(self, data):
        return self.__encrypt_obj.encrypt(data)

    def __decrypt_data(self, data):
        decrypted_data = self.__decrypt_obj.decrypt(data, object)
        if decrypted_data == object:
            raise ClientError('Decrypted data error, please check you key pair!')
        return decrypted_data class AESCipher(object):
    """AES256 加密实现。
            :param str key: 对称加密数据密钥
            :param str start: 对称加密初始随机值
        .. note::
            用户可自行实现对称加密算法，需服务如下规则：
            1、提供对称加密算法名，ALGORITHM
            2、提供静态方法，返回加密密钥和初始随机值（若算法不需要初始随机值，也需要提供）
            3、提供加密解密方法
    """

    # aes 256, key always is 32 bytes
    def __init__(self):
        self.alg = None
        self.key_len = _AES_256_KEY_SIZE
        self.block_size_len = _AES_BLOCK_LEN
        self.block_size_len_in_bits = _AES_BLOCK_BITS_LEN

    @abc.abstractmethod
    def get_key(self):
        pass

    @abc.abstractmethod
    def get_iv(self):
        pass

    @abc.abstractmethod
    def initialize(self, key, iv, off=0):
        pass

    @abc.abstractmethod
    def encrypt(self, raw):
        pass

    @abc.abstractmethod
    def decrypt(self, enc):
        pass

    @abc.abstractmethod
    def determine_part_size(self, data_size, excepted_part_size=None):
        pass

    def adjust_range(self, start, end):
        return start, end

    def is_block_aligned(self, offset):
        if offset is None:
            offset = 0
        return 0 == offset % self.block_size_len

    def is_valid_part_size(self, part_size, data_size=None):
        return True def silently_remove(filename):
    """删除文件，如果文件不存在也不报错。"""
    try:
        os.remove(filename)
    except OSError as e:
        if e.errno != errno.ENOENT:
            raise class OpenApiServerError(OssError):
    def __init__(self, status, request_id, message, error_code):
        OssError.__init__(self, status, {OSS_REQUEST_ID : request_id}, '', {'Code': error_code, 'Message': message}) class OpenApiFormatError(OssError):
    def __init__(self, message):
        OssError.__init__(self, OSS_FORMAT_ERROR_STATUS, {}, message, {})

    def __str__(self):
        return self._str_with_body() class ClientError(OssError):
    def __init__(self, message):
        OssError.__init__(self, OSS_CLIENT_ERROR_STATUS, {}, 'ClientError: ' + message, {})

    def __str__(self):
        return self._str_with_body() OSS_ID = os.getenv("OSS_TEST_ACCESS_KEY_ID")
OSS_SECRET = os.getenv("OSS_TEST_ACCESS_KEY_SECRET")
OSS_ENDPOINT = os.getenv("OSS_TEST_ENDPOINT")
OSS_TEST_BUCKET = os.getenv("OSS_TEST_BUCKET")
OSS_CNAME = os.getenv("OSS_TEST_CNAME")
OSS_REGION = os.getenv("OSS_TEST_REGION", "cn-hangzhou")
OSS_CMK = os.getenv("OSS_TEST_KMS_CMK_ID")
OSS_CMK_REGION = os.getenv("OSS_TEST_KMS_REGION")
OSS_STS_ID = os.getenv("OSS_TEST_STS_ID")
OSS_STS_KEY = os.getenv("OSS_TEST_STS_KEY")
OSS_STS_ARN = os.getenv("OSS_TEST_STS_ARN")
OSS_PAYER_UID = os.getenv("OSS_TEST_PAYER_UID")
OSS_PAYER_ID = os.getenv("OSS_TEST_PAYER_ACCESS_KEY_ID")
OSS_PAYER_SECRET = os.getenv("OSS_TEST_PAYER_ACCESS_KEY_SECRET")
OSS_INVENTORY_BUCKET_DESTINATION_ARN = os.getenv("OSS_TEST_RAM_ROLE_ARN")
OSS_INVENTORY_BUCKET_DESTINATION_ACCOUNT = os.getenv("OSS_TEST_RAM_UID")
OSS_AUTH_VERSION = None
OSS_TEST_AUTH_SERVER_HOST = os.getenv("OSS_TEST_AUTH_SERVER_HOST")
OSS_BUCKET = ''
    OSS_BUCKET = 'oss-python-sdk-'+random_string(10)
    OSS_BUCKET = OSS_TEST_BUCKET + random_string(10)
    SINGLE_THREAD_CASE = 'single thread case'
        OSS_AUTH_VERSION = os.getenv('OSS_TEST_AUTH_VERSION')
def random_string(n):
def random_bytes(n):
def clean_and_delete_bucket(bucket):
def clean_and_delete_bucket_by_prefix(bucket_prefix):
def delete_keys(bucket, key_list):
    def __init__(self, value):
def wait_meta_sync():
    def __init__(self, *args, **kwargs):
    def setUp(self):
    def tearDown(self):
    def random_key(self, suffix=''):
    def random_filename(self):
    def _prepare_temp_file(self, content):
    def _prepare_temp_file_with_size(self, size):
    def retry_assert(self, func):
    def assertFileContent(self, filename, content):
    def assertFileContentNotEqual(self, filename, content):
class NonlocalObject(object):
class OssTestCase(unittest.TestCase):
import json
import unittest
import os
import oss2
import unittests
import random
from aliyunsdkcore import client
from oss2 import LocalRsaProvider, AliKMSProvider, RsaProvider, compat
from oss2.utils import AESCipher, silently_remove
from oss2.exceptions import OpenApiServerError, OpenApiFormatError, ClientError
from mock import patch
from .common import OSS_ID, OSS_SECRET, OSS_REGION, OSS_CMK_REGION, OSS_CMK, OSS_STS_ID, OSS_STS_ARN, OSS_STS_KEY, random_string, \
    key_pair, key_pair_compact
from aliyunsdksts.request.v20150401 import AssumeRoleRequest
from Crypto.PublicKey import RSA
from Crypto.PublicKey.RSA import RsaKey
    # 测试基本key, start加/解密
    def test_rsa_provider_basic(self):
        silently_remove('./rsa-test.public_key.pem')
        silently_remove('./rsa-test.private_key.pem')

        provider = LocalRsaProvider(dir='./', key='rsa-test', passphrase=random_string(8))
        self.assertEqual(provider.wrap_alg, "RSA/NONE/OAEPWithSHA-1AndMGF1Padding")
        self.assertEqual(provider.cipher.alg, "AES/CTR/NoPadding")
        plain_key = provider.get_key()
        self.assertEqual(len(plain_key), provider.cipher.key_len)
        plain_iv = provider.get_iv()

        with patch.object(oss2.utils, 'random_key', return_value=plain_key, autospect=True):
            with patch.object(oss2.utils, 'random_iv', return_value=plain_iv, autospect=True):
                content_crypto_material = provider.create_content_material()
                self.assertFalse(content_crypto_material.is_unencrypted())
                decrypted_key = provider.decrypt_encrypted_key(content_crypto_material.encrypted_key)
                decrypted_iv = provider.decrypt_encrypted_iv(content_crypto_material.encrypted_iv)
                self.assertEqual(plain_key, decrypted_key)
                self.assertEqual(plain_iv, decrypted_iv)

        silently_remove('./rsa-test.public_key.pem')
        silently_remove('./rsa-test.private_key.pem')

        provider = RsaProvider(key_pair=key_pair, passphrase=random_string(8))
        self.assertEqual(provider.wrap_alg, "RSA/NONE/PKCS1Padding")
        self.assertEqual(provider.cipher.alg, "AES/CTR/NoPadding")
        plain_key = provider.get_key()
        self.assertEqual(len(plain_key), provider.cipher.key_len)
        plain_iv = provider.get_iv()

        with patch.object(oss2.utils, 'random_key', return_value=plain_key, autospect=True):
            with patch.object(oss2.utils, 'random_iv', return_value=plain_iv, autospect=True):
                content_crypto_material = provider.create_content_material()
                self.assertFalse(content_crypto_material.is_unencrypted())
                decrypted_key = provider.decrypt_encrypted_key(content_crypto_material.encrypted_key)
                decrypted_iv = provider.decrypt_encrypted_iv(content_crypto_material.encrypted_iv)
                self.assertEqual(plain_key, decrypted_key)
                self.assertEqual(plain_iv, decrypted_iv)

    # 测试使用不同的rsa keys的provider
    def test_local_rsa_provider_diff_keys(self):
        silently_remove('./rsa-test.public_key.pem')
        silently_remove('./rsa-test.private_key.pem')
        silently_remove('./rsa-test-diff.public_key.pem')
        silently_remove('./rsa-test-diff.private_key.pem')

        provider = LocalRsaProvider(dir='./', key='rsa-test')
        provider_diff = LocalRsaProvider(dir='./', key='rsa-test-diff')

        plain_key = provider.get_key()
        plain_iv = provider.get_iv()

        with patch.object(oss2.utils, 'random_key', return_value=plain_key, autospect=True):
            with patch.object(oss2.utils, 'random_iv', return_value=plain_iv, autospect=True):
                content_crypto_material = provider.create_content_material()
                self.assertFalse(content_crypto_material.is_unencrypted())
                self.assertRaises(ClientError, provider_diff.decrypt_encrypted_key,
                                  content_crypto_material.encrypted_key)
                self.assertRaises(ClientError, provider_diff.decrypt_encrypted_iv,
                                  content_crypto_material.encrypted_iv)

        silently_remove('./rsa-test.public_key.pem')
        silently_remove('./rsa-test.private_key.pem')
        silently_remove('./rsa-test-diff.public_key.pem')
        silently_remove('./rsa-test-diff.private_key.pem')

        provider = RsaProvider(key_pair=key_pair)
        provider_diff = RsaProvider(key_pair=key_pair_compact)

        plain_key = provider.get_key()
        plain_iv = provider.get_iv()

        with patch.object(oss2.utils, 'random_key', return_value=plain_key, autospect=True):
            with patch.object(oss2.utils, 'random_iv', return_value=plain_iv, autospect=True):
                content_crypto_material = provider.create_content_material()
                self.assertFalse(content_crypto_material.is_unencrypted())
                self.assertRaises(ClientError, provider_diff.decrypt_encrypted_key,
                                  content_crypto_material.encrypted_key)
                self.assertRaises(ClientError, provider_diff.decrypt_encrypted_iv,
                                  content_crypto_material.encrypted_iv)

    def test_rsa_provider_adapter(self):
        silently_remove('./rsa-test.public_key.pem')
        silently_remove('./rsa-test.private_key.pem')

        provider = LocalRsaProvider(dir='./', key='rsa-test')
        content = b'a' * random.randint(1, 100) * 1024
        content_crypto_material = provider.create_content_material()
        plain_key = provider.decrypt_encrypted_key(content_crypto_material.encrypted_key)
        plain_iv = provider.decrypt_encrypted_iv(content_crypto_material.encrypted_iv)
        cipher = content_crypto_material.cipher

        stream_encrypted = provider.make_encrypt_adapter(content, cipher)
        encrypted_content = stream_encrypted.read()
        # reset cipher
        cipher.initialize(plain_key, plain_iv)
        stream_decrypted = provider.make_decrypt_adapter(encrypted_content, cipher)
        self.assertEqual(content, stream_decrypted.read())

        silently_remove('./rsa-test.public_key.pem')
        silently_remove('./rsa-test.private_key.pem')

        provider = RsaProvider(key_pair)
        content = b'b' * random.randint(1, 100) * 1024
        content_crypto_material = provider.create_content_material()
        plain_key = provider.decrypt_encrypted_key(content_crypto_material.encrypted_key)
        plain_iv = provider.decrypt_encrypted_iv(content_crypto_material.encrypted_iv)
        cipher = content_crypto_material.cipher

        stream_encrypted = provider.make_encrypt_adapter(content, cipher)
        encrypted_content = stream_encrypted.read()
        # reset cipher
        cipher.initialize(plain_key, plain_iv)
        stream_decrypted = provider.make_decrypt_adapter(encrypted_content, cipher)
        self.assertEqual(content, stream_decrypted.read())

    # 测试初始化AliKMSProvider时未初始化cipher，此时应该抛出异常
    def test_ali_kms_provider_init_cipher_is_none(self):
        id, key, token = self.get_sts()

        self.assertRaises(ClientError, AliKMSProvider, access_key_id=id, access_key_secret=key, region=OSS_REGION,