LINEログイン PKCE対応メモ

LINEログインをPKCE対応を試してみたのでメモしておく。
ACG +PKCEでアクセストークンを取得→ユーザー情報取得まで試した。

PKCEとは

Proof Key Code Exchange の略称。
認可コード横取り攻撃への対策を目的として定義されているOAuth2.0拡張仕様。
- 認可コード横取り攻撃：悪意のあるアプリが何らかの方法で認可コードを含むカスタムURIを取得し、ユーザー固有のアクセストークンを横取りする。
通常のLINEログインとの変更点(クライアント側)

認可パラメータにcode_challenge_methodとcode_challengeを含める。
1. code_verifierを生成する。
2. code_verifierを元にcode_challenge_method(S256)を利用してcode_challengeを生成する。
トークンリクエスト時にリクエストにcode_verifierを含める。

作成するAPI

認可リクエスト作成API

LINEログイン認可エンドポイントへアクセスするためのURLをパラメータをつけて生成・返却する。

state,code_verifierをAPI呼び出し時に生成し、レスポンスとして返却する。
client_idなど固定の属性値は環境変数から取得する。
リクエスト例

POST /auth/authz_request/create HTTP/1.1
Host: localhost:8000

レスポンス例

{
    "authz_request_url": "https://access.line.me/dialog/oauth/weblogin?client_id=YOUR_CHANNEL_ID&redirect_uri=http%3A%2F%2Flocalhost%3A8000%2Fauth%2Fcallback&state=AAABBBCCC&response_type=code&code_challenge=XXXYYYZZZ&code_challenge_method=S256",
    "state": "AAABBBCCC",
    "code_verifier": "ZZZYYYXXX",
}

トークンリクエスト+ユーザー情報取得API

認可レスポンスとしてLINEからのリダイレクトで受け取った認可コード(code)と前述の認可リクエスト作成APIレスポンスのコードベリフィア(code_verifier)を指定し、トークンリクエスト＋ユーザー情報取得を行う。
- 今回は簡略化のため認可リクエストパラメータのDB保存まで実装しておらず、検証処理は行っていないが、stateパラメータの検証は必ず行うこと。
- リクエスト例
```
POST /auth/authz_request/complete HTTP/1.1
Host: localhost:8000
Content-Type: application/json
Content-Length: 128

{
    "code_verifier":"fugafuga",
    "code":"hogehoge"
}
```
- レスポンス例
```
{
    "userId": "hogefuga",
    "displayName": "hoge",
    "pictureUrl": "https://profile.line-scdn.net/fugahoge"
}
```

実装

root    -   docker-compose.yml
        -   line.env
        -   be                  - Dockerfile
                                - requirements.txt
                                - api   -   main.py

docker-compose.yml

  version: "3"

  services:
    api:
      container_name: "api"
      build: ./be
      env_file: line.env
      ports:
        - "8000:8000"
      volumes:
        - ./be/api:/usr/src/server

line.env
- クライアント＋LINEエンドポイント情報

  LINE_CLIENT_ID=YOUR_CHANNEL_ID
  LINE_CLIENT_SECRET=YOUR_CHANNEL_SECRET
  LINE_REDIRECT_URI=YOUR_CALLBACK_URL
  LINE_AUTHZ_ENDPOINT=https://access.line.me/dialog/oauth/weblogin
  LINE_TOKEN_ENDPOINT=https://api.line.me/v2/oauth/accessToken
  LINE_USER_INFO_ENDPOINT=https://api.line.me/v2/profile

Dockerfile

  FROM python:3.8

  WORKDIR /usr/src/server
  ADD requirements.txt .
  RUN pip install -r requirements.txt

  # uvicornのオプションに--reloadを付与すると、
  # main.pyを編集と同時に変更内容が反映される。
  CMD ["uvicorn", "main:app", "--reload", "--host", "0.0.0.0", "--port", "8000"]

