FirebaseのLINEカスタム認証をAWSのサーバーレスで実装してみた

Severless Frameworkを使ってAWS環境にサーバーレスで実装してみたいと思います

#Firebase

#LINE

西田将幸

2019.01.11

この記事は公開されてから1年以上経過しています。情報が古い可能性がありますので、ご注意ください。

西田@大阪です

FirebaseにはFacebookなど代表的なSNS認証がデフォルトで用意されていますが、LINEを使ってのSNS認証は用意されておらずカスタム認証サーバーを用意する必要があります。参考

今回はSeverless Frameworkを使ってAWS環境にサーバーレスで実装してみたいと思います

実装環境

Serverless Framework

Serverless Framework 1.32.0
serverless-python-requirements 4.2.4

Python

python 1.6
boto3 1.9.8
firebase-admin 2.13.0

概要

アプリでLINEのアクセストークンを取得します
1で取得したLINEのアクセストークンを今回作成するカスタム認証APIに渡します
LINEのアクセストークンを使ってトークンの検証しプロフィールを取得します
Firebase Admin SDKを使ってFirebaseのカスタムトークンを取得します
4で取得したカスタムトークンをアプリに返します
カスタムトークンを使ってFirebaseにログインします

Serverless Frameworkのプロジェクトを作成

※本記事は serverless => slsとエイリアスが作成されていることを前提としています

Serverless Frameworkの雛形を用意します

sls create --template aws-python3 --path ${path/to/project}

Pythonのライブラリの依存関係を管理するためのツールとして、今回はPipenvを利用します

Pipenvで管理してるライブラリをLambdaにデプロイするために、Serverless Frameworkのプラグインserverless-python-requirementsをインストールします

sls plugin install -n serverless-python-requirements

使用するPythonのライブラリをインストールします

pipenv install firebase-admin
pipenv install boto3

Firebaseの認証情報をKMSで暗号化

Firebaseのカスタムトークンを生成するのにFirebase Admin SDK を使います

Firebase Admin SDKを利用するには認証情報が必要です

認証情報には秘密鍵が含まれるので、そのままgit等のリポジトリにアップしてしまうのはためらわれます

今回はKMSで暗号化したファイルをリポジトリにアップしLambdaに割り当てられたIAM Roleで復号化できるようにします

サーバーに Firebase Admin SDK を追加するこちらのリンク先を参考にサービスアカウントJSONファイルを作成し開発用マシンにダウンロードします

ダウンロードしたサービスアカウントJSONファイルをプロジェクトのフォルダに移動しKMSで暗号化します

aws kms encrypt --key-id 'alias/line-login' --plaintext file://./path/to/serviceAccountKey.json --output text > path/to/serviceAccountKey.json.encrypted

serverless.yaml を編集

暗号化されたサービスアカウントJSONファイルがLambdaにデプロイするときにパッケージングされるよう設定します

package:
include:
- path/to/serviceAccountKey.json.encrypted

ルーティングの設定をします

GET line/loginというパスでtoken(LINEのアクセストークン)をQuery Stringパラメーターで受け取ります

functions:
lineLogin:
handler: handler.line_login

events:
- http:
path: line/login
method: get
request:
parameters:
querystring:
token: true

コード

Lambdaが呼び出されたときに実行されるエントリポイントの関数

主な処理として以下を行います

アプリから渡されたLINEアクセストークンの検証
Firebaseのカスタムトークンの発行

import json
import urllib.error
import urllib.parse
import urllib.request
import urllib.response

import boto3
import os
from firebase_admin import initialize_app, auth, credentials, _apps

# 1
firebase_credentials = None

def line_login(event, context):
line_access_token = query_params['token']

# 2
if not verify_line_access_token(line_access_token):
return dict(
statusCode=401,
body="invalid access token"
)

# 3
uid = get_line_user_id(line_access_token)

# 4
if len(_apps) == 0:
initialize_app(credentials.Certificate(get_firebase_credentials()))

custom_token = auth.create_custom_token('line:' + uid, dict(provider='LINE'))

# 5
return dict(
statusCode=200,
firebase_token: custom_token.decode(),
)

復号化されたサービスアカウントJSONファイルのキャッシュをハンドラー外に宣言
アプリから渡されたLINEのアクセストークンを検証
アプリから渡されたLINEのアクセストークンを使ってLINEのユーザーIDを取得
Firebase Admin SDK をサービスアカウントJSONファイルの内容を使って初期化し、Firebaseのカスタムトークンを生成。デフォルトAPPが初期化済みの場合、エラーになってしまうので、そのチェック処理が必要
クライアントに生成されたFirebaseのカスタムアクセストークンを返却

LINEのアクセストークンを検証する関数

def verify_line_access_token(access_token):
# 1
query = dict(access_token=access_token)
query_string = urllib.parse.urlencode(query).encode('utf-8')
req = urllib.request.Request(url=f'https://api.line.me/v2/oauth/verify', data=query_string)

try:
res = urllib.request.urlopen(req)
except urllib.error.HTTPError as e:
if e.code == 401:
return False
else:
# エラー処理
raise e
# 2
if res['client_id'] != 'YOUR LINE CLIENT ID':
return False

return True

v2/oauth/verify APIはHTTP HeaderでなくRequest Bodyに含める必要があります
アプリから渡されたLINEのアクセストークンが自分のチャネルのものかチェックし、なりすましを防ぐ必要があります

LINEのユーザーIDを取得する関数

def get_line_user_id(access_token):
headers = dict(access_token=access_token)
req = urllib.request.Request(url=f'https://api.line.me/v2/profile', headers=headers)
res = urllib.request.urlopen(req)

return res['userId']

FirebaseのサービスアカウントJSONファイルの情報を取得する関数

暗号化されたサービスアカウントJSONファイルを復号化し関数の呼び出し元に返しています

def get_firebase_credentials
global firebase_credentials

# 1
if not firebase_credentials:
with open('path/to/serviceAccountKey.json.encrypted', 'rb') as f:
kms = boto3.client('kms', region_name='ap-northeast-1')
cipher_blob = f.read()
plain_text = kms.decrypt(CiphertextBlob=cipher_blob)['Plaintext']

firebase_credentials = json.loads(plain_text)

return firebase_credentials