Object Storage Service:Client-side encryption (Python SDK V2)

The following sample code shows how to use an RSA master key for simple object upload and download:

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

# Create a command-line argument parser to receive user input.
parser = argparse.ArgumentParser(description="encryption put object sample")

# Add the --region command-line argument, which specifies the region where the bucket is located. This argument is required.
parser.add_argument('--region', help='The region in which the bucket is located.', required=True)

# Add the --bucket command-line argument, which specifies the name of the bucket. This argument is required.
parser.add_argument('--bucket', help='The name of the bucket.', required=True)

# Add the --endpoint command-line argument, which specifies the domain name that other services use to access OSS. This argument is optional.
parser.add_argument('--endpoint', help='The domain names that other services can use to access OSS')

# Add the --key command-line argument, which specifies the name of the object. This argument is required.
parser.add_argument('--key', help='The name of the object.', required=True)

# Define the RSA public and private keys for encryption and decryption.
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():
    # Parse the command-line arguments.
    args = parser.parse_args()

    # Load credentials (AccessKey ID and AccessKey secret) from environment variables.
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()

    # Load the default configurations of the SDK.
    cfg = oss.config.load_default()

    # Set the credentials provider.
    cfg.credentials_provider = credentials_provider

    # Set the region where the bucket is located.
    cfg.region = args.region

    # If a custom endpoint is provided, set it in the configuration.
    if args.endpoint is not None:
        cfg.endpoint = args.endpoint

    # Initialize the OSS client with the configuration object.
    client = oss.Client(cfg)

    # Initialize the MasterRsaCipher object for encryption and decryption.
    mc = oss.crypto.MasterRsaCipher(
        mat_desc={"tag": "value"},
        public_key=RSA_PUBLIC_KEY,  # The RSA public key for encryption.
        private_key=RSA_PRIVATE_KEY  # The RSA private key for decryption.
    )

    # Initialize the encryption client.
    encryption_client = oss.EncryptionClient(client, mc)

    # Define the data to upload.
    data = b'hello world'

    # Call the put_object method of the encryption client to upload the encrypted object.
    result = encryption_client.put_object(
        oss.PutObjectRequest(
            bucket=args.bucket,  # The name of the destination bucket.
            key=args.key,        # The name of the object.
            body=data,           # The data to upload.
        )
    )

    # Print the status code and other information about the operation.
    print(f'status code: {result.status_code}, '  # The HTTP status code, which indicates whether the request was successful.
          f'request id: {result.request_id}, '    # The request ID, used for tracking logs and debugging.
          f'content md5: {result.content_md5}, '  # The MD5 checksum of the returned object content.
          f'etag: {result.etag}, '               # The ETag of the returned object.
          f'hash crc64: {result.hash_crc64}, '   # The CRC-64 checksum of the returned object.
          f'version id: {result.version_id}')    # The version ID of the object if versioning is enabled.


    # Call the get_object method of the encryption client to get the content of the encrypted object.
    result = encryption_client.get_object(
        oss.GetObjectRequest(
            bucket=args.bucket,  # The name of the destination bucket.
            key=args.key,        # The name of the destination object (file path).
        )
    )

    # Print information about the operation.
    print(f'status code: {result.status_code}, '  # The HTTP status code, which indicates whether the request was successful.
          f'request id: {result.request_id}, '   # The request ID, used for tracking logs and debugging.
          f'content md5: {result.content_md5}, '  # The MD5 checksum of the object content.
          f'etag: {result.etag}, '               # The ETag of the object.
          f'hash crc64: {result.hash_crc64}, '   # The CRC-64 checksum of the object.
          f'version id: {result.version_id}')    # The version ID of the object if versioning is enabled.


if __name__ == "__main__":
    # The program entry point. Call the main function to run the logic.
    main()

The following sample code shows how to use an RSA master key for multipart object upload:

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

# Create a command-line argument parser to receive user input.
parser = argparse.ArgumentParser(description="encryption put object sample")

# Add the --region command-line argument, which specifies the region where the bucket is located. This argument is required.
parser.add_argument('--region', help='The region in which the bucket is located.', required=True)

# Add the --bucket command-line argument, which specifies the name of the bucket. This argument is required.
parser.add_argument('--bucket', help='The name of the bucket.', required=True)

# Add the --endpoint command-line argument, which specifies the domain name that other services use to access OSS. This argument is optional.
parser.add_argument('--endpoint', help='The domain names that other services can use to access OSS')

# Add the --key command-line argument, which specifies the name of the object (file path). This argument is required.
parser.add_argument('--key', help='The name of the object.', required=True)

