Flask-OAuthlib 完全解説：FlaskアプリでのOAuth認証の導入と注意点

pip install Flask-OAuthlib

from flask import Flask, redirect, url_for, session, request, jsonify
from flask_oauthlib.client import OAuth

app = Flask(__name__)
# 必ず SECRET_KEY を設定してください
app.secret_key = 'your_secret_key_here' # 実際のアプリケーションでは環境変数などから読み込むべきです

oauth = OAuth(app)

# 例: GitHub OAuth2クライアントの登録
github = oauth.remote_app(
    'github',
    consumer_key='YOUR_GITHUB_CLIENT_ID',         # GitHubから取得したClient ID
    consumer_secret='YOUR_GITHUB_CLIENT_SECRET',   # GitHubから取得したClient Secret
    request_token_params={'scope': 'user:email'}, # 要求する権限 (スコープ)
    base_url='https://api.github.com/',
    request_token_url=None,                       # OAuth2では使用しない
    access_token_method='POST',
    access_token_url='https://github.com/login/oauth/access_token',
    authorize_url='https://github.com/login/oauth/authorize'
)

# トークンを取得するための関数 (セッションに保存する例)
@github.tokengetter
def get_github_oauth_token():
    return session.get('github_token')

@app.route('/login')
def login():
    # ユーザーをGitHubの認可ページにリダイレクトさせる
    # callbackには、認可後にリダイレクトされる自アプリのURLを指定
    return github.authorize(callback=url_for('authorized', _external=True))

@app.route('/login/authorized')
def authorized():
    resp = github.authorized_response()
    if resp is None or resp.get('access_token') is None:
        return 'Access denied: reason={} error={}'.format(
            request.args['error'],
            request.args['error_description']
        )

    # アクセストークンをセッションに保存
    session['github_token'] = (resp['access_token'], '') # (token, secret) の形式で保存 (OAuth2ではsecretは空)
    session['github_user'] = github.get('user').data # 例: ユーザー情報を取得

    return redirect(url_for('index'))

@app.route('/logout')
def logout():
    session.pop('github_token', None)
    session.pop('github_user', None)
    return redirect(url_for('index'))

@app.route('/')
def index():
    if 'github_token' in session:
        user_info = session.get('github_user')
        if user_info:
            return 'Logged in as id: {} name: {}  Logout'.format(user_info.get('id'), user_info.get('login'))
        else:
             # APIアクセスに失敗した場合など
             return 'Logged in, but failed to fetch user info. Logout'
    return 'You are not logged in. Log in with GitHub'

if __name__ == '__main__':
    # HTTPSが推奨されますが、開発環境ではhttpでも動作します
    # 重要: OAuthプロバイダに登録するコールバックURLと一致させる必要があります
    app.run(debug=True, port=5000) # 必要に応じて host='localhost' を指定

from flask import Flask, render_template, request, jsonify
from flask_sqlalchemy import SQLAlchemy # データモデルの例としてSQLAlchemyを使用
from flask_oauthlib.provider import OAuth2Provider
import datetime

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///oauth_provider.db' # 例: SQLite
app.config['SECRET_KEY'] = 'your_provider_secret_key' # クライアントと同様に必須
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
db = SQLAlchemy(app)
oauth = OAuth2Provider(app)

# --- データモデルの定義 (例) ---
class User(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(80), unique=True, nullable=False)
    # ... (パスワードハッシュなど、通常のユーザーモデルに必要な属性)

