客户端加密(Python SDK V2)

OSS客户端加密是在数据上传至OSS之前,由用户在本地对数据进行加密处理,确保只有密钥持有者才能解密数据,增强数据在传输和存储过程中的安全性。

注意事项

  • 本文示例代码以华东1(杭州)的地域IDcn-hangzhou为例,默认使用外网Endpoint,如果您希望通过与OSS同地域的其他阿里云产品访问OSS,请使用内网Endpoint。关于OSS支持的RegionEndpoint的对应关系,请参见OSS地域和访问域名

  • 使用客户端加密功能时,您需要对主密钥的完整性和正确性负责。

  • 在对加密数据进行复制或者迁移时,您需要对加密元数据的完整性和正确性负责。

方法定义

对于主密钥的使用,Python SDK V2目前支持如下两种方式:

  • 使用用户自主管理的主密钥(RSA)

    SDK提供了RSA的默认实现,当主密钥信息由用户提供时,用户需要将主密钥的公钥、私钥信息作为参数传递给SDK。

  • 使用用户自定义的主密钥

    RSA主密钥方式无法满足需求时,用户可以自己实现主密钥的加解密行为。

使用以上两种加密方式能够有效地避免数据泄漏,保护客户端数据安全。即使数据泄漏,其他人也无法解密得到原始数据。

重要

如果您需要了解OSS客户端加密实现的原理,请参考OSS用户指南中的客户端加密

使用客户端加密,首先您需要实例化加密客户端,然后调用其提供的接口进行操作。您的对象将作为请求的一部分自动加密和解密。

class EncryptionClient:
  ...

def __init__(self,client: Client, master_cipher: MasterCipher, decrypt_master_ciphers: Optional[List[MasterCipher]] = None)

请求参数列表

参数名

类型

说明

client

*Client

非加密客户端实例

master_cipher

MasterCipher

主密钥实例,用于加密和解密数据密钥

decrypt_master_ciphers

List[MasterCipher]

主密钥实例,用于解密数据密钥

EncryptionClient接口列举如下:

基础接口名

说明

get_object_meta

获取对象的部分元信息

head_object

获取对象的部元信息

get_object

下载对象,并自动解密

put_object

上传对象,并自动加密

initiate_multipart_upload

初始化一个分片上传事件 和 分片加密上下文(EncryptionMultiPartContext)

upload_part

初始化一个分片上传事件, 调用该接口上传分片数据,并自动加密。调用该接口时,需要设置 分片加密上下文

complete_multipart_upload

在将所有分片数据上传完成后,调用该接口合并成一个文件

abort_multipart_upload

取消分片上传事件,并删除对应的分片数据

list_parts

列举指定上传事件所属的所有已经上传成功分片

使用RSA主密钥

使用主密钥RSA简单上传和下载Object

使用主密钥RSA简单上传和下载Object示例代码如下:

import argparse
import alibabacloud_oss_v2 as oss
import alibabacloud_oss_v2.crypto
from alibabacloud_oss_v2.encryption_client import EncryptionClient, EncryptionMultiPartContext

# 创建命令行参数解析器,用于接收用户输入的参数
parser = argparse.ArgumentParser(description="encryption put object sample")

# 添加命令行参数 --region,表示存储空间所在的地域,必填项
parser.add_argument('--region', help='The region in which the bucket is located.', required=True)

# 添加命令行参数 --bucket,表示存储空间的名称,必填项
parser.add_argument('--bucket', help='The name of the bucket.', required=True)

# 添加命令行参数 --endpoint,表示其他服务访问 OSS 时使用的域名,可选项
parser.add_argument('--endpoint', help='The domain names that other services can use to access OSS')

# 添加命令行参数 --key,表示对象的名称,必填项
parser.add_argument('--key', help='The name of the object.', required=True)

# 定义 RSA 公钥和私钥,用于加密和解密操作
RSA_PUBLIC_KEY = """-----BEGIN PUBLIC KEY-----
MIGfMA0G6mse2QsIgz3******GBcom6kEF6MmR1EKixaQIDAQAB
-----END PUBLIC KEY-----"""

