msizの日記

ソフトウェア関係の覚え書きが中心になる予定

「Python3 + Flask + はてなAPIのOAuth認可」サンプルを作成（移植？）しました

gistにあるPythonではてなの OAuth 対応 API を利用する をベースに、Python3 対応の OAuth1 サンプルにしました（記事作成時点で、はてなはOAuth2に対応していない）。

フレームワークとして Flask を、OAuth ライブラリとして requests-oauthlibを利用したサンプルプログラムです。

  1. （はてな側の準備）はてなに新しいアプリケーションを登録（consumer_keyとconsumer_secretを用意）します（やり方は「[python3][プログラミング]Python3 はてな APIのOAuth認証する(ほぼコピペでOK) - lisz-works」あたりを参照ください）
    • ここで作成した「新しいアプリケーション」は削除できるか不明なので、「my-oauth-test」みたいな感じで、いつでも無効にできるような名前にすると無難かと思います
    • 権限は read_public だけで構いません（余計な権限はつけない方が無難です）
  2. python3でvenv環境を作成します
  3. Flask と requests-oauthlibをvenv環境でインストールします
  4. 下のPythonコードを保存します (oauth_consumer.py とします)
  5. 環境変数 CONSUMER_KEY, CONSUMER_SECRET に自分の consumer_key, consumer_secret を設定して実行します

（Linux, bash系の場合の実行例）

$ python3 -m venv hatena-oauth-sample
$ cd hatena-oauth-sample
$ source bin/activate # 大抵は、この後プロンプトの表示が変化します。ちなみにvenv環境ぬけるときは「deactivate」を実行
$ pip install Flask requests-oauthlib
$ vim oauth_consumer.py # vimに限らず好きな方法で、Pythonコードを保存
$ CONSUMER_KEY=<YOUR_KEY> CONSUMER_SECRET_KEY=<YOUR_SECRETE> python oauth_consumer.py

... で起動してから http://localhost:5000 に Web ブラウザでアクセスして下さい。

OAuth認証が失敗してそうな時は、コマンド実行した端末に表示されているログを確認してください。

また、 Flask のデバッグ・モードを有効にしているので、例外の stacktrace がブラウザに表示された例外画面になった場合は、 ブラウザ上で任意の frame で Python コマンドを実行できます（やり方は、例外画面のメッセージを確認してください）。

作業中に入れたコメントのいくつかは、参考までに残しています。Flask、OAuthはよく分かってないので、お気づきの点などあればコメントしてもらえるとありがたいです。

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oauth_consumer.py
import os
import urllib.parse

try:
    import json
except ImportError:
    import simplejson as json

import flask
from flask import (
    session, request, redirect, url_for, render_template_string,
)
#import oauth2 as oauth
import requests
from requests_oauthlib import OAuth1

SECRET_KEY = 'FLASK_SEC_KEY' # used by flask to encrypt session info
SCOPE = 'read_public'
CONSUMER_KEY = os.environ.get('CONSUMER_KEY', 'HATENA_CONSUMER_KEY')
CONSUMER_SECRET = os.environ.get('CONSUMER_SECRET_KEY', 'HATENA_SECRET_KEY')
DEBUG = True

REQUEST_TOKEN_URL='https://www.hatena.com/oauth/initiate'
ACCESS_TOKEN_URL='https://www.hatena.com/oauth/token'
AUTHORIZE_URL='https://www.hatena.ne.jp/oauth/authorize'
HATENA_USERINFO_URL='http://n.hatena.com/applications/my.json'

app = flask.Flask(__name__)
app.secret_key = SECRET_KEY
app.config['DEBUG'] = DEBUG

TEMPLATE = """
<html>
    <head>
    <titile></title>
    </head>
    <body>
        {% if user %}
            <p>
                Hello {{ user.display_name }}(id:{{ user.url_name }})
                <img src="{{ user.profile_image_url }}">
            </p>
            <p><a href="{{ url_for('logout') }}">LOGOUT</a></p>
        {% else %}
            <p>Hello GUEST</p>
            <p><a href="{{ url_for('login') }}">LOGIN</a></p>
        {% endif %}
    </body>
</html>
"""

@app.route('/')
def index():
    app.logger.debug("index() is called")
    ctx = { 'user': None}
    access_token = session.get('access_token')
    if access_token:
        # access_tokenなどを使ってAPIにアクセスする
        auth = OAuth1(
            CONSUMER_KEY,
            CONSUMER_SECRET,
            access_token['oauth_token'], # 'resource_owner_key' in OAuth1 param
            access_token['oauth_token_secret']) # 'resource_owner_secret'
        r = requests.post(HATENA_USERINFO_URL, auth=auth)
        if not r.ok:
            on_oauth_error(r)
        ctx['user'] = json.loads(r.text)
    return render_template_string(TEMPLATE, **ctx)

# リクエストトークン取得から認証用URLにリダイレクトするための関数
@app.route('/login')
def login():
    # リクエストトークンの取得
    auth = OAuth1(
        CONSUMER_KEY,
        CONSUMER_SECRET,
        callback_uri=url_for('on_auth', _external=True)) # need '_external' flag to avoid rejection from hatena oauth server
    app.logger.debug("auth: {}".format(vars(auth)))
    # scope を POST データで指定して、 request_token 取得 URL へ POST 実行
    r = requests.post(REQUEST_TOKEN_URL,
                        auth=auth,
                        data='scope={}'.format(urllib.parse.quote(SCOPE)))
    if not r.ok:
        on_oauth_error(r)
    request_token = dict(urllib.parse.parse_qsl(r.text))

    # セッションへリクエストトークンを保存しておく
    app.logger.debug("request_token: {}".format(request_token))
    session['request_token'] = request_token
    # 認証用URLにリダイレクトする
    return redirect('{}?oauth_token={}'.format(
                    AUTHORIZE_URL,
                    session['request_token']['oauth_token']))

# セッションに保存されたトークンを破棄しログアウトする関数
@app.route('/logout')
def logout():
    if session.get('access_token'): session.pop('access_token')
    if session.get('request_token'): session.pop('request_token')
    return redirect(url_for('index'))

# 認証からコールバックされ、アクセストークンを取得するための関数
@app.route('/on-auth')
def on_auth():
    # login()処理中に session へ保存していた request_token を取得
    request_token = session.get('request_token')
    # 認証用URLからのリダイレクト時に設定された oauth_verifier を取得
    oauth_verifier = request.args['oauth_verifier']

    # request_token と oauth_verifier を用いてアクセストークンを取得
    auth = OAuth1(
        CONSUMER_KEY,
        CONSUMER_SECRET,
        request_token['oauth_token'], # 'resource_owner_key' in OAuth1 param
        request_token['oauth_token_secret'], # 'resource_owner_secret'
        verifier=oauth_verifier)
    r = requests.post(ACCESS_TOKEN_URL, auth=auth)
    if not r.ok:
        on_oauth_error(r)
    access_token = dict(urllib.parse.parse_qsl(r.text))
    # セッションに対し、request_token を削除、アクセストークンを記録
    if session.get('request_token'): session.pop('request_token')
    session['access_token'] = access_token
    return redirect(url_for('index'))

def on_oauth_error(res):
    if not res.ok:
        app.logger.debug("response header: {}".format(res.headers))
        app.logger.debug("response content: {}".format(res.content))
        res.raise_for_status()

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5000, debug=True)

2019-11-30追記：「OAuthは認証ではなく認可」という説明をどこかで 目にしたので、タイトルを変更しました（認証→認可）。