AWS Step Functions を使って NAT Gateway 無しの VPC から外部へ HTTPS 通信
🗓 AWS Lambda と Serverless Advent Calendar 2023 の 25 日目 🗓
🔰 はじめに
対象
以下のサービスの基本的な知識がある方
- Amazon VPC や VPC エンドポイント
- AWS Step Functions
- AWS Lambda
- AWS SDK for Python (Boto3)
やること
こちらの図のように NAT Gateway 無しで、VPC 外への HTTPS リクエスト結果を取得します。この図ではユーザーアカウントの VPC へ紐付けした Lambda 関数の使用を表現していますが、EC2 インスタンスや AWS Fargate を使用している場合でも実現可能です。
背景
2023/11/26 に Step Functions が HTTPS エンドポイントのサポートをしました ! この機能を使用して、プライベートサブネット内から NAT Gateway 無しで VPC 外へ HTTPS でリクエストして結果を取得する方法を実装してみました !
🔖 要約
Boto3 の場合は Step Functions の StartSyncExecution のレスポンスから欲しい部分を取り出せる。しかし単純な dict 型でキーにアクセスできない部分があるため、欲しい部分を取り出すコードで少しだけ手間な処理を各必要がある。
手順
- プライベートサブネットのみの VPC を用意
- Amazon EventBridge で「接続」というリソースを用意する (今回は認証不要の通信の検証のため、
no-authという名前で認証タイプをAPI キーにして適当な値を入れて作成) - Step Functions でステートマシンを Express タイプ で用意
- Step Functions で同期呼び出しするための VPC エンドポイント (
com.amazonaws.<region-code>.sync-states) を用意 + HTTPS のインバウンドを許可するセキュリティグループも必要 - Step Functions の StartSyncExecution を実行するコードを書いた Lambda 関数を用意 + HTTPS のアウトバウンドを許可するセキュリティグループも必要
- Lambda 関数の IAM ロールに StartSyncExecution を許可するポリシーを追加
注意点
- HTTPS 以外のエンドポイントは未対応であるため、HTTP は使用できない
- 受け取れるレスポンスのサイズは Task の Input/Output の制限 (現時点で 256 KB) の影響を受ける
- 受け取れる レスポンス形式に制限 がある
- 現状は HTTP ヘッダーの
content-typeがapplication/octet-stream,image/*,video/*,audio/*の場合や、バイナリなど文字列で読み取れないレスポンス
- 現状は HTTP ヘッダーの
🧑🏫 実施した手順の詳細と解説
🌟 Step Functions でステートマシンを Express タイプ で用意
Step Functions でステートマシンを作成するときは、Standard か Express のどちらかを選択 しますが、今回は Express タイプを選択します。
その理由は、Express タイプの場合だと StartSyncExecution を使用して、ステートマシンの実行してその結果をレスポンスで受け取れるためです。
Standard タイプの場合は StartExecution でステートマシンを非同期実行しかできないため、結果を受け取るには DescribeExecution が必要です。そのため今回はコード量を少なくできるように Express タイプにしています。
ステートマシンを Amazon States Language で定義した内容はこちらです。
-
"Retry"の設定は必要に応じて変更
{
"Comment": "Request HTTPS endpoint using input",
"StartAt": "Call HTTPS API",
"States": {
"Call HTTPS API": {
"Type": "Task",
"Resource": "arn:aws:states:::http:invoke",
"Parameters": {
"ApiEndpoint.$": "$.apiEndpoint",
"Method.$": "$.method",
"Authentication": {
"ConnectionArn.$": "$.connectionArn"
},
"RequestBody.$": "$.requestBody",
"QueryParameters.$": "$.queryParameters",
"Headers.$": "$.headers"
},
"Retry": [
{
"ErrorEquals": [
"States.ALL"
],
"BackoffRate": 2,
"IntervalSeconds": 1,
"MaxAttempts": 3,
"JitterStrategy": "FULL"
}
],
"End": true
}
}
}
ここでのポイントをいくつか挙げます。
State 定義の "Resource" は "arn:aws:states:::http:invoke"
Task の種類は こちらのドキュメント にある通りですが、新しく追加された An HTTP Task を使用します。
"Parameters" 内の要素に HTTPS 接続に必要な以下の情報をステートマシン実行時の Input から取得するように設定
固定値の使用や必須要素以外を削除しても良いですが、外部 API のリクエストに必要な値を実行時の Input から受け取れるようにしています。
| パラメーター名 | 簡易的な説明 |
|---|---|
ApiEndpoint |
https:// から始まる文字列 |
Method |
HTTP メソッドの GET や POST などの文字列 |
ConnectionArn |
接続の ARN の文字列 |
RequestBody |
JSON |
QueryParameters |
JSON |
Headers |
JSON |
🌟 Step Functions の StartSyncExecution を実行するコードを書いた Lambda 関数を用意
簡易的な検証のため、ステートマシンの実行結果から外部 API 呼び出しのレスポンス部分だけを取得します。
認証なしでサイズの少ない json を取得したかったため、GitHub にある AWS Config の定義ファイル を raw データで取得できる URL を使用しています。
import json
import os
import boto3
sfn_client = boto3.client('stepfunctions')
def lambda_handler(event, context):
res = sfn_client.start_sync_execution(
stateMachineArn=os.environ.get('STATE_MACHINE_ARN'),
# API に必要なパラメータを文字列で用意、少し変更して動的に値を設定するコードにできるかも
input='{"apiEndpoint": "' + os.environ.get('HTTPS_ENDPOINT') + '",'
+ '"method": "GET",'
+ '"requestBody": "",'
+ '"queryParameters": {},'
+ '"headers": {},'
+ '"connectionArn": "' + os.environ.get('CONNECTION_ARN') + '"}'
)
# start_sync_execution のレスポンスの型と中身を表示
print(type(res)) # <class 'dict'>
print(res)
# res['output'] に StateMachine の Task が出力した最終結果が str 型で格納されている
# str 型ではあるが json 文字列なので、dict 型として扱うには json.loads で変換
print(type(res['output'])) # <class 'str'>
output_dict = json.loads(res['output'])
# output の子要素の 'ResponseBody' に apiEndpoint からのレスポンスが格納されている
print(type(output_dict['ResponseBody'])) # <class 'str'>
# レスポンスが json の前提であれば json.loads で dict 型に変換して
response_body_dict = json.loads(output_dict['ResponseBody'])
return {
'result': response_body_dict
}
この Lambda 関数を実行すると以下のような結果を受け取れます。
{
"result": {
"accountId": "string",
"arn": "string",
"availabilityZone": "string",
"awsRegion": "string",
:
"version": "string"
}
}
ここでのポイントをいくつか挙げます。
Boto3 で start_sync_execution を実行した戻り値は dict 型だけど、子要素に str 型の json 文字列が入っている
res = sfn_client.start_sync_execution の戻り値の内容は少し加工していますが以下のような形式であり、res['output']['ResponseBody'] のように必要な部分だけを取り出したかったのですができません。
{
"output": {
"Headers": {
:
},
"ResponseBody": {
"accountId": "string",
"arn": "string",
"availabilityZone": "string",
:
},
"StatusCode": 206,
"StatusText": "Partial Content"
},
"outputDetails": {
"truncated": false
}
}
なぜかというと、実は "output" の値部分は文字列型だからです。
実際の res = sfn_client.start_sync_execution の戻り値は以下のような形式で、エスケープが行われているのがわかります。
{
"output": "{
\"Headers\": {
:
},
\"ResponseBody\": \"{
\\\"accountId\\\": \\\"string\\\",
\\\"arn\\\": \\\"string\\\",
\\\"availabilityZone\\\": \\\"string\\\",
:
}\",
\"StatusCode\": 206,
\"StatusText\": \"Partial Content\"
},
"outputDetails": {
"truncated": false
}
}
つまり res['output']['ResponseBody'] とすると string indices must be integers, not 'str' のエラーが出ます。そのため output_dict = json.loads(res['output']) のように、まずは "output" を dict 型にしています。
さらに output_dict['ResponseBody'] で取得した値も文字列型であるため、外部 API のレスポンスを json で受け取る場合、それを dict として扱いたい場合は再度 response_body_dict = json.loads(output_dict['ResponseBody']) のような処理が必要です。これで response_body_dict を dict 型として扱えるようになりました。
✅ まとめ
簡単にできるかなと思って試してみたら、StartSyncExecution の戻り値の扱いにけっこう苦労しました。
あとステートマシンの結果を受け取るために Step Functions の Standard と Express タイプの違いを改めて整理できたのでよかったです !
同期と非同期実行の場合で、VPC エンドポイントが異なる点は知らなかったので勉強になりました 💡
今回は HTTP Task を使用しましたが、Step Functions では AWS SDK の統合 があるので、NAT Gateway がなくてもプライベートサブネットの Lambda 関数などから AWS サービスを使用したりもできそうですね !
Discussion