Zenn
🔐

[Python] ChannelAccessToken v2.1 で LINE BOTを作る

2024/01/28に公開
LINE
Python
LINE Bot
tech

LINE Messasging API を活用して LINE Bot を作るとき、ネットでドキュメントや作例をみると「チャネルアクセストークン（長期）」と「チャネルシークレット」を使って構築する例が多くあります。
私も今までその構成で BOT を開発していたのですが、今は 「ChannelAccessToken v2.1」なるものがあり、現在の推奨として公式ドキュメントに案内されています。

ただ、公式ドキュメントや公式SDKの情報だけだと初学者にはハマりどころや情報不足感も否めないため、備忘録として私なりの管理方法を記載します。

前提

  • LINE Developers Console でチャネルを開設済み
  • Python + FastAPI

用語と用途

いろいろと登場人物が多く、混乱しやすいのでかいつまんで説明します。

チャネルアクセストークン

LINE の API にアクセスするためのトークン。
４種類あり、ネットで見かけるのは管理画面上で取得可能な「チャネルアクセストークン(長期)」というもの。

お手軽だが、トークン管理の方法が脆弱だと不正利用される恐れがあるため、できればセキュアに運用したい場合に他の方法を検討します。
今回はChannel Access Token v2.1 という、トークンの有効期間を自分で設定できるトークンを発行します。

アサーションキーペア

「公開鍵」と「秘密鍵」のペアのことを指します。
このキーペアは漏洩などしない限りは長期で管理していくもので、特に「秘密鍵」は環境変数経由で読み込ませたり、暗号化してデータベースに保存するなどの対応が必要です。

アサーション署名キー

JWTを生成するための「公開鍵」側のほうです。 LINE Developers コンソールで登録するものです。

kid

アサーション署名キーのIDです。JWT を生成する際に使います。この値も保存しておく必要がありますが、セキュアに保存する必要はありません。

JWT

チャネルアクセストークン v2.1 を発行する際、LINEに送信するトークンです。

生成するためには、以下情報が必要です。

  • チャネルID
  • kid
  • アサーションキーペアの「秘密鍵」

チャネルシークレット

チャネルアクセストークンを発行する際には必要ありませんが、Webhookでユーザーからのメッセージを受け取る際、正規サーバーからのリクエストか確認するために用います。
そのため、この値もアサーションキーペアと同じく、環境変数もしくはデータベースなどに保存しておき、呼び出せるようにしておく必要があります。

今回ここを扱う項目（アクセス元の検証）については詳しく取り上げません

Channel Access Token v2.1を発行する流れ

アサーションキーペアの生成

まず、アサーションキーペアと呼ばれる公開鍵、秘密鍵を生成します。
この手順では省略していますが、秘密鍵は上述の通り安全に管理する必要があります。

import jwk
import json

def generate_assertion_keypair() -> tuple(dict[str, str], dict[str, str]):
	key = jwk.JWK.generate(kty="RSA", size=2048, use="sig", alg="RS256")

	private_key = key.export_private()
	public_key = key.export_public()

	return private_key, public_key

生成した public_key は dict 型のため、以下のように JSON 形式にしてください

import json

keypair = generate_assertion_keypair()
print(json.dumps(keypair[0],  indent=2))

出力されたJSONをアサーション署名キーの項目から登録します。
Alt text

登録後、表示された kid をコピーしておきます。また、JWTを生成する際に必要になるためデータベースなどに保存しておく必要があります。(1つのチャネルしか管理しない場合はベタ書きでも問題ないかとは思います)

JWT の生成

LINE Developers Consoleで取得できる「チャネルID」、「kid」と、上記で生成した「秘密鍵」の3つを用いて JWT を生成します。
この JWT は、ChannelAccessToken v2.1 を生成するために必要なものです。

import json
import jwt
from jwt.algorithms import RSAAlgorithm
from datetime import datetime

