前回、Rails + Doorkeeperを使ってOAuth2.0の認可サーバを、Rails + OmniAuth を使ってクライアントを作ってみました。
Rails + Devise + OmniAuthにて、Deviseで認証を、OmniAuthでOAuth2.0の認可コードグラントフローの認可だけを扱えるか試してみた - メモ的な思考的な

ただ、せっかくなので、別の言語でOAuth2.0のクライアントを作ってみたくなりました。

　
PythonでOAuth2.0を扱えるパッケージがないかを探したところ、 requests と組み合わせて使える requests-oauthlib がありました。
requests/requests-oauthlib: OAuthlib support for Python-Requests!

そこで、Python + requests-oauthlib を使い、「前回作成した認可サーバ兼リソースサーバからデータを取得する」Python スクリプトを作成してみました。

環境
認可サーバに、今回使うOAuth Applicationを登録する
実装
動作確認
ソースコード

環境

OAuth2.0 クライアント
- Python 3.10.4
- requests 2.27.1
- requests-oauthlib 1.3.1
- python-dotenv 0.20.0
  - 認可サーバで管理しているクライアントIDとクライアントシークレットをハードコーディングしないようにするため
認可サーバ兼リソースサーバ
- 前回の記事のサーバを流用
  - rails 7.0.2.3
  - devise 4.8.1
  - doorkeeper 5.5.4

認可サーバに、今回使うOAuth Applicationを登録する

前回作成した認可サーバを起動し、以下のURLにアクセスします。
http://localhost:3801/oauth/applications/

以下の内容でOAuth Applicationを登録し、クライアントIDとクライアントシークレットを取得します。

項目	値
Name	任意(今回は `oauthlib_app`)
Redirect URI	`urn:ietf:wg:oauth:2.0:oob`
Confidential	チェックを入れる
Scopes	`read`

　
なお、Redirect URIについてですが、前回のWebアプリとは異なり、今回のPython スクリプトではコールバックURLを持っていません。

どうすればよいかを調べたところ、RFC8252 (OAuth 2.0 for Native Apps) の 7.1. Private-Use URI Scheme Redirection を使えば良さそうに見えました。

一方、DoorkeeperのWikiによると、デフォルトでは urn:ietf:wg:oauth:2.0:oob の値を入れることで、ブラウザ上に認可コードを表示させることができるようです。
Authorization Code Flow · doorkeeper-gem/doorkeeper Wiki

そこで、今回はDoorkeeperの仕様に従い、 urn:ietf:wg:oauth:2.0:oob を Redirect URI に設定しています。

以上で、認可サーバ側の設定は完了です。

実装

認可コードグラントフローを使ってアクセストークンを取得する

requests-oauthlibのドキュメントに従い、実装してみます。
https://requests-oauthlib.readthedocs.io/en/latest/oauth2_workflow.html

なお、ドキュメントの該当ページはPython2な書き方に見えるので、今回はPython3な書き方で実装します。

requests-oauthlibでHTTP通信を許可する

OAuth2.0はHTTPS(TLS/SSL)通信を前提にしているため、今回のlocalhostにある認可サーバとの間でHTTP通信をすると、 requests-oauthlib でエラーになってしまいます。

そのため、環境変数 OAUTHLIB_INSECURE_TRANSPORT に '1' を設定しておきます。

os.environ['OAUTHLIB_INSECURE_TRANSPORT'] = '1'

認可コードの取得

認可コードは

requests-oauthlib の OAuth2Session インスタンスを生成する
生成したインスタンスの authorization_url() を使い、認可エンドポイントのURLを取得する
取得したURLをブラウザに入力し、認可サーバにログインすることで、認可コードを取得する

という流れで取得します。

　
そこで、まずは OAuth2Sessionのインスタンスを生成します。

生成するときには

client_id
scope
redirect_uri

を指定します。

import os
from dotenv import load_dotenv
from requests_oauthlib import OAuth2Session

