はじめに

ちょっとしたAPIを作りたいとき、関数URL（Function URLs）を有効化したAWS LambdaはHTTPエンドポイントを設定するためのAPI Gatewayを都度建てる必要がないので便利です。
一方で、リクエスト元がIAM認証に利用するSigV4署名付きリクエストの生成に対応できない場合は関数URLがサポートするのAWS_IAM認証機構は利用できないため、ある程度安心してエンドポイントを利用するには引き続きAPI Gatewayを前段に用意したうえでAPIキー認証の設定ないしオーソライザーを実装するか、関数側で独自のリクエスト検証機構を設定する必要があります。

この記事ではセキュリティに関する要件があまり求められないシンプルなタスク向けに、Python実装の単一のLambda関数で簡易的なリクエスト検証機構を実装する例を紹介します。マネジメントコンソールで実装をコピペできるレベルのシンプルさをコンセプトとしています。

この記事で紹介する方法は無認証よりは相対的にマシ程度のTipsです。リクエスト元がSigV4署名付きリクエストに対応している場合はAWS_IAM認証を利用するに越したことはありません。独自実装によるリクエスト検証のリスクを承知し、エンドポイント要件やタスクに応じたセキュリティや認証機構を選定してください。
執筆者は本記事を参考に実装したことで発生するいかなる事象について一切の責任を負いません。

関数URL有効なリクエストのペイロード

Lambdaハンドラでは以下のような構造でリクエストが格納されてきます（便宜的にJSON表記です）。

{
  "version": "2.0",
  "routeKey": "$default",
  "rawPath": "/",
  "rawQueryString": "",
  "headers": {
    "content-length": "123",
    "x-amzn-tls-cipher-suite": "ECDHE-RSA-AES128-GCM-SHA256",
    "x-amzn-tls-version": "TLSv1.2",
    "x-amzn-trace-id": "Root=1-XXX...",
    "x-forwarded-proto": "https",
    "host": "XXX....lambda-url.ap-northeast-1.on.aws",
    "x-forwarded-port": "443",
    "content-type": "application/json",
    "x-forwarded-for": "XXX...",
    "accept": "*/*",
    "user-agent": "curl/7.79.1"
  },
  "requestContext": {
    "accountId": "anonymous",
    "apiId": "XXX...",
    "domainName": "XXX....lambda-url.ap-northeast-1.on.aws",
    "domainPrefix": "XXX...",
    "http": {
      "method": "POST",
      "path": "/",
      "protocol": "HTTP/1.1",
      "sourceIp": "XXX...",
      "userAgent": "curl/7.79.1"
    },
    "requestId": "XXX...",
    "routeKey": "$default",
    "stage": "$default",
    "time": "XX/XXX/XXXX:XX:XX:XX +0000",
    "timeEpoch": 1234567890123
  },
  "body": "{\"hello\": \"world\"}",
  "isBase64Encoded": false
}

この構造をもとにリクエストの内容を検証する実装を作成していきます。

リクエスト検証

関数に到達したリクエスト内容を検証していきます。ここでは最低限HTTPメソッドと環境変数であらかじめ設定した情報をもとに認証情報の検証機構を実装します。

メソッド

エンドポイントが許容するメソッド以外のリクエストを拒否します。

def validate(event):
  if (event.get('requestContext', {}).get('http', {}).get('method') is not None):
    # ここでは POST のみを許可します
    if (event.get('requestContext').get('http').get('method') == 'POST'):
      return True
  
  return False

認証ヘッダ

事前に定義されたキーないしID/PW定義との一致を検証します。

説明を単純にするために環境変数から平文で認証情報を取得していますが、設定する情報は必要に応じてAWS KMSなどによる暗号化・復号化を実施してください。

API キー

環境変数で定義したキーとの一致を確認します。


def authorize(event):
  api_key = os.getenv("API_KEY")
  if (event.get('headers', {}).get('x-api-key') is not None):
    if event['headers']['x-api-key'] == api_key:
      return True
  
  return False

Basic 認証

Authorization: Basic XXXXX... のようなBasic認証リクエストを検証します。

def authorize(event):
  user = os.getenv("BASIC_AUTH_USER")
  password = os.getenv("BASIC_AUTH_PASS")

  encoded_user_and_pass = base64.b64encode('{}:{}'.format(user, password).encode('utf-8')).decode('utf-8')

  if (event.get('headers', {}).get('authorization') is not None):
    if event['headers']['authorization'] == 'Basic {}'.format(encoded_user_and_pass):
      return True
  
  return False

実装例

上記までの検証をハンドラを含めて実装したサンプルが下記になります。ID/PWの取得やレスポンスの取り回しが少し異なります。

from urllib import request
import urllib
import json
import os
import base64
import boto3

def validate(event):
  if (event.get('requestContext', {}).get('http', {}).get('method') is not None):
    if (event.get('requestContext').get('http').get('method') == 'POST'):
      return True
  
  return False

def authorize(event, user, password):
  encoded_user_and_pass = base64.b64encode('{}:{}'.format(user, password).encode('utf-8')).decode('utf-8')
  if (event.get('headers', {}).get('authorization') is not None):
    if event['headers']['authorization'] == 'Basic {}'.format(encoded_user_and_pass):
      return True
  
  return False

def lambda_handler(event, context):

  basic_auth_user = os.environ["BASIC_AUTH_USER"]
  # KMSで暗号化されたパスワードを復号する例
  _basic_auth_pass = os.environ["BASIC_AUTH_PASS"]
  basic_auth_pass = boto3.client('kms').decrypt(CiphertextBlob=base64.b64decode(_basic_auth_pass), EncryptionContext={'LambdaFunctionName': os.environ['AWS_LAMBDA_FUNCTION_NAME']})['Plaintext'].decode('utf-8')

  if (validate(event) is False):
    return {'statusCode': 405, 'headers': {'Content-Type': 'application/json', 'Allow': 'POST'}, 'body': json.dumps({'message': 'Method not allowed'})}
    
  if (authorize(event, basic_auth_user, basic_auth_pass) is False):
    return {'statusCode': 401, 'headers': {'Content-Type': 'application/json', 'WWW-Authenticate': 'Basic'}, 'body': json.dumps({'message': 'Unauthorized'})}
    
  return {'statusCode': 200, 　'headers': {'Content-Type': 'application/json'}, 'body': json.dumps({'message': 'OK'}) }

おわりに

この記事では最低限のリクエスト検証を実施するアイデアを紹介しました。実際には他の要素やボディ等の内容も検証の対象とすることでより安全にAPIを利用できるようになると思います。

参考資料