RSA_PRIVATE_KEY = """-----BEGIN PRIVATE KEY-----
MIICdQIBADANBgk******ItewfwXIL1Mqz53lO/gK+q6TR92gGc+4ajL
-----END PRIVATE KEY-----"""

def main():
    # 解析命令行参数
    args = parser.parse_args()

    # 从环境变量中加载凭证信息(AccessKeyId 和 AccessKeySecret)
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()

    # 加载 SDK 的默认配置
    cfg = oss.config.load_default()

    # 设置凭证提供者
    cfg.credentials_provider = credentials_provider

    # 设置存储空间所在的地域
    cfg.region = args.region

    # 如果用户提供了自定义的 endpoint,则设置到配置中
    if args.endpoint is not None:
        cfg.endpoint = args.endpoint

    # 使用配置对象初始化 OSS 客户端
    client = oss.Client(cfg)

    # 初始化 MasterRsaCipher 对象,用于加密和解密操作
    mc = oss.crypto.MasterRsaCipher(
        mat_desc={"tag": "value"},
        public_key=RSA_PUBLIC_KEY,  # RSA 公钥,用于加密
        private_key=RSA_PRIVATE_KEY  # RSA 私钥,用于解密
    )

    # 初始化加密客户端
    encryption_client = oss.EncryptionClient(client, mc)

    # 定义要上传的数据
    data = b'hello world'

    # 调用加密客户端的 put_object 方法上传加密对象
    result = encryption_client.put_object(
        oss.PutObjectRequest(
            bucket=args.bucket,  # 指定目标存储空间的名称
            key=args.key,        # 指定对象的名称
            body=data,           # 指定要上传的数据
        )
    )

    # 打印操作结果的状态码和其他相关信息
    print(f'status code: {result.status_code}, '  # HTTP 状态码,表示请求是否成功
          f'request id: {result.request_id}, '    # 请求 ID,用于追踪请求日志和调试
          f'content md5: {result.content_md5}, '  # 返回的对象内容的 MD5 校验值
          f'etag: {result.etag}, '               # 返回的对象的 ETag 值
          f'hash crc64: {result.hash_crc64}, '   # 返回的对象的 CRC64 校验值
          f'version id: {result.version_id}')    # 如果启用了版本控制,返回对象的版本 ID


    # 调用加密客户端的 get_object 方法获取加密对象的内容
    result = encryption_client.get_object(
        oss.GetObjectRequest(
            bucket=args.bucket,  # 指定目标存储空间的名称
            key=args.key,        # 指定目标对象的名称(文件路径)
        )
    )

    # 打印操作结果的相关信息
    print(f'status code: {result.status_code}, '  # HTTP 状态码,表示请求是否成功
          f'request id: {result.request_id}, '   # 请求 ID,用于追踪请求日志和调试
          f'content md5: {result.content_md5}, '  # 对象内容的 MD5 校验值
          f'etag: {result.etag}, '               # 对象的 ETag 值
          f'hash crc64: {result.hash_crc64}, '   # 对象内容的 CRC64 校验值
          f'version id: {result.version_id}')    # 对象的版本 ID(如果启用了版本控制)


if __name__ == "__main__":
    # 程序入口,调用 main 函数执行逻辑
    main()

使用主密钥RSA分片上传Object

使用主密钥RSA分片上传Object示例代码如下:

import argparse
import alibabacloud_oss_v2 as oss
import os
import alibabacloud_oss_v2.crypto
from alibabacloud_oss_v2.encryption_client import EncryptionClient, EncryptionMultiPartContext

# 创建命令行参数解析器,用于接收用户输入的参数
parser = argparse.ArgumentParser(description="encryption put object sample")

# 添加命令行参数 --region,表示存储空间所在的地域,必填项
parser.add_argument('--region', help='The region in which the bucket is located.', required=True)

# 添加命令行参数 --bucket,表示存储空间的名称,必填项
parser.add_argument('--bucket', help='The name of the bucket.', required=True)

