Zenn
💪

AWSアクセスを偽装し、Lambda→パラメータストア/Secrets Mangerの値をローカルサーバーから取得

2023/12/06に公開
AWS
Python
Lambda
SystemsManager
#
secretsmanager
tech

※ローカルでのテスト環境についてであり、実環境は対象外。

TL;DR

  • ssm.ap-northeast-1.amazonaws.com,
    secretsmanager.ap-northeast-1.amazonaws.comを
    /etc/hostsで127.0.0.1に対応させる
  • 自己署名証明書でsubjectAltName=DNS:*.ap-northeast-1.amazonaws.comを指定して443ポートで偽装サーバーを立てる
  • /lambda-entrypoint.sh実行シェルの環境変数REQUESTS_CA_BUNDLEに自己署名証明書のcrtファイルを指定する

背景

※Python前提
Lambda PowertoolsでParameter StoreやSecrets Mangerの値取得を実装したLambda関数のローカルでのテストは、単体テストならunittestやpytestでget_parameterやget_secretをMock化等すればサーバーを立てずとも簡単にできる。

しかし、実際のLambdaに近い環境としてLambda Runtime Interface Emulator (RIE)でハンドラを立ち上げてAPIを叩くような結合テストでは、仕組み上ハンドラ側で呼び出される関数をMock化することができないのでAWSアカウントを用いて実際の値を取得しにいってしまうため、完全にローカル環境で閉じたテストにはなっていない。

これを回避するためには、AWSへのHTTPSアクセスを偽装したうえで、期待値を返すサーバーを用意する必要がある。

手順

自己署名証明書の作成

リージョンは適宜変えること。

command
export KEY_NAME=server.key && \
    export CSR_NAME=server.csr && \
    export CRT_NAME=server.crt && \
    export SUBJ_NAME=subject.txt && \
    echo "subjectAltName=DNS:*.ap-northeast-1.amazonaws.com" > $SUBJ_NAME && \
    openssl genrsa 2048 > $KEY_NAME && \
    openssl req -new -key $KEY_NAME -out $CSR_NAME -subj "/CN=amazonaws.com" && \
    openssl x509 -req -days 36500 -in $CSR_NAME -signkey $KEY_NAME -out $CRT_NAME -extfile $SUBJ_NAME

Python Dockerコンテナ起動

リージョンは適宜変えること。

command
export WORK_DIR=/root/workspace && \
    docker run \
    -itd \
    --name lambda \
    -v .:$WORK_DIR \
    -p 9000:8080 \
    -e AWS_DEFAULT_REGION=ap-northeast-1 \
    -e AWS_ACCESS_KEY_ID=dummyAWSID \
    -e AWS_SECRET_ACCESS_KEY=dummyAWSSecret \
    -e LAMBDA_TASK_ROOT=$WORK_DIR \
    -e REQUESTS_CA_BUNDLE=$WORK_DIR/$CRT_NAME \
    --add-host=ssm.ap-northeast-1.amazonaws.com:127.0.0.1 \
    --add-host=secretsmanager.ap-northeast-1.amazonaws.com:127.0.0.1 \
    --workdir=$WORK_DIR \
    --entrypoint=/bin/bash \
    public.ecr.aws/lambda/python:3.11
  • アカウント情報はダミー値を設定
  • 今回は同一コンテナ内でサーバーを立てるようにしたため127.0.0.1に対応させている
    もしコンテナ外にするならホスト名やIPを適宜指定
  • REQUESTS_CA_BUNDLEに自己署名証明書の.crtファイルへのパスを指定
    • pipなどの実行でSSLエラーとなるので、その際の実行シェルではenv -u REQUESTS_CA_BUNDLE等で環境変数を無効にしておくこと

以降の実装はすべてコンテナ内で作成する。

パラメータ取得Lambdaサンプル

パラメータストアからはパラメータ名Paramを、Sercrets Managerからはキー名SecretTextをテキストで、SecretJsonをJSONで取得する例

get_params.py
from aws_lambda_powertools.utilities.parameters import get_parameter, get_secret


def lambda_handler(event, context):
    param: str = get_parameter("Param")
    secret_txt: str = get_secret("SecretText")
    secret_json: dict = get_secret("SecretJson", transform="json")
    return dict(param=param, secret_txt=secret_txt, secret_json=secret_json)

偽装サーバー作成

本手順ではFlaskで実装

dummy_server.py
import json
import re
from base64 import b64encode
from re import Match

