AWS Parameters and Secrets Lambda ExtensionをPythonのrequestsモジュールなしで使ってみた

こんにちは、つくぼし(tsukuboshi0755)です!

LambdaでSSM Parameter StoreやSecrets Managerから値を取得する際に、AWS Parameters and Secrets Lambda Extension(以降Lambda Extension)を使うと、自前で実装しなくともキャッシュを利用でき、コストの削減やレイテンシーの改善を実現できます。

本機能をPythonで用いる場合、上記含め多くの記事でrequestsモジュールを用いたコードが記載されています。

requestsモジュールはコード量が少ないという特徴がある一方で、Python3.8以降では外部ライブラリになってしまっているため、使用する場合は別途レイヤーを作成する必要があります。これはちょっと面倒ですよね..。

そこで、今回はrequestsモジュールの代わりに、Python3.8以降も標準ライブラリに存在するurllib.requestモジュールを用いて、Lambda Extensionを使ってみたいと思います！

コード内容

以下のコードを用いる事で、SSM Parameter Store(String/Secure String)及びSecrets Managerから値を取得できます。

import urllib.request
import os
import json

# Lambda関数ハンドラー
def lambda_handler(event, context):
  string = get_string(os.environ.get('PARAM_NAME'))
  secure_string = get_secure_string(os.environ.get('SECURE_PARAM_NAME'))
  secret = get_secret(os.environ.get('SECRET_NAME'), os.environ.get('SECRET_KEY'))

  # 取得した情報を用いて、以下にLambdaで実施したい処理内容を記載
  ...

# Parameter StoreからStringを取得する関数
def get_string(param_name):
    params_extension_endpoint = "http://localhost:2773/systemsmanager/parameters/get/?name=" + param_name
    headers = {"X-Aws-Parameters-Secrets-Token": os.environ.get('AWS_SESSION_TOKEN')}
    params_extension_req = urllib.request.Request(params_extension_endpoint, headers=headers)
    with urllib.request.urlopen(params_extension_req) as response:
        param_config = response.read()
    param_value = json.loads(param_config)['Parameter']['Value']
    return param_value

# Parameter StoreからSecure Stringを取得する関数
def get_secure_string(secure_param_name):
    params_extension_endpoint = "http://localhost:2773/systemsmanager/parameters/get/?name=" + secure_param_name + "&withDecryption=true"
    headers = {"X-Aws-Parameters-Secrets-Token": os.environ.get('AWS_SESSION_TOKEN')}
    params_extension_req = urllib.request.Request(params_extension_endpoint, headers=headers)
    with urllib.request.urlopen(params_extension_req) as response:
        secure_param_config = response.read()
    secure_param_value = json.loads(secure_param_config.decode("utf-8"))['Parameter']['Value']
    return secure_param_value

# Secret Managerから取得する関数
def get_secret(secret_name, secret_key):
    secrets_extension_endpoint = "http://localhost:2773/secretsmanager/get?secretId=" + secret_name
    headers = {"X-Aws-Parameters-Secrets-Token": os.environ.get('AWS_SESSION_TOKEN')}
    secrets_extension_req = urllib.request.Request(secrets_extension_endpoint, headers=headers)
    with urllib.request.urlopen(secrets_extension_req) as response:
        secret_config = response.read()
    secret_json = json.loads(secret_config)['SecretString']
    secret_value = json.loads(secret_json)[secret_key]
    return secret_value

今回のコードではリクエストを送信する際に、requestsモジュールの代わりに、以下のurllib.requestモジュールを用いたリソース取得方法を使用しています。

urllib パッケージを使ってインターネット上のリソースを取得するには — Python 3.11.4 ドキュメント

上記のコードを用いる事で、現在LambdaでサポートされているPython3.7~3.11の全てで、追加のレイヤーを作成する事なくLambda Extensionを利用できます。

やってみた

実際に上記のコードで、Lambda ExtensionからSSM Parameter Store及びSecrets Managerの値を取得できるか確認してみます。

コードの動作検証をするため、適当な値を、Parameter Store及びSecret Managerに保存します。

今回は以下の内容で、3つのサービスに分けて値を保存し、Lambdaで取り出せるか試します。

まず以下の通り、Lambda関数をPythonで作成します。

次に以下の通り、Lambda関数のレイヤーにLambda Extensionを追加します。

続いて、Lambdaの環境変数を以下の通り設定します。

さらにLambda実行ロールに対して、SSM Parameter Store及びSecrets Managerへのアクセス権が付与されたポリシーを作成しアタッチします。