load_dotenv()
CLIENT_ID = os.environ.get('CLIENT_ID')
REDIRECT_URI = 'urn:ietf:wg:oauth:2.0:oob'
REQUEST_SCOPE = 'read'

session = OAuth2Session(
    client_id=CLIENT_ID,
    scope=REQUEST_SCOPE,
    redirect_uri=REDIRECT_URI
)

　
続いて authorization_url() メソッドを使い、認可エンドポイントのURLを取得します。

AUTHORIZATION_SERVER_BASE_URL = 'http://localhost:3801'
AUTHORIZE_URL = f'{AUTHORIZATION_SERVER_BASE_URL}/oauth/authorize'

authorization_url, state = session.authorization_url(AUTHORIZE_URL)

print('URLをブラウザにコピペし、認可コードを取得してください:', authorization_url)
code = input('表示されている認可コードをコピペしてください: ')

　
ここまでを実行すると、

URLをブラウザにコピペし、認可コードを取得してください: http://localhost:3801/oauth/authorize?response_type=code&client_id=***&redirect_uri=urn%3Aietf%3Awg%3Aoauth%3A2.0%3Aoob&scope=read&state=***
表示されている認可コードをコピペしてください:

と認可エンドポイントのURLが表示されるとともに、認可コードの入力を待ち受けます。

　
認可エンドポイントのURLをブラウザにペーストして認可サーバにログイン・アプリケーションに対する認可処理を実行すると、認可コードがブラウザ上に表示されます。

トークン類の取得

アクセストークンを始めとするトークン類は、 fetch_token() メソッドで取得します。

CLIENT_SECRET = os.environ.get('CLIENT_SECRET')
TOKEN_URL = f'{AUTHORIZATION_SERVER_BASE_URL}/oauth/token'

token = session.fetch_token(
    TOKEN_URL,
    client_secret=CLIENT_SECRET,
    code=code,
)

　
なお、今回の fetch_token() メソッドの今回の戻り値 token の中身には

{
    "access_token": "***",
    "token_type": "Bearer",
    "expires_in": 60,
    "refresh_token": "***",
    "scope": [
        "read"
    ],
    "created_at": 1648867286,
    "expires_at": 1648867346.5057201
}

のように、アクセストークンやリフレッシュトークンが含まれています。

アクセストークンを使い、リソースサーバのAPIからデータを取得

アクセストークンが取得できたため、次はリソースサーバのAPIからデータを取得します。

また、 requests-oauthlib ではアクセストークンの有効期限が切れた場合は自動で更新する機能が備わっているため、今回はそれも使います。
https://requests-oauthlib.readthedocs.io/en/latest/oauth2_workflow.html#third-recommended-define-automatic-token-refresh-and-update

自動で更新するには OAuth2Session のインスタンスを生成する際の引数として

auto_refresh_url
- トークンエンドポイントのURLを指定
auto_refresh_kwargs
- dictとして、クライアントIDとクライアントシークレットを指定
token_updater
- アクセストークンを更新したときに実行されるコールバック関数を指定
  - コールバック関数では、更新後のアクセストークンを保管することを想定

を渡します。

def save_token(token):
  with open(OAUTH_TOKEN_FILE_PATH, 'wb') as f:
    pickle.dump(token, f, protocol=5)
    print('===== tokenを保存しました =====')
      
session = OAuth2Session(
    CLIENT_ID,
    redirect_uri=REDIRECT_URI,
    token=token,
    scope=REQUEST_SCOPE,
    # アクセストークンが有効期限切れの場合、リフレッシュトークンを使って自動更新する
    auto_refresh_url=TOKEN_URL,
    auto_refresh_kwargs={
        'client_id': CLIENT_ID,
        'client_secret': CLIENT_SECRET,
    },
    token_updater=save_token,
)

response = session.get('http://localhost:3801/api/memos/').json()

ソースコード全体

以上で個別の実装が終わりましたので、それらを組み合わせてみます。

import os
import pathlib
import pickle

from dotenv import load_dotenv
from requests_oauthlib import OAuth2Session

