トークンで簡単なログイン保持機能をPythonで実装する方法

はじめに

• エンジニアリング歴半年
• 備忘録的な感じ
• 深夜テンション
• Python信者

やりたいこと

ブラウザでログインを行った時に、LambdaからDynamoDBでの確認ができた場合、Lambdaより固定トークンを発行しブラウザに一時保存させる。

↓

ブラウザでトークンを保持しておき、サーバ側で確認することで各機能でいちいち認証しなくても良くなる！！！！

考えた人天才か？

準備したこと

• AWSのLambdaとDynamoDB + API Gatewayを使ってサーバーレスな環境を準備
• Postmanのインストール
• DynamoDBにテストデータを用意

やってみよう（トークンの発行ロジック）

仕様

• リクエストはJSON、パラメータとしてuserId (string)、password (string)を持つ。（どっちも必須）
• レスポンスはJSON、token (string)、今回は固定で”token”という文字列を返す。
• DynamoDBでリクエストに一致するものがある場合、レスポンスを返す。
• 一致しない場合は、エラー401。

1. API GatewayにHTTP APIを作成、Lambdaを作成してアタッチ

なぜGETじゃなくてPOSTなのか？

• GETメゾットを利用するとURLクエリパラメータとして表示されるためログがブラウザなどに残る可能性がある。POSTならリクエストボディだから大丈夫って訳なのよ。初めて知った…

AWSの操作は割愛

2. Lambdaの編集

import json
import boto3

dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table('User')

def lambda_handler(event, context):
    try:
        # リクエストボディの解析
        body = json.loads(event.get('body', '{}'))
        userId = body.get('userId')
        password = body.get('password')

        # userId または password が存在しない場合のエラーハンドリング
        if not userId or not password:
            return {
                'statusCode': 400,
                'body': json.dumps({
                    'message': 'userId または password がリクエストに含まれていません。'
                })
            }

        response = table.get_item(
            Key={
                'userId': userId
            }
        )

        if 'Item' in response:
            stored_password = response['Item']['password']
            if stored_password == password:
                return {
                    'statusCode': 200,
                    'body': json.dumps({
                        'token': 'token'
                    })
                }
            else:
                return {
                    'statusCode': 401,
                    'body': json.dumps({
                        'message': 'userIdまたはpasswordが一致しません。'
                    })
                }
        else:
            return {
                'statusCode': 401,
                'body': json.dumps({
                    'message': 'userIdまたはpasswordが一致しません。'
                })
            }
    except Exception as e:
        # 予期しないエラーのハンドリング
        return {
            'statusCode': 500,
            'body': json.dumps({
                'message': 'Internal Server Error',
                'error': str(e)
            })
        }

今回はPython3で実装、JavaScriptに比べてコードがわかりやすい（当社比）

userIdで取得→passwordが一致してるか確認っていう簡単なロジックで動いてます。

GetItemを使ってコード内で処理するほか、ScanやQueryを使ってクエリで処理するロジックを組むことも可能！！（ScanやQueryの方がお勧め）

ちなエラーハンドリングはめっちゃ重要！！！最後のやつ追加しないとそもそも何のエラー起きてるかわからん

今更ながらパラメータtokenの中身tokenにしたのバカ設計だし、パラメータかけてる時のエラーハンドリングを実装してなくて草

3. Postmanでレスポンスの確認

メゾットをPOSTにしてリクエストボディから、

raw → JSON → 内容を入力 → Send

例：

{
	"userId": "test",
	"password": "test"
}

やったぁ！！{ “token”: “token” } が返ってきましたね！！！！

やってみよう（トークン確認のロジック）

さて、ここからは作ったtokenの使い方を一緒に作ってみましょう！

仕様（バックエンドで実装する場合）

• リクエストはJSON、パラメータとしてuserId (string)を持つ。（必須）
• レスポンスはJSON、age (number)。
• headersの’authorization’で”token”を受け取る。
• DynamoDBでリクエストに一致するものがある場合、レスポンスを返す。
• ユーザがいない場合は、エラー404。
• tokenが無いと認証エラー401を返す。

1. APIの設定とアタッチ

割愛〜

2. Lambdaの編集

import json
import boto3

dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table('User')

