【AWS】Cognitoを用いてシステム間認可してみた

はじめに

背景・目的

• M2M認可(システム間認可)の実装方式について、有効期限やアクセス範囲を持つことによるセキュリティ観点から、OAuth 2.0を採用することがあるかと思います。
• Amazon Cognitoを用いることで、OAuth 2.0 Client Credentials GrantでのM2M認可を簡易に実装することができます。
• 本記事では、OAuth 2.0 Client Credentials Grantの概要、及びAmazon Cognitoを用いたM2M認可の実装方法について解説します。

想定読者

• M2M認可の実装方式に関心がある、OAuth 2.0の概要を理解している、Amazon Cognitoのサービス概要を理解している方を想定して、本記事を執筆しています。

OAuth 2.0 Client Credentials Grantの概要

• OAuth 2.0のアクセス権限付与において、クライアント認証のみを行い、アクセストークンを取得する方式です。
  • クライアントは認可サーバのトークンエンドポイントへリクエストし、クライアント認証を行います。
  • 認証情報が有効な場合、認可サーバはアクセストークンを発行し、クライアントへ返却します。
• ユーザ認証は行われないことから、主にM2M認可で利用されます。
• なお、Client Credentials Grantはコンフィデンシャルクライアントのみに許可されています。
  (参考元: RFC 6749: The OAuth 2.0 Authorization Framework, OAuth 2.0 grants)

Amazon Cognitoを用いたM2M認可の実装

• CDKを用いてAmazon CognitoやAmazon API Gateway、AWS Lambda資源を構築していきます。
  (参考元: amazon-cognito-and-api-gateway-based-machine-to-machine-authorization-using-aws-cdk, baseline-environment-on-aws)

Amazon Cognito構築

• Cognitoユーザープールに関する資源を構築します。
• clientCredentials、及びgenerateSecretの有効化を忘れないように留意ください。

    // ------------ Prerequisite ---------------
    const prefix: string = `${props.appName}-${props.envName}`

    // ------------ Amazon Cognito ---------------
    // ---- User Pool
    // Create User Pool
    const m2mUserPool = new cognito.UserPool(this, "M2mUserPool", {
      userPoolName: `${prefix}-cognito-userpool-m2m`,
      selfSignUpEnabled: false,
    })

    // Create Domain Name
    const m2mDomainName = m2mUserPool.addDomain("M2mDomainName", {
      cognitoDomain: {
        domainPrefix: `${prefix}-m2m`
      }
    })

    // Create Custom Scope
    const readScope = new cognito.ResourceServerScope({
      scopeName: "read",
      scopeDescription: "Allow Read access"
    })

    // Add Resource Server
    const resourceServer = m2mUserPool.addResourceServer("ResourceServer", {
      userPoolResourceServerName: `${prefix}-cognito-resource-server`,
      identifier: prefix,
      scopes: [readScope]
    })
    const readScopeName = `${resourceServer.userPoolResourceServerId}/${readScope.scopeName}`
    const oAuthScopes: cdk.aws_cognito.OAuthScope[] = [{ scopeName: readScopeName }]

    // Add Client setting
    const m2mClient = m2mUserPool.addClient("M2mClient", {
      userPoolClientName: `${prefix}-cognito-client-m2m`,
      idTokenValidity: cdk.Duration.days(1),
      accessTokenValidity: cdk.Duration.days(1),
      authFlows: {
        userPassword: false,
        userSrp: false,
        custom: false
      },
      oAuth: {
        flows: {
          authorizationCodeGrant: false,
          implicitCodeGrant: false,
          clientCredentials: true
        },
        scopes: oAuthScopes
      },
      preventUserExistenceErrors: true,
      generateSecret: true,
    })

• 余談ですが、Advanced security機能はユーザープールのユーザアカウントとパスワードのみを監視していることから、M2M認可の保護には対応していないようです。

Client credentials grants are intended for machine-to-machine (M2M) authorization with no connection to user accounts. Advanced security features only monitor user accounts and passwords in your user pool. To implement security features with your M2M activity, consider the capabilities of AWS WAF for monitoring request rates and content.

(引用元: Adding advanced security to a user pool)

Amazon API Gateway・AWS Lambda構築