{
	"Version": "2012-10-17",
	"Statement": [
		{
			"Effect": "Allow",
			"Action": [
				"ssm:GetParameter*"
			],
			"Resource": [
				"arn:aws:secretsmanager:<region>:<account-id>:parameter:<parameter-name>"
			]
    },
		{
			"Effect": "Allow",
			"Action": [
				"ssm:GetParameter*"
			],
			"Resource": [
				"arn:aws:secretsmanager:<region>:<account-id>:parameter:<secure-parameter-name>"
			]
    },
		{
			"Effect": "Allow",
			"Action": [
				"kms:Decrypt"
			],
			"Resource": [
				"arn:aws:kms:<region>:<account-id>:key/<default-ssm-key-id>"
			]
    },
    {
			"Effect": "Allow",
			"Action": [
				"secretsmanager:GetSecretValue"
			],
			"Resource": [
				"arn:aws:secretsmanager:<region>:<account-id>:secret:<secret-name>"
			]
		}
	]
}

そしてLambdaコードを以下の内容でデプロイします。

import urllib.request
import os
import json

# Lambda関数ハンドラー
def lambda_handler(event, context):
  string = get_string(os.environ.get('PARAM_NAME'))
  secure_string = get_secure_string(os.environ.get('SECURE_PARAM_NAME'))
  secret = get_secret(os.environ.get('SECRET_NAME'), os.environ.get('SECRET_KEY'))

  # CloudWatch Logsに取得した内容を出力
  print("From SSM Parameter Store (String): " + string)
  print("From SSM Parameter Store (SecureString): " + secure_string)
  print("From Secret Manager: " + secret)

# Parameter StoreからStringを取得する関数
def get_string(param_name):
    params_extension_endpoint = "http://localhost:2773/systemsmanager/parameters/get/?name=" + param_name
    headers = {"X-Aws-Parameters-Secrets-Token": os.environ.get('AWS_SESSION_TOKEN')}
    params_extension_req = urllib.request.Request(params_extension_endpoint, headers=headers)
    with urllib.request.urlopen(params_extension_req) as response:
        param_config = response.read()
    param_value = json.loads(param_config)['Parameter']['Value']
    return param_value

# Parameter StoreからSecure Stringを取得する関数
def get_secure_string(secure_param_name):
    params_extension_endpoint = "http://localhost:2773/systemsmanager/parameters/get/?name=" + secure_param_name + "&withDecryption=true"
    headers = {"X-Aws-Parameters-Secrets-Token": os.environ.get('AWS_SESSION_TOKEN')}
    params_extension_req = urllib.request.Request(params_extension_endpoint, headers=headers)
    with urllib.request.urlopen(params_extension_req) as response:
        secure_param_config = response.read()
    secure_param_value = json.loads(secure_param_config.decode("utf-8"))['Parameter']['Value']
    return secure_param_value

# Secret Managerから取得する関数
def get_secret(secret_name, secret_key):
    secrets_extension_endpoint = "http://localhost:2773/secretsmanager/get?secretId=" + secret_name
    headers = {"X-Aws-Parameters-Secrets-Token": os.environ.get('AWS_SESSION_TOKEN')}
    secrets_extension_req = urllib.request.Request(secrets_extension_endpoint, headers=headers)
    with urllib.request.urlopen(secrets_extension_req) as response:
        secret_config = response.read()
    secret_json = json.loads(secret_config)['SecretString']
    secret_value = json.loads(secret_json)[secret_key]
    return secret_value

なお今回のコードは動作確認が目的のため、一目で分かるようにprint関数を用いて、CloudWatch Logsに保存した値を出力しています。
本番環境で上記コードを使用する場合、基本的に取得した秘密情報はCloudWatch Logsに出力しないでください。

以上が完了した後、Lambdaのテストボタンで関数を実行し、CloudWatch Logsで結果を確認します。

SSM Parameter Store及びSecrets Managerから、hoge,fuga,piyoの値が取得できている事が分かりますね。

最後に

今回はrequestsモジュールの代わりにurllib.requestモジュールを用いて、Lambda Extensionを使ってみました。

requestsモジュールと比較して少し記述量は多くなってしまいますが、現状LambdaがサポートしているPythonの全てのバージョンで標準ライブラリとして使用できるのは嬉しいですね。

Lambda Extensionは便利な機能ですので、Lambdaを書く機会がある方は活用を検討してみると良いかもしれません。

以上、つくぼし(tsukuboshi0755)でした！