Cloudflare R2 파일 업로드 파이썬 코드 예제

• Boto3에 이슈(urllib3)가 있어서 직접 AWS V4 Signiture를 꾸역꾸역 구현
• Cloudflare R2에 도메인을 붙여서 사용하고, 더불어 구현에서의 유연성을 위해 환경변수를 넣어서 설정

예제 코드

'''
%env r2_account_id = ...
%env r2_access_key_id = ...
%env r2_secret_access_key = ...
%env r2_bucket = ...
%env r2_domain = ...
!echo $r2_account_id $r2_access_key_id $r2_secret_access_key $r2_bucket $r2_domain

%%writefile sample.csv
col1, col2, col3
1, 2, 3
4, 5, 6
7, 8, 9
'''

import io
import os
import hashlib
import hmac
import datetime
import requests
from requests.auth import AuthBase

class AWSV4Auth(AuthBase):
    @classmethod
    def sign(cls, key, message):
        return hmac.new(key, message.encode('utf-8'), hashlib.sha256).digest()

    @classmethod
    def get_signature_key(cls, secret_key, date_stamp, region, service):
        k_date = cls.sign(('AWS4' + secret_key).encode('utf-8'), date_stamp)
        k_region = cls.sign(k_date, region)
        k_service = cls.sign(k_region, service)
        k_signing = cls.sign(k_service, 'aws4_request')
        return k_signing

    def __init__(self, access_id, secret_key, region, service):
        self.access_id = access_id
        self.secret_key = secret_key
        self.region = region
        self.service = service

    def __call__(self, r):
        host = r.url.split('://')[1].split('/')[0]
        payload_hash = hashlib.sha256(r.body or b'').hexdigest()
        date_time = datetime.datetime.utcnow()
        amz_date = date_time.strftime('%Y%m%dT%H%M%SZ')
        date_stamp = date_time.strftime('%Y%m%d')

        canonical_request = f"{r.method}\n{r.path_url}\n\nhost:{host}\nx-amz-content-sha256:{payload_hash}\nx-amz-date:{amz_date}\n\nhost;x-amz-content-sha256;x-amz-date\n{payload_hash}"
        algorithm = 'AWS4-HMAC-SHA256'
        credential_scope = f"{date_stamp}/{self.region}/{self.service}/aws4_request"

        string_to_sign = f"{algorithm}\n{amz_date}\n{credential_scope}\n{hashlib.sha256(canonical_request.encode('utf-8')).hexdigest()}"
        signing_key = self.get_signature_key(self.secret_key, date_stamp, self.region, self.service)
        signature = hmac.new(signing_key, (string_to_sign).encode('utf-8'), hashlib.sha256).hexdigest()

        authorization_header = f"{algorithm} Credential={self.access_id}/{credential_scope}, SignedHeaders=host;x-amz-content-sha256;x-amz-date, Signature={signature}"

        r.headers['x-amz-content-sha256'] = payload_hash
        r.headers['x-amz-date'] = amz_date
        r.headers['Authorization'] = authorization_header

        return r

class R2:
    @classmethod
    def put_file(cls, filename, obj):
        aws_auth = AWSV4Auth(
            access_id=os.getenv('r2_access_key_id'),
            secret_key=os.getenv('r2_secret_access_key'),
            region='auto', service='s3')
        URL = 'https://{}.r2.cloudflarestorage.com/{}/{}'.format(
            os.getenv('r2_account_id'), os.getenv('r2_bucket'), filename,
        )
        response = requests.put(URL, data=obj, auth=aws_auth)
        return f'{os.getenv("r2_domain")}/{filename}'

filename = 'sample.csv'
with open(filename, 'rb') as f:
    print(R2.put_file(filename, f.read()))