boto3からの解放。python3の標準ライブラリのみでAWSサービスを取り扱うには

概要

こんにちは! KDDIアジャイル開発センターの小板橋です。
今回の記事は、KDDI Engineer & Designer Advent Calendar 2021の3日目の記事となります。

ある日、こんな要望が舞い降りてきました。
「python3の標準ライブラリのみでAWSサービスに対して操作(今回は、S3に対するCRUD操作-GetとPostに限定)できるようにして欲しいな。」

そんな時に、どうやってあの便利なboto3を使用せずにこの要件を満たせるかを検証してみたというものです。

boto3のDEBUGログ

• boto3依存をやめるとなると、まずやらなければならないのはboto3が何を隠蔽してくれているのかを確認する必要があります。
• まあ、単純に言えばBoto3がどんなREST APIのリクエストを投げているのかを確認すればよいのです。
• そこで、Boto3でDEBUGログを出力してあげると簡単にREST APIのリクエストの内容が確認できます。

DEBUGログの設定

debug.py

import boto3

# ログ出力設定
boto3.set_stream_logger()

# boto3/botocoreの詳細指定まで可能で、ログレベルの変更も可能
boto3.set_stream_logger('boto3.resources', logging.WARN)

# パッケージの指定を''にすると
# boto3/botocore全てのログが出力される。
boto3.set_stream_logger('')

S3に対するGet処理

• まず、AWSへのAPIリクエストを行う場合(AWS SDKや、AWS CLI、Boto3などのAWSツールを使わない場合)、リクエストの署名するためのコードを含める必要があります。
• 基本的には、この署名を気にすることはありません。(AWS SDK、AWS CLI、Boto3などのAWSツールは、ツールの設定時に指定するアクセスキーを使用してAPIリクエストに署名します。)

そもそも何で署名が必要なの？？

簡単に言えば、署名によってリクエストのセキュリティ確保をしたい為です。
AWSのリクエストに対する署名では以下の点でセキュリティの確保を行なっています。

• リクエスタのIDの確認
  署名により、有効なアクセスキーを持っている人がリクエストを送信したことを確認できます。

• 送信中のデータの保護
  送信中のリクエストの改ざんを防ぐために、リクエストの要素からハッシュ値を計算し、得られたハッシュ値をリクエストの一部として含めます。
  AWSがリクエストを受け取ると、同じ情報を使用してハッシュを計算し、リクエストに含まれているハッシュ値と比較します。ハッシュ値が一致しない場合、AWSはそのリクエストを拒否します。 => Canonical Request
  この時、HTTP Authorization ヘッダーを使用します。

• 潜在的なリプレイ攻撃の防止
  リクエストに含まれるタイムスタンプの5分以内にAWSに到達する必要があります。その条件を満たさない場合、AWSはリクエストを拒否します。

署名のバージョン

• AWSでは、署名バージョン4と署名バージョン2がサポートされています。AWS CLI, AWS SDKは、署名バージョン4をサポートするすべてのサービスに対して自動で署名バージョン4を使用します。
• 今回も、使用する署名のバージョンは署名バージョン4にします。

S3におけるCanonical Requestでは何が必要？？

