S3 server-side encryption SSE with Python boto3

I would like to use boto3 (Python) to encrypt the data to be written to S3.

reference https://qiita.com/ot-nemoto/items/66cc783e8d8714f88bd8#%E9%9D%9E%E6%9A%97%E5%8F%B7%E5%8C%96%E3%83%90%E3%82%B1%E3%83%83%E3%83%88%E3%81%AB%E3%82%B5%E3%83%BC%E3%83%90%E3%82%B5%E3%82%A4%E3%83%89%E6%9A%97%E5%8F%B7%E5%8C%96%E3%82%92%E6%8C%87%E5%AE%9A%E3%81%97%E3%81%A6%E3%83%95%E3%82%A1%E3%82%A4%E3%83%AB%E3%82%92%E3%82%A2%E3%83%83%E3%83%97%E3%83%AD%E3%83%BC%E3%83%89

The types of SSE server-side encryption are:

Default encryption

1. SSE with AES-256
2. SSE with KMS AWS Managed Keys
3. SSE with KMS CMK(Customer Managed Keys)

Other than default encryption

1. SSE with AES-256
2. SSE with KMS AWS Managed Keys
3. SSE with KMS CMK(Customer Managed Keys)
4. SSE with Customer Key(AES-256 etc.)

sample

Preparation

Change the value of BUCKET_NAME in the source to your S3 bucket name. Change SSEKMSKeyId to the Key ID of your Customer Managed Keys. Create an SSECustomerKey so that it can be encrypted with AES-256. I'm using Ubuntu 18.04 and I used the following command execution result. SSE_CUSTOMER_KEY=$(cat /dev/urandom | base64 -i | fold -w 32 | head -n 1)

event = {
    "BUCKET_NAME" : "xxxxxxxxx",
    "encryption_mode" : default_encryption,
    "SSEKMSKeyId" : "yyyyyyyy",
    "SSECustomerKey" : "zzzzzzzzzzzz"
}

Switching between default encryption and non-default encryption

Default encryption

For default encryption, on the S3 console, press Properties → Default Encryption on your S3 bucket and select AES-256 or AWS-KMS. Set the encryption_mode value of the above event variable to default_encryption.

Non-default encryption pattern

To encrypt with a pattern that is not the default encryption, on the S3 console, press Properties → Default Encryption on the S3 bucket and select None. Set the encryption_mode value of the above event variable to non_default_encryption.

#-*- encoding:utf-8 -*-
from datetime import datetime,timedelta,timezone
import json
import os,os.path
import sys
#Third Party
import boto3

#kms
kms = boto3.client("kms")

#s3
s3 = boto3.client("s3")

def default_encryption(**event) -> None:
    """
    Check the "Default encryption" on the S3 bucket Properties
    Automatically encrypt objects when stored in Amazon S3
    Args:
        event
    Returns:
        None
    """
    #SSE with AES-256
    #SSE with KMS AWS Managed Keys
    #SSE with KMS CMK(Customer Managed Keys)
    response = s3.put_object(
        Bucket = event["BUCKET_NAME"],
        Key = "test",
        Body = "Encrypted".encode("UTF-8")
    )
    print(f'ServerSideEncryption'.ljust(20) + f' = {response["ServerSideEncryption"]}')

    #just only for KMS. check the KeyManager
    if response["ServerSideEncryption"] == "aws:kms":
        KeyManager = kms.describe_key(
            KeyId = response["SSEKMSKeyId"]
        )["KeyMetadata"]["KeyManager"]
        print(f"KeyManager".ljust(20) + f" = {KeyManager}")

    #Body
    Body = s3.get_object(
        Bucket = event["BUCKET_NAME"],
        Key = "test"
    )["Body"].read().decode("UTF-8")
    print(f"Body".ljust(20) + f" = {Body}")


def non_default_encryption(**event) -> None:
    """
    Encrypt the data on your behalf
    Args:
        event
    Returns:
        None
    """
    #SSE with AES-256
    #SSE with KMS AWS Managed Keys
    #SSE with KMS CMK(Customer Managed Keys)
    #SSE with Client operations key. This is not the key which S3 or KMS operates
    l = [
        {"ServerSideEncryption" : "AES256"},
        {"ServerSideEncryption" : "aws:kms" },
        {"ServerSideEncryption" : "aws:kms","SSEKMSKeyId" : event["SSEKMSKeyId"]},
        {"SSECustomerAlgorithm" : "AES256","SSECustomerKey" : event["SSECustomerKey"]}
    ]
    for item in l:
        params = {
            "Bucket" : event["BUCKET_NAME"],
            "Key" : "test",
            "Body" : "Encrypted".encode("UTF-8")
        }
        for key in item:
            params[key] = item[key]
        
        response = s3.put_object(**params)
        if "ServerSideEncryption" in response:
            print(f'ServerSideEncryption'.ljust(20) + f' = {response["ServerSideEncryption"]}')
            #just only for KMS. check the KeyManager
            if response["ServerSideEncryption"] == "aws:kms":
                KeyManager = kms.describe_key(
                    KeyId = response["SSEKMSKeyId"]
                )["KeyMetadata"]["KeyManager"]
                print(f"KeyManager".ljust(20) + f" = {KeyManager}")

        elif "SSECustomerAlgorithm" in response:
            print(f'SSECustomerAlgorithm'.ljust(20) + f' = {response["SSECustomerAlgorithm"]}')

        
        #Body
        params = {
            "Bucket" : event["BUCKET_NAME"],
            "Key" : "test"
        }
        if "SSECustomerAlgorithm" in item:
            params["SSECustomerAlgorithm"] = item["SSECustomerAlgorithm"]
            params["SSECustomerKey"] = item["SSECustomerKey"]

        Body = s3.get_object(
            **params    
        )["Body"].read().decode("UTF-8")
        print(f"Body".ljust(20) + f" = {Body}")



if __name__ == "__main__":
    event = {
        "BUCKET_NAME" : "xxxxxxxxx",
        "encryption_mode" : default_encryption,
        "SSEKMSKeyId" : "yyyyyyyy",
        "SSECustomerKey" : "zzzzzzzzzzzz"
    }
    
    event["encryption_mode"](**event)