Configure client-side encryption by using OSS SDK for Python to locally encrypt an object before you upload the object to OSS - Object Storage Service

Precautions

In this topic, the region ID cn-hangzhou of China (Hangzhou) is used as an example. By default, the public endpoint is used. If you want to access OSS from other Alibaba Cloud services in the same region as OSS, use the internal endpoint. For more information about the mapping between regions and endpoints of OSS, see OSS regions and endpoints.
When you use client-side encryption, you must ensure the integrity and validity of the customer master key (CMK).
When you copy or migrate encrypted data, you are responsible for the integrity and validity of object metadata.

Method definition

OSS SDK for Python V2 supports the following methods to use CMKs:

RSA-based CMKs managed by yourself

The SDK provides the default Rivest-Shamir-Adleman (RSA) implementation. If you provide a CMK, you must pass the public key and private key to the SDK.
Custom CMKs

If the RSA-based CMK method cannot meet your requirements, you can implement encryption and decryption based on custom CMKs.

You can use the preceding encryption methods to prevent data leaks and protect your data on the client. Even if your data is leaked, it cannot be decrypted by others.

Important

For more information about the principles of client-side encryption in OSS, see Client-side encryption in the OSS User Guide.

To use client-side encryption, you must create an instance for the client for client-side encryption and call related API operations. Your objects are automatically encrypted and decrypted as part of the request.

class EncryptionClient:
  ...

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

Request parameters

Parameter	Type	Description
client	*Client	An instance of a non-encryption client.
master_cipher	MasterCipher	The CMK instance for encrypting and decrypting data keys.
decrypt_master_ciphers	List[MasterCipher]	The CMK instance for decrypting data keys.

The following table lists the operations provided by EncryptionClient:

Basic operation	Description
get_object_meta	Queries standard HTTP metadata headers of the object.
head_object	Queries all metadata information of the object.
get_object	Downloads and decrypts the object.
put_object	Uploads and encrypts the object.
initiate_multipart_upload	Initiates a multipart upload and the multipart encryption context (EncryptionMultiPartContext).
upload_part	Initiates a multipart upload and automatically performs encryption. When you call the API operation, you must set the context for encryption in multipart upload.
complete_multipart_upload	Combines all parts into a complete object after all parts are uploaded.
abort_multipart_upload	Cancels a multipart upload task and deletes all parts uploaded by the multipart upload task.
list_parts	Lists all parts that were uploaded by the multipart upload task.

Use RSA-based CMKs

Use an RSA-based CMK to encrypt objects in simple uploads and decrypt objects in simple downloads

The following sample code provides an example on how to use an RSA-based CMK to encrypt objects in simple uploads and decrypt objects in simple downloads:

import argparse
import alibabacloud_oss_v2 as oss

# Parse the command line parameters
parser = argparse.ArgumentParser(description="encryption put object sample")

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

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

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

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

# Define the RSA public key and private key for encryption and decryption operations
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 credential information (AccessKeyId and AccessKeySecret) from environment variables
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()

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

    # Specify the credential provider
    cfg.credentials_provider = credentials_provider

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

    # If the user provides a custom endpoint, 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 operations
    mc = oss.crypto.MasterRsaCipher(
        mat_desc={"tag": "value"},
        public_key=RSA_PUBLIC_KEY,  # RSA public key for encryption
        private_key=RSA_PRIVATE_KEY  # 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,  # Specify the name of the target bucket
            key=args.key,        # Specify the name of the object
            body=data,           # Specify the data to upload
        )
    )

    # Print the status code and other related information of the operation result
    print(f'status code: {result.status_code}, '  # HTTP status code, indicating whether the request was successful
          f'request id: {result.request_id}, '    # Request ID, used to track request logs and debug
          f'content md5: {result.content_md5}, '  # MD5 checksum of the returned object content
          f'etag: {result.etag}, '               # ETag value of the returned object
          f'hash crc64: {result.hash_crc64}, '   # CRC64 checksum of the returned object
          f'version id: {result.version_id}')    # 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,  # Specify the name of the target bucket
            key=args.key,        # Specify the name of the target object (file path)
        )
    )

    # Print the related information of the operation result
    print(f'status code: {result.status_code}, '  # HTTP status code, indicating whether the request was successful
          f'request id: {result.request_id}, '   # Request ID, used to track request logs and debug
          f'content md5: {result.content_md5}, '  # MD5 checksum of the object content
          f'etag: {result.etag}, '               # ETag value of the object
          f'hash crc64: {result.hash_crc64}, '   # CRC64 checksum of the object content
          f'version id: {result.version_id}')    # Version ID of the object (if versioning is enabled)


if __name__ == "__main__":
    # Program entry, call the main function to execute the logic
    main()

Use an RSA-based CMK to encrypt an object in multipart upload

The following code provides an example on how to use an RSA-based CMK to encrypt an object in multipart upload:

import argparse
import alibabacloud_oss_v2 as oss
import os

# Parse the command line parameters
parser = argparse.ArgumentParser(description="encryption put object sample")

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

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

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

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

