Zenn
😀

ローカル環境のノリでAWSの署名v4付きリクエストをしよう！

2023/10/01に公開
AWS
Python
Lambda
tech

概要

本記事では、Pythonでローカルに簡易的なWebサーバーを立て、そこにリクエストすることで、AWS署名v4が必要なサービスに、送られてきた内容にそのまま署名を付けてリクエストを投げるプログラムを紹介します。

動機

ローカル環境で動作確認をするのと同じワークフローでAWSに乗っけたdev環境等にアクセスする方法を用意したかったため。
これを使うとテスト等について、ドメインを変えるだけで実行できるのがかなりうれしい。世間によくあるAPIテストのフレームワークで、AWSの署名v4が実装されていそうなのは、postmanぐらいしかなさそうだったため、CI等であったりしても便利かなといった背景がありました。

実装

以下が実装になります。エンドポイントやホストさえ書けば、コピペでほぼ動くはずです。

import http.server
import http.client
import socketserver
import os
from botocore.awsrequest import AWSRequest
from botocore.auth import SigV4Auth
from botocore.credentials import Credentials
import requests

class AWSSigV4ProxyHandler(http.server.SimpleHTTPRequestHandler):
    def forward_request(self):
        end_point = "接続先"
        url = end_point+self.path
        credentials = Credentials(os.environ['AWS_ACCESS_KEY_ID'], os.environ['AWS_SECRET_ACCESS_KEY'])
        headers = dict(self.headers)
        del req_headers['Host']
   
        data = self.rfile.read(int(self.headers.get("content-length", 0)))
        request = AWSRequest(method=self.command, url=url, headers=req_headers, data=data)


        SigV4Auth(credentials, 'execute-api', "ap-northeast-1").add_auth(request)

        headers = {
            'Authorization': request.headers['Authorization'],
            'Host':'接続先ホスト',
            'X-Amz-Date':request.context['timestamp'],
            'X-Amz-Security-Token':os.environ['AWS_SESSION_TOKEN'],
            **req_headers
        }

        if self.command == 'GET':
            response = requests.get(url,headers=headers)
        elif self.command == 'POST':
            response = requests.post(url,headers=headers, data=data)
        
        self.send_response(response.status_code)
        for key, value in response.headers.items():
            self.send_header(key, value)
        self.end_headers()
        self.wfile.write(response.content)

    def do_GET(self):
        self.forward_request()

    def do_POST(self):
        self.forward_request()


if __name__ == "__main__":
    port = 9999
    handler = AWSSigV4ProxyHandler
    httpd = socketserver.TCPServer(("", port), handler)
    print(f"Serving on port {port}")
    httpd.serve_forever()

必要なライブラリ群は次で入るはずです。(なるべく少なめにしています)

pip install botocore
pip install requests

また、環境変数に次の内容をセットしておく必要があります。

AWS_ACCESS_KEY_ID=AWSのアクセスID
AWS_SECRET_ACCESS_KEY=AWSのシークレット
AWS_SESSION_TOKEN=セッショントークン

終わりに

AWSの署名はかなり煩雑な作業が必要で、新しいcurlにはデフォルトで機能が実装されたりもしています。ところが、ほかのフレームワークやアプリケーションで利用したいときには、その中に実装を追加するのは面倒ということもあるかもしれません。急ごしらえの力技ですが、中間サーバーによる署名は一つの選択肢になるかなと思います。

Discussion