LINEログイン PKCE対応メモ

# LINEログイン PKCE対応メモ

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



## PKCEとは

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

* 通常のLINEログインとの変更点(クライアント側)

  1. 認可パラメータに`code_challenge_method`と`code_challenge`を含める。
     1. `code_verifier`を生成する。
     2. `code_verifier`を元に`code_challenge_method`(S256)を利用して`code_challenge`を生成する。

  2. トークンリクエスト時にリクエストに`code_verifier`を含める。



## 作成するAPI

### 認可リクエスト作成API

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

  * `state`,`code_verifier`をAPI呼び出し時に生成し、レスポンスとして返却する。

  * `client_id`など固定の属性値は環境変数から取得する。

  * リクエスト例

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

  * レスポンス例

    ```json
    {
        "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パラメータの検証は必ず行うこと。**

  * リクエスト例

    ```http
    POST /auth/authz_request/complete HTTP/1.1
    Host: localhost:8000
    Content-Type: application/json
    Content-Length: 128
    
    {
        "code_verifier":"fugafuga",
        "code":"hogehoge"
    }
    ```

  * レスポンス例

    ```json
    {
        "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`

  ```yaml
  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`

  ```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を用意。

  ```python
  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)
  
  ```

  

## 起動

```shell
docker-compose up -d
```





## 参考情報

* [ LINEログインをPKCE対応する](https://developers.line.biz/ja/docs/line-login/integrate-pkce/#how-to-integrate-pkce)