【AWS】API Gateway + Lambda Authorizerで沼らないために知っておきたいHTTP APIとREST APIの決定的な違い

はじめに

こんにちは！KDDIアジャイル開発センターのTooMeです！
普段，AWSに関する初心者向けの記事をQiitaで投稿しています！

突然ですが，API GatewayのHTTP APIとREST APIの仕様の違い，ご存じでしょうか？
コストやレイテンシー，機能の豊富さが違う．という記事はよく見かけます．
しかし，Lambda Authorizerにおける挙動の違いについて解説した記事はなかなか見つかりませんでした．

先日，まさにこの「挙動の違い」が原因で，私は大きな沼にハマってしまいました...．
Lambda Authorizer関数単体のテストは成功するのに，フロントエンドから呼び出すと延々とエラーが返ってくる...．この謎を解くのに，かなりの時間を溶かしてしまいました．

そこでこの記事では，私の失敗談をもとに，HTTP APIとREST APIでLambda Authorizerの何がどう違うのかに焦点を当て，同じ沼にハマる人を1人でも減らそうと思います．

そもそもHTTP APIとREST APIって何が違うの？

簡単にこの2つの違いを説明するため，HTTP APIのメリット，デメリットに着目すると以下のようになります．

HTTP API

✅ メリット

• 低レイテンシー：REST APIと比べて最大60％低レイテンシー
• 低コスト：REST APIと比べて71%安価
• JWT/OIDCネイティブ対応：CognitoやAuth0などのJWT認証を簡単に組み込める
• 自動デプロイ機能
• シンプルさ：複雑な設定が不必要

❌ デメリット

• 機能が限定的：WAF統合，リクエスト検証，キャッシュなど，REST APIにある高度な機能が利用できない

これらの機能は，一部CloudFrontを経由するなどして代替することができますが，HTTP APIのシンプルさを損なう形になってしまいます．

REST APIでは，このメリット，デメリットがほぼ逆転します．高機能で柔軟ですが，その分コストと設定の複雑さが増します．

そのため，AWS公式のブログを見てみると，まずHTTP APIをデフォルトの選択肢とし，REST APIの追加コストと複雑性を正当化できる機能要件がある場合にREST APIを使用するのがベターとされているみたいです．

Lambda Authorizerとは

AWS公式ドキュメントから抜粋すると，

Lambda オーソライザー (以前はカスタムオーソライザーと呼ばれていました) は、
API へのアクセスを制御するために使用します。
クライアントが API の メソッドをリクエストすると、
API Gateway は Lambda オーソライザーを呼び出します。
Lambda オーソライザーは、発信者の ID を入力として受け取り、IAM ポリシーを出力として返します。

日本語に翻訳されたものということもあり，正直何を言っているのかよくわからないですね．
Geminiに聞いてみました．とてもわかりやすいですね☺️

Lambda Authorizerとは？
**Lambda Authorizer（ラムダ・オーソライザー）は、一言で言うと「APIの門番」**のようなものです。あなたの作ったAPI（Webサービスなどの機能の呼び出し口）に誰でもアクセスできてしまうと困りますよね。そこで、アクセスしてきた人が「正当な利用者かどうか」をチェックする仕組みがLambda Authorizerです。
このチェック処理を、AWSのLambdaというサービスを使って行うため、この名前で呼ばれています。

APIエンドポイントを無防備に公開すると，誰でもアクセスし放題になってしまいます．それはまずいです．そこでLambda Authorizerを設定し，アクセスしてきた人は「正当な利用者か？」をLambda関数でチャックします．

また，Lambda Authorizerでは複数の認証処理（IPアドレス制限，Cognito認証，独自トークンを使用した共通鍵方式等）を1つの関数にまとめたり，認証の仕組みを自由に作ることができます．とっても便利！

私が沼った原因：レスポンス形式の勘違い

原因を一言でいうならば，API GatewayでHTTP APIを作成したにもかかわらず，Lambda AuthorizerでREST APIが期待する形式のレスポンスを返していたことです．

この2つのAPI方式では，それぞれLambda Authorizerに渡される入力データ（イベント）と，Lambda Authorizerが返すべき出力データ（レスポンス）の形式（スキーマ）が全く異なるのです．

2つの形式を表にすると以下のようになります．

