0
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

@masanori0(尾崎 正典)in朝日新聞社

Terraform で作る API Gateway + Lambda：GET クエリ（`apikey`）だけで API キー認証する

0
Posted at

CTO室の尾崎です。
Terraform で API Gateway（REST API）+ Lambda を構築し、認証に カスタムオーソライザー（REQUEST） を使いながら、GET のクエリストリング（apikey）だけで認証できる構成を紹介します。

（「URL だけで叩ける」要件があるときに、現実的にどう組むかにフォーカスした記事です）

TL;DR（結論だけ先に）

  • REQUEST 型カスタムオーソライザーで identity_source = "method.request.querystring.apikey" を使う
  • REST API 側で api_key_source = "AUTHORIZER" とし、オーソライザーが usageIdentifierKey を返して Usage Plan をカウントさせる
  • メソッド側は api_key_required = true + request_parameters で クエリ apikey 必須にする
  • 管理 API（boto3）の一覧系は件数制限に注意し、limit=500 + position で ページネーションする

目次

この記事でわかること

  • Terraform で API Gateway（REST API）+ Lambda を構築する考え方
  • **カスタムオーソライザー（REQUEST）**で、GET のクエリ apikey を認証の入力にする方法
  • Usage Plan を usageIdentifierKey でカウントさせる構成のポイント
  • 実装でハマりやすい点（ページネーション、キャッシュ、/favicon.ico など）

先に注意

クエリに API キーを載せるのは、URL のログ（アクセスログ、履歴、監視、共有、リファラ等）に残りやすく、漏洩リスクが上がります。
それでも「URL だけで叩ける」要件がある場合に限って採用するのが現実的です。

全体構成（このリポジトリの実装）

登場要素はシンプルで、役割は次のとおりです。

  • API Gateway（REST API）
    • GET /api：CUSTOM（カスタムオーソライザー）
  • Lambda（API 本体）：API 本体の処理
  • Lambda（オーソライザー）：API キー検証・Usage Plan 連携

リクエストの流れ（ざっくり）はこうなります。

  • GET /api?apikey=...
  • API Gateway → オーソライザー（REQUEST でクエリから apikey を読む）
  • オーソライザーが Allow/Deny（Allow 時は usageIdentifierKey を返す）
  • Allow のときだけ API 本体 Lambda が呼ばれる

Terraform：カスタムオーソライザーを「クエリ apikey」で動かす

この構成の肝は、以下の 2 点です。

  • オーソライザーを REQUEST 型にする（ヘッダ/クエリ等から入力を取れる）
  • identity_source をクエリ apikey にする（認証入力・キャッシュキー）

（ここから先は「1 行ずつ何をしているか」をコメントで説明します）

resource "aws_api_gateway_rest_api" "api" {
  # オーソライザーが返す usageIdentifierKey をUsage Planのカウントに使う設定
  api_key_source = "AUTHORIZER"
}

resource "aws_api_gateway_authorizer" "custom" {
  # REQUEST型オーソライザー（ヘッダ/クエリ等から値を取れる）
  type = "REQUEST"

  # 認証の入力（およびキャッシュキー）としてクエリ apikey を使う
  identity_source = "method.request.querystring.apikey"

  # API Gateway側の「オーソライザー結果キャッシュ」は無効（常にオーソライザーを呼ぶ）
  authorizer_result_ttl_in_seconds = 0
}

Terraform：GET メソッド側（クエリ必須 + Usage Plan カウント）

/api は、カスタムオーソライザー必須の GET メソッドとして定義します。

resource "aws_api_gateway_method" "method" {
  # カスタムオーソライザーを適用する
  authorization = "CUSTOM"
  authorizer_id = aws_api_gateway_authorizer.custom.id

  # Usage Planのカウントを有効化する（api_key_source = "AUTHORIZER" とセットで効く）
  api_key_required = true

  request_parameters = {
    # クエリ apikey を必須にする（ここが「クエリだけで認証」を担保する）
    "method.request.querystring.apikey" = true
  }
}

Terraform：authorizer_function.py / lambda_function.py を ZIP 化して Lambda に定義する

このリポジトリでは、./modules/lambda/api ディレクトリを 1 つの ZIP に固め、
その ZIP を API 本体 Lambda と Authorizer Lambda の両方で使っています（違いは handler だけです）。