# 添加命令行参数 --endpoint,表示其他服务访问 OSS 时使用的域名,可选项
parser.add_argument('--endpoint', help='The domain names that other services can use to access OSS')

# 添加命令行参数 --key,表示对象的名称(文件路径),必填项
parser.add_argument('--key', help='The name of the object.', required=True)

# 定义 RSA 公钥和私钥,用于加密和解密操作
RSA_PUBLIC_KEY = """-----BEGIN PUBLIC KEY-----
MIGfMA0G6mse2QsIgz3******GBcom6kEF6MmR1EKixaQIDAQAB
-----END PUBLIC KEY-----"""

RSA_PRIVATE_KEY = """-----BEGIN PRIVATE KEY-----
MIICdQIBADANBgk******ItewfwXIL1Mqz53lO/gK+q6TR92gGc+4ajL
-----END PRIVATE KEY-----"""

def main():
    # 解析命令行参数
    args = parser.parse_args()

    # 从环境变量中加载凭证信息(AccessKeyId 和 AccessKeySecret)
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()

    # 加载 SDK 的默认配置
    cfg = oss.config.load_default()

    # 设置凭证提供者
    cfg.credentials_provider = credentials_provider

    # 设置存储空间所在的地域
    cfg.region = args.region

    # 如果用户提供了自定义的 endpoint,则设置到配置中
    if args.endpoint is not None:
        cfg.endpoint = args.endpoint

    # 使用配置对象初始化 OSS 客户端
    client = oss.Client(cfg)

    # 初始化主密钥加密对象(MasterRsaCipher),用于加密和解密操作
    mc = oss.crypto.MasterRsaCipher(
        mat_desc={"tag": "value"},# 创建一个主密钥的描述信息,创建后不允许修改。主密钥描述信息和主密钥一一对应。
        public_key=RSA_PUBLIC_KEY,  # RSA 公钥,用于加密
        private_key=RSA_PRIVATE_KEY  # RSA 私钥,用于解密
    )

    # 创建加密客户端
    encryption_client = oss.EncryptionClient(client, mc)

    # 定义分片上传的每个分片大小(单位:字节),这里设置为 100 KB
    part_size = 100 * 1024

    # 获取本地文件的大小(单位:字节)
    data_size = os.path.getsize("/local/dir/example")  # 替换为实际的本地文件路径

    # 初始化分片上传任务,返回上传任务的初始信息
    result = encryption_client.initiate_multipart_upload(
        oss.InitiateMultipartUploadRequest(
            bucket="example_bucket",  # 指定目标存储空间的名称
            key="example_key",        # 指定目标对象的名称(文件路径)
            cse_part_size=part_size,  # 每个分片的大小
            cse_data_size=data_size   # 文件的总大小
        )
    )

    # 打印初始化分片上传任务的结果
    print(vars(result))

    # 初始化分片编号和分片列表
    part_number = 1
    upload_parts = []

    # 打开本地文件并按分片大小逐片读取内容
    with open("/local/dir/example", 'rb') as f:  # 替换为实际的本地文件路径
        for start in range(0, data_size, part_size):  # 按分片大小迭代文件内容
            n = part_size  # 当前分片的大小
            if start + n > data_size:  # 如果最后一片不足分片大小,则调整大小
                n = data_size - start

            # 使用 SectionReader 读取文件的当前分片内容
            reader = oss.io_utils.SectionReader(
                oss.io_utils.ReadAtReader(f),  # 将文件包装为支持随机读取的对象
                start,  # 当前分片的起始位置
                n       # 当前分片的大小
            )

            # 上传当前分片
            up_result = encryption_client.upload_part(
                oss.UploadPartRequest(
                    bucket="example_bucket",  # 指定目标存储空间的名称
                    key="example_key",        # 指定目标对象的名称(文件路径)
                    upload_id=result.upload_id,  # 分片上传任务的唯一标识符
                    part_number=part_number,  # 当前分片的编号
                    cse_multipart_context=result.cse_multipart_context,  # 加密上下文信息
                    body=reader  # 当前分片的数据内容
                )
            )

            # 打印上传分片的结果
            print(vars(result))

            # 将当前分片的编号和 ETag 添加到分片列表中
            upload_parts.append(
                oss.UploadPart(
                    part_number=part_number,  # 当前分片的编号
                    etag=up_result.etag       # 当前分片上传后的 ETag 值
                )
            )

            # 更新分片编号
            part_number += 1

    # 对分片列表按分片编号排序
    parts = sorted(upload_parts, key=lambda p: p.part_number)

    # 完成分片上传任务,合并所有分片并生成最终对象
    result = encryption_client.complete_multipart_upload(
        oss.CompleteMultipartUploadRequest(
            bucket="example_bucket",  # 指定目标存储空间的名称
            key="example_key",        # 指定目标对象的名称(文件路径)
            upload_id=result.upload_id,  # 分片上传任务的唯一标识符
            complete_multipart_upload=oss.CompleteMultipartUpload(
                parts=parts  # 排序后的分片列表
            )
        )
    )

    # 打印完成分片上传任务的结果
    print(vars(result))