class Client(db.Model):
    client_id = db.Column(db.String(40), primary_key=True)
    client_secret = db.Column(db.String(55), unique=True, index=True, nullable=False)
    user_id = db.Column(db.ForeignKey('user.id'))
    user = db.relationship('User')
    _redirect_uris = db.Column(db.Text)
    _default_scopes = db.Column(db.Text)
    # ... (クライアント名、クライアントタイプなど)

    @property
    def redirect_uris(self):
        if self._redirect_uris:
            return self._redirect_uris.split()
        return []

    @property
    def default_redirect_uri(self):
        return self.redirect_uris[0] if self.redirect_uris else None

    @property
    def default_scopes(self):
        if self._default_scopes:
            return self._default_scopes.split()
        return []

    @property
    def client_type(self):
        # 'public' または 'confidential' を返す
        return 'public' # 例

    @property
    def allowed_grant_types(self):
        # サポートするグラントタイプをリストで返す
        return ['authorization_code', 'password', 'client_credentials', 'refresh_token']

    @property
    def allowed_response_types(self):
        # サポートするレスポンスタイプをリストで返す
        return ['code', 'token']


class Token(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    client_id = db.Column(db.String(40), db.ForeignKey('client.client_id'), nullable=False)
    client = db.relationship('Client')
    user_id = db.Column(db.Integer, db.ForeignKey('user.id'))
    user = db.relationship('User')
    token_type = db.Column(db.String(40)) # 'Bearer' など
    access_token = db.Column(db.String(255), unique=True)
    refresh_token = db.Column(db.String(255), unique=True)
    expires = db.Column(db.DateTime)
    _scopes = db.Column(db.Text)

    @property
    def scopes(self):
        if self._scopes:
            return self._scopes.split()
        return []

class Grant(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    user_id = db.Column(db.Integer, db.ForeignKey('user.id'), nullable=False)
    user = db.relationship('User')
    client_id = db.Column(db.String(40), db.ForeignKey('client.client_id'), nullable=False)
    client = db.relationship('Client')
    code = db.Column(db.String(255), index=True, nullable=False)
    redirect_uri = db.Column(db.String(255))
    expires = db.Column(db.DateTime)
    _scopes = db.Column(db.Text)

    @property
    def scopes(self):
        if self._scopes:
            return self._scopes.split()
        return []

# --- OAuth Provider の設定 ---

# 現在ログインしているユーザーを取得する関数 (Flask-Loginなどを使うのが一般的)
def get_current_user():
    # ここで実際のログインユーザーを取得するロジックを実装
    # 例: Flask-Login の current_user を使う
    # from flask_login import current_user
    # if current_user.is_authenticated:
    #     return User.query.get(current_user.id)
    # return None
    # ダミー実装
    user = User.query.filter_by(username='testuser').first()
    if not user:
        user = User(username='testuser')
        db.session.add(user)
        db.session.commit()
    return user


@oauth.clientgetter
def load_client(client_id):
    return Client.query.filter_by(client_id=client_id).first()

@oauth.grantgetter
def load_grant(client_id, code):
    return Grant.query.filter_by(client_id=client_id, code=code).first()

@oauth.grantsetter
def save_grant(client_id, code, request, *args, **kwargs):
    # 認可コードを保存
    expires = datetime.datetime.utcnow() + datetime.timedelta(seconds=100)
    grant = Grant(
        client_id=client_id,
        code=code['code'],
        redirect_uri=request.redirect_uri,
        _scopes=' '.join(request.scopes),
        user=get_current_user(), # 現在のユーザーを取得
        expires=expires
    )
    db.session.add(grant)
    db.session.commit()
    return grant

@oauth.tokengetter
def load_token(access_token=None, refresh_token=None):
    if access_token:
        return Token.query.filter_by(access_token=access_token).first()
    elif refresh_token:
        return Token.query.filter_by(refresh_token=refresh_token).first()
    return None

@oauth.tokensetter
def save_token(token, request, *args, **kwargs):
    # 既存のトークンを削除 (プロバイダのポリシーによる)
    toks = Token.query.filter_by(client_id=request.client.client_id,
                                 user_id=request.user.id).all()
    for t in toks:
        db.session.delete(t)

    expires_in = token.get('expires_in')
    expires = datetime.datetime.utcnow() + datetime.timedelta(seconds=expires_in)

    new_token = Token(
        access_token=token['access_token'],
        refresh_token=token.get('refresh_token'),
        token_type=token['token_type'],
        _scopes=token['scope'],
        expires=expires,
        client_id=request.client.client_id,
        user_id=request.user.id
    )
    db.session.add(new_token)
    db.session.commit()
    return new_token

@oauth.usergetter
def get_user(username, password, *args, **kwargs):
    # Password Credentials Grant のためのユーザー検証ロジック
    # 本番環境ではパスワードハッシュを比較する
    user = User.query.filter_by(username=username).first()
    if user and user.check_password(password): # check_password は自分で実装する必要がある
        return user
    return None

@app.route('/oauth/authorize', methods=['GET', 'POST'])
@oauth.authorize_handler
def authorize(*args, **kwargs):
    # ユーザーがログインしていなければログインページにリダイレクト
    user = get_current_user()
    if not user:
        # Flask-Loginなどのログイン処理へ
        return redirect(url_for('login_page', next=request.url))

    if request.method == 'GET':
        # ユーザーに認可を求めるフォームを表示
        client_id = kwargs.get('client_id')
        client = Client.query.filter_by(client_id=client_id).first()
        kwargs['client'] = client
        kwargs['user'] = user
        # scopes は kwargs['scopes'] にリストで入っている
        return render_template('authorize_form.html', **kwargs) # 認可フォーム用のテンプレート

    # POSTリクエストはユーザーがフォームで許可/拒否した場合
    confirm = request.form.get('confirm', 'no')
    return confirm == 'yes'


@app.route('/oauth/token', methods=['POST'])
@oauth.token_handler
def access_token():
    # この関数自体は通常空で良い
    # @oauth.token_handler がトークン発行処理を行う
    return None

@app.route('/api/me')
@oauth.require_oauth('email') # 要求するスコープを指定
def me(req):
    # req.user にはアクセストークンに対応するユーザーオブジェクトが入る
    user = req.user
    return jsonify(id=user.id, username=user.username)

@app.route('/api/client')
@oauth.require_oauth() # クライアントクレデンシャル用のエンドポイント例
def client_api(req):
    # req.client にはクライアントオブジェクトが入る
    client = req.client
    return jsonify(client_id=client.client_id, client_secret=client.client_secret) # 注意: シークレットを返すのは通常非推奨

# データベースの初期化 (初回実行時など)
@app.cli.command('init-db')
def init_db_command():
    """Clear the existing data and create new tables."""
    db.create_all()
    print('Initialized the database.')
    # 必要に応じてテスト用のクライアントやユーザーを作成
    if not User.query.filter_by(username='testuser').first():
        test_user = User(username='testuser')
        # test_user.set_password('password') # パスワード設定処理を実装する
        db.session.add(test_user)

    if not Client.query.filter_by(client_id='test_client').first():
         client = Client(
             client_id='test_client',
             client_secret='test_secret', # 本番ではランダム生成する
             _redirect_uris='http://localhost:5000/login/authorized http://127.0.0.1:5000/login/authorized', # 開発用コールバックURL例
             _default_scopes='email profile',
             user=User.query.filter_by(username='testuser').first() # クライアント所有者
         )
         db.session.add(client)
    db.session.commit()
    print('Added test user and client.')


if __name__ == '__main__':
    app.run(debug=True, port=8000) # クライアントと異なるポートで実行

# --- Flask-OAuthlib ---
# from flask_oauthlib.client import OAuth
# oauth = OAuth(app)
# github = oauth.remote_app(...)

# --- Authlib ---
from authlib.integrations.flask_client import OAuth
oauth = OAuth(app)
# または oauth = OAuth(); oauth.init_app(app) でも可
github = oauth.register(
    name='github',
    client_id='YOUR_GITHUB_CLIENT_ID',
    client_secret='YOUR_GITHUB_CLIENT_SECRET',
    access_token_url='https://github.com/login/oauth/access_token',
    access_token_params=None,
    authorize_url='https://github.com/login/oauth/authorize',
    authorize_params=None,
    api_base_url='https://api.github.com/',
    client_kwargs={'scope': 'user:email'} # スコープ指定方法などが異なる場合がある
)