項目	REST API	HTTP API
入力(Event)	{ "type": "TOKEN", "methodArn": "arn:...", "authorizationToken": "Bearer eyJhbGci..." }	{ "version": "2.0", "type": "REQUEST", "headers": { "authorization": "Bearer eyJhbGci..." }
出力(Response)	{ "principalId": "user123", "policyDocument": { ... }	{ "isAuthorized": true, "context": { "userId": "user123", ... } }

両者の違いは一目瞭然だと思います．
特に出力（レスポンス）に注目してください．

• REST API：認証が成功すると，そのユーザーがAPIを呼び出す権限を持つことを示す**IAMポリシー(JSON形式)**を返します．
• HTTP API：認証が成功したか否かを，単純なtrue/falseで返します．

私はHTTP APIでAPI Gatewayを構築したにもかかわらず，Lambda AuthorizerではREST API用のIAMポリシーを返していました．API Gatewayからすれば，「trueかfalseが欲しいのに，全然違う形式のJSONが来たぞ？」となり，認証が通らないのです．

これが，単体テストでは気づけなかった理由です．Lambda関数単体のテストでは，API Gatewayを介しません．そのため，関数が正しいIAMポリシーを生成している限り，テストは「成功」と判断されてしまったのです．

おわりに

今回は，私がAPI GatewayとLambda Authorizerで沼った原因について解説しました．
この失敗から，サービスの仕様，特に入力と出力の形式を正確に把握することの重要性を改めて痛感しました．

Lambda Authorizerはハマりどころもありますが，使いこなせれば間違いなく便利です．Lambda最高です！

最後の付録として，Cognito認証行うLambda Authorizerのサンプルコード（Python）を，HTTP API用とREST API用の両方載せておきます．レスポンス形式の違いを見比べてください．

【付録】Cognito認証 Lambda Authorizer サンプルコード（Python）

HTTP API ver

import json
import base64
import os
from datetime import datetime, timezone

def lambda_handler(event, context):
    """
    HTTP API用 Cognito オーソライザー
    """
    try:
        
        # 1. 環境変数の確認
        USER_POOL_ID = os.environ.get('USER_POOL_ID')# cognitoのやつ
        CLIENT_ID = os.environ.get('CLIENT_ID')# cognitoのやつ
        REGION = os.environ.get('REGION', 'ap-northeast-1')
        
        if not USER_POOL_ID or not CLIENT_ID:
            print("ERROR: Missing required environment variables")
            return generate_deny_response("Missing environment variables")
        
        # 2. ヘッダーの確認（HTTP APIの形式）
        headers = event.get('headers', {})
        if not headers:
            print("ERROR: No headers found")
            return generate_deny_response("No headers")
        
        # 大文字小文字を無視してAuthorizationヘッダーを検索
        auth_header = None
        for key, value in headers.items():
            if key.lower() == 'authorization':
                auth_header = value
                print(f"Found authorization header with key '{key}': {value[:50]}...")
                break
        
        if not auth_header:
            print("ERROR: No Authorization header found")
            print(f"Available headers: {list(headers.keys())}")
            return generate_deny_response("No authorization header")
        
        # 3. トークンの抽出
        if not auth_header.startswith('Bearer '):
            print(f"ERROR: Authorization header must start with 'Bearer '. Got: {auth_header[:50]}")
            return generate_deny_response("Invalid authorization format")
        
        token = auth_header.split(' ', 1)[1]
        if not token:
            print("ERROR: Empty token")
            return generate_deny_response("Empty token")
        
        print(f"Extracted token: {token[:50]}...{token[-10:]}")
        
        # 4. JWTの基本構造チェック
        token_parts = token.split('.')
        if len(token_parts) != 3:
            print(f"ERROR: Invalid JWT structure, parts: {len(token_parts)}")
            return generate_deny_response("Invalid JWT structure")
        
        print(f"JWT parts lengths: header={len(token_parts[0])}, payload={len(token_parts[1])}, signature={len(token_parts[2])}")
        
        # 5. JWTのデコードと検証
        try:
            # Base64URLデコード（パディング修正）
            def decode_base64url(data):
                missing_padding = len(data) % 4
                if missing_padding:
                    data += '=' * (4 - missing_padding)
                return base64.urlsafe_b64decode(data)
            
            header = json.loads(decode_base64url(token_parts[0]))
            payload = json.loads(decode_base64url(token_parts[1]))
            
            print(f"JWT Header: {json.dumps(header, indent=2)}")
            print(f"JWT Payload: {json.dumps(payload, indent=2)}")
            
        except Exception as e:
            print(f"ERROR: Failed to decode JWT: {str(e)}")
            return generate_deny_response("JWT decode error")
        
        # 6. 必須クレームの確認
        required_claims = ['iss', 'aud', 'exp']
        for claim in required_claims:
            if claim not in payload:
                print(f"ERROR: Missing required claim: {claim}")
                return generate_deny_response(f"Missing claim: {claim}")
        
        # 7. 発行者の確認
        expected_issuer = f'https://cognito-idp.{REGION}.amazonaws.com/{USER_POOL_ID}'
        actual_issuer = payload['iss']
        
        print(f"Expected issuer: {expected_issuer}")
        print(f"Actual issuer: {actual_issuer}")
        
        if actual_issuer != expected_issuer:
            print("ERROR: Invalid issuer")
            return generate_deny_response("Invalid issuer")
        
        # 8. オーディエンスの確認
        expected_audience = CLIENT_ID
        actual_audience = payload['aud']
        
        print(f"Expected audience: {expected_audience}")
        print(f"Actual audience: {actual_audience}")
        
        if actual_audience != expected_audience:
            print("ERROR: Invalid audience")
            return generate_deny_response("Invalid audience")
        
        # 9. 有効期限の確認
        current_time = datetime.now(timezone.utc).timestamp()
        exp_time = payload['exp']
        
        if current_time >= exp_time:
            print("ERROR: Token expired")
            return generate_deny_response("Token expired")
        
        # 10. 成功
        username = payload.get('cognito:username', payload.get('username', 'unknown'))
        print(f"SUCCESS: Authentication successful for user: {username}")
        
        return generate_allow_response(username, payload)
        
    except Exception as e:
        print(f"CRITICAL ERROR: {str(e)}")
        import traceback
        print(f"Traceback: {traceback.format_exc()}")
        return generate_deny_response(f"Internal error: {str(e)}")

def generate_allow_response(username, payload):
    """HTTP API用の許可レスポンス"""
    response = {
        "isAuthorized": True,
        "context": {
            "username": username,
            "sub": payload.get('sub', ''),
            "email": payload.get('email', ''),
            "token_use": payload.get('token_use', '')
        }
    }
    print(f"Generated ALLOW response: {json.dumps(response, indent=2)}")
    return response

def generate_deny_response(reason):
    """HTTP API用の拒否レスポンス"""
    response = {
        "isAuthorized": False,
        "context": {
            "reason": reason
        }
    }
    print(f"Generated DENY response: {json.dumps(response, indent=2)}")
    return response

REST API ver

import json
import base64
import os
from datetime import datetime, timezone

def lambda_handler(event, context):
    """
    REST API用 Cognito オーソライザー
    """
    try:
        
        # 1. 環境変数の確認
        USER_POOL_ID = os.environ.get('USER_POOL_ID')# cognitoのやつ
        CLIENT_ID = os.environ.get('CLIENT_ID')# cognitoのやつ
        REGION = os.environ.get('REGION', 'ap-northeast-1')
        
        if not USER_POOL_ID or not CLIENT_ID:
            print("ERROR: Missing required environment variables")
            raise Exception('Unauthorized')
        
        # 2. REST APIのイベント構造から認証トークンを取得
        auth_token = event.get('authorizationToken', '')
        method_arn = event.get('methodArn', '')
        
        if not auth_token:
            print("ERROR: No authorizationToken found")
            raise Exception('Unauthorized')
        
        if not method_arn:
            print("ERROR: No methodArn found")
            raise Exception('Unauthorized')
        
        print(f"Method ARN: {method_arn}")
        print(f"Authorization Token: {auth_token[:50]}...")
        
        # 3. トークンの抽出
        if not auth_token.startswith('Bearer '):
            print(f"ERROR: Authorization token must start with 'Bearer '. Got: {auth_token[:50]}")
            raise Exception('Unauthorized')
        
        token = auth_token.split(' ', 1)[1]
        if not token:
            print("ERROR: Empty token")
            raise Exception('Unauthorized')
        
        print(f"Extracted token: {token[:50]}...{token[-10:]}")
        
        # 4. JWTの基本構造チェック
        token_parts = token.split('.')
        if len(token_parts) != 3:
            print(f"ERROR: Invalid JWT structure, parts: {len(token_parts)}")
            raise Exception('Unauthorized')
        
        print(f"JWT parts lengths: header={len(token_parts[0])}, payload={len(token_parts[1])}, signature={len(token_parts[2])}")
        
        # 5. JWTのデコードと検証
        try:
            # Base64URLデコード（パディング修正）
            def decode_base64url(data):
                missing_padding = len(data) % 4
                if missing_padding:
                    data += '=' * (4 - missing_padding)
                return base64.urlsafe_b64decode(data)
            
            header = json.loads(decode_base64url(token_parts[0]))
            payload = json.loads(decode_base64url(token_parts[1]))
            
            print(f"JWT Header: {json.dumps(header, indent=2)}")
            print(f"JWT Payload: {json.dumps(payload, indent=2)}")
            
        except Exception as e:
            print(f"ERROR: Failed to decode JWT: {str(e)}")
            raise Exception('Unauthorized')
        
        # 6. 必須クレームの確認
        required_claims = ['iss', 'aud', 'exp']
        for claim in required_claims:
            if claim not in payload:
                print(f"ERROR: Missing required claim: {claim}")
                raise Exception('Unauthorized')
        
        # 7. 発行者の確認
        expected_issuer = f'https://cognito-idp.{REGION}.amazonaws.com/{USER_POOL_ID}'
        actual_issuer = payload['iss']
        
        print(f"Expected issuer: {expected_issuer}")
        print(f"Actual issuer: {actual_issuer}")
        
        if actual_issuer != expected_issuer:
            print("ERROR: Invalid issuer")
            raise Exception('Unauthorized')
        
        # 8. オーディエンスの確認
        expected_audience = CLIENT_ID
        actual_audience = payload['aud']
        
        print(f"Expected audience: {expected_audience}")
        print(f"Actual audience: {actual_audience}")
        
        if actual_audience != expected_audience:
            print("ERROR: Invalid audience")
            raise Exception('Unauthorized')
        
        # 9. 有効期限の確認
        current_time = datetime.now(timezone.utc).timestamp()
        exp_time = payload['exp']
        
        if current_time >= exp_time:
            print("ERROR: Token expired")
            raise Exception('Unauthorized')
        
        # 10. 成功
        username = payload.get('cognito:username', payload.get('username', 'unknown'))
        sub = payload.get('sub', username)
        print(f"SUCCESS: Authentication successful for user: {username}")
        
        # 11. IAMポリシーの生成（REST API用）
        return generate_policy(sub, 'Allow', method_arn, username, payload)
        
    except Exception as e:
        print(f"CRITICAL ERROR: {str(e)}")
        # REST APIの場合、'Unauthorized'をraiseすることで403を返す
        raise Exception('Unauthorized')

def generate_policy(principal_id, effect, resource, username, payload):
    """REST API用のIAMポリシーレスポンスを生成"""
    
    # リソースARNからAPI全体へのアクセスを許可するARNを作成
    tmp = resource.split(':')
    api_gateway_arn_tmp = tmp[5].split('/')
    aws_account_id = tmp[4]
    
    # API ID、ステージ、メソッド、リソースパスを取得
    api_id = api_gateway_arn_tmp[0]
    stage = api_gateway_arn_tmp[1] if len(api_gateway_arn_tmp) > 1 else '*'
    
    # すべてのメソッドとリソースを許可するARNを作成
    policy_resource = f'arn:aws:execute-api:{tmp[3]}:{aws_account_id}:{api_id}/{stage}/*/*'
    
    auth_response = {
        'principalId': principal_id,
        'policyDocument': {
            'Version': '2012-10-17',
            'Statement': [
                {
                    'Action': 'execute-api:Invoke',
                    'Effect': effect,
                    'Resource': policy_resource
                }
            ]
        },
        'context': {
            'username': username,
            'sub': payload.get('sub', ''),
            'email': payload.get('email', ''),
            'token_use': payload.get('token_use', ''),
            'stringKey': username,  # REST APIではcontextのキー名に制限があるため
            'numberKey': int(datetime.now().timestamp()),
            'booleanKey': True
        }
    }
    
    # usageIdentifierKeyを追加（API使用量の追跡用）
    if 'sub' in payload:
        auth_response['usageIdentifierKey'] = payload['sub']
    
    print(f"Generated {effect} policy: {json.dumps(auth_response, indent=2)}")
    return auth_response