Web API 連携の必須知識 OAuth を Python で簡単に実装!
はじめに:requests-oauthlib とは?🤔
現代の Web サービス開発において、外部の API と連携することは一般的です。例えば、Twitter や Google、GitHub などのサービスが提供する API を利用して、ログイン機能やデータ連携を実現するケースは非常に多いでしょう。これらの API の多くは、安全にリソースへのアクセスを許可するために OAuth (Open Authorization) という認証・認可プロトコルを採用しています。
しかし、OAuth の仕様は複雑で、特に OAuth 1.0a は署名プロセスなどが必要となり、自力で実装するのは骨が折れます 😩。そこで登場するのが requests-oauthlib
です!
requests-oauthlib
は、Python で非常に人気のある HTTP クライアントライブラリ requests
と、OAuth のコアロジックを実装する oauthlib
を組み合わせ、OAuth 1.0a と OAuth 2.0 のクライアント実装を劇的に簡単にしてくれる Python ライブラリです。Requests-OAuthlib を使うことで、開発者は OAuth の複雑な詳細を意識することなく、シンプルかつ直感的に OAuth 認証が必要な API と連携できます。
このライブラリは、requests
の使いやすさをそのままに、OAuth 認証に必要な処理をカプセル化して提供します。これにより、わずか数行のコードで OAuth 認証付きの HTTP リクエストを送信できるようになります🚀。
このブログ記事では、requests-oauthlib
のインストール方法から、OAuth 1.0a と OAuth 2.0 の基本的な使い方、具体的なユースケース、注意点まで、詳細に解説していきます。
インストール方法 💻
requests-oauthlib
のインストールは、Python のパッケージ管理ツール pip
を使って簡単に行えます。ターミナルまたはコマンドプロンプトで以下のコマンドを実行してください。
pip install requests requests-oauthlib
requests-oauthlib
は requests
と oauthlib
に依存しているため、通常は一緒にインストールされますが、明示的に指定しておくと確実です。
OAuth の基本を押さえよう 📜
requests-oauthlib
を使いこなす前に、OAuth の基本的な概念を理解しておくと、ライブラリの機能をより深く理解できます。OAuth には主に二つのバージョン、OAuth 1.0a と OAuth 2.0 があります。
OAuth 1.0a
- 2010年に RFC 5849 として標準化されました(OAuth 1.0 のセキュリティ問題を修正したバージョン)。
- 署名ベースの認証方式を採用しています。リクエストごとに、クライアントキー、クライアントシークレット、リクエストトークン(またはアクセストークン)、トークンシークレットなどを使って複雑な署名を生成し、リクエストに含める必要があります。
- 主に Twitter API (v1.1) などで利用されていました(v2 では OAuth 2.0 もサポート)。
- 認証フローは、リクエストトークンの取得 → ユーザー認可 → アクセストークンの取得、というステップを踏みます。
- HTTPS (TLS/SSL) は必須ではありませんが、推奨されます。
OAuth 2.0
- 2012年に RFC 6749 として標準化され、現在の主流となっています。
- OAuth 1.0a との後方互換性はありません。
- トークンベースの認証方式で、署名生成は不要になりました。代わりに HTTPS (TLS/SSL) が必須となり、通信経路の暗号化によってトークンの盗聴を防ぎます。
- アクセストークンのみを使用し、認証フローがシンプル化されました。
- 様々なユースケースに対応するため、複数の「グラントタイプ (Grant Type)」と呼ばれる認証フローが定義されています。
- Authorization Code Grant (認可コードグラント): Web アプリケーションで最も一般的に使われるフロー。サーバーサイドで安全にクライアントシークレットを扱える場合に適しています。
- Implicit Grant (インプリシットグラント): JavaScript などで実装されるシングルページアプリケーション (SPA) やモバイルアプリ向けでしたが、セキュリティ上の懸念から現在では非推奨とされ、Authorization Code Grant with PKCE の利用が推奨されます。
- Resource Owner Password Credentials Grant (リソースオーナーパスワードクレデンシャルグラント): ユーザー名とパスワードを直接クライアントアプリケーションに入力するフロー。信頼できるアプリケーション(公式アプリなど)以外での使用は非推奨です。
- Client Credentials Grant (クライアントクレデンシャルグラント): アプリケーション自身の認証に使われるフロー。ユーザーの認可を必要とせず、クライアント ID とクライアントシークレットのみでアクセストークンを取得します。
- アクセストークンの有効期限が切れた場合に、ユーザーの再認可なしに新しいアクセストークンを取得するためのリフレッシュトークンの仕組みがあります(一部のグラントタイプ)。
- モバイルアプリや SPA でのセキュリティを高めるための拡張仕様 PKCE (Proof Key for Code Exchange) (RFC 7636) が広く使われています。
OAuth 1.0a と OAuth 2.0 の主な違い(まとめ)
項目 | OAuth 1.0a | OAuth 2.0 |
---|---|---|
標準化 | RFC 5849 (2010年) | RFC 6749 (2012年) |
認証方式 | 署名ベース | トークンベース |
HTTPS (TLS/SSL) | 推奨 | 必須 |
トークン種類 | リクエストトークン、アクセストークン | アクセストークン (リフレッシュトークンあり) |
複雑さ | 複雑(署名生成) | 比較的シンプル(フローによる) |
対応アプリ | 主に Web アプリ | Web アプリ、モバイルアプリ、SPA、サーバー間通信など多様 |
主なグラントタイプ | – (単一フロー) | Authorization Code, Implicit, Password Credentials, Client Credentials など |
セキュリティ拡張 | – | PKCE など |
現在、新規に OAuth 認証を実装する場合、特別な理由がない限り OAuth 2.0 を採用するのが一般的です。多くの主要な API プロバイダー (Google, Facebook, GitHub, Microsoft など) が OAuth 2.0 を採用しています。
requests-oauthlib の使い方 (OAuth 1.0a) 🔑🐦
OAuth 1.0a は複雑ですが、requests-oauthlib
を使えば比較的簡単に扱うことができます。主に OAuth1Session
クラスを利用します。
OAuth 1.0a を利用するには、事前に API プロバイダー (例: Twitter) にアプリケーションを登録し、以下のクレデンシャル情報を取得しておく必要があります。
- Client Key (Consumer Key): アプリケーションを識別するためのキー
- Client Secret (Consumer Secret): アプリケーションの秘密鍵
- Resource Owner Key (Access Token): ユーザーがアプリケーションを認可した後に得られるトークン
- Resource Owner Secret (Access Token Secret): アクセストークンに対応する秘密鍵
1. 認証済みリクエストの送信 (アクセストークン取得済みの場合)
すでに上記4つのクレデンシャルを持っている場合、OAuth1Session
を初期化して requests
と同じように get
や post
メソッドで API にリクエストを送信できます。
from requests_oauthlib import OAuth1Session
# Twitter API v1.1 の例
client_key = 'YOUR_CLIENT_KEY' # あなたの Client Key
client_secret = 'YOUR_CLIENT_SECRET' # あなたの Client Secret
resource_owner_key = 'USER_ACCESS_TOKEN' # ユーザーの Access Token
resource_owner_secret = 'USER_ACCESS_TOKEN_SECRET' # ユーザーの Access Token Secret
# OAuth1Session オブジェクトを作成
twitter = OAuth1Session(client_key,
client_secret=client_secret,
resource_owner_key=resource_owner_key,
resource_owner_secret=resource_owner_secret)
# 保護されたリソースにアクセス (例: ユーザー設定の取得)
url = 'https://api.twitter.com/1.1/account/settings.json'
try:
response = twitter.get(url)
response.raise_for_status() # エラーがあれば例外を発生させる
user_settings = response.json()
print("ユーザー設定を取得しました:")
print(user_settings)
except Exception as e:
print(f"エラーが発生しました: {e}")
if hasattr(e, 'response') and e.response is not None:
print(f"ステータスコード: {e.response.status_code}")
print(f"レスポンスボディ: {e.response.text}")
OAuth1Session
が自動的にリクエストに必要な OAuth 署名を計算し、Authorization
ヘッダーに追加してくれます。
また、requests
の auth
パラメータに OAuth1
オブジェクトを渡す方法もあります。
import requests
from requests_oauthlib import OAuth1
auth = OAuth1(client_key, client_secret, resource_owner_key, resource_owner_secret)
response = requests.get(url, auth=auth)
# ...以降の処理は同様
OAuth1Session
を使う方が、セッション管理 (クッキーの保持など) が容易になるため、一般的にはこちらが推奨されます。
2. 認証フローの実装 (アクセストークン未取得の場合)
アプリケーションが初めてユーザーのリソースにアクセスする場合や、アクセストークンを持っていない場合は、OAuth 1.0a の認証フローを実行する必要があります。
ステップ 1: リクエストトークンの取得
まず、Client Key と Client Secret を使って、API プロバイダーからリクエストトークンを取得します。OAuth1Session
の fetch_request_token
メソッドを使います。コールバック URL を指定することもできます。
from requests_oauthlib import OAuth1Session
import webbrowser
import os
client_key = 'YOUR_CLIENT_KEY'
client_secret = 'YOUR_CLIENT_SECRET'
request_token_url = 'https://api.twitter.com/oauth/request_token' # Twitter の例
authorization_url = 'https://api.twitter.com/oauth/authorize'
access_token_url = 'https://api.twitter.com/oauth/access_token'
callback_uri = 'oob' # Out-of-band (OOB): PINコードをユーザーが手入力する場合
# 1. リクエストトークンを取得
print("リクエストトークンを取得中...")
oauth_session = OAuth1Session(client_key, client_secret=client_secret, callback_uri=callback_uri)
try:
fetch_response = oauth_session.fetch_request_token(request_token_url)
resource_owner_key = fetch_response.get('oauth_token')
resource_owner_secret = fetch_response.get('oauth_token_secret')
print("リクエストトークン取得成功!")
print(f" oauth_token: {resource_owner_key}")
print(f" oauth_token_secret: {resource_owner_secret}")
except Exception as e:
print(f"リクエストトークン取得エラー: {e}")
exit()
# 2. ユーザー認可のためにリダイレクト
print("\nユーザー認可が必要です。")
auth_link = oauth_session.authorization_url(authorization_url)
print(f"以下のURLにアクセスして、アプリケーションを認可し、表示されるPINコードを入力してください:\n{auth_link}")
# ブラウザを自動で開く (環境によっては動作しない場合があります)
try:
webbrowser.open(auth_link)
except webbrowser.Error:
print("ブラウザを自動で開けませんでした。手動でURLを開いてください。")
verifier = input("PINコードを入力してください: ")
# 3. アクセストークンを取得
print("\nアクセストークンを取得中...")
oauth_session = OAuth1Session(client_key,
client_secret=client_secret,
resource_owner_key=resource_owner_key,
resource_owner_secret=resource_owner_secret,
verifier=verifier)
try:
oauth_tokens = oauth_session.fetch_access_token(access_token_url)
access_token = oauth_tokens.get('oauth_token')
access_token_secret = oauth_tokens.get('oauth_token_secret')
user_id = oauth_tokens.get('user_id')
screen_name = oauth_tokens.get('screen_name')
print("アクセストークン取得成功! 🎉")
print(f" Access Token: {access_token}")
print(f" Access Token Secret: {access_token_secret}")
print(f" User ID: {user_id}")
print(f" Screen Name: {screen_name}")
# 取得したアクセストークンを使ってAPIリクエストを実行
# (ここでは例として再度セッションを作成)
final_twitter_session = OAuth1Session(client_key,
client_secret=client_secret,
resource_owner_key=access_token,
resource_owner_secret=access_token_secret)
# 例: 自分のタイムラインを取得
timeline_url = 'https://api.twitter.com/1.1/statuses/home_timeline.json'
response = final_twitter_session.get(timeline_url, params={'count': 5})
if response.status_code == 200:
tweets = response.json()
print("\n最新のタイムライン (5件):")
for tweet in tweets:
print(f"- @{tweet['user']['screen_name']}: {tweet['text']}")
else:
print(f"\nタイムライン取得エラー: {response.status_code} - {response.text}")
except Exception as e:
print(f"アクセストークン取得またはAPIアクセスエラー: {e}")
if hasattr(e, 'response') and e.response is not None:
print(f"ステータスコード: {e.response.status_code}")
print(f"レスポンスボディ: {e.response.text}")
ステップ 2: ユーザー認可
取得したリクエストトークン (oauth_token
) を使って、認可 URL を生成します (authorization_url
メソッド)。ユーザーをこの URL にリダイレクトし、アプリケーションへのアクセス許可を求めます。ユーザーが許可すると、多くの場合 PIN コード (Verifier) が表示されるか、指定したコールバック URL にリダイレクトされます。
ステップ 3: アクセストークンの取得
ユーザーが入力した Verifier (PIN コード) と、ステップ 1 で取得したリクエストトークン、リクエストトークンシークレットを使って、fetch_access_token
メソッドを呼び出し、最終的なアクセストークンとアクセストークンシークレットを取得します。
取得したアクセストークンとシークレットは、次回以降の API リクエストのために安全な場所に保存しておく必要があります。
callback_uri
にその URL を指定し、リダイレクト先で oauth_verifier
パラメータを受け取る処理が必要です。
requests-oauthlib の使い方 (OAuth 2.0) 🔑🔒
OAuth 2.0 は現在の主流であり、requests-oauthlib
では OAuth2Session
クラスを使って簡単に扱うことができます。OAuth 2.0 には複数のグラントタイプがありますが、ここでは主要なものを中心に解説します。
OAuth 2.0 を利用するには、事前に API プロバイダー (例: Google, GitHub) にアプリケーションを登録し、以下の情報を取得しておく必要があります(グラントタイプによって必要な情報は異なります)。
- Client ID: アプリケーションを識別するための ID
- Client Secret: アプリケーションの秘密鍵 (Confidential Client の場合)
- Redirect URI (Callback URL): ユーザー認可後にリダイレクトされる URL
- Scope: アプリケーションが要求するアクセス権限の範囲
- Authorization Endpoint URL: ユーザー認可リクエストを送る URL
- Token Endpoint URL: アクセストークンを取得・更新する URL
oauthlib.oauth2.InsecureTransportError
が発生します。テスト目的で一時的に無効にするには、環境変数 OAUTHLIB_INSECURE_TRANSPORT=1
を設定します。本番環境では絶対にこの設定を使用しないでください。
1. Authorization Code Grant (認可コードグラント) – Web アプリ向け
Web アプリケーションで最も一般的に使用されるフローです。サーバーサイドで Client Secret を安全に管理できる場合に適しています。
ステップ 1: ユーザーを認可エンドポイントにリダイレクト
OAuth2Session
を初期化し、authorization_url
メソッドを使って認可 URL を生成します。この URL にユーザーをリダイレクトします。
from requests_oauthlib import OAuth2Session
from flask import Flask, request, redirect, session, url_for, jsonify
import os
import webbrowser
# Flask アプリケーションのセットアップ (例)
app = Flask(__name__)
# セッション管理のために SECRET_KEY を設定 (実際には安全な値を設定)
app.secret_key = os.urandom(24)
# --- Google API の例 ---
client_id = "YOUR_GOOGLE_CLIENT_ID.apps.googleusercontent.com" # あなたの Client ID
client_secret = "YOUR_GOOGLE_CLIENT_SECRET" # あなたの Client Secret
# Google Cloud Console で設定したリダイレクト URI
# この例では Flask の /callback ルートを指す
redirect_uri = 'http://localhost:5000/callback'
# 要求するスコープ (例: メールアドレスとプロファイル情報)
scope = ['https://www.googleapis.com/auth/userinfo.email',
'https://www.googleapis.com/auth/userinfo.profile']
authorization_base_url = "https://accounts.google.com/o/oauth2/v2/auth"
token_url = "https://www.googleapis.com/oauth2/v4/token"
# --- ここまで Google API の例 ---
@app.route("/")
def index():
"""ステップ 1: ユーザー認可のためのURLを生成しリダイレクト"""
google = OAuth2Session(client_id, scope=scope, redirect_uri=redirect_uri)
# 認可URLと state (CSRF対策) を生成
authorization_url, state = google.authorization_url(
authorization_base_url,
# offline アクセスを要求するとリフレッシュトークンが取得できる (初回のみ)
access_type="offline",
# 毎回同意画面を表示させたい場合 (通常は不要)
# prompt="consent"
)
# state をセッションに保存して、コールバックで検証する
session['oauth_state'] = state
print(f"認可URL: {authorization_url}")
print(f"生成された state: {state}")
# ユーザーを認可URLへリダイレクト
return redirect(authorization_url)
@app.route("/callback")
def callback():
"""ステップ 2: 認可サーバーからのコールバックを受け取り、アクセストークンを取得"""
# state の検証 (CSRF対策)
print(f"コールバック受信。 request.url: {request.url}")
print(f"セッションに保存された state: {session.get('oauth_state')}")
# セッションから state を取得し、比較後すぐに削除
expected_state = session.pop('oauth_state', None)
if expected_state is None:
return 'Error: state がセッションに見つかりません。', 400
google = OAuth2Session(client_id, redirect_uri=redirect_uri, state=expected_state)
try:
# 認可コードを使ってアクセストークンを取得
# redirect_uri に含まれる完全な URL を渡す
token = google.fetch_token(token_url,
client_secret=client_secret,
# リクエストURL全体を渡すことが推奨される
authorization_response=request.url)
# トークンをセッションに保存 (本番環境ではDBなどに安全に保存)
session['oauth_token'] = token
print("トークン取得成功!")
print(token)
return redirect(url_for('profile'))
except Exception as e:
print(f"トークン取得エラー: {e}")
# エラーの詳細を表示 (デバッグ用)
if hasattr(e, 'description'):
print(f"エラー詳細: {e.description}")
return f'アクセストークンの取得に失敗しました: {e}', 500
@app.route("/profile")
def profile():
"""ステップ 3: 取得したアクセストークンを使って保護されたリソースにアクセス"""
if 'oauth_token' not in session:
return redirect(url_for('index'))
token = session['oauth_token']
google = OAuth2Session(client_id, token=token)
try:
# Google UserInfo API にアクセス
user_info = google.get('https://www.googleapis.com/oauth2/v1/userinfo').json()
print("ユーザー情報取得成功:")
print(user_info)
return jsonify(user_info)
except Exception as e:
print(f"APIアクセスエラー: {e}")
# トークン有効期限切れの可能性もある (リフレッシュ処理が必要)
return f'APIへのアクセスに失敗しました: {e}', 500
if __name__ == "__main__":
# HTTPSが利用できないローカル環境でのテスト用
# 本番環境では絶対に True にしないこと!
os.environ['OAUTHLIB_INSECURE_TRANSPORT'] = '1'
# Flask 開発サーバーを起動
app.run(port=5000, debug=True) # debug=True は開発時のみ
上記のコードは Flask を使った例です。/
ルートにアクセスすると、Google の認可画面にリダイレクトされます。ユーザーが許可すると、/callback
ルートにリダイレクトされ、そこで認可コード (URL の code
パラメータ) を使って fetch_token
メソッドでアクセストークンを取得します。取得したトークンはセッションに保存され、/profile
ルートでそのトークンを使って Google API からユーザー情報を取得しています。
CSRF 対策の state
パラメータ: authorization_url
メソッドは、ランダムな state
文字列を生成します。これをセッションなどに保存しておき、コールバック時に返される state
パラメータと比較することで、クロスサイトリクエストフォージェリ (CSRF) 攻撃を防ぎます。一致しない場合はリクエストを拒否する必要があります。
2. Authorization Code Grant with PKCE – SPA・モバイルアプリ向け
シングルページアプリケーション (SPA) やモバイルアプリのように、Client Secret を安全に保持できないクライアントのためのフローです。PKCE (Proof Key for Code Exchange, RFC 7636) という仕組みを使い、認可コードの傍受によるなりすましを防ぎます。
requests-oauthlib
バージョン 1.3.1 以降で PKCE がサポートされています。
基本的な流れは通常の Authorization Code Grant と似ていますが、以下の点が異なります。
- コードベリファイア (Code Verifier) の生成: クライアントはまず、ランダムで推測困難な文字列「コードベリファイア」を生成します。
- コードチャレンジ (Code Challenge) の生成: コードベリファイアをハッシュ化 (SHA256 が推奨) し、Base64 URL エンコードした「コードチャレンジ」と、使用したハッシュアルゴリズム (
code_challenge_method='S256'
) を認可リクエストに含めます。 - 認可コードの取得: ユーザー認可後、通常通り認可コードが返されます。
- アクセストークンの取得: トークンリクエスト時に、認可コードに加えて、ステップ 1 で生成した元のコードベリファイアを送信します。
- サーバーでの検証: 認可サーバーは、受け取ったコードベリファイアからコードチャレンジを再計算し、認可リクエスト時に受け取ったコードチャレンジと一致するか検証します。一致すればアクセストークンを発行します。
from requests_oauthlib import OAuth2Session
import secrets
import hashlib
import base64
import os
import webbrowser
# --- Auth0 API の例 (PKCEが推奨される) ---
# Auth0 の Application 設定で Application Type を Native または SPA にする
client_id = "YOUR_AUTH0_CLIENT_ID"
# SPA や Native アプリでは Client Secret は通常不要 (または設定しない)
# client_secret = "YOUR_AUTH0_CLIENT_SECRET" # 不要
redirect_uri = 'http://localhost:8080/callback' # Auth0 Dashboard で許可された Callback URL
scope = ['openid', 'profile', 'email'] # OpenID Connect スコープ
authorization_base_url = "https://YOUR_AUTH0_DOMAIN/authorize"
token_url = "https://YOUR_AUTH0_DOMAIN/oauth/token"
# --- ここまで Auth0 API の例 ---
# ステップ 1: コードベリファイア生成
code_verifier = secrets.token_urlsafe(64)
print(f"Code Verifier: {code_verifier}")
# ステップ 2: コードチャレンジ生成 (S256)
hashed = hashlib.sha256(code_verifier.encode('utf-8')).digest()
code_challenge = base64.urlsafe_b64encode(hashed).decode('utf-8').replace('=', '')
print(f"Code Challenge (S256): {code_challenge}")
# OAuth2Session を初期化
oauth = OAuth2Session(client_id, redirect_uri=redirect_uri, scope=scope)
# 認可URLを生成 (PKCE パラメータを追加)
authorization_url, state = oauth.authorization_url(
authorization_base_url,
code_challenge=code_challenge,
code_challenge_method='S256' # 使用したハッシュアルゴリズムを指定
)
print(f"\n認可URL:\n{authorization_url}")
print(f"State: {state}")
# --- ここからユーザーをリダイレクトし、コールバックを受け取る処理 (Flask 等が必要) ---
# 例として、ブラウザを開き、認可コードを手動で入力する想定
try:
webbrowser.open(authorization_url)
except webbrowser.Error:
print("ブラウザを自動で開けませんでした。手動でURLを開いて認可してください。")
authorization_response = input("\nブラウザで認可後、リダイレクトされた URL 全体を入力してください:\n")
# ステップ 4 & 5: アクセストークン取得 (コードベリファイアを渡す)
print("\nアクセストークンを取得中...")
try:
# fetch_token に code_verifier を渡す
# client_secret は PKCE では通常不要
token = oauth.fetch_token(
token_url,
authorization_response=authorization_response,
# Client Secret が不要な Public Client の場合は include_client_id=True を指定することがある
# include_client_id=True,
# client_secret=client_secret, # Public Client では不要
code_verifier=code_verifier # ★★★ ここで元のコードベリファイアを渡す ★★★
)
print("\nトークン取得成功! 🎉")
print(token)
# 取得したトークンでAPIアクセス
# 例: Auth0 UserInfo エンドポイント
userinfo_url = f"https://YOUR_AUTH0_DOMAIN/userinfo"
user_info_resp = oauth.get(userinfo_url)
if user_info_resp.ok:
print("\nユーザー情報:")
print(user_info_resp.json())
else:
print(f"\nユーザー情報取得エラー: {user_info_resp.status_code} - {user_info_resp.text}")
except Exception as e:
print(f"\nトークン取得またはAPIアクセスエラー: {e}")
if hasattr(e, 'response') and e.response is not None:
print(f"ステータスコード: {e.response.status_code}")
print(f"レスポンスボディ: {e.response.text}")
if hasattr(e, 'description'):
print(f"エラー詳細: {e.description}")
重要なのは、authorization_url
生成時に code_challenge
と code_challenge_method
を指定し、fetch_token
時に元の code_verifier
を渡す点です。これにより、認可コードが途中で漏洩しても、コードベリファイアを知らない攻撃者はアクセストークンを取得できません。
3. Client Credentials Grant (クライアントクレデンシャルグラント) – サーバー間通信向け
ユーザーの介在なしに、アプリケーション自身が API にアクセスする場合に使用します。例えば、自社サービスの分析データを取得する API などです。Client ID と Client Secret を使って直接アクセストークンを取得します。リフレッシュトークンは通常発行されません。
from requests_oauthlib import OAuth2Session
from oauthlib.oauth2 import BackendApplicationClient
import os
client_id = 'YOUR_SERVICE_CLIENT_ID'
client_secret = 'YOUR_SERVICE_CLIENT_SECRET'
token_url = 'https://api.example.com/oauth/token' # トークンエンドポイント
scope = ['read:stats', 'write:logs'] # 要求するスコープ (例)
try:
# BackendApplicationClient を使ってセッションを準備
# scope はここで指定することも、fetch_token で指定することも可能
client = BackendApplicationClient(client_id=client_id, scope=scope)
oauth_session = OAuth2Session(client=client)
# アクセストークンを取得
# client_id と client_secret は通常 Basic 認証ヘッダーで送信されるが、
# プロバイダーによっては body に含める必要がある場合もある (その場合は fetch_token の引数で指定)
token = oauth_session.fetch_token(token_url=token_url,
client_id=client_id, # body に含める必要がある場合
client_secret=client_secret,
scope=scope # fetch_token で scope を指定する場合
)
print("アクセストークン取得成功!")
print(token)
# 取得したトークンを使ってAPIリクエスト
api_endpoint = 'https://api.example.com/v1/stats'
# OAuth2Session は取得したトークンを自動でヘッダーに付与してくれる
response = oauth_session.get(api_endpoint)
response.raise_for_status()
stats_data = response.json()
print("\nAPIからのデータ:")
print(stats_data)
except Exception as e:
print(f"エラーが発生しました: {e}")
if hasattr(e, 'response') and e.response is not None:
print(f"ステータスコード: {e.response.status_code}")
print(f"レスポンスボディ: {e.response.text}")
if hasattr(e, 'description'):
print(f"エラー詳細: {e.description}")
このフローではユーザー認可のステップがないため、非常にシンプルです。oauthlib.oauth2.BackendApplicationClient
を OAuth2Session
に渡して初期化し、fetch_token
を呼び出すだけでアクセストークンを取得できます。
4. Resource Owner Password Credentials Grant (パスワードグラント) – 非推奨 ⚠️
ユーザーの ID とパスワードを直接アプリケーションに入力させてアクセストークンを取得するフローです。ユーザーは自分の認証情報をアプリケーションに預けることになるため、セキュリティリスクが高く、現在ではほとんどのケースで非推奨とされています。公式アプリなど、ユーザーがアプリケーションを完全に信頼できる場合に限定して使用されるべきです。
from requests_oauthlib import OAuth2Session
import os
import getpass # パスワード入力用
client_id = 'YOUR_APP_CLIENT_ID'
client_secret = 'YOUR_APP_CLIENT_SECRET' # 必要であれば
token_url = 'https://api.example.com/oauth/token'
scope = ['read', 'write']
# ユーザーから認証情報を取得
username = input("ユーザー名を入力してください: ")
password = getpass.getpass("パスワードを入力してください: ") # パスワードを隠して入力
try:
# OAuth2Session を初期化
oauth = OAuth2Session(client_id=client_id, scope=scope)
# パスワードグラントでトークンを取得
token = oauth.fetch_token(token_url=token_url,
username=username,
password=password,
client_id=client_id, # body に含める必要があれば
client_secret=client_secret # body に含める必要があれば
)
print("\nアクセストークン取得成功!")
print(token)
# API アクセス (省略)
# api_endpoint = 'https://api.example.com/v1/me'
# response = oauth.get(api_endpoint)
# ...
except Exception as e:
print(f"\nエラーが発生しました: {e}")
if hasattr(e, 'response') and e.response is not None:
print(f"ステータスコード: {e.response.status_code}")
print(f"レスポンスボディ: {e.response.text}")
if hasattr(e, 'description'):
print(f"エラー詳細: {e.description}")
このフローを使う場合は、ユーザー認証情報の取り扱いに最大限の注意が必要です。可能な限り、Authorization Code Grant や PKCE 付きのフローを利用することを検討してください。
5. アクセストークンの自動更新 (リフレッシュトークン) 🔄
OAuth 2.0 のアクセストークンには通常、有効期限があります(数分~数時間程度)。有効期限が切れると API にアクセスできなくなるため、ユーザーに再度認可を求めることなく新しいアクセストークンを取得する仕組みとしてリフレッシュトークンがあります。
Authorization Code Grant で access_type='offline'
を指定した場合など、条件を満たすとアクセストークンと一緒にリフレッシュトークン (refresh_token
) が発行されます。
requests-oauthlib
は、このリフレッシュトークンを使ったアクセストークンの自動更新をサポートしています。
from requests_oauthlib import OAuth2Session
import time
import json # トークン保存/読み込み用
client_id = "YOUR_CLIENT_ID"
client_secret = "YOUR_CLIENT_SECRET"
token_url = "https://api.example.com/oauth/token" # トークンエンドポイント
token_file = "my_token.json" # トークンを保存するファイル
# --- トークン更新用の関数 ---
def token_saver(token):
"""取得・更新されたトークンをファイルに保存する関数"""
print("トークンをファイルに保存します...")
with open(token_file, 'w') as f:
json.dump(token, f)
print("保存完了。")
# --- 以前取得したトークンをファイルから読み込む ---
try:
with open(token_file, 'r') as f:
token = json.load(f)
print("保存されたトークンを読み込みました。")
print(token)
except FileNotFoundError:
print("トークンファイルが見つかりません。初回認証が必要です。")
# ここで Authorization Code Grant などの初回認証フローを実行する
# token = perform_initial_auth() # 例: 初回認証フローを実行する関数
# token_saver(token) # 取得したトークンを保存
token = None # この例では終了
exit()
except json.JSONDecodeError:
print("トークンファイルの形式が正しくありません。")
token = None
exit()
# 自動更新の設定
# 'expires_in' (秒数) または 'expires_at' (Unixタイムスタンプ) がトークンに含まれている必要がある
extra = {
'client_id': client_id,
'client_secret': client_secret,
}
# OAuth2Session を初期化 (自動更新を設定)
oauth = OAuth2Session(
client_id,
token=token,
auto_refresh_url=token_url, # トークン更新用のURL
auto_refresh_kwargs=extra, # 更新時に追加で送信するパラメータ
token_updater=token_saver # 更新されたトークンを保存する関数
)
# --- API リクエストを実行 ---
# アクセストークンの有効期限が近い、または切れている場合、
# リクエスト前に自動的にリフレッシュトークンを使って更新が行われる
api_endpoint = "https://api.example.com/v1/resource"
print(f"\nAPI ({api_endpoint}) にアクセスします...")
try:
# 1回目のリクエスト
response = oauth.get(api_endpoint)
response.raise_for_status()
print("1回目のAPIアクセス成功!")
print(response.json())
# --- トークンの有効期限が切れるのを待つ (テスト用) ---
# expires_in = token.get('expires_in', 3600) # デフォルト1時間とする
# print(f"\n{expires_in}秒待機してトークンを期限切れにさせます...")
# time.sleep(expires_in + 5) # 期限切れ + 5秒待つ
# print("待機完了。")
# --- ここまでテスト用待機 ---
# 2回目のリクエスト (もし期限切れなら自動更新されるはず)
# print("\n再度APIにアクセスします...")
# response2 = oauth.get(api_endpoint)
# response2.raise_for_status()
# print("2回目のAPIアクセス成功!(トークンが自動更新された可能性があります)")
# print(response2.json())
# 更新されたトークンを確認 (token_saver が呼ばれていればファイルも更新されている)
# print("\n現在のセッションのトークン:")
# print(oauth.token)
except Exception as e:
print(f"\nAPIアクセスエラー: {e}")
if hasattr(e, 'response') and e.response is not None:
print(f"ステータスコード: {e.response.status_code}")
print(f"レスポンスボディ: {e.response.text}")
if hasattr(e, 'description'):
print(f"エラー詳細: {e.description}")
ポイントは OAuth2Session
の初期化時に以下の引数を設定することです。
token
: 読み込んだトークン情報 (リフレッシュトークンを含む)。auto_refresh_url
: トークンエンドポイントの URL。auto_refresh_kwargs
: トークン更新リクエストに追加で送信するパラメータ (通常はclient_id
とclient_secret
)。token_updater
: 更新されたトークンを受け取り、保存するためのコールバック関数。
これにより、oauth.get()
などを呼び出した際に、ライブラリが内部でトークンの有効期限 (expires_at
または expires_in
から計算) をチェックし、期限切れまたは期限が近い場合に自動的にリフレッシュトークンを使って新しいアクセストークンを取得し、指定された token_updater
関数を呼び出してくれます。
よくあるユースケースと注意点 💡✅⚠️
各種 Web API との連携例
requests-oauthlib
は、様々な Web API との連携に使用できます。
- Google API: OAuth 2.0 (Authorization Code Grant, PKCE, Service Account など)。UserInfo, Gmail, Calendar, Drive, YouTube Data API など多数。リフレッシュトークン対応。公式ライブラリ
google-auth-oauthlib
もrequests-oauthlib
を内部で使用しています。 - Twitter API: v1.1 は OAuth 1.0a、v2 は OAuth 2.0 (Authorization Code with PKCE, Client Credentials) と OAuth 1.0a をサポート。
- GitHub API: OAuth 2.0 (Authorization Code Grant)。
- Facebook Graph API: OAuth 2.0。
- Microsoft Graph API: OAuth 2.0 (Authorization Code Grant, PKCE, Client Credentials など)。
- Slack API: OAuth 2.0。
- Auth0: OAuth 2.0 (Authorization Code Grant with PKCE など)。
- その他多数…
各 API プロバイダーのドキュメントで、サポートされている OAuth フロー、エンドポイント URL、必要なスコープなどを確認してください。requests-oauthlib
のドキュメントには、いくつかのプロバイダー (Google, GitHub, Twitter, Facebook, Fitbit, Slack など) 向けの Compliance Fix (準拠のための修正) やサンプルも含まれています。
トークンの永続化
取得したアクセストークンやリフレッシュトークンは、アプリケーションの再起動後も利用できるように、永続化する必要があります。
- Web アプリケーション (サーバーサイド): データベース (ユーザーアカウントに関連付けて保存)、セキュアなファイルストレージ、キャッシュシステム (Redis など) に保存します。
- デスクトップアプリケーション: OS のセキュアストレージ (Windows Credential Manager, macOS Keychain など)、暗号化された設定ファイルなどに保存します。
- モバイルアプリケーション: OS 提供のセキュアストレージ (Android Keystore, iOS Keychain) に保存します。
- SPA (ブラウザ): セキュリティリスクが高いため、リフレッシュトークンをブラウザストレージ (LocalStorage, SessionStorage) に保存することは強く非推奨です。可能であれば、アクセストークンのみをメモリ上に保持し、有効期限が切れたら再度認証フローを開始するか、バックエンド (BFF: Backend for Frontend) 経由でトークンを管理するなどの対策が必要です。HttpOnly 属性付きのクッキーにバックエンドでトークンを保存するアプローチもあります。
いずれの場合も、トークン、特にリフレッシュトークンは機密情報として扱い、適切なアクセス制御と暗号化を施して安全に保管してください。
エラーハンドリング
OAuth フローや API リクエストでは様々なエラーが発生する可能性があります。
- ネットワークエラー:
requests.exceptions.RequestException
など。タイムアウトや接続エラー。 - OAuth 関連エラー:
oauthlib.oauth2.OAuth2Error
のサブクラス。InvalidGrantError
: 認可コードやリフレッシュトークンが無効。InvalidClientError
: Client ID や Client Secret が無効。UnsupportedGrantTypeError
: サポートされていないグラントタイプ。AccessDeniedError
: ユーザーがアクセスを拒否。TokenExpiredError
: アクセストークンの有効期限切れ (自動更新が有効でない場合など)。InsecureTransportError
: HTTP を使用している (HTTPS が必須)。- その他 (
invalid_request
,invalid_scope
,server_error
,temporarily_unavailable
など)
- API 固有のエラー: API プロバイダーが返すエラーレスポンス (4xx, 5xx ステータスコードとエラーメッセージ)。
requests.exceptions.HTTPError
(response.raise_for_status()
で発生) や、レスポンスボディを確認して処理します。
try...except
ブロックを使ってこれらのエラーを適切に捕捉し、ログ記録、ユーザーへのフィードバック、リトライ処理などを行う必要があります。特にトークン関連のエラー (InvalidGrantError
, TokenExpiredError
など) が発生した場合は、ユーザーに再認証を促すなどの対応が必要です。
デバッグ方法
- ログ出力の有効化:
oauthlib
の詳細なログを出力すると、内部的な処理の流れやエラーの原因特定に役立ちます。import logging import sys log = logging.getLogger('oauthlib') log.addHandler(logging.StreamHandler(sys.stdout)) log.setLevel(logging.DEBUG)
- リクエストとレスポンスの確認:
requests
のレスポンスオブジェクト (response
) のstatus_code
,headers
,text
(またはjson()
) を確認して、API サーバーとの通信内容を詳細に調べます。特にエラーレスポンスのボディには原因を示す情報が含まれていることが多いです。 state
やcode_verifier
の確認: 認可フローでstate
や PKCE の値が正しく生成・検証されているか確認します。
セキュリティに関する注意点 🚨
- Client Secret の保護: 絶対に漏洩させないように厳重に管理してください。パブリッククライアント (SPA, モバイルアプリ) では使用せず、PKCE を利用します。
- HTTPS の強制: OAuth 2.0 では必須です。通信経路を暗号化し、トークンや認証情報の盗聴を防ぎます。
- Redirect URI の厳密な検証: 認可サーバーに登録する Redirect URI は、できるだけ限定的 (特定のパスまで指定) にし、悪意のあるサイトへのリダイレクトを防ぎます。ワイルドカードの使用は避けるべきです。
- State パラメータによる CSRF 対策: Authorization Code Grant では必ず
state
パラメータを使用し、コールバック時に検証してください。 - PKCE の利用: 可能であれば、Authorization Code Grant では常に PKCE を利用することを推奨します (特にパブリッククライアント)。認可コードの傍受リスクを軽減します。
- スコープの最小化: アプリケーションが必要とする最小限の権限 (スコープ) のみを要求してください (最小権限の原則)。
- トークンの安全な保管: 上記「トークンの永続化」を参照。特にリフレッシュトークンの扱いに注意してください。
- ライブラリの更新:
requests
,oauthlib
,requests-oauthlib
を常に最新の安定版に保ち、セキュリティ修正を取り込んでください。
まとめ ✨
requests-oauthlib
は、Python で OAuth 1.0a および OAuth 2.0 認証を扱うための強力で使いやすいライブラリです。面倒な署名生成や複雑な認証フローの実装を隠蔽し、開発者が API 連携の本質的な部分に集中できるようにしてくれます。
この記事では、基本的な使い方から、主要な OAuth 2.0 グラントタイプの実装例、トークンの自動更新、PKCE、そして注意点まで幅広く解説しました。
外部 API との連携は現代のアプリケーション開発に不可欠です。requests-oauthlib
を活用して、安全かつ効率的に OAuth 認証を実装し、より高度なアプリケーション開発に挑戦してみてください!💪
さらに詳しい情報や最新情報については、公式ドキュメントを参照することをお勧めします。
コメント