################################
# Lambda
################################
# apiディレクトリにLambdaのソースコードがある前提
# apiディレクトリを api.zip という名前に固めて resource "aws_lambda_function" "api" から参照できるようにする
data "archive_file" "lambda_zip" {
  type        = "zip"
  source_dir  = "./modules/lambda/api"
  output_path = "./.terraform/tmp/api.zip"
}

resource "aws_lambda_function" "lambda_api" {
  depends_on       = [aws_iam_role.lambda_role,aws_cloudwatch_log_group.lambda_accesslog]
  filename         = data.archive_file.lambda_zip.output_path
  function_name    = "${var.resource_prefix}-api-${var.stage_name}"
  role             = aws_iam_role.lambda_role.arn
  handler          = "lambda_function.lambda_handler"
  architectures    = ["arm64"]
  runtime          = "python3.12"
  timeout          = 30
  layers           = [aws_lambda_layer_version.requests.arn]
  #kms_key_arn      = var.kms_key_arn
  tags             = "${var.tags}"
  source_code_hash = data.archive_file.lambda_zip.output_base64sha256
  environment {
    variables = {
      API_KEY      = var.apikey
      RESOURCE_PREFIX = var.resource_prefix
      STAGE_NAME = var.stage_name
    }
  }
}

# カスタムオーソライザー用のLambda関数
resource "aws_lambda_function" "api_authorizer" {
  depends_on       = [aws_iam_role.lambda_role,aws_cloudwatch_log_group.authorizer_accesslog]
  filename         = data.archive_file.lambda_zip.output_path
  function_name    = "${var.resource_prefix}-authorizer-${var.stage_name}"
  role             = aws_iam_role.lambda_role.arn
  handler          = "authorizer_function.lambda_handler"
  architectures    = ["arm64"]
  runtime          = "python3.12"
  timeout          = 10
  tags             = "${var.tags}"
  source_code_hash = data.archive_file.lambda_zip.output_base64sha256
  environment {
    variables = {
      # Usage Plan名を動的に構築（${resource_prefix}-usage-plan-${stage_name}の形式）
      RESOURCE_PREFIX = var.resource_prefix
      STAGE_NAME = var.stage_name
    }
  }
}

1 行ずつ説明（要点）

  • data "archive_file" "lambda_zip": ディレクトリを ZIP 化するためのデータソースです。
  • type = "zip": ZIP 形式で固めます。
  • source_dir = "./modules/lambda/api": authorizer_function.py / lambda_function.py が置かれたディレクトリを指定します。
  • output_path = "./.terraform/tmp/api.zip": 生成する ZIP の出力先です（Terraform 実行時に作られます）。
  • filename = data.archive_file.lambda_zip.output_path: aws_lambda_function に ZIP を渡します。
  • source_code_hash = data.archive_file.lambda_zip.output_base64sha256: ZIP の変更検知用で、更新デプロイのトリガーになります。
  • handler = "lambda_function.lambda_handler": API 本体 Lambda のエントリポイントです（lambda_function.py の lambda_handler）。
  • handler = "authorizer_function.lambda_handler": Authorizer Lambda のエントリポイントです（authorizer_function.py の lambda_handler）。

オーソライザー：usageIdentifierKey を返す理由（Usage Plan 連携）

REST API で api_key_source = "AUTHORIZER" を使う場合、Usage Plan のカウントに使われる値は、
リクエストヘッダ x-api-key ではなく、オーソライザーが返す usageIdentifierKey です。

オーソライザーの Allow レスポンスでは、以下のように usageIdentifierKey を含めます。

# 文字列が半角の英数字のみで構成されているかどうかを判定
def is_alphanumeric_ascii(s):
    return True if re.fullmatch('[\\d\\w]+', s, re.ASCII) else False

def generate_allow_response(api_key, method_arn, usage_identifier_key=None):
    # principalId はAPI Gatewayの仕様上必須
    response = {
        "principalId": api_key,
        "policyDocument": {
            "Version": "2012-10-17",
            "Statement": [{"Action": "execute-api:Invoke", "Effect": "Allow", "Resource": method_arn}],
        },
        "context": {"apiKey": api_key},
    }

    # api_key_source = "AUTHORIZER" のとき、Usage Planカウントのために usageIdentifierKey を返す
    if usage_identifier_key:
        response["usageIdentifierKey"] = usage_identifier_key

    return response