# Define the RSA public and private keys for encryption and decryption.
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():
    # Parse the command-line arguments.
    args = parser.parse_args()

    # Load credentials (AccessKey ID and AccessKey secret) from environment variables.
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()

    # Load the default configurations of the SDK.
    cfg = oss.config.load_default()

    # Set the credentials provider.
    cfg.credentials_provider = credentials_provider

    # Set the region where the bucket is located.
    cfg.region = args.region

    # If a custom endpoint is provided, set it in the configuration.
    if args.endpoint is not None:
        cfg.endpoint = args.endpoint

    # Initialize the OSS client with the configuration object.
    client = oss.Client(cfg)

    # Initialize the master key cipher (MasterRsaCipher) for encryption and decryption.
    mc = oss.crypto.MasterRsaCipher(
        mat_desc={"tag": "value"},# Create a description for the master key. The description cannot be modified after creation. The master key description corresponds one-to-one with the master key.
        public_key=RSA_PUBLIC_KEY,  # The RSA public key for encryption.
        private_key=RSA_PRIVATE_KEY  # The RSA private key for decryption.
    )

    # Create an encryption client.
    encryption_client = oss.EncryptionClient(client, mc)

    # Define the size of each part for the multipart upload in bytes. Here, it is set to 100 KB.
    part_size = 100 * 1024

    # Get the size of the local file in bytes.
    data_size = os.path.getsize("/local/dir/example")  # Replace with the actual path of the local file.

    # Initialize a multipart upload task and return the initial information of the upload task.
    result = encryption_client.initiate_multipart_upload(
        oss.InitiateMultipartUploadRequest(
            bucket="example_bucket",  # The name of the destination bucket.
            key="example_key",        # The name of the destination object (file path).
            cse_part_size=part_size,  # The size of each part.
            cse_data_size=data_size   # The total size of the file.
        )
    )

    # Print the result of initializing the multipart upload task.
    print(vars(result))

    # Initialize the part number and the list of parts.
    part_number = 1
    upload_parts = []

    # Open the local file and read its content part by part based on the part size.
    with open("/local/dir/example", 'rb') as f:  # Replace with the actual path of the local file.
        for start in range(0, data_size, part_size):  # Iterate through the file content by part size.
            n = part_size  # The size of the current part.
            if start + n > data_size:  # If the last part is smaller than the part size, adjust the size.
                n = data_size - start

            # Use SectionReader to read the content of the current part of the file.
            reader = oss.io_utils.SectionReader(
                oss.io_utils.ReadAtReader(f),  # Wrap the file as an object that supports random reads.
                start,  # The starting position of the current part.
                n       # The size of the current part.
            )

            # Upload the current part.
            up_result = encryption_client.upload_part(
                oss.UploadPartRequest(
                    bucket="example_bucket",  # The name of the destination bucket.
                    key="example_key",        # The name of the destination object (file path).
                    upload_id=result.upload_id,  # The unique identifier of the multipart upload task.
                    part_number=part_number,  # The number of the current part.
                    cse_multipart_context=result.cse_multipart_context,  # The encryption context information.
                    body=reader  # The data content of the current part.
                )
            )

            # Print the result of uploading the part.
            print(vars(result))

            # Add the number and ETag of the current part to the list of parts.
            upload_parts.append(
                oss.UploadPart(
                    part_number=part_number,  # The number of the current part.
                    etag=up_result.etag       # The ETag value after the current part is uploaded.
                )
            )

            # Update the part number.
            part_number += 1

    # Sort the list of parts by part number.
    parts = sorted(upload_parts, key=lambda p: p.part_number)

    # Complete the multipart upload task, which merges all parts to generate the final object.
    result = encryption_client.complete_multipart_upload(
        oss.CompleteMultipartUploadRequest(
            bucket="example_bucket",  # The name of the destination bucket.
            key="example_key",        # The name of the destination object (file path).
            upload_id=result.upload_id,  # The unique identifier of the multipart upload task.
            complete_multipart_upload=oss.CompleteMultipartUpload(
                parts=parts  # The sorted list of parts.
            )
        )
    )

    # Print the result of completing the multipart upload task.
    print(vars(result))


if __name__ == "__main__":
    # The program entry point. Call the main function to run the logic.
    main()

If the default RSA implementation provided by the SDK does not meet your requirements, you can implement your own encryption and decryption behavior for the master key. The following sample code uses Alibaba Cloud KMS as an example to demonstrate how to use a custom master key for simple object upload and download.

import argparse
import base64
import json
from aliyunsdkkms.request.v20160120.DecryptRequest import DecryptRequest
from aliyunsdkkms.request.v20160120.EncryptRequest import EncryptRequest
from aliyunsdkcore.client import AcsClient
from typing import Optional, Dict
import alibabacloud_oss_v2 as oss

# Create a command-line argument parser to receive user input.
parser = argparse.ArgumentParser(description="encryption kms sample")

# Add the --region command-line argument, which specifies the region where the bucket is located. This argument is required.
parser.add_argument('--region', help='The region in which the bucket is located.', required=True)

