oauth-2-0-setup

OAuth 2.0 のセットアップ

このスキルを使うと、AI エージェントが API 統合用の OAuth 2.0 認証を実装できます。エージェントはユースケースに応じて適切なグラントタイプを選択します。ユーザー向けアプリには PKCE を使用した認可コード、マシン間認証にはクライアントクレデンシャル、入力に制限のあるデバイスにはデバイスコードを使用します。フロー全体を通じて、トークンの保存、リフレッシュトークンのローテーション、state パラメーターによる CSRF 保護、安全な認証情報管理を処理します。

ワークフロー

1. 適切なグラントタイプを選択する： クライアントの種類に基づいて OAuth 2.0 のフローを選択します。ユーザーが存在する Web アプリやモバイルアプリには、PKCE を使用した認可コードを使います。PKCE はクライアントシークレットに代わるもので、認可コードのインターセプト攻撃を防ぎます。ユーザーコンテキストのないサーバー間通信には、クライアントクレデンシャルを使います。ブラウザーでのログインができない CLI ツールやスマート TV には、デバイスコードフローを使います。インプリシットフローは非推奨であり、使用しないでください。

2. プロバイダーにアプリケーションを登録する： プロバイダーの開発者コンソール（Google、GitHub、Auth0 など）で OAuth アプリケーションを作成します。リダイレクト URI を正確に設定してください。不一致の URI は、セットアップで最もよくあるエラーです。PKCE フローでは、アプリケーションをパブリッククライアントとして設定します。クライアント ID、クライアントシークレット（該当する場合）、認可エンドポイント、トークンエンドポイント、スコープを記録します。

3. 認可リクエストを実装する： 必須パラメーターである client_id、redirect_uri、response_type=code、scope と、CSRF 保護用の暗号学的にランダムな state パラメーターを含む認可 URL を構築します。PKCE の場合は、ランダムな code_verifier（43～128 文字）を生成し、SHA-256 を使って code_challenge を導出して、リクエストに code_challenge と code_challenge_method=S256 の両方を含めます。state と code_verifier はセッションに保存します。

4. コールバックを処理してトークンを交換する： プロバイダーが認可コードを付けてリダイレクトしてきたら、まず state パラメーターがセッションに保存した値と一致することを確認します。次に、grant_type=authorization_code、認可コード、redirect_uri、client_id、および code_verifier（PKCE の場合）を指定してトークンエンドポイントに POST し、コードをトークンと交換します。レスポンスから access_token、refresh_token、expires_in、token_type を解析します。

5. トークンを安全に保存する： トークンを localStorage（XSS に対して脆弱）や URL パラメーター（サーバーのアクセスログに記録される）には決して保存しないでください。Web アプリでは HTTP 専用のセキュア Cookie、デスクトップアプリではシステムのキーチェーン、モバイルアプリでは暗号化ストレージを使います。可能な場合は、リフレッシュトークンをサーバー側に保存します。有効期限前に更新できるよう、トークンの有効期限のタイムスタンプを記録します。

6. トークンの更新とローテーションを実装する： 各 API 呼び出しの前に、アクセストークンが期限切れか、期限切れ間近（60 秒以内）かを確認します。該当する場合は、リフレッシュトークンを使って新しいアクセストークンを取得します。リフレッシュトークンのローテーションに対応します。プロバイダーが新しいアクセストークンとともに新しいリフレッシュトークンを発行した場合は、新しいリフレッシュトークンを保存し、古いものを無効化します。invalid_grant エラーで更新に失敗した場合は、ユーザーが再認証する必要があります。

対応技術

• プロバイダー： Auth0、Google、GitHub、Microsoft Entra ID、Okta、AWS Cognito、Keycloak
• サーバーフレームワーク： Express.js、FastAPI、Django、Spring Security、ASP.NET Core
• ライブラリ： passport.js（Node.js）、authlib（Python）、oauthlib（Python）、Spring Security OAuth
• トークン形式： JWT（自己完結型）、不透明トークン（イントロスペクションが必要）
• 標準： RFC 6749（OAuth 2.0）、RFC 7636（PKCE）、RFC 8628（デバイスコード）、RFC 9449（DPoP）

使用方法

エージェントに OAuth プロバイダー、アプリケーションの種類（Web アプリ、SPA、CLI、サーバー間）、必要なスコープを指定します。エージェントは適切なグラントタイプを選択し、認可フロー、トークン交換、安全な保存、更新ロジックを含む実装一式を作成します。

例

例 1：PKCE を使用した認可コードフロー（Node.js/Express）

const express = require("express");
const crypto = require("crypto");
const session = require("express-session");

const app = express();
app.use(session({
  secret: process.env.SESSION_SECRET,
  resave: false,
  saveUninitialized: false,
  cookie: { secure: true, httpOnly: true, sameSite: "lax", maxAge: 3600000 },
}));

const OAUTH_CONFIG = {
  clientId: process.env.OAUTH_CLIENT_ID,
  authorizationEndpoint: "https://accounts.google.com/o/oauth2/v2/auth",
  tokenEndpoint: "https://oauth2.googleapis.com/token",
  redirectUri: "https://myapp.com/auth/callback",
  scopes: ["openid", "email", "profile"],
};

