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:
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"
}
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.
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)
Recommended Posts