AWS S3 객체 업로드 시 KMS Key 를 이용하여 암호화 후 다운로드 시 복호화 하는 방법

안녕하세요.
AWS Support에 문의 주셔서 감사합니다. 저는 고객님의 케이스를 담당하게 된 AWS 백향목입니다.
고객님께서는 Unload 시 암호화 및 s3에 업로드된 파일에 대해서 복호화/압축해제에 대한 방법에 대해서 문의주신것으로 이해했습니다.
제가 잘못이해했다면 정정해주시기 바랍니다.

우선적으로 암호화에 대한 부분은 공식문서 기반으로 아래와 같이 전달드립니다.

언로드하는 동안 AWS 키 관리 서비스 키 (SSE-KMS) 를 사용한 서버 측 암호화 또는 고객 관리 키를 사용한 클라이언트 측 암호화 (CSE-CMK) 를 지정하지 않는 경우, UNLOAD는 기본적으로 AWS 관리형 암호화 키 (SSE-S3) 를 사용한 Amazon S3 서버 측 암호화, 즉 “AES-256" 암호화를 사용하여 파일을 자동으로 생성합니다.

S3 버킷에 기본 암호화가 설정되어 있고 암호화된 S3 버킷에 객체를 업로드하면 설명서에 나와 있는 대로 다음 암호화 동작이 적용됩니다. [1]
--PUT 요청 헤더에 암호화 정보가 포함되어 있지 않은 경우 Amazon S3는 버킷의 기본 암호화 설정을 사용하여 객체를 암호화합니다.

--PUT 요청 헤더에 암호화 정보가 포함된 경우 Amazon S3는 PUT 요청의 암호화 정보를 사용하여 객체를 암호화한 후 Amazon S3 에 저장합니다.

참고로 UNLOAD 중에는 PUT 요청 헤더가 기본 암호화와 함께 제공되므로 이 경우 S3는 UNLOAD 명령과 함께 제공되는 암호화 정보를 사용하여 객체를 Amazon S3 저장하기 전에 암호화합니다.

프로세스는 다음과 같습니다.

• 프라이빗 암호화 키, 즉 루트 대칭 키로 사용할 base64 인코딩 256비트 AES 키를 생성합니다.
• 루트 대칭 키와 ENCRYPTED 옵션을 추가하여 UNLOAD 명령을 실행합니다.
• UNLOAD가 1회용 대칭키(봉투 대칭 키로도 불림)와 데이터 암호화에 사용되는 초기화 벡터(IV)를 생성합니다.
• UNLOAD가 루트 대칭 키를 사용하여 봉투 대칭 키를 암호화합니다.
• UNLOAD가 암호화된 데이터 파일을 Amazon S3에 저장하고, 각 파일과 함께 암호화된 봉투 키와 IV를 객체 메타데이터로 저장합니다. 암호화된 봉투 키는 객체 메타데이터 x-amz-meta-x-amz-key로 저장되고, IV는 객체 메타데이터 x-amz-meta-x-amz-iv로 저장됩니다.

아래의 전달드리는 링크를 참고 부탁드리겠습니다.
[+] https://docs.aws.amazon.com/ko_kr/redshift/latest/dg/t_unloading_encrypted_files.html
[+] https://docs.amazonaws.cn/en_us/redshift/latest/dg/t_uploading-encrypted-data.html

공식문서 및 내부에서 확인했으나, 암호화 및 복호화 과정에 대한 상세한 로직을 제공해드리기 어려운점 양해 부탁드리겠습니다.
Best effort 로서 내부의 유사 사례 및 S3 사례들을 찾아보았고, 고객님께서 참고하실 수 있는 python 코드를 이용한 암/복호화 테스트 코드를 전달드리며 참고 부탁드리겠습니다.

======================================================
encrypt :


import boto3
import base64
import redshift_connector


def unload(key):
conn = redshift_connector.connect(
host=‘examplecluster.nnnn.us-west-1.redshift.amazonaws.com’,
database=‘dev’,
port=5439,
user=‘awsuser’,
password=‘my_password’
)
cursor = conn.cursor()
query = (
f"unload (‘select venuename, venuecity from venue’) to ‘s3://mybuckets/encrypted/nnn’ "
f"iam_role ‘arn:aws:iam::nnnnnnn:role/MyRedshiftRole’ "
f"master_symmetric_key {key} "
f"manifest "
f"encrypted;"
)
cursor.execute(query)


def main():
client = boto3.client(‘kms’)
key_id = ‘arn:aws:kms:us-east-2:nnnnnnnn:alias/MyKeyAlias’
resp = client.generate_data_key(KeyId=key_id, KeySpec=‘AES_256’)
unload_key = base64.b64encode(resp[‘Plaintext’])
unload(unload_key)


if name == ‘main’:
main()
======================================================

decrypt :


import boto3
import base64


client = boto3.client(‘s3’)
BUCKET = ‘bucketss-upload-us-east-2’
KEY = ‘redshift/dataset123.txt’


def mock_unload(data_key):
try:
resp = client.put_object(
Bucket=BUCKET,
Key=KEY,
SSECustomerAlgorithm=‘AES256’,
SSECustomerKey=data_key,
Body=b’Hello,World!’
)
except Exception as e:
print(f’Error uploading file. {e}')
raise e


def main():
client_kms = boto3.client(‘kms’)
key_id = ‘arn:aws:kms:us-east-2:nnnnnnnnn:alias/MyKeyAlias’
resp = client_kms.generate_data_key(KeyId=key_id, KeySpec=‘AES_256’)
# In this example I don’t need to encode with base64 like redshift API.
unload_key = resp[‘Plaintext’]
mock_unload(unload_key)


try:
    resp = client.get_object(
        Bucket='koiker-upload-us-east-2',
        Key='redshift/dataset123.txt',
        SSECustomerAlgorithm='AES256',
        SSECustomerKey=unload_key
    )
    print(resp['Body'].read())
except Exception as e:
    print(f'Exception while downloading file. {e}')

if name == ‘main’:
main()

======================================================

고객님께 암호화 복호화에 대한 내부 상세로직 및 해제 방법에 대해서 제공해드리기 어려운점 양해 부탁드립니다.
또한 추가적으로 S3 의 암/복호화 방법에 대해서 궁금한 부분이 있으신 경우 S3 케이스로 열어주시면 해당 전문엔지니어가 도움을 드릴 수 있습니다.

오늘도 편안한 하루 되시길 바랍니다.
감사합니다.