• Amazon API Gateway(REST API)、及びAWS Lambda関数に関する資源を構築します。
• Authorizerを設定し、ユーザープールからアクセストークンを取得済みのシステムのみがAmazon API Gateway背後のAWS Lambda関数にリクエストできるようにします。

    // ------------ Amazon API Gateway ---------------
    // ---- REST API
    // Create LogGroup
    const apiGatewayLogGroup = new logs.LogGroup(this, 'ApiGatewayLogGroup', {
      retention: logs.RetentionDays.ONE_MONTH,
    })
    
    // Create REST API
    const restApi = new apigateway.RestApi(this, "RestApi", {
      restApiName: `${prefix}-apigw-rest-backend`,
      description: "Backend REST API",
      deployOptions: {
        accessLogDestination: new apigateway.LogGroupLogDestination(apiGatewayLogGroup),
        accessLogFormat: apigateway.AccessLogFormat.jsonWithStandardFields(),
        loggingLevel: props.logLevel ?? apigateway.MethodLoggingLevel.INFO,
        tracingEnabled: true,
        metricsEnabled: true,
      }
    })

    // Create Authorizer
    const cognitoAuthorizer = new apigateway.CognitoUserPoolsAuthorizer(this, "CognitoAuthorizer", {
      cognitoUserPools: [m2mUserPool]
    })

    // ------------ AWS Lambda ---------------
    // Create Lambda Function
    const itemLambda = new lambda.Function(this, "ItemLambda", {
      functionName: `${prefix}-lambda-backend-item`,
      runtime: lambda.Runtime.PYTHON_3_12,
      code: lambda.Code.fromAsset('lambda/python/backend'),
      handler: 'item.lambda_handler',
      memorySize: 256,
      timeout: cdk.Duration.seconds(25),
      tracing: lambda.Tracing.ACTIVE,
      insightsVersion: lambda.LambdaInsightsVersion.VERSION_1_0_98_0,
    })

    // Create Lambda Integration
    const itemResource = restApi.root.addResource("item");
    const itemIntegration = new apigateway.LambdaIntegration(itemLambda)
    itemResource.addMethod('GET', itemIntegration, {
      authorizationType: apigateway.AuthorizationType.COGNITO,
      authorizer: cognitoAuthorizer,
      authorizationScopes: [readScopeName]
    })

• 今回はCognito Authorizerの挙動を確認できれば良かったので、バックエンドのLambda関数は簡易な実装としています。

import json


def lambda_handler(event, context):
    response = get_items()
    return {
        "statusCode": 200,
        "body": json.dumps(response["Items"])
    }


def get_items():
    response = {
        "Items": [
            {"test": "hoge"},
        ]}
    return response

動作確認

• せっかくなので、Pythonアプリケーションを用いて、アクセストークンの取得、及びAPI呼び出しを試みます。

import os
import requests
from requests.auth import HTTPBasicAuth
import traceback

def get_access_token(client_id, client_secret, token_url, grant_type):
    # Grant Typeを設定
    data = {
        'grant_type': grant_type
    }

    try:
        # トークンリクエストを送信
        response = requests.post(token_url, data=data,
                                 auth=HTTPBasicAuth(client_id, client_secret))

        # レスポンスからアクセストークンを取得
        if response.status_code == 200:
            token_info = response.json()
            access_token = token_info.get('access_token')
            return access_token
        else:
            print(f"Failed to obtain token: {response.status_code}")
            print(response.json())
            raise Exception
    except:
        traceback.print_exc()

def make_api_request(url, access_token):
    try:
        headers = {'Authorization': f'Bearer {access_token}'}
        response = requests.get(url, headers=headers)

        # レスポンスのステータスコードをチェック
        response.raise_for_status()
        return response.json()
    except:
        traceback.print_exc()

if __name__ == '__main__':
    # クライアントIDとクライアントシークレットを環境変数から取得
    client_id = os.environ["CLIENT_ID"]
    client_secret = os.environ["CLIENT_SECRET"]

    # トークンエンドポイントURLを環境変数から取得
    token_url = os.environ["TOKEN_ENDPOINT_URL"]

    # アクセストークンを取得
    access_token = get_access_token(
        client_id, client_secret, token_url, grant_type="client_credentials")

    # API呼び出しURLを環境変数から取得
    api_url = os.environ["API_URL"]

    # API呼び出し
    response = make_api_request(api_url, access_token)
    print(response)

• 上記コードを実行したところ、無事にバックエンドのAWS Lambda関数からレスポンスが返却されました！

$ python client.py
[{'test': 'hoge'}]

注意事項

• 本記事は万全を期して作成していますが、お気づきの点がありましたら、ご連絡よろしくお願いします。
• なお、本記事の内容を利用した影響について、筆者は一切の責任を負いませんので、予めご了承ください。