def generate_deny_response(error_message, method_arn):
    """
    認証失敗時のDenyレスポンスを生成
    API Gatewayが403 Forbiddenエラーを返し、バックエンドのLambda関数を呼び出さないようにする
    """
    # methodArnからリソースARNを構築
    if method_arn and method_arn != '*':
        arn_parts = method_arn.split('/')
        if len(arn_parts) >= 2:
            resource_arn = f"{arn_parts[0]}/{arn_parts[1]}/*/*"
        else:
            resource_arn = method_arn
    else:
        resource_arn = '*'

    # Denyポリシーを返すことで、API Gatewayが403 Forbiddenエラーを返し、
    # バックエンドのLambda関数を呼び出さないようにする
    return {
        'principalId': 'unauthorized',
        'policyDocument': {
            'Version': '2012-10-17',
            'Statement': [
                {
                    'Action': 'execute-api:Invoke',
                    'Effect': 'Deny',
                    'Resource': resource_arn
                }
            ]
        },
        'context': {
            'error': error_message
        }
    }

authorizer_function.py：API キー認証の「標準フロー」

このリポジトリの authorizer_function.py は、**「GET のクエリ apikey を優先して読む」**ことを前提に、次の流れで認証します。

1) 入力（API キー）をどこから取るか

実装は クエリ優先で、無ければ ヘッダを見ます（REQUEST 型オーソライザーのイベント入力を利用）。

# まずクエリパラメータからAPIキーを取得（優先）
query_params = event.get('queryStringParameters')
if query_params:
    api_key = query_params.get('apikey')

# クエリパラメータにない場合はヘッダーから取得
if not api_key:
    headers = event.get('headers') or {}
    # ヘッダー名は小文字に変換される可能性があるため、両方をチェック
    api_key = headers.get('x-api-key') or headers.get('X-Api-Key')

1 行ずつ説明

  • query_params = event.get('queryStringParameters'): クエリパラメータ辞書を取得します（無い場合はNone）。
  • if query_params:: クエリが存在する場合だけ処理します。
  • api_key = query_params.get('apikey'): apikey の値を取り出します（存在しない場合はNone）。
  • if not api_key:: クエリに無い（または空）場合だけヘッダを見ます。
  • headers = event.get('headers') or {}: ヘッダ辞書を取得し、無い場合は空辞書にします。
  • headers.get('x-api-key') or headers.get('X-Api-Key'): 大文字小文字揺れを考慮してヘッダから API キーを取ります。

2) 必須チェックと形式チェック（英数字のみ）

if not api_key:
    return generate_deny_response("Missing API key", event.get('methodArn', '*'))

if not is_alphanumeric_ascii(api_key):
    return generate_deny_response("Invalid API key format", event.get('methodArn', '*'))

1 行ずつ説明

  • if not api_key:: キーが取得できなかったら即 Deny します。
  • generate_deny_response(...): API Gateway に 403 を返させ、バックエンド Lambda を呼ばせません。
  • is_alphanumeric_ascii(api_key): 半角英数字のみかをチェックします（正規表現）。
  • 形式不正も即 Deny します。

3) Usage Plan に紐づくキーかチェックし、usageIdentifierKey を返す

Usage Plan 名は環境変数から組み立てます（Terraform 側の命名と一致させる前提）。

resource_prefix = os.environ.get('RESOURCE_PREFIX')
stage_name = os.environ.get('STAGE_NAME')
target_usage_plan_name = f"{resource_prefix}-usage-plan-{stage_name}"

1 行ずつ説明

  • RESOURCE_PREFIX / STAGE_NAME: リソース命名・環境を表す環境変数です。
  • target_usage_plan_name = f"...": Usage Plan 名を固定フォーマットで生成します。

Usage Plan ID と、紐づく API キー一覧を取得して照合します。

target_usage_plan_id = get_usage_plan_id(apigateway, target_usage_plan_name)
api_keys_set, api_key_mapping, api_key_id_mapping = get_api_keys_from_usage_plan(apigateway, target_usage_plan_id)

if api_key_for_validation in api_keys_set:
    usage_plan_api_key = api_key_mapping.get(api_key_for_validation)
    return generate_allow_response(api_key, event.get('methodArn', '*'), usage_plan_api_key)

