S3 serverseitige Verschlüsselung SSE mit Python boto3

Ich möchte boto3 (Python) verwenden, um die in S3 geschriebenen Daten zu verschlüsseln.

Referenz 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

Die Arten der serverseitigen SSE-Verschlüsselung sind:

Standardverschlüsselung

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

Andere als Standardverschlüsselung

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

Stichprobe

Vorbereitung

Ändern Sie den Wert von BUCKET_NAME in der Quelle in Ihren S3-Bucket-Namen. Ändern Sie SSEKMSKeyId in die Schlüssel-ID Ihrer vom Kunden verwalteten Schlüssel. Erstellen Sie einen SSECustomerKey, damit er mit AES-256 verschlüsselt werden kann. Ich verwende Ubuntu 18.04 und habe die folgenden Ergebnisse der Befehlsausführung verwendet. 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"
}

Umschalten zwischen Standardverschlüsselung und nicht standardmäßiger Verschlüsselung

Standardverschlüsselung

Für die Standardverschlüsselung drücken Sie in der S3-Konsole im S3-Bucket Eigenschaften → Standardverschlüsselung und wählen Sie AES-256 oder AWS-KMS aus. Setzen Sie den Wert für encryption_mode der obigen Ereignisvariablen auf default_encryption.

Nicht standardmäßiges Verschlüsselungsmuster

Um mit einem anderen Muster als der Standardverschlüsselung zu verschlüsseln, drücken Sie in der S3-Konsole auf Eigenschaften → Standardverschlüsselung des S3-Buckets und wählen Sie Keine. Setzen Sie den Wert für encryption_mode der obigen Ereignisvariablen auf 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)