Object Storage Service:Client-side encryption

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()

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()

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()