1 行ずつ説明

  • get_usage_plan_id(...): Usage Plan 一覧から名前一致で ID を探します（ページネーション対応）。
  • get_api_keys_from_usage_plan(...): Usage Plan に紐づくキーをまとめて取得します（ページネーション対応）。
  • api_keys_set: 存在判定を高速化するための set です。
  • api_key_mapping: “検証用キー”→“Usage Plan 登録値”のマップです（usageIdentifierKey に使う）。
  • if api_key_for_validation in api_keys_set:: Usage Plan に紐づくかの判定です。
  • generate_allow_response(..., usage_plan_api_key): usageIdentifierKey に “Usage Plan 登録値” を入れて Allow します。

参考：get_usage_plan_id() の実装（ページネーション + キャッシュ）

このリポジトリの get_usage_plan_id() は、Usage Plan を全件取得して名前一致で探す実装です。
また、Lambda 実行環境の再利用を前提に、Usage Plan ID をグローバル変数でキャッシュします。

def get_usage_plan_id(apigateway, usage_plan_name):
    """
    Usage Plan IDを取得（キャッシュを活用）
    Lambdaの実行環境が再利用されるため、グローバル変数でキャッシュ可能
    """
    global _cached_usage_plan_id, _cached_usage_plan_name

    # キャッシュが有効な場合はそれを返す
    if _cached_usage_plan_id and _cached_usage_plan_name == usage_plan_name:
        return _cached_usage_plan_id

    # キャッシュがない、または異なるUsage Plan名の場合は取得
    # すべてのUsage Planを取得して、名前で検索
    # limit=500を指定して最大500件まで1回で取得、それ以上ある場合はページネーションで全件取得
    try:
        usage_plans = []
        position = None

        # ページネーションループ: limit=500で取得し、positionが返された場合は次のページを取得
        while True:
            if position:
                # 次のページを取得
                usage_plans_response = apigateway.get_usage_plans(limit=500, position=position)
            else:
                # 最初のページを取得（limit=500で最大500件まで1回で取得可能）
                usage_plans_response = apigateway.get_usage_plans(limit=500)

            # 取得したUsage Planをリストに追加
            items = usage_plans_response.get('items', [])
            usage_plans.extend(items)

            # positionが返されていない場合は全件取得完了
            position = usage_plans_response.get('position')
            if not position:
                break

        # 指定された名前のUsage Planを検索
        target_usage_plan_id = None
        for plan in usage_plans:
            plan_name = plan.get('name')
            plan_id = plan.get('id')
            if plan_name == usage_plan_name:
                target_usage_plan_id = plan_id
                break

        # 見つかった場合はキャッシュに保存
        if target_usage_plan_id:
            _cached_usage_plan_id = target_usage_plan_id
            _cached_usage_plan_name = usage_plan_name
        else:
            print(f"Usage plan not found: {usage_plan_name}")

        return target_usage_plan_id
    except Exception as e:
        print(f"Error getting usage plans: {str(e)}")
        traceback.print_exc()
        return None

1 行ずつ説明（要点）

  • if _cached_usage_plan_id ...: 同じ Usage Plan 名なら、前回の ID をそのまま返して管理 API の呼び出しを省きます。
  • apigateway.get_usage_plans(limit=500, position=...): デフォルト 25 件制限を回避しつつ、呼び出し回数を最小化します。
  • position = usage_plans_response.get('position'): 次ページがある場合に返るトークンで、無ければ全件取得完了です。
  • for plan in usage_plans: ... if plan_name == usage_plan_name: 取得した一覧から「名前一致」で目的の ID を特定します。

実務でハマりがち：get_usage_plans() は 25 件で止まる

API Gateway の管理 API（boto3）は、一覧系 API がデフォルトで 25 件 のことがあります。

このリポジトリのオーソライザーでは、以下の方針で対応しています。

  • limit=500（最大）で取得して、500 件以下なら 1 回で終わらせる
  • position が返る場合は ページネーションで全件取得する

この方式は、高速（API コール回数が最小） と 堅牢（500 件超でも全件取得） を両立します。

まとめ

  • Terraform で API Gateway + Lambda + カスタムオーソライザー を構築し、GET のクエリ apikey だけで認証する実装パターンを紹介しました
  • Usage Plan をカウントさせるには、api_key_source = "AUTHORIZER" と usageIdentifierKey の返却が重要です
  • 一覧取得 API のデフォルト件数制限（25 件）には、limit=500 + positionページネーション が堅牢です
0
0
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
0
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?