보토 3에서 해방.python 3의 표준 라이브러리에서만 AWS 서비스를 처리할 때
개요
안녕하세요!KDDI 아시아 개발센터의 작은 판교입니다.
이번 기사는 KDDI Engineer & Designer Advent Calendar 2021 3일째 보도입니다.
어느 날, 나는 이런 요구를 받았다.
"python 3의 표준 라이브러리만 사용하면 AWS 서비스를 조작할 수 있기를 바랍니다. (이번에는 S3에 대한 CRUD 조작 - Get과 Post에 한정됩니다.)"
이럴 때 우리는 어떻게 그 편리한boto3을 사용하지 않고 이 조건을 만족시킬 수 있는지 검증해 보았다.
boto 3의 DEBUG 로그
import boto3
# boto3が対象で、ログレベルはDEBUG
boto3.set_stream_logger()
# boto3/botocoreの詳細指定まで可能で、ログレベルの変更も可能
boto3.set_stream_logger('boto3.resources', logging.WARN)
# パッケージの指定を''にすると
# boto3/botocore全てのログが出力される。
boto3.set_stream_logger('')
S3에 대한 Get 처리
도대체 왜 사인이 필요해??
간단하게 말하면 서명을 통해 요청의 안전을 확보하기 위해서다.
AWS가 요청한 서명에 대해 다음과 같은 몇 가지 점에서 안전성을 확보했다.
Recrester ID 확인
서명을 통해 키에 유효한 접근자가 요청을 보냈는지 확인할 수 있습니다.
전송 중인 데이터 보호
발송 중인 요청이 변경되는 것을 방지하기 위해 요청 요소에 따라 해시 값을 계산하고 얻은 해시 값을 요청의 일부분으로 포함합니다.
AWS가 요청을 받았을 때 같은 정보를 사용하여 해시 값을 계산하고 요청에 포함된 해시 값과 비교합니다.해시 값이 일치하지 않으면 AWS에서 요청을 거부합니다. =>Canonical Request
이제 HTTP Authorization 머리글을 사용합니다.
잠재적 재방공 방지
요청에 포함된 타임 스탬프의 5분 이내에 AWS에 도착해야 합니다.이 조건이 충족되지 않으면 AWS는 요청을 거부합니다.
서명 버전
S3의 Cannonical Request에는 무엇이 필요합니까?
①: Canonical Request 제작
먼저 S3의 canonical request 형식은 다음과 같습니다.
<HTTPMethod>\n
<CanonicalURI>\n
<CanonicalQueryString>\n
<CanonicalHeaders>\n
<SignedHeaders>\n
<HashedPayload>
HTTPmethod는 GET/PUT/HEAD/DELETE 등 HTTP 방법 중의 하나다.
CanonicalUri는 Uri의 Uri 인코딩 버전입니다.도메인 이름의 "/"에서 시작하여 문자열의 끝 또는 물음표 문자("?")까지의 전부를 지정합니다.
Cannonical QueryString은 URI 인코딩에 대한 질의 매개변수를 지정합니다.
Cannonical Headers는 머리글과 값을 요청하는 목록입니다.개별 머리글 이름 및 값 쌍은 행 분리("\n")로 구분됩니다.머리글 이름은 소문자여야 합니다.또한 Cannoniccal Headers에는 다음이 포함되어야 합니다.
host:s3.amazonaws.com
x-amz-content-sha256:e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b785
2b855
x-amz-date:20130708T220855Z
SignedHeaders는 문자순으로 정렬된 세미콜론으로 구분된 소문자 요청 헤더 이름의 목록입니다.목록의 요청 머리글은 Cannonical Headers 문자열에 포함된 것과 같습니다.
HashedPaylad는 유효한 하중 SHA256 해싱을 요청하는 16진수입니다.
②: 서명할 문자열 만들기
다음은 서명 문자열의 예입니다.
"AWS4-HMAC-SHA256" + "\n" +
timeStampISO8601Format + "\n" +
<Scope> + "\n" +
Hex(SHA256Hash(<CanonicalRequest>))
AWS4-HMAC-SHA 256은 해시 알고리즘인 HMAC-SHA 256을 사용함을 나타냅니다.
timeStamp ISO8601 형식의 현재 UTC 시간을 입력합니다.
Scope는 결과 서명을 특정 날짜, AWS 영역 및 서비스 이름과 연결합니다.결과 서명은 특정 구역과 특정 서비스에서만 유효하며 서명은 지정된 날짜부터 7일 이내에 유효하다.
date.Format(<YYYYMMDD>) + "/" + <region> + "/" + <service> + "/aws4_request"
③: 서명 계산
DateKey = HMAC-SHA256("AWS4"+"<SecretAccessKey>", "<YYYYMMDD>")
DateRegionKey = HMAC-SHA256(<DateKey>, "<aws-region>")
DateRegionServiceKey = HMAC-SHA256(<DateRegionKey>, "<aws-service>")
SigningKey = HMAC-SHA256(<DateRegionServiceKey>, "aws4_request")
코드 설치(Python 3)
import sys, os, base64, datetime, hashlib, hmac
import urllib.request, urllib.response
import urllib.parse
method = 'GET'
service = 's3'
host = 'xxxxbucket.s3.xxxregion.amazonaws.com'
region = 'us-east-1'
request_parameters = ''
# Key derivation functions. See:
# http://docs.aws.amazon.com/general/latest/gr/signature-v4-examples.html#signature-v4-examples-python
def sign(key, msg):
return hmac.new(key, msg.encode('utf-8'), hashlib.sha256).digest()
def getSignatureKey(key, dateStamp, regionName, serviceName):
kDate = sign(('AWS4' + key).encode('utf-8'), dateStamp)
kRegion = sign(kDate, regionName)
kService = sign(kRegion, serviceName)
kSigning = sign(kService, 'aws4_request')
return kSigning
access_key = os.environ.get('AWS_ACCESS_KEY_ID')
secret_key = os.environ.get('AWS_SECRET_ACCESS_KEY')
if access_key is None or secret_key is None:
print('No access key is available.')
sys.exit()
# Create a date for headers and the credential string
t = datetime.datetime.utcnow()
amzdate = t.strftime('%Y%m%dT%H%M%SZ')
datestamp = t.strftime('%Y%m%d') # Date w/o time, used in credential scope
# ************* TASK 1: CREATE A CANONICAL REQUEST *************
canonical_uri = "https://%s%s" % (host, "s3_bucketのkey")
canonical_querystring = request_parameters
payload_hash = hashlib.sha256(("").encode("utf-8")).hexdigest()
canonical_headers = 'host:' + host + '\n' + 'x-amz-content-sha256:' + payload_hash + '\n' + 'x-amz-date:' + amzdate + '\n'
signed_headers = 'host;x-amz-content-sha256;x-amz-date'
canonical_request = method + '\n' + canonical_uri + '\n' + canonical_querystring + '\n' + canonical_headers + '\n' + signed_headers + '\n' + payload_hash
# ************* TASK 2: CREATE THE STRING TO SIGN*************
algorithm = 'AWS4-HMAC-SHA256'
credential_scope = datestamp + '/' + region + '/' + service + '/' + 'aws4_request'
string_to_sign = algorithm + '\n' + amzdate + '\n' + credential_scope + '\n' + hashlib.sha256(canonical_request.encode('utf-8')).hexdigest()
# ************* TASK 3: CALCULATE THE SIGNATURE *************
signing_key = getSignatureKey(secret_key, datestamp, region, service)
signature = hmac.new(signing_key, (string_to_sign).encode('utf-8'), hashlib.sha256).hexdigest()
# ************* TASK 4: ADD SIGNING INFORMATION TO THE REQUEST *************
authorization_header = algorithm + ' ' + 'Credential=' + access_key + '/' + credential_scope + ', ' + 'SignedHeaders=' + signed_headers + ', ' + 'Signature=' + signature
headers = {'x-amz-date':amzdate, 'x-amz-content-sha256': payload_hash, 'Authorization':authorization_header}
request = urllib.request.Request(
canonical_uri,
headers=headers,
method="GET",
)
try:
with urllib.request.urlopen(request) as response:
print("リクエスト送信に成功" + response)
except (ValueError, Exception):
print("リクエスト送信に失敗")
raise
S3 포스트 처리
multippart/form-data에서 보내기
Content-Type: multipart/form-data; boundary=----WebKitFormBoundaryO5quBRiT4G7Vm3R7
Requests를 사용하지 않고 urllib로 멀티퍼스널 / form-data를 보내기
#!/usr/bin/env python3
__all__ = ["MultipartPostHandler"]
from email.generator import _make_boundary
from os.path import basename
from io import IOBase as FILE_TYPE
from urllib.parse import urlencode
from urllib.request import BaseHandler
def b(str_or_bytes):
if not isinstance(str_or_bytes, bytes):
return str_or_bytes.encode("utf-8")
else:
return str_or_bytes
NEWLINE = "\r\n"
class MultipartPostHandler(BaseHandler):
handler_order = BaseHandler.handler_order - 10
def _encode_form_data(self, fields, files):
boundary = _make_boundary()
parts = []
for name, value in fields:
parts.append(b("--%s" % boundary))
parts.append(b("Content-Disposition: form-data; name=\"%s\""
% name))
parts.append(b(""))
parts.append(b(value))
for name, fp in files:
filename = basename(fp.name)
fp.seek(0)
parts.append(b("--%s" % boundary))
parts.append(b("Content-Disposition: form-data; name=\"%s\"; " \
"filename=\"%s\"" % (name, filename)))
parts.append(b(""))
parts.append(fp.read())
parts.append(b("--%s--" % boundary))
data = b(NEWLINE).join(parts)
return boundary, data
def http_request(self, req):
data = req.data
if data and isinstance(data, dict):
fields = []
files = []
for key, value in data.items():
if isinstance(value, FILE_TYPE):
files.append((key, value))
else:
fields.append((key, value))
if files:
boundary, data = self._encode_form_data(fields, files)
req.add_header("Content-Type", "multipart/form-data; " \
"boundary=\"%s\"" % boundary)
req.add_header("Content-Length", len(data))
else:
data = urlencode(fields, doseq=True)
req.data = data
return req
https_request = http_request
#!/usr/bin/env python3
from typing import Any, Dict, Optional, Tuple, BinaryIO
from module.multipart_post_handler import MultipartPostHandler
import urllib.request, urllib.response
import urllib.parse
def main():
response = s3_presigned_post("urlを入れてください", "formを入れてください", "file(バイナリで入れてください。)")
logging.debug(response)
def s3_presigned_post(
url: str, form: Dict[str, str], file: BinaryIO, verify: bool = True
):
opener = urllib.request.build_opener(MultipartPostHandler())
params = form.copy()
params["key"] = params["key"].encode("utf-8")
params["file"] = file
return opener.open(url, params)
if __name__ == "__main__":
main()
총결산
어때?
만약 참고 가치가 있다면 정말 좋겠다.
Reference
이 문제에 관하여(보토 3에서 해방.python 3의 표준 라이브러리에서만 AWS 서비스를 처리할 때), 우리는 이곳에서 더 많은 자료를 발견하고 링크를 클릭하여 보았다 https://zenn.dev/yoshii0110/articles/28e705cc7f4a5d텍스트를 자유롭게 공유하거나 복사할 수 있습니다.하지만 이 문서의 URL은 참조 URL로 남겨 두십시오.
우수한 개발자 콘텐츠 발견에 전념
(Collection and Share based on the CC Protocol.)
좋은 웹페이지 즐겨찾기