# Define the RSA public key and private key for encryption and decryption operations
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 credential information (AccessKeyId and AccessKeySecret) from environment variables
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()

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

    # Specify the credential provider
    cfg.credentials_provider = credentials_provider

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

    # If the user provides a custom endpoint, 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 encryption object (MasterRsaCipher) for encryption and decryption operations
    mc = oss.crypto.MasterRsaCipher(
        mat_desc={"tag": "value"},# Create a description for the master key, which cannot be modified after creation. You can specify only one description for a CMK.
        public_key=RSA_PUBLIC_KEY,  # RSA public key for encryption
        private_key=RSA_PRIVATE_KEY  # RSA private key for decryption
    )

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

    # Define the size of each part for multipart upload (in bytes), set to 100 KB here
    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 local file path

    # 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",  # Specify the name of the target bucket
            key="example_key",        # Specify the name of the target object (file path)
            cse_part_size=part_size,  # Size of each part
            cse_data_size=data_size   # Total size of the file
        )
    )

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

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

    # Open the local file and read the content by part size
    with open("/local/dir/example", 'rb') as f:  # Replace with the actual local file path
        for start in range(0, data_size, part_size):  # Iterate through the file content by part size
            n = part_size  # 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 current part content of the file
            reader = oss.io_utils.SectionReader(
                oss.io_utils.ReadAtReader(f),  # Wrap the file as an object that supports random reading
                start,  # Starting position of the current part
                n       # Size of the current part
            )

            # Upload the current part
            up_result = encryption_client.upload_part(
                oss.UploadPartRequest(
                    bucket="example_bucket",  # Specify the name of the target bucket
                    key="example_key",        # Specify the name of the target object (file path)
                    upload_id=result.upload_id,  # Unique identifier of the multipart upload task
                    part_number=part_number,  # Number of the current part
                    cse_multipart_context=result.cse_multipart_context,  # Encryption context information
                    body=reader  # 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 part list
            upload_parts.append(
                oss.UploadPart(
                    part_number=part_number,  # Number of the current part
                    etag=up_result.etag       # ETag value after uploading the current part
                )
            )

            # Update the part number
            part_number += 1

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

    # Complete the multipart upload task, combine all parts and generate the final object
    result = encryption_client.complete_multipart_upload(
        oss.CompleteMultipartUploadRequest(
            bucket="example_bucket",  # Specify the name of the target bucket
            key="example_key",        # Specify the name of the target object (file path)
            upload_id=result.upload_id,  # Unique identifier of the multipart upload task
            complete_multipart_upload=oss.CompleteMultipartUpload(
                parts=parts  # Sorted part list
            )
        )
    )

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


if __name__ == "__main__":
    # Program entry, call the main function to execute the logic
    main()

Use custom CMKs

Use a custom CMK to encrypt objects in simple uploads and decrypt objects in simple downloads

The SDK provides the default RSA implementation. If the default implementation does not meet your needs, you can implement encryption and decryption based on custom CMKs. The following sample code uses Alibaba Cloud KMS as an example to demonstrate how to customize CMK encryption and decryption for simple uploads and downloads of objects.

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


# Parse the command line parameters
parser = argparse.ArgumentParser(description="encryption kms sample")

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

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

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

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

# Add the --kms_id command line parameter, which specifies the user's CMK (Customer Master Key) ID. This parameter is required.
parser.add_argument('--kms_id', help='The id of the your CMK ID.', required=True)


# Custom master key encryptor class, inheriting from 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

        # If a description of the master key is provided, serialize it to 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:
        # Return the encryption algorithm name, fixed as 'KMS/ALICLOUD'
        return 'KMS/ALICLOUD'

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

    def encrypt(self, data: bytes) -> bytes:
        """
        Encrypt data using the KMS service
        :param data: Original data to be encrypted (in bytes format)
        :return: Encrypted data (in bytes format)
        """
        # Encode the original data in Base64 format
        base64_crypto = base64.b64encode(data)

        # Construct an encryption request object
        request = EncryptRequest()
        request.set_KeyId(self.kms_id)  # Set the CMK ID
        request.set_Plaintext(base64_crypto)  # Set the Base64 data to be encrypted

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

        # Parse the encrypted data field in the response and decode it to bytes format
        return base64.b64decode(json.loads(response).get('CiphertextBlob'))

    def decrypt(self, data: bytes) -> bytes:
        """
        Decrypt data using the KMS service
        :param data: Encrypted data (in bytes format)
        :return: Decrypted original data (in bytes format)
        """
        # Encode the encrypted data in Base64 format
        base64_crypto = base64.b64encode(data)

        # Construct a decryption request object
        request = DecryptRequest()
        request.set_CiphertextBlob(base64_crypto)  # Set the encrypted data

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

        # Parse the plaintext field in the response and decode it to bytes format
        return base64.b64decode(json.loads(response).get('Plaintext'))


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

    # Load credential information (AccessKeyId and AccessKeySecret) from environment variables
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()

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

    # Specify the credential provider
    cfg.credentials_provider = credentials_provider

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

    # If the user provides a custom endpoint, 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 KMS client for interacting with the KMS service
    kms_client = KmsTransferAcsClient(
        ak=credentials_provider._credentials.access_key_id,  # Get AccessKeyId from the credential provider
        secret=credentials_provider._credentials.access_key_secret,  # Get AccessKeySecret from the credential provider
        region_id=args.region  # Specify region information
    )

    # Initialize the master key encryptor (MasterKmsCipher) for encryption and decryption operations
    mc = MasterKmsCipher(
        mat_desc={"desc": "your master encrypt key material describe information"},  # Master key description information
        kms_client=kms_client,  # KMS client instance
        kms_id=args.kms_id  # 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,  # Specify the name of the target bucket
            key=args.key,        # Specify the name of the object (file path)
            body=data,           # Specify 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,  # Specify the name of the target bucket
            key=args.key,        # Specify the name of the object (file path)
        )
    )

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

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


if __name__ == "__main__":
    # Program entry, call the main function to execute the logic
    main()

References

For more information about the principles of client-side encryption in OSS, see Client-side encryption.
For more information about client-side encryption operations in OSS SDK for Python V2, see Operation Guide.
For the complete sample code on how to use an RSA-based CMK to encrypt objects in simple uploads and decrypt objects in simple downloads, see Github sample.
For the complete sample code on how to use a KMS-based CMK to encrypt objects in simple uploads and decrypt objects in simple downloads, see Github sample.