if __name__ == "__main__":
    # 程序入口,调用 main 函数执行逻辑
    main()

使用自定义主密钥

使用自定义主密钥简单上传和下载Object

SDK提供了RSA默认实现, 当这个方式不满足用户的需求时,用户可以自己实现主密钥的加解密行为。以下示例代码以阿里云KMS为例,演示如何自定义主密钥加解密进行简单上传和下载Object。

import argparse
import base64
import json
from aliyunsdkkms.request.v20160120.DecryptRequest import DecryptRequest
from aliyunsdkkms.request.v20160120.EncryptRequest import EncryptRequest
from alibabacloud_dkms_transfer.kms_transfer_acs_client import KmsTransferAcsClient
from typing import Optional, Dict
import alibabacloud_oss_v2 as oss
import alibabacloud_oss_v2.crypto
from alibabacloud_oss_v2.encryption_client import EncryptionClient, EncryptionMultiPartContext

# 创建命令行参数解析器,用于接收用户输入的参数
parser = argparse.ArgumentParser(description="encryption kms sample")

# 添加命令行参数 --region,表示存储空间所在的地域,必填项
parser.add_argument('--region', help='The region in which the bucket is located.', required=True)

# 添加命令行参数 --bucket,表示存储空间的名称,必填项
parser.add_argument('--bucket', help='The name of the bucket.', required=True)

# 添加命令行参数 --endpoint,表示其他服务访问 OSS 时使用的域名,可选项
parser.add_argument('--endpoint', help='The domain names that other services can use to access OSS')

# 添加命令行参数 --key,表示对象的名称(文件路径),必填项
parser.add_argument('--key', help='The name of the object.', required=True)

# 添加命令行参数 --kms_id,表示用户的 CMK(Customer Master Key)ID,必填项
parser.add_argument('--kms_id', help='The id of the your CMK ID.', required=True)


