手を動かしてOAuthを理解したかったので、お試しクライアントアプリを書いてみた時の備忘録。
Flaskで実装されている例が見当たらなかったのでちょうどよいと思って試してみた。
(python3.10.9)
OAuth2.0とは
OAuthとは一言で言えば、利用者に外部のサービスを使う(Gmailの新着メールを見る、Twitterのタイムラインを見る)ための許可を安全に得るための技術である。
仕様の流れは以下の通り。
登場人物はサービス利用者・サーバー・外部サービス認可サーバー・外部サービスリソースサーバー。
①サーバーは外部サービス認可サーバーにリダイレクトさせ、サービス利用者にログイン情報を入力してもらう
②ログイン情報が正しければ、外部サービス認可サーバーはサーバーにアクセストークンを発行する
③サーバーがアクセストークンを使って外部サービスリソースサーバーのリソースを使用する
似たような技術にOpenID Connectがある。
OpenID Connectはリソースの使用許可を取るというよりID管理を外部サービスに委ねるための技術である。
（GoogleアカウントでYoutubeにログインする、Facebookアカウントでゲームにログインする）
シングルサインオンとも少し似ているが、
シングルサインオンは複数のサービスを一元的に管理するための技術である。
（社内の認可サーバーを一度通ればSalesforceにもMicrosoftにもサインインする）
事前準備
今回はGoogleのGmailを使う。
Gmailの未読メッセージ一覧の取得を目指してみる。
ただし今回は勉強のためもあるので自前実装したが、GoogleのOAuthはそれ専用のラッパーがありOAuthの処理を自前で書くのは推奨されていないことに注意。
事前準備として任意のGoogleアカウントでクライアントIDと秘密鍵を発行する必要がある。
以下を参考にする。
future-architect.github.io
上の手順にも書いてあるがリダイレクトさせるURLは事前に申請する必要がある。
これがないと400番のエラーが返ってくることになるが、私はここでちょっと詰まってしまった。
今回の例でいうと http://127.0.0.1:5000/callback がそれにあたる。
実装
フォルダ構成
├── app.py
├── requirements.txt
└── templates
├── index.html
└── result.htmlrequirements.txt
Flask jinja2 requests
ソースコード
index.html
<!DOCTYPE html> <html lang="ja"> <head> <title>Oauth sample app</title> <meta charset="utf-8"/> </head> <body> <h1>{{title}}</h1> <p>{{message}}</p> <a href="/oauth"> <button> Connect to Gmail </button> </a> </body> </html>||< result.html >|html| <!DOCTYPE html> <html lang="ja"> <head> <title>Oauth sample app</title> <meta charset="utf-8"/> </head> <body> {% for message in messages %} <li> {{message}} </li> {% endfor %} </body> </html>
app.py
from flask import redirect, render_template, session from urllib.parse import urlencode from requests.auth import HTTPBasicAuth import flask, requests, os, hashlib, base64 app = flask.Flask(__name__) app.secret_key = os.urandom(32).hex() class GoogleOAuth(): client_id: str = "YOUR CLIENT ID" client_secret: str = "YOUR CLIENT SECRET" @app.route("/") def index(): session["csrf"] = os.urandom(32).hex() return render_template('index.html', title="OAuth sample", message="Connect to Gmail") @app.route("/oauth") def oauth(): if session.get("csrf") is None: return redirect("/") # Googleの認可サーバーのURL authorization_endpoint = "https://accounts.google.com/o/oauth2/v2/auth" # 利用したいリソースの権限をスペース区切りで設定する。今回は情報取得のみなのでreadonly属性のみを指定 scope = "https://www.googleapis.com/auth/gmail.readonly" # code_vefierとcode_challengeを設定。S256の仕様に従う。 code_verifier = os.urandom(43).hex() code_challenge = base64.urlsafe_b64encode(hashlib.sha256(code_verifier.encode('utf-8')).digest()).decode('utf-8').replace('=','') session["code_verifier"] = code_verifier params: dict = { "response_type": "code", "client_id": GoogleOAuth.client_id, "state": session.get("csrf"), "scope": scope, "redirect_uri": "http://127.0.0.1:5000/callback", "code_challenge": code_challenge, "code_challenge_method": "S256" } res = requests.get(authorization_endpoint, params=params) return redirect(authorization_endpoint + "?" + urlencode(params)) @app.route("/callback") def callback(): if (session.get("csrf") is None) or (session["csrf"] != state): return redirect("/") if (session.get("code_verifier") is None): return redirect("/") # "state"と"code"がクエリパラメータとしてリダイレクトされてくる state = flask.request.args.get("state") code = flask.request.args.get("code") params = { "client_id": GoogleOAuth.client_id, "client_secret": GoogleOAuth.client_secret, "redirect_uri": "http://127.0.0.1:5000/callback", "grant_type": "authorization_code", "code": code, "code_verifier": session["code_verifier"] } headers = {"content-type": "application/x-www-form-urlencoded"} response = requests.post( "https://oauth2.googleapis.com/token", params=params, headers=headers, auth=HTTPBasicAuth(GoogleOAuth.client_id, GoogleOAuth.client_secret)) data = response.json() access_token = data["access_token"] session["access_token"] = access_token return redirect("/home") @app.route("/home") def home(): if session.get("access_token") is None: return redirect("/") access_token = session.get("access_token") # GmailのAPI resource_endpoint = 'https://gmail.googleapis.com/gmail/v1/users/me/messages' params = {'maxResults': 50, 'q': 'is:unread'} headers = {'Authorization': 'Bearer {}'.format(access_token)} response = requests.get( resource_endpoint, params=params, headers=headers ) data = response.json() # 正常にレスポンスではmessagesというフィールドに値が格納 messages = data["messages"] if "messages" in data.keys() else [data] return render_template("result.html", messages=messages) if __name__=="__main__": app.debug = True app.run(host="127.0.0.1", port=5000)
ターミナルで python app.py を実行後、
ブラウザのアドレスバーに 127.0.0.1:5000 と入力する。
こんな感じの結果が返ってくると思う。
・{'id': '185d801397a28c7f', 'threadId': '185d801397a28c7f'}
・{'id': '185d7f65f403db40', 'threadId': '185d7f65f403db40'}
・{'id': '185d75989607f13e', 'threadId': '185d75989607f13e'}
・{'id': '185d6834381417f6', 'threadId': '185d6834381417f6'}
・{'id': '185d26d791a422cf', 'threadId': '185d26d791a422cf'}
・{'id': '185ee6de3d4d53ee', 'threadId': '185ee6de3d4d53ee'}
・{'id': '185ee186a7fddac1', 'threadId': '185ee186a7fddac1'}本当はメッセージの中身まで取得して表示したかったが、あくまでサンプルアプリなのでここでやめることにした。
