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

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

requestsモジュールの代わりにurllib.requestモジュールを使おう
Lambda ExtensionsAWS LambdaPythonAWS
FacebookHatena blogX
2023.08.11

この記事は公開されてから1年以上経過しています。情報が古い可能性がありますので、ご注意ください。

こんにちは、つくぼし(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で取り出せるか試します。

サービス 種類 名前 キー 値
Parameter Store String /cm-tsukuboshi/param - hoge
Parameter Store Secure String /cm-tsukuboshi/secure-param - huga
Secrets Manager - /cm-tsukuboshi/secret sample piyo

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

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

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

キー 値 補足
PARAM_NAME /cm-tsukuboshi/param SSM Parameter Storeに保存したパラメータ(String)の名前
SECURE_PARAM_NAME /cm-tsukuboshi/secure-param SSM Parameter Storeに保存したパラメータ(Secure String)の名前
SECRET_NAME /cm-tsukuboshi/secret Secrets Managerに保存したシークレットの名前
SECRET_KEY sample Secrets Managerに保存したシークレットのキー

さらに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)でした！

この記事をシェアする

FacebookHatena blogX

関連記事

AWS CDK v2.84.0 で AWS Parameter and Secrets Lambda extension を L2 Construct で構築可能になりました
若槻龍太
2023.06.15
AWS Parameters and Secrets Lambda Extensionを使用してみた
SUMANGALAS
2023.05.18
【アップデート】Lambda Extensionsのオブザーバビリティをさらに強化するTelemetry APIが利用可能になりました
岩田智哉
2022.11.12
AWS Parameters and Secrets Lambda Extensionの実装について色々確認してみた
岩田智哉
2022.10.23