// Generate PKCE code verifier and challenge
function generatePKCE() {
  const verifier = crypto.randomBytes(32).toString("base64url");
  const challenge = crypto
    .createHash("sha256")
    .update(verifier)
    .digest("base64url");
  return { verifier, challenge };
}

// Step 1: Start authorization — redirect user to provider
app.get("/auth/login", (req, res) => {
  const state = crypto.randomBytes(16).toString("hex");
  const { verifier, challenge } = generatePKCE();

  // Store in session for verification on callback
  req.session.oauthState = state;
  req.session.codeVerifier = verifier;

  const params = new URLSearchParams({
    client_id: OAUTH_CONFIG.clientId,
    redirect_uri: OAUTH_CONFIG.redirectUri,
    response_type: "code",
    scope: OAUTH_CONFIG.scopes.join(" "),
    state: state,
    code_challenge: challenge,
    code_challenge_method: "S256",
    access_type: "offline",  // Request refresh token
    prompt: "consent",
  });

  res.redirect(`${OAUTH_CONFIG.authorizationEndpoint}?${params}`);
});

// Step 2: Handle callback — verify state and exchange code for tokens
app.get("/auth/callback", async (req, res) => {
  const { code, state, error } = req.query;

  if (error) {
    console.error(`OAuth error: ${error}`);
    return res.redirect("/auth/error");
  }

  // CSRF protection: verify state matches
  if (state !== req.session.oauthState) {
    console.error("State mismatch — possible CSRF attack");
    return res.status(403).send("Invalid state parameter");
  }

  try {
    const tokenResponse = await fetch(OAUTH_CONFIG.tokenEndpoint, {
      method: "POST",
      headers: { "Content-Type": "application/x-www-form-urlencoded" },
      body: new URLSearchParams({
        grant_type: "authorization_code",
        code: code,
        redirect_uri: OAUTH_CONFIG.redirectUri,
        client_id: OAUTH_CONFIG.clientId,
        code_verifier: req.session.codeVerifier,
      }),
    });

    if (!tokenResponse.ok) {
      const err = await tokenResponse.json();
      throw new Error(`Token exchange failed: ${err.error_description || err.error}`);
    }

    const tokens = await tokenResponse.json();

    // Store tokens securely in session (server-side)
    req.session.accessToken = tokens.access_token;
    req.session.refreshToken = tokens.refresh_token;
    req.session.tokenExpiry = Date.now() + tokens.expires_in * 1000;

    // Clean up PKCE and state from session
    delete req.session.oauthState;
    delete req.session.codeVerifier;

    res.redirect("/dashboard");
  } catch (err) {
    console.error("Token exchange error:", err.message);
    res.redirect("/auth/error");
  }
});

// Token refresh middleware
async function ensureValidToken(req, res, next) {
  if (!req.session.accessToken) {
    return res.redirect("/auth/login");
  }

  // Refresh if token expires within 60 seconds
  if (Date.now() > req.session.tokenExpiry - 60000) {
    try {
      const refreshResponse = await fetch(OAUTH_CONFIG.tokenEndpoint, {
        method: "POST",
        headers: { "Content-Type": "application/x-www-form-urlencoded" },
        body: new URLSearchParams({
          grant_type: "refresh_token",
          refresh_token: req.session.refreshToken,
          client_id: OAUTH_CONFIG.clientId,
        }),
      });

      if (!refreshResponse.ok) throw new Error("Refresh failed");

      const tokens = await refreshResponse.json();
      req.session.accessToken = tokens.access_token;
      req.session.tokenExpiry = Date.now() + tokens.expires_in * 1000;
      // Handle refresh token rotation
      if (tokens.refresh_token) {
        req.session.refreshToken = tokens.refresh_token;
      }
    } catch {
      // Refresh token is invalid — user must re-authenticate
      req.session.destroy();
      return res.redirect("/auth/login");
    }
  }
  next();
}

app.get("/dashboard", ensureValidToken, async (req, res) => {
  const profile = await fetch("https://www.googleapis.com/oauth2/v2/userinfo", {
    headers: { Authorization: `Bearer ${req.session.accessToken}` },
  });
  const user = await profile.json();
  res.json({ message: `Welcome, ${user.name}` });
});

app.listen(3000);

例 2：クライアントクレデンシャルフロー（マシン間）

import os
import time
import threading
import requests