def lambda_handler(event, context):
    try:
        # 受け取ったeventをログに出力
        print(event)
        
        # トークンの取得
        token = event.get('headers', {}).get('authorization')
        
        # トークンの認証
        if not token:
            return {
                'statusCode': 401,
                'body': json.dumps({
                    'message': '認証エラー。トークンが提供されていません。'
                })
            }
        elif token != "token":
            return {
                'statusCode': 401,
                'body': json.dumps({
                    'message': '認証エラー。トークンが無効です。'
                })
            }
        else:
            # トークンが正しい場合、リクエストボディの解析を行う
            body = json.loads(event.get('body', '{}'))
            userId = body.get('userId')

            # userId が存在しない場合のエラーハンドリング
            if not userId:
                return {
                    'statusCode': 400,
                    'body': json.dumps({
                        'message': 'userId がリクエストに含まれていません。'
                    })
                }

            response = table.get_item(
                Key={
                    'userId': userId
                }
            )

            if 'Item' in response and 'age' in response['Item']:
                return {
                    'statusCode': 200,
                    'body': json.dumps({
                        'age': response['Item']['age']
                    })
                }
            else:
                return {
                    'statusCode': 404,
                    'body': json.dumps({
                        'message': '指定された userId に一致するユーザが見つかりません。'
                    })
                }
    except Exception as e:
        # 予期しないエラーのハンドリング
        return {
            'statusCode': 500,
            'body': json.dumps({
                'message': 'Internal Server Error',
                'error': str(e)
            })
        }

仕組みは単純、ヘッダーから’Authorization’としてトークンを取得、それが有効なものか確認するという簡単なロジックで動いてます。

3. Postmanでレスポンスの確認

Headersに”Authorization” Keyを設定してValueに”Token”を入れます。

リクエストボディをお好みで入れて、

例：

{
	"userId": "test"
}

望んだリクエストが返って来れば成功です！！

お疲れ様でした！

俺のエラーポイント（備忘録）

• tokenの受け渡し時にheadersのkeyで’Authorization’を受け渡したとしても、受け渡し時に強制的に小文字になるため、コード上はget('authorization')としないと一生認証できません（俺が馬鹿すぎてつまずいたポイント）

• LambdaがDynamoDBにアクセスする際に権限エラーが出る可能性があります。権限が変更できないユーザを与えられている人は管理者権限を持つ人に付与してもらってください。

"message": "Internal Server Error"
"error": "An error occurred (AccessDeniedException) when calling the GetItem operation: [Lambda関数の識別番号と名前] is not authorized to perform: dynamodb:GetItem on resource: [DynamoDBの識別番号と名前] because no identity-based policy allows the dynamodb:GetItem action"

IAMサービスからLambda関数のロールに権限（AmazonDynamoDBFullAccess）を与えると動作します。

が、これはDynamoDBにアクセスするFull権限なので本番環境などでは権限は最小に調節することをお勧めします。

• DynamoDBからのレスポンスには、数値をDecimal型で返されるため以下のエラーが発生する可能性が考えられます。

"message": "Internal Server Error"
"error": "Object of type Decimal is not JSON serializable"

この場合、Decimal型をJSONシリアライズ可能な型（intやfloatなど）に変換します。

例：（int型にシリアライズ）

import json
import boto3
from decimal import Decimal

dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table('User')

class DecimalEncoder(json.JSONEncoder):
    def default(self, obj):
        if isinstance(obj, Decimal):
            return int(obj)  # Decimalをintに変換
        return super(DecimalEncoder, self).default(obj)

def lambda_handler(event, context):
    try:
        # ... [その他のコードは変更なし] ...

        if 'Item' in response and 'age' in response['Item']:
            return {
                'statusCode': 200,
                'body': json.dumps({
                    'age': response['Item']['age']
                }, cls=DecimalEncoder)  # ここでカスタムエンコーダを使用
            }
        else:
            return {
                'statusCode': 404,
                'body': json.dumps({
                    'message': '指定された userId に一致するユーザが見つかりません。'
                })
            }
    except Exception as e:
        # 予期しないエラーのハンドリング
        return {
            'statusCode': 500,
            'body': json.dumps({
                'message': 'Internal Server Error',
                'error': str(e)
            })
        }