requirements.txt
- Python依存ライブラリ

  uvicorn==0.11.5
  uvloop==0.16.0
  fastapi==0.68.1

main.py
- FastAPIで上記2APIを用意。

  from fastapi import FastAPI
  import hashlib
  import urllib.parse as parse
  import urllib.request as req
  import urllib.error as error
  import json
  import os
  import base64
  from pydantic import BaseModel

  # Client Param
  client_id = os.getenv('LINE_CLIENT_ID')
  client_secret = os.getenv('LINE_CLIENT_SECRET')
  redirect_uri = os.getenv('LINE_REDIRECT_URI')

  # LINE Endpoint
  authz_endpoint = os.getenv('LINE_AUTHZ_ENDPOINT')
  token_endpoint = os.getenv('LINE_TOKEN_ENDPOINT')
  user_info_endpoint = os.getenv('LINE_USER_INFO_ENDPOINT')

  app = FastAPI()

  # Create Authorization Request Endpoint
  @app.post("/auth/authz_request/create")
  def create():
      # Generate state
      state = hashlib.sha256(os.urandom(32)).hexdigest()

      # https://developers.line.biz/ja/docs/line-login/integrate-pkce/#how-to-integrate-pkce
      # Generate code_verifier
      code_verifier = hashlib.sha256(os.urandom(43)).hexdigest()
      # Generate code_challenge
      code_challenge = hashlib.sha256(code_verifier.encode()).hexdigest()
      code_challenge = base64.b64encode(code_challenge.encode()).decode().replace('+', '-').replace('\/', '-').replace('=','')

      # Create Authz Request URL
      authz_request_url = authz_endpoint+'?{}'.format(parse.urlencode({
          'client_id': client_id,
          'redirect_uri': redirect_uri,
          'state': state,
          'response_type': 'code',
          'code_challenge':code_challenge,
          'code_challenge_method':'S256'
      }))
      res_body = {
          "authz_request_url": authz_request_url,
          "state": state,
          "code_verifier":code_verifier
      }
      return json.loads(json.dumps(res_body))


  # Token Request And Get User Info Endpoint
  # Request Body
  class ReqBody(BaseModel):
      code: str
      code_verifier:str

  @app.post("/auth/authz_request/complete")
  def complete(reqBody: ReqBody):
      # Parse Req Body
      code = reqBody.code
      code_verifier = reqBody.code_verifier

      # Token Request
      # https://developers.line.biz/ja/docs/line-login/integrate-line-login-v2/#get-access-token
      token_req_body = parse.urlencode({
          'code': code,
          'client_id': client_id,
          'client_secret': client_secret,
          'redirect_uri': redirect_uri,
          'grant_type': 'authorization_code',
          'code_verifier':code_verifier
      }).encode('utf-8')
      token_req = req.Request(token_endpoint)
      err_str = ''
      try:
          with req.urlopen(token_req, data=token_req_body) as token_res:
              token_res_body = token_res.read()
      except error.HTTPError as err:
          err_str = str(err.code) + ':' + err.reason + ':' + str(err.read())
          pprint(err_str)
      except error.URLError as err:
          err_str = err.reason
          pprint(err_str)
      access_token = json.loads(token_res_body)['access_token']

      # Get User Profile
      # https://developers.line.biz/ja/docs/line-login/managing-users/#get-profile
      headers = {
          'Authorization': 'Bearer ' + access_token
      }
      user_info_req = req.Request(
          user_info_endpoint, headers=headers, method='GET')
      try:
          with req.urlopen(user_info_req) as user_info_res:
              user_info_res_body = user_info_res.read()
      except error.HTTPError as err:
          err_str = str(err.code) + ':' + err.reason + ':' + str(err.read())
          pprint(err_str)
      except error.URLError as err:
          err_str = err.reason
          pprint(err_str)
      return json.loads(user_info_res_body)

起動

docker-compose up -d

参考情報

LINEログインをPKCE対応する