class ClientCredentialsAuth:
    """OAuth 2.0 client credentials flow for server-to-server auth.

    Manages token lifecycle with thread-safe caching and automatic refresh.
    """

    def __init__(self, token_endpoint: str, client_id: str, client_secret: str,
                 scopes: list[str] | None = None):
        self.token_endpoint = token_endpoint
        self.client_id = client_id
        self.client_secret = client_secret
        self.scopes = scopes or []
        self._access_token: str | None = None
        self._token_expiry: float = 0
        self._lock = threading.Lock()

    def get_access_token(self) -> str:
        """Get a valid access token, refreshing if necessary."""
        with self._lock:
            if self._access_token and time.time() < self._token_expiry - 60:
                return self._access_token
            return self._fetch_new_token()

    def _fetch_new_token(self) -> str:
        response = requests.post(
            self.token_endpoint,
            data={
                "grant_type": "client_credentials",
                "client_id": self.client_id,
                "client_secret": self.client_secret,
                "scope": " ".join(self.scopes),
            },
            headers={"Content-Type": "application/x-www-form-urlencoded"},
            timeout=10,
        )
        response.raise_for_status()
        token_data = response.json()

        self._access_token = token_data["access_token"]
        self._token_expiry = time.time() + token_data.get("expires_in", 3600)
        return self._access_token

    def authorized_request(self, method: str, url: str, **kwargs) -> requests.Response:
        """Make an HTTP request with automatic bearer token injection."""
        token = self.get_access_token()
        headers = kwargs.pop("headers", {})
        headers["Authorization"] = f"Bearer {token}"
        return requests.request(method, url, headers=headers, **kwargs)


# Usage
auth = ClientCredentialsAuth(
    token_endpoint="https://auth.example.com/oauth/token",
    client_id=os.environ["M2M_CLIENT_ID"],
    client_secret=os.environ["M2M_CLIENT_SECRET"],
    scopes=["read:data", "write:data"],
)

# Token is fetched automatically and cached
response = auth.authorized_request("GET", "https://api.example.com/v1/reports")
print(response.json())

# Subsequent calls reuse the cached token until it nears expiry
response = auth.authorized_request("POST", "https://api.example.com/v1/exports", json={
    "format": "csv",
    "date_range": "2024-01-01/2024-12-31",
})
print(f"Export started: {response.json()['export_id']}")

ベストプラクティス

• 認可コードフローでは必ず PKCE を使用してください。機密クライアントの場合でも同様です。コストをかけずに、認可コードの傍受に対する多層防御を追加できます。インプリシットフローは非推奨です。使用しないでください。
• 認可コードを処理する前に、すべてのコールバックで state パラメーターを検証してください。これは CSRF 攻撃に対する最も重要な防御策です。128 ビット以上の暗号学的に安全なランダム値を使用してください。
• 可能な限り トークンはサーバー側に保存してください。ウェブアプリでは、トークンをサーバーセッションに保持し、セッション ID には HTTP-only、secure、SameSite Cookie を使用してください。アクセストークンをクライアント側の JavaScript に公開してはいけません。
• リフレッシュトークンのローテーションを実装し、各リフレッシュトークンを 1 回だけ使用できるようにしてください。トークンレスポンスで新しいリフレッシュトークンを受け取ったら、古いものを直ちに置き換えてください。ローテーション済みのトークンが再利用された場合は、漏えいしたものと見なし、そのセッションのすべてのトークンを無効にしてください。
• 必要最小限のスコープを要求し、利用可能な場合は段階的な認可を使用してください。最小権限から始め、追加のスコープが必要な操作をユーザーが実行するときにだけ要求してください。
• アクセストークンの有効期間を短く設定してください（機密性の高い API では 5～15 分）。長期間の利用にはリフレッシュトークンを使用します。これにより、アクセストークンが漏えいした場合の被害期間を抑えられます。

エッジケース

• リフレッシュトークンの有効期限切れ： リフレッシュトークンは期限切れになったり、無効化されたりすることがあります。リフレッシュを試みて invalid_grant が返された場合は、セッションを破棄し、ユーザーを再認証に誘導してください。無期限に再試行してはいけません。
• リフレッシュリクエストの同時実行： 複数のスレッドが同時にトークンをリフレッシュしようとする場合、1 つだけがリフレッシュを実行し、ほかのスレッドは待機するようにしてください。ミューテックスまたはシングルフライトパターンを使用し、複数のリフレッシュリクエストが競合して互いのトークンを無効化するのを防いでください。
• プロバイダーの停止： トークンエンドポイントに接続できない場合、現在のアクセストークンがまだ期限切れでなければ、そのまま使用してください。すぐに失敗させるのではなく、指数バックオフを使ってトークンのリフレッシュを再試行するようキューに入れてください。
• クロックスキュー： トークンの有効期限チェックには、正確な時刻が必要です。サーバーと OAuth プロバイダーの時刻のずれを考慮し、有効期限を確認する際に 30～60 秒の余裕を設けてください。
• リダイレクト URI の不一致： OAuth プロバイダーでは、リダイレクト URI を完全一致させる必要があります。登録済みの URI と、プロトコル、ポート、末尾のスラッシュまで含めて完全に一致していることを確認してください。不一致があると、原因の特定が難しいまま処理が失敗します。
• ログ内のトークン漏えい： アクセストークンがログ、エラーメッセージ、URL に含まれることは、よくある脆弱性です。すべてのログ記録ミドルウェアでトークンをマスキングしてください。トークンを URL のクエリパラメーターとして渡してはいけません。Authorization ヘッダーのみを使用してください。