from flask import Flask, jsonify, request

PARAMETERS: dict[str, str] = {"Param": "param data"}
SECRETS: dict[str, str] = {
    "SecretText": b64encode(b"secret text").decode(),
    "SecretJson": b64encode(
        json.dumps({"SecretKey": "secret value"}).encode()
    ).decode(),
}

SSM_HOST_PATTERN, SECRETS_MANAGER_HOST_PATTERN = [
    re.compile(f"^{sub}\.[a-z]{{2}}-[a-z]+-\d\.amazonaws.com$")
    for sub in ["ssm", "secretsmanager"]
]


def start_server() -> None:
    app: Flask = Flask(__name__)

    @app.post("/")
    def get_param():
        data: dict = json.loads(request.data.decode())
        return (
            jsonify(
                dict(Parameter=dict(Value=PARAMETERS.get(data["Name"])))
                if isinstance(SSM_HOST_PATTERN.match(request.host), Match)
                else dict(SecretBinary=SECRETS.get(data["SecretId"]))
                if isinstance(SECRETS_MANAGER_HOST_PATTERN.match(request.host), Match)
                else {}
            ),
            200,
        )

    app.run(host="0.0.0.0", port=443, ssl_context=("server.crt", "server.key"))


if __name__ == "__main__":
    start_server()
  • 期待値をPARAMETERS、SECRETSで定義
  • SECRETSはbase64でエンコードされている必要あり
  • PowertoolsからのリクエストはJSONではないためデコードしてJSON化
  • 要求ホスト名を正規表現でチェックして動作切り分け

偽装サーバー起動

command
python dummy_server.py

 * Serving Flask app 'dummy_server'
 * Debug mode: off
WARNING: This is a development server. Do not use it in a production deployment. Use a production WSGI server instead.
 * Running on all addresses (0.0.0.0)
 * Running on https://127.0.0.1:443
 * Running on https://172.17.0.3:443
Press CTRL+C to quit

Lambda Runtime Interface Emulator起動

command
/lambda-entrypoint.sh get_params.lambda_handler

** *** **** **:**:**,*** [INFO] (rapid) exec '/var/runtime/bootstrap' (cwd=/root/workspace, handler=)

ホストからAPI呼び出し

command
curl http://localhost:9000/2015-03-31/functions/function/invocations -d '{}'

{"param": "param data", "secret_txt": "secret text", "secret_json": {"SecretKey": "secret value"}}

補足

Powertools(Python)でのHTTPSリクエストはbotocoreのクラスを用いており、SSL検証をデフォルトでは有効としている。
そのため、ただ偽装サーバーを立てるだけでは証明書チェックで弾かれてAPI呼び出しに失敗する。

ただし、環境変数REQUESTS_CA_BUNDLEに証明書のパスを指定することで自前の証明書で検証させることができるので、アクセスしようとするドメインに対する証明書を作成してしまえば弾かれることがなくなる。

/var/lang/lib/python3.11/site-packages/botocore/endpoint.py
class EndpointCreator:
    ・・・
    def _get_verify_value(self, verify):
        # This is to account for:
        # https://github.com/kennethreitz/requests/issues/1436
        # where we need to honor REQUESTS_CA_BUNDLE because we're creating our
        # own request objects.
        # First, if verify is not None, then the user explicitly specified
        # a value so this automatically wins.
        if verify is not None:
            return verify
        # Otherwise use the value from REQUESTS_CA_BUNDLE, or default to
        # True if the env var does not exist.
        return os.environ.get('REQUESTS_CA_BUNDLE', True)

この仕様はおそらくrequestsモジュールを踏襲しているように思われる。

※REQUESTS_CA_BUNDLEに空文字を設定すればどんな自己署名証明書であっても検証が無視されるので取得できるが、アクセスのたびに警告が出る。

他のサービスやライブラリについては確認していないが、例えばboto3等でも同様にAWSへのアクセスを偽装して完全にローカルで閉じたテスト環境を構築することはできるはずで、
もっと言えばAWSや言語に限らず特定のホストにアクセスするどんな機能に対しても、自前の証明書で検証させることができる仕組みさえあれば同じ手段で対応できると思われる。

ただし、当然偽装のために環境変数が変わっていることで実環境とは異なる挙動となっているため完全に環境をエミュレートしたテストにはなっていないが、少なくともpytestのような単体テストだけでなく上記のような方法でテストの幅を広げることは可能である。

Discussion