• Canonical Requestでは、次のStepで署名の検証が行われます。
• ①: 署名するための文字列を決めます。
• ②: 署名キーを使用して、署名する文字列のHMAC-SHA256ハッシュを計算します。
• ③: s3は認証されたリクエストを受信すると、署名を計算しリクエストで指定した署名と比較します。 (そのため、s3と同じ方法で署名を計算する必要があります。=> ここで、署名のために合意された形式でリクエストを送信するプロセスは、正規化と呼ばれます。

①:Canonical Requestの作成

まず、下記にあるのがS3におけるcanonical requestのフォーマットになります。

<HTTPMethod>\n
<CanonicalURI>\n
<CanonicalQueryString>\n
<CanonicalHeaders>\n
<SignedHeaders>\n
<HashedPayload>

• HTTPMethodは、GET/PUT/HEAD/DELETE等のHTTPメソッドの1つです。
• CanonicalURIは、URIのURIエンコードバージョンです。ドメイン名に続く「/」で始まり、文字列末尾まで、または疑問符文字（ '？'）までのすべてを指定します。
• CanonicalQueryStringは、URIエンコーディングされたクエリパラメータを指定します。
• CanonicalHeadersは、リクエストヘッダーとその値のリストです。個々のヘッダー名と値のペアは、改行文字（ "\ n"）で区切られます。 **ヘッダー名は小文字にする必要があります。**また、CanonicalHeadersは下記のものを必ず含めなければなりません。
  • HTTPホストヘッダー
  • Content-Typeヘッダーがリクエストに存在する場合は、追加
  • リクエストに含める予定のx-amz-*ヘッダーも追加。たとえば、一時的なセキュリティクレデンシャルを使用している場合は、リクエストにx-amz-security-tokenを含める必要があります。 下記、CanonicalHeadersのサンプル。

host:s3.amazonaws.com
x-amz-content-sha256:e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b785
2b855
x-amz-date:20130708T220855Z

• SignedHeadersは、アルファベット順にソートされたセミコロンで区切られた小文字のリクエストヘッダー名のリストです。 リスト内のリクエストヘッダーは、CanonicalHeaders文字列に含めたものと同じヘッダーです。
• HashedPayloadは、リクエストペイロードのSHA256ハッシュの16進値です。
  • ちなみに、GETリクエストを使用してオブジェクトを取得する場合、空の文字列ハッシュを計算します。

②: 署名する文字列を作成

下記が署名する文字列の例になります。

"AWS4-HMAC-SHA256" + "\n" +
timeStampISO8601Format + "\n" +
<Scope> + "\n" +
Hex(SHA256Hash(<CanonicalRequest>))

• AWS4-HMAC-SHA256は、ハッシュアルゴリズムHMAC-SHA256を使用していることを示します。
• timeStampは、ISO8601形式の現在のUTC時刻を入れます。
• Scopeは、結果の署名を特定の日付、AWSリージョン、およびサービス名を連結したものを入れます。 結果の署名は、特定の地域および特定のサービスでのみ機能し、署名は指定された日付から7日間有効です。

ちなみに、Scopeで連結したものの例が下記になります。

date.Format(<YYYYMMDD>) + "/" + <region> + "/" + <service> + "/aws4_request"

③: 署名を計算

• AWS署名バージョン4では、AWSアクセスキーを使用してリクエストに署名する代わりに、最初に特定のリージョンとサービスを対象とする署名キーを作成します。

下記が作成する署名キーの例になります。

DateKey              = HMAC-SHA256("AWS4"+"<SecretAccessKey>", "<YYYYMMDD>")
DateRegionKey        = HMAC-SHA256(<DateKey>, "<aws-region>")
DateRegionServiceKey = HMAC-SHA256(<DateRegionKey>, "<aws-service>")
SigningKey           = HMAC-SHA256(<DateRegionServiceKey>, "aws4_request")

実装コード(Python3)

• それでは、上記の内容を元にPython3の標準ライブラリのみでS3の特定のbucketに対してGetのリクエストを投げてみましょう。
• 下記がその時のコードになります。

sample.py

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に対するPost処理

• S3に対するアップロードの処理で気をつけなければならないのは、アップロードするデータの種類を固定させるわけにはいかないので、複合データ型（=multipart）を扱える、multipart/form-dataという形式でアップロードしなければなりません。

multipart/form-dataでの送信

• multipart/form-dataは、複数の種類のデータを一度に扱える形式です。
• 気をつけなければならないのは、下記のようにboundaryをコンテンツの境界を示す文字列として入れてあげなければなりません。

Content-Type: multipart/form-data; boundary=----WebKitFormBoundaryO5quBRiT4G7Vm3R7

• 今回は、Python3の標準ライブラリのみを使用して、s3に対してアップロードしたいと考えています。
• 外部ライブラリで有名なrequestsというものがありますが、こちらを使うと簡単にmultipart/form-data形式の送信ができますが、今回はurllibで頑張って実装していきます。

requestsを使わずにurllibでmultipart/form-data送信する

• multipart/form-data送信するクラスが下記になります。

multipart_post_handler.py

#!/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

• あとは、このクラスを使用しS3に対してmultipart/form-dataでアップロードを実行してあげれば完成です。

sample_post.py

#!/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()

まとめ

いかがだったでしょうか。
もし、何かの参考になれば幸いです。