def generate_jwt(
		channel_id: str, kid: str, private_key: dict[str, str]
) -> str:

	# 30 minutes(JWT TTL)
	exp = int(datetime.now().timestamp() + 30 * 60)

	# 30 days(Channel Acess Token TTL)
	token_exp = 30 * 24 * 60 * 60

	header = {
		"alg": "RS256",
		"typ": "JWT",
		"kid": kid,
	}
	payload = {
		"iss": channel_id,
		"sub": channel_id,
		"aud": "https://api.line.me/",
		"exp": exp,
		"token_exp": token_exp,
	}

	private_key = assertion_key_pair.decrypt_private_key()
	key = RSAAlgorithm.from_jwk(json.dumps(private_key))

	return jwt.encode(
		payload, key=key, algorithm="RS256", headers=header, json_encoder=None
	)
  • JWT を使い捨てで運用する場合、 exp はなるべく短くすることをおすすめします。今回の例では使い捨てを想定したコードにしていますが、使い捨てせず、発行したJWTも一時保存するケースがあまり思い浮かびませんでした。
  • Channel Access Tokenの TTL token_exp は、最大 30日です。要件に応じて変更してください。また、expと異なり、期限切れになるまでの秒数を指定します。

ここまでで前準備は完了です。

ChannelAccessToken v2.1 の発行

以降は、SDKを通じてBOTからメッセージを送信する際に呼び出していく、「ChannelAccesToken v2.1」の発行に関するコードの例を見ていきます。

from abc import abstractmethod, ABC
from dataclasses import dataclass

@dataclass(frozen=True)
class ChannelAccessToken:
	access_token: str
	expires_in: int
	token_type: str
	key_id: str

	def to_dict(self):
		return {
			"access_token": self.access_token,
			"expires_in": self.expires_in,
			"token_type": self.token_type,
			"key_id": self.key_id,
		}


class ChannelAccessTokenIssuer:
	def __init__(self, channel_id: str, kid, str, private_key[str, str]):
		self.channel_id = channel_id
		self.kid = kid
		self.private_key = private_key

	def get(self) -> ChannelAccessToken:
		# 1: Redis やデータベースから、有効期限内のチャネルアクセストークンを取得する
		channel_access_token = SomeRepositoryClass.get(self.channel_id) 

		if channel_access_token is not None:
			return channel_access_token

		# 2: まだチャネルアクセストークンを発行していない場合は LINE SDK を用いて発行する
		jwt = _get_jwt(self.channel_id, self.kid, self.private_key)
		SomeRepositoryClass.save(
			self.channel_id, channel_access_token,
		)

		return channel_access_token

	def _get_jwt(self, channel_id: str, kid: str, private_key: dict[str, str]) -> str:
		return generate_jwt(
			channel_id=channel_id,
			kid=kid,
			assertion_key_pair=private_key,
		)

	def _issue(jwt: str) -> ChannelAccessToken:
		line_api = LineChannelAccessToken()
		response = line_api.issue_channel_token_by_jwt(
			grant_type="client_credentials",
			client_assertion_type="urn:ietf:params:oauth:client-assertion-type:jwt-bearer",
			client_assertion=jwt,
		)

		return ChannelAccessToken(
			access_token=response.access_token,
			expires_in=response.expires_in,
			token_type=response.token_type,
			key_id=response.key_id,
		)

最初にどこかのリポジトリ層からチャネルアクセストークンがないか確認し、あればそれをそのまま返却するようにしています。
私の場合、Redis に保存しており、Redis側で有効期限を設定しているので、期限切れのトークンは自動で None が返される仕組みにしていますが、今回の主題からは外れるため割愛します。

発行したChannelAccessTokenを用いてメッセージを送信する

from linebot.v3 import WebhookParser
from linebot.v3.exceptions import InvalidSignatureError
from linebot.v3.messaging import (
	Configuration,
	ApiClient,
	MessagingApi,
	TextMessage,
	ReplyMessageRequest,
)
from linebot.v3.webhooks import MessageEvent, TextMessageContent