# 自定义主密钥加密器类,继承自 oss.crypto.MasterCipher
class MasterKmsCipher(oss.crypto.MasterCipher):

    def __init__(
        self,
        mat_desc: Optional[Dict] = None,
        kms_client: Optional[KmsTransferAcsClient] = None,
        kms_id: Optional[str] = None,
    ):
        self.kms_client = kms_client
        self.kms_id = kms_id
        self._mat_desc = None

        # 如果提供了主密钥的描述信息,则将其序列化为 JSON 字符串
        if mat_desc is not None and len(mat_desc.items()) > 0:
            self._mat_desc = json.dumps(mat_desc)

    def get_wrap_algorithm(self) -> str:
        # 返回加密算法名称,固定为 'KMS/ALICLOUD'
        return 'KMS/ALICLOUD'

    def get_mat_desc(self) -> str:
        return self._mat_desc or ''

    def encrypt(self, data: bytes) -> bytes:
        """
        使用 KMS 服务加密数据
        :param data: 待加密的原始数据(字节格式)
        :return: 加密后的数据(字节格式)
        """
        # 将原始数据编码为 Base64 格式
        base64_crypto = base64.b64encode(data)

        # 构造加密请求对象
        request = EncryptRequest()
        request.set_KeyId(self.kms_id)  # 设置 CMK ID
        request.set_Plaintext(base64_crypto)  # 设置待加密的 Base64 数据

        # 调用 KMS 客户端执行加密操作,并获取响应
        response = self.kms_client.do_action_with_exception(request)

        # 解析响应中的加密数据字段,并解码为字节格式
        return base64.b64decode(json.loads(response).get('CiphertextBlob'))

    def decrypt(self, data: bytes) -> bytes:
        """
        使用 KMS 服务解密数据
        :param data: 已加密的数据(字节格式)
        :return: 解密后的原始数据(字节格式)
        """
        # 将加密数据编码为 Base64 格式
        base64_crypto = base64.b64encode(data)

        # 构造解密请求对象
        request = DecryptRequest()
        request.set_CiphertextBlob(base64_crypto)  # 设置加密数据

        # 调用 KMS 客户端执行解密操作,并获取响应
        response = self.kms_client.do_action_with_exception(request)

        # 解析响应中的明文字段,并解码为字节格式
        return base64.b64decode(json.loads(response).get('Plaintext'))


def main():
    # 解析命令行参数
    args = parser.parse_args()

    # 从环境变量中加载凭证信息(AccessKeyId 和 AccessKeySecret)
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()

    # 加载 SDK 的默认配置
    cfg = oss.config.load_default()

    # 设置凭证提供者
    cfg.credentials_provider = credentials_provider

    # 设置存储空间所在的地域
    cfg.region = args.region

    # 如果用户提供了自定义的 endpoint,则设置到配置中
    if args.endpoint is not None:
        cfg.endpoint = args.endpoint

    # 使用配置对象初始化 OSS 客户端
    client = oss.Client(cfg)

    # 初始化 KMS 客户端,用于与 KMS 服务交互
    kms_client = KmsTransferAcsClient(
        ak=credentials_provider._credentials.access_key_id,  # 从凭证提供者中获取 AccessKeyId
        secret=credentials_provider._credentials.access_key_secret,  # 从凭证提供者中获取 AccessKeySecret
        region_id=args.region  # 指定地域信息
    )

    # 初始化主密钥加密器(MasterKmsCipher),用于加密和解密操作
    mc = MasterKmsCipher(
        mat_desc={"desc": "your master encrypt key material describe information"},  # 主密钥描述信息
        kms_client=kms_client,  # KMS 客户端实例
        kms_id=args.kms_id  # 用户的 CMK ID
    )

    # 创建加密客户端
    encryption_client = oss.EncryptionClient(client, mc)

    # 定义要上传的数据
    data = b'hello world'

    # 调用加密客户端的 put_object 方法上传加密对象
    result = encryption_client.put_object(
        oss.PutObjectRequest(
            bucket=args.bucket,  # 指定目标存储空间的名称
            key=args.key,        # 指定对象的名称(文件路径)
            body=data,           # 指定要上传的数据
        )
    )

    # 打印上传加密对象的结果
    print(vars(result))

    # 调用加密客户端的 get_object 方法获取加密对象的内容
    result = encryption_client.get_object(
        oss.GetObjectRequest(
            bucket=args.bucket,  # 指定目标存储空间的名称
            key=args.key,        # 指定对象的名称(文件路径)
        )
    )

    # 打印获取加密对象的结果
    print(vars(result))

    # 打印解密后的对象内容
    print(result.body.read())


if __name__ == "__main__":
    # 程序入口,调用 main 函数执行逻辑
    main()

相关文档

  • 关于OSS客户端加密实现的原理,请参见客户端加密

  • 关于客户端加密的Python SDK V2操作指南,请参见操作指南

  • 关于使用主密钥RSA简单上传和下载Object的完整代码示例,请参见Github示例

  • 关于使用主密钥KMS简单上传和下载Object的完整代码示例,请参见Github示例