# Add the --bucket command-line argument, which specifies the name of the bucket. This argument is required.
parser.add_argument('--bucket', help='The name of the bucket.', required=True)

# Add the --endpoint command-line argument, which specifies the domain name that other services use to access OSS. This argument is optional.
parser.add_argument('--endpoint', help='The domain names that other services can use to access OSS')

# Add the --key command-line argument, which specifies the name of the object (file path). This argument is required.
parser.add_argument('--key', help='The name of the object.', required=True)

# Add the --kms_id command-line argument, which specifies the user's customer master key (CMK) ID. This argument is required.
parser.add_argument('--kms_id', help='The id of the your CMK ID.', required=True)


# Custom master key cipher class that inherits from oss.crypto.MasterCipher.
class MasterKmsCipher(oss.crypto.MasterCipher):

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

        # If a description for the master key is provided, serialize it into a JSON string.
        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:
        # Returns the name of the wrap algorithm, which is fixed as 'KMS/ALICLOUD'.
        return 'KMS/ALICLOUD'

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

    def encrypt(self, data: bytes) -> bytes:
        """
        Encrypts data using the KMS service.
        :param data: The raw data to be encrypted, in bytes.
        :return: The encrypted data, in bytes.
        """
        # Encodes the raw data in Base64 format.
        base64_crypto = base64.b64encode(data)

        # Constructs an encryption request object.
        request = EncryptRequest()
        request.set_KeyId(self.kms_id)  # Sets the CMK ID.
        request.set_Plaintext(base64_crypto)  # Sets the Base64-encoded data to be encrypted.

        # Calls the KMS client to perform the encryption operation and obtain the response.
        response = self.kms_client.do_action_with_exception(request)

        # Parses the encrypted data field from the response and decodes it into bytes.
        return base64.b64decode(json.loads(response).get('CiphertextBlob'))

    def decrypt(self, data: bytes) -> bytes:
        """
        Decrypts data using the KMS service.
        :param data: The encrypted data, in bytes.
        :return: The decrypted raw data, in bytes.
        """
        # Encodes the encrypted data in Base64 format.
        base64_crypto = base64.b64encode(data)

        # Constructs a decryption request object.
        request = DecryptRequest()
        request.set_CiphertextBlob(base64_crypto)  # Sets the encrypted data.

        # Calls the KMS client to perform the decryption operation and obtain the response.
        response = self.kms_client.do_action_with_exception(request)

        # Parses the plaintext field from the response and decodes it into bytes.
        return base64.b64decode(json.loads(response).get('Plaintext'))


def main():
    # Parse the command-line arguments.
    args = parser.parse_args()

    # Load credentials (AccessKey ID and AccessKey secret) from environment variables.
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()

    # Load the default configurations of the SDK.
    cfg = oss.config.load_default()

    # Set the credentials provider.
    cfg.credentials_provider = credentials_provider

    # Set the region where the bucket is located.
    cfg.region = args.region

    # If a custom endpoint is provided, set it in the configuration.
    if args.endpoint is not None:
        cfg.endpoint = args.endpoint

    # Initialize the OSS client with the configuration object.
    client = oss.Client(cfg)

    # Initializes the KMS client to interact with the KMS service.
    kms_client = AcsClient(
        ak=credentials_provider._credentials.access_key_id,  # Obtains the AccessKey ID from the credentials provider.
        secret=credentials_provider._credentials.access_key_secret,  # Obtains the AccessKey secret from the credentials provider.
        region_id=args.region  # Specifies the region.
    )

    # Initializes the master key cipher (MasterKmsCipher) for encryption and decryption operations.
    mc = MasterKmsCipher(
        mat_desc={"desc": "your master encrypt key material describe information"},  # The description of the master key.
        kms_client=kms_client,  # The KMS client instance.
        kms_id=args.kms_id  # The user's CMK ID.
    )

    # Create an encryption client.
    encryption_client = oss.EncryptionClient(client, mc)

    # Define the data to upload.
    data = b'hello world'

    # Call the put_object method of the encryption client to upload the encrypted object.
    result = encryption_client.put_object(
        oss.PutObjectRequest(
            bucket=args.bucket,  # The name of the destination bucket.
            key=args.key,        # The name of the object (file path).
            body=data,           # The data to upload.
        )
    )

    # Print the result of uploading the encrypted object.
    print(vars(result))

    # Call the get_object method of the encryption client to get the content of the encrypted object.
    result = encryption_client.get_object(
        oss.GetObjectRequest(
            bucket=args.bucket,  # The name of the destination bucket.
            key=args.key,        # The name of the object (file path).
        )
    )

    # Print the result of getting the encrypted object.
    print(vars(result))

    # Prints the decrypted object content.
    print(result.body.read())


if __name__ == "__main__":
    # The program entry point. Call the main function to run the logic.
    main()