async def get_body(request: Request):
	return await request.body()


@router.post(
	"/webhook",
	status_code=200,
	tags=["Line"],
)
def process_webhook(
	x_line_signature: str = Header(None, alias="X-Line-Signature"),
	body: bytes = Depends(get_body),
):
	"""An endpoint to process LINE webhook."""
	body_text = body.decode("utf-8")

	channel_id = "..." # 何かしらでチャネルIDを取得する
	kid = "..." # 何かしらで kid を取得する
	channel_secret = "..." # 何かしらでチャネルシークレットを取得する
	private_key = {} # 何かしらで秘密鍵を取得する

	# チャネルアクセストークンを発行または取得する
	channel_access_token_issuer = ChannelAccessTokenIssuer(
		channel_id=channel_id,
		kid=kid,
		private_key=private_key
	)
	channel_assess_token = channel_access_token_issuer.get()

	# チャネルアクセストークンを用いてMessaging API のインスタンス化
	configuration = Configuration(access_token=channel_access_token.access_token)
	api_client = ApiClient(configuration)
	line_bot_api = MessagingApi(api_client)

	parser = WebhookParser(channel_secret)

	try:
		events = parser.parse(body_text, x_line_signature)
	except InvalidSignatureError:
		raise HTTPException(status_code=400, detail="Invalid signature")

	# 受け取ったメッセージの処理
	for event in events:
		logger.info(event)
		if not isinstance(event, MessageEvent):
			continue
		if not isinstance(event.message, TextMessageContent):
			continue

		line_bot_api.reply_message(
			ReplyMessageRequest(
				reply_token=event.reply_token,
				messages=[TextMessage(text="こんにちは！")],
			),
		)

	return "ok"
  • 公式ドキュメントをはじめとした多くのサンプルコードでは非同期(async)ルートで request.body を取得し、X-Line_Signature と組み合わせて検証する例を記載していますが、今回は事故が少ない同期ルートでrequest.bodyを取得しています。
  • チャネルアクセストークンを取得・発行してしまえば、 ConfigurationクラスにセットしてApiClient、MessagingApiのインスタンス化ができるようになります。ここまでできてしまえばあとは他のドキュメントに示しているようなサンプルコードは動かせるはずです。

ハマったポイント

JWT生成時の aud を正確に記述していなかった

そこまでドツボにハマったわけでもないですが、JWT を生成する際は aud というキーに https://api.line.me/を指定する必要があります。
この時 Trailing slash やプロトコル違いは考慮されていないので正確に入力する必要がありました。
私の場合、 https://api.line.me とスラッシュ無しで入力していたために、トークンの生成に失敗しました。

SDK のバージョンが古い場合、LINE SDK を用いた ChannelAccessToken が発行できない

対応しているプロジェクトで使っていた SDK バージョンは 3.1.0 だったのですが、3.5以上でないと上記コードは動きません。
理由は ChanelAccessToken v2.1 を発行する際、この時点ではアクセストークンは持っていないはずですが、発行時に「チャネルアクセストークンがない」というエラーが表示されます。

エラーの内容と、SDK側のコードを読んでみたところ、SDK側では内部的にアクセストークンがないと LINE 側の API を呼び出せない仕様となっており、矛盾が生じていました。
念の為 JavaScript 版のコードも読んでみたのですが、ChannelAccessToken を発行するためのクラス、APIに関してはこれを必要としないため、問題なく発行できる仕様でした。

私のおっちょこちょいで、これに関してはバージョンアップで解決すると知らず、公式のリポジトリで問い合わせました。

https://github.com/line/line-bot-sdk-python/issues/587

利用しているライブラリの挙動に違和感やバグを見つけた際、まずは最新バージョンで解決するか確認する、という基本を怠っていてので、改め直す良い機会でした。

参考

チャネルアクセストークンv2.1を発行する

GoでLINE Messaging APIのチャネルアクセストークンv2.1を取得する

GitHubで編集を提案

Discussion