load_dotenv()

CLIENT_ID = os.environ.get('CLIENT_ID')
CLIENT_SECRET = os.environ.get('CLIENT_SECRET')

REDIRECT_URI = 'urn:ietf:wg:oauth:2.0:oob'
REQUEST_SCOPE = 'read'

AUTHORIZATION_SERVER_BASE_URL = 'http://localhost:3801'
AUTHORIZE_URL = f'{AUTHORIZATION_SERVER_BASE_URL}/oauth/authorize'
TOKEN_URL = f'{AUTHORIZATION_SERVER_BASE_URL}/oauth/token'

BASE_DIR = pathlib.Path(__file__).resolve().parent
OAUTH_TOKEN_FILE_PATH = f'{BASE_DIR}/oauth_token.pickle'

# 本来、OAuth2.0はhttps通信が必要
# ただ、今回はlocalhostのASなため、http通信可能にする設定を行っておく
# https://requests-oauthlib.readthedocs.io/en/latest/examples/real_world_example.html
os.environ['OAUTHLIB_INSECURE_TRANSPORT'] = '1'


def save_token(token):
    with open(OAUTH_TOKEN_FILE_PATH, 'wb') as f:
        pickle.dump(token, f, protocol=5)
        print('===== tokenを保存しました =====')


def fetch_token():
    # すでにトークン類を取得済の場合は、そのトークン類を使う
    if pathlib.Path(OAUTH_TOKEN_FILE_PATH).is_file():
        with open(OAUTH_TOKEN_FILE_PATH, 'rb') as f:
            return pickle.load(f)

    # トークン類がない場合は、認可コードグラントフローにより取得
    session = OAuth2Session(
        client_id=CLIENT_ID,
        scope=REQUEST_SCOPE,
        redirect_uri=REDIRECT_URI
    )

    authorization_url, state = session.authorization_url(AUTHORIZE_URL)

    print('URLをブラウザにコピペし、認可コードを取得してください:', authorization_url)
    code = input('表示されている認可コードをコピペしてください: ')

    token = session.fetch_token(
        TOKEN_URL,
        client_secret=CLIENT_SECRET,
        code=code,
    )

    return token


def fetch_resource_server(token):
    session = OAuth2Session(
        CLIENT_ID,
        redirect_uri=REDIRECT_URI,
        token=token,
        scope=REQUEST_SCOPE,
        # アクセストークンが有効期限切れの場合、リフレッシュトークンを使って自動更新する
        auto_refresh_url=TOKEN_URL,
        auto_refresh_kwargs={
            'client_id': CLIENT_ID,
            'client_secret': CLIENT_SECRET,
        },
        token_updater=save_token,
    )

    return session.get('http://localhost:3801/api/memos/').json()


def main():
    # トークン類の取得
    token = fetch_token()

    # もしトークン類を保存していない場合、保存しておく
    if not pathlib.Path(OAUTH_TOKEN_FILE_PATH).is_file():
        save_token(token)

    # リソースサーバよりデータを取得
    response_body = fetch_resource_server(token)
    print(response_body)


if __name__ == '__main__':
    main()

　
なお、上記のソースコードを動かすためには、 .env ファイルをPython スクリプトと同じディレクトリに保存しておきます。

中身はこんな感じで、環境に合わせて値を指定します。

CLIENT_ID=
CLIENT_SECRET=

動作確認

Python スクリプトを実行して動作を確認してみます。

初回実行時

URLをブラウザにコピペし、認可コードを取得してください: http://localhost:3801/oauth/authorize?response_type=code&***
表示されている認可コードをコピペしてください: ***
===== tokenを保存しました =====
{'message': 'ふじ'}

アクセストークンが有効な間

{'message': 'ふじ'}

トークンの有効期限が切れた時

===== tokenを保存しました =====
{'message': 'ふじ'}

ソースコード

Githubに上げました。
https://github.com/thinkAmi-sandbox/requests-oauthlib-sample